diff --git a/backend/core/state.py b/backend/core/state.py
index c8d3ce096..87f60fd6c 100644
--- a/backend/core/state.py
+++ b/backend/core/state.py
@@ -37,7 +37,13 @@
 
 # ── Connected node state tracking ─────────────────────────────────────────────
 connected_nodes: dict[str, dict] = {}
-# node_id → {config_hash, config, status, last_heartbeat, peer, is_synthetic, capabilities}
+# node_id → {config_hash: str, config: dict, status: str,
+#            last_heartbeat: ISO-8601 str, peer: str, is_synthetic: bool,
+#            capabilities: dict, node_ref: str}
+# node_ref is present on mirrored nodes only; see services/node_refs._mirrored_ref.
+# status is set locally to "active" on ingest and "disconnected" when the TCP
+# peer drops, but a heartbeat overwrites it with whatever the node reported, so
+# it is not a closed set — do not branch on it as if it were.
 
 # Empirical FOV (see retina_analytics.empirical_coverage / manager /
 # association).  off/shadow/active, ASSOC_CLAIM_MODE precedent — an
@@ -396,6 +402,10 @@ def _adsb_for_seeding() -> dict[str, dict]:
 
 # ── ADS-B positions reported inside detection frames ──────────────────────────
 adsb_aircraft: dict[str, dict] = {}
+# normalize_hex_key'd hex → one record, built in services/tcp_handler and
+# extended by adsb_derived_fields; see services/feed_helpers.adsb_store for the
+# staleness rule that orders writes. last_seen_ms is the node's clock and
+# recv_ms is ours, so the two are not interchangeable.
 
 # ── Known-target claims registry (KNOWN_LANE_MODE) ────────────────────────────
 # Written by services/known_claiming.py once per frame; the interface between
diff --git a/backend/core/types.py b/backend/core/types.py
deleted file mode 100644
index d812720cb..000000000
--- a/backend/core/types.py
+++ /dev/null
@@ -1,73 +0,0 @@
-"""Typed definitions for core data structures.
-
-These TypedDicts describe the shapes of dicts stored in state.py.  They are
-documentation-first — not enforced at runtime — so introducing them is
-backwards-compatible with existing code that constructs plain dicts.
-"""
-
-from __future__ import annotations
-
-from typing import TypedDict
-
-
-class NodeState(TypedDict, total=False):
-    """Shape of entries in ``state.connected_nodes``."""
-
-    config_hash: str
-    config: dict
-    status: str  # "active" | "disconnected"
-    last_heartbeat: str  # ISO-8601
-    peer: str
-    is_synthetic: bool
-    capabilities: dict
-    node_ref: str  # mirrored nodes only; see services/node_refs._mirrored_ref
-
-
-class AircraftPosition(TypedDict, total=False):
-    """Shape of entries in ``state.adsb_aircraft``."""
-
-    hex: str
-    callsign: str
-    lat: float
-    lon: float
-    alt_baro: int
-    gs: float  # ground speed (knots)
-    track: float  # heading (degrees)
-    baro_rate: int
-    squawk: str
-    category: str
-    seen: float  # epoch of last update
-    rssi: float
-    node_id: str
-
-
-class GeoAircraft(TypedDict, total=False):
-    """Shape of entries in the aircraft.json ``aircraft`` list."""
-
-    hex: str
-    flight: str
-    lat: float
-    lon: float
-    alt_baro: int
-    alt_geom: int
-    gs: float
-    track: float
-    baro_rate: int
-    squawk: str
-    category: str
-    seen: float
-    rssi: float
-    type: str  # "adsb_icao" | "radar" | ...
-    node_id: str
-    multi_node: bool
-    anomaly: bool
-    anomaly_types: list[str]
-    trail: list[list[float]]
-    arc: list[dict]
-
-
-class TaskHealth(TypedDict, total=False):
-    """Shape of entries exposed in ``/api/test/dashboard`` → ``task_health``."""
-
-    last_success: dict[str, float]  # task_name → epoch
-    error_counts: dict[str, int]  # task_name → cumulative
diff --git a/backend/main.py b/backend/main.py
index a24696216..480dc9da6 100644
--- a/backend/main.py
+++ b/backend/main.py
@@ -58,7 +58,10 @@
 from routes.test import router as test_router
 from services import detection_mirror, publication
 from services.alerting import log_destination
