fix(edge): stop hiding a failed camera; fix ghost dedup leak + silent zone drop

- Heartbeat reports the CONFIGURED camera count, not just the started ones: a
  camera that failed to construct at boot (wrong index / unseated ribbon) now
  shows as total/erroring instead of vanishing, so the server can see it is dark.
- GHOST emits without an engine cooldown key (ghost.update already fires once per
  never-reused track id) — removes a permanent _last_emit key that forget() could
  never reclaim (a slow 24/7 leak).
- Malformed zone rows are logged loudly instead of silently swallowed, matching
  the pixel-space branch's "don't fail dark" — a dead zone is no longer discovered
  only after a worker enters it.

edge/daemon.py

@@ -18,7 +18,7 @@ def _default_log(msg, exc=None):  # pragma: no cover (thin stderr wrapper)
 def run_loop(runtime, client, queue, *, clock, sleep, iterations=None,
              sync_every=150, heartbeat_every=900, flush_every=300,
              frame_interval=0.0, zones_for=None, log=None,
-             heartbeat_interval_s=None, health_fn=None):
+             heartbeat_interval_s=None, health_fn=None, configured_cameras=None):
     # `runtime` may be a single PipelineRuntime or a list of them (multi-camera). All
     # cameras share this loop, one shared clock read per tick, and the periodic
     # sync/heartbeat/flush. A per-camera step exception is counted and rate-limit-logged,
@@ -102,9 +102,16 @@ def run_loop(runtime, client, queue, *, clock, sleep, iterations=None,
                 t0, f0 = beat_anchor
                 elapsed = now - t0
                 fps = (counters["frames"] - f0) / elapsed if elapsed > 0 else 0.0
-                erroring = sum(1 for rt in runtimes if id(rt) in streak)
+                # cameras_total is the CONFIGURED count, not just the started one: a camera that
+                # failed to construct at boot (wrong /dev index, unseated ribbon) is skipped by
+                # build_runtimes and would otherwise vanish from the heartbeat entirely — the
+                # server could never see that a configured camera is missing. Count the not-started
+                # cameras as erroring so a 4-cam device that lost one reports 4 total / 1 erroring.
+                total = configured_cameras if configured_cameras is not None else len(runtimes)
+                not_started = max(0, total - len(runtimes))
+                erroring = not_started + sum(1 for rt in runtimes if id(rt) in streak)
                 metrics = {"fps": round(fps, 2), "frames": counters["frames"],
-                           "cameras_total": len(runtimes), "cameras_erroring": erroring,
+                           "cameras_total": total, "cameras_erroring": erroring,
                            "errors": counters["errors"], "queue_pending": queue.pending()}
                 # Pi thermal telemetry (None off-Pi or on failure → field omitted, never a fake 0).
                 health = health_fn()
@@ -222,10 +229,11 @@ def main() -> None:  # pragma: no cover (integration entrypoint, exercised on de
         return PipelineRuntime(cam_cfg, source, pose, queue, wall_clock=wall)
 
     # A missing/unseated camera is skipped (logged), not fatal — the others still run.
-    runtimes = build_runtimes(parse_cameras(dict(os.environ), cfg.camera_id),
-                              build_source, make_runtime)
+    cameras = parse_cameras(dict(os.environ), cfg.camera_id)
+    runtimes = build_runtimes(cameras, build_source, make_runtime)
 
     run_loop(runtimes, client, queue, clock=mono, sleep=time.sleep, iterations=None,
              frame_interval=frame_interval_from_env(dict(os.environ)),
              heartbeat_interval_s=30,  # predictable cadence; server marks offline after 120s (4x)
+             configured_cameras=len(cameras),  # so a camera that failed to start stays visible
              zones_for=lambda c, rt: zones_from_config(c, rt.camera_id))

edge/engine.py

@@ -87,12 +87,15 @@ class EventEngine:
                     events.append(self._mk(etype, occurred_at, tr.track_id, zone_id=hit["zone_id"]))
 
         for g in self.ghost.update(active_tracks, now):
-            key = (EventType.GHOST, g["track_id"])
-            if self._allow(key, now):
-                events.append(self._mk(EventType.GHOST, occurred_at, track_id=g["track_id"],
-                                       spatial=g["spatial"],
-                                       metrics={"last_state": g["last_state"],
-                                                "waited_s": g["waited_s"]}))
+            # No _allow cooldown for GHOST: ghost.update pops each incident from its candidates as it
+            # emits, and track ids are never reused, so a given track fires GHOST at most once ever.
+            # Routing it through _allow would only insert a permanent (GHOST, track_id) dedup key for
+            # a track that was already forgotten reappear_s ago — a slow 24/7 memory leak that
+            # forget() can never reclaim (the track never appears in lost_tracks again).
+            events.append(self._mk(EventType.GHOST, occurred_at, track_id=g["track_id"],
+                                   spatial=g["spatial"],
+                                   metrics={"last_state": g["last_state"],
+                                            "waited_s": g["waited_s"]}))
 
         # ghost.on_lost already consumed lost tracks' state above, so forgetting here is safe.
         for tr in lost_tracks:

edge/zones.py

@@ -41,7 +41,14 @@ def zones_from_config(config: dict, camera_id: str | None = None) -> list[Zone]:
                     row.get("zone_id"), row.get("camera_id"), polygon[:2])
                 continue
             out.append(Zone(str(row["zone_id"]), str(row["level"]), polygon))