-from services.background import (
+from services.runtime_coverage import start as _start_coverage
+from services.runtime_coverage import stop as _stop_coverage
+from services.state_snapshot import SAVE_INTERVAL_S, restore_snapshot, save_snapshot
+from services.tasks import (
     adsb_truth_fetcher,
     aircraft_flush_task,
     analytics_refresh_task,
@@ -77,9 +80,6 @@
     track_flush_task,
     users_backup_task,
 )
-from services.runtime_coverage import start as _start_coverage
-from services.runtime_coverage import stop as _stop_coverage
-from services.state_snapshot import SAVE_INTERVAL_S, restore_snapshot, save_snapshot
 from services.tasks.executor import task_executor, unfinished_task_executors
 from services.tasks.solver import solver_workers_stopping
 from services.tcp_handler import handle_tcp_client
diff --git a/backend/pipeline/passive_radar.py b/backend/pipeline/passive_radar.py
index d95bba0ff..43db45fff 100644
--- a/backend/pipeline/passive_radar.py
+++ b/backend/pipeline/passive_radar.py
@@ -6,8 +6,6 @@
 Pipeline: detection data → retina-tracker → retina-geolocator → tar1090 JSON
 """
 
-import glob
-import json
 import logging
 import math
 import os
@@ -819,28 +817,6 @@ def process_frame(self, frame: dict):
             self._last_prune_time = _now_mono
             self._prune_stale_tracks()
 
-    def process_file(self, filepath: str) -> list:
-        """Process an entire .detection file. Returns geolocated tracks."""
-        with open(filepath) as f:
-            content = f.read().strip()
-            if not content.startswith("["):
-                content = "[" + content + "]"
-            frames = json.loads(content)
-
-        # Feed all frames to tracker only (skip per-frame geolocation for batch)
-        for frame in frames:
-            ts = frame["timestamp"]
-            delays = frame.get("delay", [])
-            dopplers = frame.get("doppler", [])
-            snrs = frame.get("snr", [])
-            detections = [{"delay": d, "doppler": f, "snr": s} for d, f, s in zip(delays, dopplers, snrs)]
-            self.tracker.process_frame(detections, ts)
-
-        # Run geolocation once after all frames are processed
-        self._run_geolocation()
-
-        return list(self.geolocated_tracks.values())
-
     def generate_aircraft_json(self) -> dict:
         """Generate tar1090-compatible aircraft.json from geolocated tracks."""
         now = time.time()
@@ -903,38 +879,3 @@ def generate_receiver_json(self) -> dict:
             "lat": lat,
             "lon": lon,
         }
-
-
-def process_detection_folder(folder: str, output_dir: str, node_config: dict = None):
-    """Process all .detection files in a folder and write tar1090 JSON to output_dir."""
-    pipeline = PassiveRadarPipeline(node_config)
-
-    detection_files = sorted(glob.glob(os.path.join(folder, "*.detection")))
-    if not detection_files:
-        logger.info("No .detection files found in %s", folder)
-        return
-
-    os.makedirs(output_dir, exist_ok=True)
-
-    receiver = pipeline.generate_receiver_json()
-    with open(os.path.join(output_dir, "receiver.json"), "w") as f:
-        json.dump(receiver, f)
-
-    for filepath in detection_files:
-        logger.info("Processing: %s", os.path.basename(filepath))
-        pipeline.process_file(filepath)
-
-    aircraft_data = pipeline.generate_aircraft_json()
-    with open(os.path.join(output_dir, "aircraft.json"), "w") as f:
-        json.dump(aircraft_data, f)
-
-    logger.info("Output: %d geolocated targets", len(aircraft_data["aircraft"]))
-    return aircraft_data
-
-
-if __name__ == "__main__":
-    import sys
-
-    folder = sys.argv[1] if len(sys.argv) > 1 else "."
-    output = sys.argv[2] if len(sys.argv) > 2 else "./tar1090_data"
-    process_detection_folder(folder, output)
diff --git a/backend/pyproject.toml b/backend/pyproject.toml
index 2fa86a78e..f14c8b0fd 100644
--- a/backend/pyproject.toml
+++ b/backend/pyproject.toml
@@ -33,8 +33,6 @@ omit = [
     "routes/auth.py",
     # Simulation/test-only endpoints — not production code paths:
     "routes/test.py",
-    # Pure TypedDict definitions — no runtime code to cover:
-    "core/types.py",
 ]
 
 [tool.coverage.report]
diff --git a/backend/routes/admin.py b/backend/routes/admin.py
index 1a9fefe26..1a76285e0 100644
--- a/backend/routes/admin.py
+++ b/backend/routes/admin.py
@@ -51,11 +51,6 @@
 logger = logging.getLogger(__name__)
 
 
-# Was a byte-identical copy of routes/test.py's; the rule now lives beside the
-# interval table it reads.
-_get_stale_tasks = get_stale_tasks
-
-
 def _mn_pos_history_size() -> int:
     """Size of the solver's per-hex smoothing buffer (soak observability)."""
     from services.tasks import solver as _solver
@@ -769,7 +764,7 @@ async def system_metrics(_user=Depends(require_admin)):
         "ws_clients": len(state.ws_clients),
         "ws_send_timeouts": state.ws_send_timeouts,
         "ws_live_clients": len(state.ws_live_clients),
-        "stale_tasks": _get_stale_tasks(),
+        "stale_tasks": get_stale_tasks(),
         "process_rss_mb": round(rss_mb, 1),
         "load_avg": list(os.getloadavg()),
         "disk_total_gb": round(disk.total / (1024**3), 1),
diff --git a/backend/routes/radar.py b/backend/routes/radar.py
index f875fc2c7..7de980365 100644
--- a/backend/routes/radar.py
+++ b/backend/routes/radar.py
@@ -6,13 +6,12 @@
 import time
 from datetime import datetime, timezone
 
-from fastapi import APIRouter, Depends, Header, HTTPException, Request
+from fastapi import APIRouter, Header, HTTPException, Request
 from fastapi.responses import Response
 from pydantic import BaseModel, Field
 
 from config.constants import RATE_BUCKETS_MAX_IPS
 from core import state
-from core.users import require_admin
 from pipeline.passive_radar import PassiveRadarPipeline
 from routes.node_schemas import NodeRef
 from services import node_refs, node_registration
@@ -51,10 +50,6 @@ class BulkDetectionRequest(BaseModel):
     nodes: list[BulkNodeEntry] = Field(..., max_length=500)
 
 
-class LoadFileRequest(BaseModel):
-    path: str = Field(..., min_length=1)
-
-
 RADAR_API_KEY = os.getenv("RADAR_API_KEY", "")
 _RETINA_ENV = os.getenv("RETINA_ENV", "").lower()
 if not RADAR_API_KEY:
@@ -65,9 +60,6 @@ class LoadFileRequest(BaseModel):
         logging.warning(_msg)
 _RATE_LIMIT = int(os.getenv("RADAR_RATE_LIMIT", "60"))
 _RATE_WINDOW = int(os.getenv("RADAR_RATE_WINDOW", "60"))
-_ALLOWED_DETECTION_DIR = os.path.realpath(
-    os.path.join(os.path.dirname(os.path.dirname(__file__)), "coverage_data", "archive")
-)
 
 # Module-level reference to default pipeline; set from main.py at startup
 _default_pipeline: PassiveRadarPipeline | None = None
@@ -272,26 +264,6 @@ async def ingest_detections_bulk(
     return {"status": "ok", "nodes_registered": registered, "frames_queued": queued}
 
 
-@router.post("/api/radar/load-file")
-async def load_detection_file(body: LoadFileRequest, _admin=Depends(require_admin)):
-    filepath = os.path.realpath(body.path)
-    if not filepath.startswith(_ALLOWED_DETECTION_DIR + os.sep):
-        raise HTTPException(status_code=400, detail="Path must be inside the coverage archive directory")
-    if not os.path.isfile(filepath):
-        raise HTTPException(status_code=400, detail="File not found")
-    if not filepath.endswith(".detection"):
-        raise HTTPException(status_code=400, detail="Only .detection files accepted")
-
-    tracks = _default_pipeline.process_file(filepath)
-    aircraft_data = _default_pipeline.generate_aircraft_json()
-    # Do NOT write aircraft.json here.  The flush task is the file's single
-    # writer (a different schema was being raced in from this request
-    # thread); marking dirty makes the next 1 Hz flush pick the load up.
-    state.aircraft_dirty = True
-
-    return {"status": "ok", "tracks": len(tracks), "aircraft": aircraft_data["aircraft"]}
-
-
 @router.get("/api/radar/status")
 async def radar_status():
     # Unauthenticated, and the config block quotes the default pipeline's
diff --git a/backend/routes/test.py b/backend/routes/test.py
index 8fd34963a..827295ac0 100644
--- a/backend/routes/test.py
+++ b/backend/routes/test.py
@@ -57,11 +57,6 @@ def _verify_sim_key(x_api_key: str = Header(default="", alias="X-API-Key")):
         raise HTTPException(status_code=401, detail="Invalid or missing X-API-Key")
 
 
-# Was a byte-identical copy of routes/admin.py's; the rule now lives beside the
-# interval table it reads.
-_get_stale_tasks = get_stale_tasks
-
-
 # Module-level reference set from main.py at startup
 _default_pipeline = None
 
@@ -300,7 +295,7 @@ def _build_dashboard_data() -> bytes:
             "task_health": {
                 "last_success": dict(state.task_last_success),
                 "error_counts": state.task_error_snapshot(),
-                "stale_tasks": _get_stale_tasks(),
+                "stale_tasks": get_stale_tasks(),
             },
         }
     )
@@ -471,21 +466,6 @@ async def get_ground_truth_trail(hex_code: str):
     }
 
 
-@router.get("/api/test/anomalies")
-async def get_anomaly_log():
-    """Return the anomaly event log and currently flagged hex codes."""
-    return Response(
-        content=orjson.dumps(
-            {
-                "flagged_count": len(state.anomaly_hexes),
-                "flagged_hexes": sorted(state.anomaly_hexes),
-                "events": state.anomaly_log[-100:],
-            }
-        ),
-        media_type="application/json",
-    )
-
-
 # ── Known-track hold (path H) ─────────────────────────────────────────────────
 
 
diff --git a/backend/scripts/smoke_parquet_archive.py b/backend/scripts/smoke_parquet_archive.py
deleted file mode 100644
index b2d8399a8..000000000
--- a/backend/scripts/smoke_parquet_archive.py
+++ /dev/null
@@ -1,39 +0,0 @@
-"""Smoke test: drive a frame through archive_detections and read it back."""
-
-import shutil
-from pathlib import Path
-
-import services.storage as st
-from services.storage import archive_detections, read_archived_file
-
-
-def main():
-    base = Path("/tmp/parquet_smoke")
-    if base.exists():
-        shutil.rmtree(base)
-    base.mkdir()
-
-    st._LOCAL_ARCHIVE_DIR = str(base)
-
-    frame = {
-        "timestamp": 1700000000000,
-        "delay": [12.34, 56.78],
-        "doppler": [-100.5, 33.3],
-        "snr": [15.0, 22.0],
-        "adsb": [
-            {"hex": "abcdef", "lat": 40.71, "lon": -74.0, "alt_baro": 35000, "gs": 480, "track": 270, "flight": "UAL1"},
-            None,
-        ],
-        "_signing_mode": "unknown",
-        "_signature_valid": False,
-    }
-    key = archive_detections("smoke-node", [frame])
-    print("wrote:", key)
-    decoded = read_archived_file(key)
-    print("decoded count:", decoded["count"])
-    print("first detection delay:", decoded["detections"][0]["delay"])
-    print("adsb match:", decoded["detections"][0]["adsb"][0])
-
-
-if __name__ == "__main__":
-    main()
diff --git a/backend/services/background.py b/backend/services/background.py
deleted file mode 100644
index d205d7255..000000000
--- a/backend/services/background.py
+++ /dev/null
@@ -1,28 +0,0 @@
-"""Background async tasks — thin re-export shim.
-
-The actual implementations live in ``services.tasks.*`` sub-modules.
-This file keeps the original import surface so ``main.py`` and tests
-continue to work without changes::
-
-    from services.background import frame_processor_loop, start_solver_workers
-"""
-
-from services.tasks import (  # noqa: F401
-    adsb_truth_fetcher,
-    aircraft_flush_task,
-    analytics_refresh_task,
-    archive_flush_task,
-    archive_lifecycle_task,
-    coverage_constraints_task,
-    feed_gc_task,
-    frame_processor_loop,
-    health_monitor_task,
-    heartbeat_task,
-    prune_synthetic_nodes,
-    reputation_evaluator,
-    start_solver_workers,
-    stop_solver_workers,
-    storage_refresh_task,
-    track_flush_task,
-    users_backup_task,
-)
diff --git a/backend/services/tasks/__init__.py b/backend/services/tasks/__init__.py
index ded5d8639..61cf330a6 100644
--- a/backend/services/tasks/__init__.py
+++ b/backend/services/tasks/__init__.py
@@ -1,7 +1,11 @@
-"""Background task sub-modules split from services/background.py.
+"""Background async tasks, one sub-module per task.
 