-        except (KeyError, ValueError, TypeError, IndexError, AttributeError, json.JSONDecodeError):
+        except (KeyError, ValueError, TypeError, IndexError, AttributeError, json.JSONDecodeError) as e:
+            # Same principle as the pixel-space branch above: never fail dark. A row missing a key
+            # (level/zone_id), with unparseable polygon_json, or degenerate points would otherwise be
+            # dropped silently — intrusion detection for that zone dead with no log, no dashboard
+            # signal. Flag it so it can be fixed instead of discovered after a worker enters the zone.
+            zid = row.get("zone_id") if isinstance(row, dict) else None
+            log.warning("skipping malformed zone row %r (%s: %s) — its intrusion detection is "
+                        "inactive until the row is fixed in the dashboard.", zid, type(e).__name__, e)
             continue
     return out
 

tests/test_daemon.py

@@ -122,6 +122,20 @@ def test_no_frame_counts_as_erroring_not_a_healthy_frame():
     assert m["cameras_erroring"] == 1         # the dead camera is visible to the server
 
 
+def test_heartbeat_reports_configured_camera_count_when_one_failed_to_start():
+    # build_runtimes skips a camera that fails to construct at boot, so `runtime` holds only the
+    # started cameras. The heartbeat must still report the CONFIGURED total and count the missing
+    # one as erroring — otherwise the server can never see that a configured camera is dark.
+    rt, cl, q = _Runtime(), _Client(), _Queue()   # 1 runtime started...
+    ticks = iter(float(t) for t in range(1000))
+    run_loop(rt, cl, q, clock=lambda: next(ticks), sleep=lambda s: None,
+             iterations=10, sync_every=99, heartbeat_every=10, flush_every=99,
+             configured_cameras=2)                # ...but 2 were configured
+    m = cl.metrics[0]
+    assert m["cameras_total"] == 2                 # the missing camera is not hidden
+    assert m["cameras_erroring"] == 1              # and shows as erroring
+
+
 def test_heartbeat_reports_queue_depth_and_echoes_config_version():
     class _VersionClient(_Client):
         def fetch_config(self):

tests/test_engine.py

@@ -117,6 +117,21 @@ def test_pinch_zone_emits_pinch_event_not_warning():
     assert not any(e.event_type == EventType.ZONE_WARNING for e in events)
 
 
+def test_ghost_emit_does_not_leak_a_permanent_dedup_key():
+    # A GHOST fires reappear_s AFTER the track was lost (and already forgotten). Routing it through
+    # the engine cooldown used to insert a (GHOST, track_id) key that forget() could never reclaim
+    # (the track never appears in lost_tracks again) — a slow 24/7 memory leak. Emitting directly
+    # must leave no such key behind.
+    eng = _engine()
+    tr = Track(1, deque(maxlen=90))
+    _lying_track(tr, 0.0); eng.process([tr], [], now=0.0)
+    _lying_track(tr, 0.6); eng.process([tr], [], now=0.6)   # confirmed fallen -> dangerous state
+    eng.process([], [tr], now=1.0)                          # lost -> ghost candidate registered
+    events = eng.process([], [], now=12.0)                  # > reappear_s (10s), no reappearance
+    assert any(e.event_type == EventType.GHOST for e in events)
+    assert not any(k[0] == EventType.GHOST for k in eng._last_emit)   # no leaked ghost dedup key
+
+
 def test_two_simultaneous_tracks_have_independent_fall_state():
     # Per-track detector state must not leak between two people monitored at once: a fall on track A
     # must arm and emit for A only, never for a standing track B. All other engine/detector tests

tests/test_zones.py

@@ -16,9 +16,14 @@ def test_parses_zone_rows_json_string_polygon():
     assert zones[0].polygon == [(0.0, 0.0), (0.9, 0.0), (0.9, 0.9)]
     assert thresholds_from_config(cfg)["fall_prone_s"] == 3.0
 
-def test_skips_malformed_row():
+def test_skips_malformed_row_and_warns(caplog):
+    # A malformed row (missing polygon_json/level here) must be DROPPED and LOGGED — never silently
+    # swallowed, matching the pixel-space branch's "don't fail dark" principle: a dead zone with no
+    # signal is only discovered after a worker enters it.
     cfg = {"zones": [{"zone_id": "bad"}], "thresholds": {}}
-    assert zones_from_config(cfg) == []
+    with caplog.at_level(logging.WARNING):
+        assert zones_from_config(cfg) == []
+    assert any("bad" in r.message and "malformed" in r.message for r in caplog.records)
 
 def test_empty_config():
     assert zones_from_config({}) == []