-All public names previously exported by background.py are re-exported here
-so ``from services.tasks import frame_processor_loop`` works.
+The task entry points are re-exported here so schedulers can name them
+without knowing which sub-module each lives in::
+
+    from services.tasks import frame_processor_loop, start_solver_workers
+
+Anything else a sub-module exposes is imported from that sub-module directly.
 """
 
 from services.tasks.aircraft_flush import aircraft_flush_task
diff --git a/backend/tests/test_admin_routes.py b/backend/tests/test_admin_routes.py
index ca1d6cfbb..f93ec55e5 100644
--- a/backend/tests/test_admin_routes.py
+++ b/backend/tests/test_admin_routes.py
@@ -463,17 +463,17 @@ def test_check_detects_offline(self):
 
 class TestStaleTasks:
     def test_no_stale_when_recent(self):
-        from routes.admin import _get_stale_tasks
+        from core.task_registry import get_stale_tasks
 
         state.task_last_success["frame_processor"] = time.time()
-        result = _get_stale_tasks()
+        result = get_stale_tasks()
         assert "frame_processor" not in result
 
     def test_stale_when_old(self):
-        from routes.admin import _get_stale_tasks
+        from core.task_registry import get_stale_tasks
 
         state.task_last_success["frame_processor"] = time.time() - 9999
-        result = _get_stale_tasks()
+        result = get_stale_tasks()
         assert "frame_processor" in result
 
 
diff --git a/backend/vulture_whitelist.py b/backend/vulture_whitelist.py
index fe7aa82bf..5ed2f006e 100644
--- a/backend/vulture_whitelist.py
+++ b/backend/vulture_whitelist.py
@@ -20,26 +20,6 @@
 # file's own preamble.  ANALYTICS_REFRESH_INTERVAL_S is now wired;
 # ASSOC_MIN_INTERVAL_S is passed by core/state.)
 
-# ── core/types.py ─────────────────────────────────────────────────────────────
-
-# TypedDict definitions — used as type annotations; fields are accessed via
-# dict keys at runtime, not attribute access, so vulture misses the usage.
-NodeState
-_.capabilities
-AircraftPosition
-_.hex
-_.baro_rate
-_.squawk
-_.rssi
-GeoAircraft
-_.flight
-_.alt_geom
-_.multi_node
-_.anomaly
-TaskHealth
-_.last_success
-_.error_counts
-
 # ── pipeline/passive_radar.py ─────────────────────────────────────────────────
 
 # EventWriter public API — tested in tests/test_pipeline.py