fix: seed trajectories from recent source motion

README_zh.md

@@ -183,6 +183,10 @@ occupancy_reflection_dark_fraction = 0.10
 occupancy_reflection_bright_dark_ratio = 2.0
 occupancy_confirm_frames = 2
 empty_confirm_frames = 2
+lighting_shift_guard_enabled = true
+lighting_shift_min_regions = 3
+lighting_shift_region_fraction = 0.6
+lighting_shift_mean_delta = 45.0
 trash_motion_delta = 18.0
 trash_sustained_motion_delta = 8.0
 trash_sustained_motion_frames = 2
@@ -191,6 +195,8 @@ trajectory_enabled = true
 trajectory_window_seconds = 8
 trajectory_sample_interval_seconds = 1.0
 trajectory_min_points = 3
+trajectory_segmented_enabled = true
+trajectory_segmented_min_points = 2
 trajectory_min_confidence = 0.72
 trajectory_motion_delta = 20.0
 trajectory_min_blob_area = 12
@@ -208,7 +214,8 @@ diagnostics_path = "logs/runtime_diagnostics.jsonl"
 运行诊断写入 `logs/runtime_diagnostics.jsonl`。每行包含顶层 `disposal_evidence`，以及 `diagnostics.trajectory`：
 
 - 顶层 `disposal_evidence`：本帧实际输出给状态机的来源区域到垃圾桶证据。
-- `diagnostics.trajectory`：轻量轨迹 backend 的候选、过期、拒绝和已发出证据等调试信息。
+- `diagnostics.lighting_shift`：多数区域同时同方向亮度漂移时启用，防止灯光/曝光变化被当成食品进出。
+- `diagnostics.trajectory`：轻量轨迹 backend 的候选、来源 motion 预缓存、过期、拒绝、分段轨迹和已发出证据等调试信息。
 
 ## 本地测试
 

config/example.toml

@@ -48,6 +48,10 @@ occupancy_reflection_dark_fraction = 0.10
 occupancy_reflection_bright_dark_ratio = 2.0
 occupancy_confirm_frames = 2
 empty_confirm_frames = 2
+lighting_shift_guard_enabled = true
+lighting_shift_min_regions = 3
+lighting_shift_region_fraction = 0.6
+lighting_shift_mean_delta = 45.0
 trash_motion_delta = 18.0
 trash_sustained_motion_delta = 8.0
 trash_sustained_motion_frames = 2
@@ -56,6 +60,8 @@ trajectory_enabled = true
 trajectory_window_seconds = 8
 trajectory_sample_interval_seconds = 1.0
 trajectory_min_points = 3
+trajectory_segmented_enabled = true
+trajectory_segmented_min_points = 2
 trajectory_min_confidence = 0.72
 trajectory_motion_delta = 20.0
 trajectory_min_blob_area = 12

docs/project.md

@@ -50,10 +50,13 @@ The `v1.2 轨迹识别` batch adds source-zone trajectory evidence for disposal
   - Stored under `[trash] roi`.
   - Does not use a food zone number.
 - v1.2 trajectory settings:
+  - `lighting_shift_guard_enabled`: freezes occupancy changes when many regions shift brightness in the same direction.
+  - `lighting_shift_min_regions`, `lighting_shift_region_fraction`, `lighting_shift_mean_delta`: tune the global lighting/exposure guard.
   - `trajectory_enabled`: enables source-zone trajectory evidence.
   - `trajectory_window_seconds`: seconds after a zone clears where movement can confirm disposal.
   - `trajectory_sample_interval_seconds`: faster runtime delay while a candidate is active.
   - `trajectory_min_points`: minimum sampled motion points required before evidence can emit.
+  - `trajectory_segmented_enabled`, `trajectory_segmented_min_points`: allow a source point plus trash-entry point to confirm disposal when the middle of the path is occluded.
   - `trajectory_min_confidence`: minimum confidence before evidence can close pending disposal.
   - `trajectory_motion_delta`: frame-difference threshold for trajectory motion points.
   - `trajectory_min_blob_area`: minimum connected motion area to keep as a point.
@@ -66,7 +69,7 @@ The `v1.2 轨迹识别` batch adds source-zone trajectory evidence for disposal
 
 1. `RTSPFrameSource` captures a resized RGB frame.
 2. `ZoneOccupancyDetector` updates per-zone binary occupancy and generic trash-motion count from calibrated ROIs.
-3. `TrajectoryTracker` watches zones that just cleared, follows lightweight motion points toward the trash ROI, and emits source-specific `DisposalEvidence` when confidence passes the configured threshold.
+3. `TrajectoryTracker` caches recent source-region motion while a zone is occupied, watches zones that just cleared, follows lightweight motion points toward the trash ROI, and emits source-specific `DisposalEvidence` when confidence passes the configured threshold. If the middle of the path is occluded, a segmented source-to-trash track can still emit evidence.
 4. `BatchEngine` processes `Observation(zone_counts, trash_deposit_count, disposal_evidence)`.
 5. For pending disposal, matching `disposal_evidence.source_zone_id` confirms `batch_discarded` before generic FIFO `trash_deposit_count` fallback is used.
 6. Runtime writes events to `logs/events.jsonl` and diagnostics to `logs/runtime_diagnostics.jsonl`.
@@ -78,8 +81,9 @@ The current tracker is a motion backend only. A later trained YOLO detector shou
 - Runtime diagnostics JSONL records one item per runtime iteration.
 - Root `disposal_evidence` is the exact evidence list passed into the engine for that iteration.
 - `diagnostics.zones` contains occupancy metrics used to derive `zone_counts`.
+- `diagnostics.lighting_shift` reports whether global brightness drift suppressed occupancy transitions.
 - `diagnostics.trash` contains generic trash-motion metrics and cooldown state.
-- `diagnostics.trajectory` contains v1.2 candidate counts, emitted evidence count, motion point count, and per-candidate emitted/rejected/expired records.
+- `diagnostics.trajectory` contains v1.2 candidate counts, emitted evidence count, motion point count, source-seeded and segmented-track flags, and per-candidate emitted/rejected/expired records.
 - Capture failures still keep the v1.2 schema with root `disposal_evidence: []` and `diagnostics.trajectory.reason = "frame_capture_failed"`.
 
 ## Event Model
@@ -122,6 +126,7 @@ In v1.2, `batch_discarded` can be triggered by zone-scoped `disposal_evidence` b
 ## Known Risks
 
 - The current vision detector is heuristic and reports binary occupancy, not item counts.
+- The lighting-shift guard suppresses multi-zone brightness/exposure jumps; if operators intentionally fill most zones at once under a large lighting change, diagnostics should be reviewed before treating that interval as clean data.
 - v1.2 motion tracking improves disposal matching but can still miss movement if the hand/object path is occluded, too broad, too small, or sampled too sparsely.
 - YOLO config fields are present for compatibility, but no trained YOLO model is part of the current runtime.
 - If food is already present during baseline collection, those regions may be treated as empty baseline until visual changes occur.

memories.md

@@ -29,6 +29,9 @@
 - Stage 2 added the lightweight motion trajectory runtime path: ROI occupancy still drives occupied/empty state, `TrajectoryTracker` emits source-zone-to-trash evidence, and generic trash-motion count remains as a fallback.
 - Stage 3 added diagnostics and tests for runtime evidence propagation, trajectory sampling interval behavior, capture-failure schema, and trajectory/yolo runtime config parsing.
 - Final review fixes: matched evidence now only subtracts the trash fallback budget by the number of batches it actually closed, and trajectory candidates reject outside-before-source motion with `motion_started_outside_source`.
+- 2026-06-01 false events across zones 1-7 were caused by a global brightness/exposure drop around 04:55 and recovery around 08:16; the fix is a lighting-shift guard that freezes occupancy transitions when many regions shift brightness in the same direction.
+- 2026-06-01 trajectory update allows segmented source-to-trash tracks: after a source-zone motion point is seen, the middle of the path may be occluded, and a later trash-entry point can still emit `disposal_evidence` when direction/confidence pass.
+- 2026-06-01 follow-up fixed empty-confirmation lag: `TrajectoryTracker` now caches recent source-region motion while a zone is still occupied and seeds the disposal candidate when the stable zone count finally clears.
 - Current v1.2 does not use YOLO. `yolo_enabled`, `yolo_model_path`, and `yolo_min_confidence` are reserved for a future trained model backend that should keep emitting the same `disposal_evidence` shape.
 
 ## Remote Deployment Notes

src/cold_display_guard/vision.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 from dataclasses import dataclass
 from datetime import datetime, timedelta
+from math import ceil
 from typing import Any
 
 from cold_display_guard.models import DisposalEvidence
@@ -39,6 +40,10 @@ class RuntimeVisionSettings:
     occupancy_reflection_bright_dark_ratio: float = 2.0
     occupancy_confirm_frames: int = 2
     empty_confirm_frames: int = 2
+    lighting_shift_guard_enabled: bool = True
+    lighting_shift_min_regions: int = 3
+    lighting_shift_region_fraction: float = 0.6
+    lighting_shift_mean_delta: float = 45.0
     trash_motion_delta: float = 18.0
     trash_sustained_motion_delta: float = 8.0
     trash_sustained_motion_frames: int = 2
@@ -47,6 +52,8 @@ class RuntimeVisionSettings:
     trajectory_window_seconds: int = 8
     trajectory_sample_interval_seconds: float = 1.0
     trajectory_min_points: int = 3
+    trajectory_segmented_enabled: bool = True
+    trajectory_segmented_min_points: int = 2
     trajectory_min_confidence: float = 0.72
     trajectory_motion_delta: float = 20.0
     trajectory_min_blob_area: int = 12
@@ -84,6 +91,8 @@ class _TrajectoryCandidate:
     points: list[_MotionPoint] | None = None
     source_motion_seen: bool = False
     pre_source_motion_seen: bool = False
+    source_seeded: bool = False
+    forced_rejection_reason: str | None = None
 
     def __post_init__(self) -> None:
         if self.points is None:
@@ -125,7 +134,12 @@ class ZoneOccupancyDetector:
         self._update_baseline(metrics_by_region)
 
         zone_counts: dict[str, int] = {}
-        diagnostics: dict[str, Any] = {"zones": {}, "baseline_ready": self.baseline_ready}
+        lighting_shift = self._lighting_shift(metrics_by_region)
+        diagnostics: dict[str, Any] = {
+            "zones": {},
+            "baseline_ready": self.baseline_ready,
+            "lighting_shift": lighting_shift,
+        }
         for region in self.regions:
             metrics = metrics_by_region[region.region_id]
             baseline = self._baseline.get(region.region_id)
@@ -133,8 +147,12 @@ class ZoneOccupancyDetector:
             if baseline is not None:
                 mean_delta = abs(metrics.mean_luma - baseline.mean_luma)
                 texture_delta = metrics.texture - baseline.texture
-                raw_occupied = self._raw_occupied(metrics, baseline, mean_delta, texture_delta)
-                occupied = self._confirmed_occupancy(region.region_id, raw_occupied)
+                if lighting_shift["active"]:
+                    raw_occupied = False
+                    occupied = self._stable_occupancy.get(region.region_id, False)
+                else:
+                    raw_occupied = self._raw_occupied(metrics, baseline, mean_delta, texture_delta)
+                    occupied = self._confirmed_occupancy(region.region_id, raw_occupied)
                 diagnostics["zones"][region.region_id] = {
                     "mean_luma": round(metrics.mean_luma, 3),
                     "baseline_mean_luma": round(baseline.mean_luma, 3),
@@ -151,6 +169,7 @@ class ZoneOccupancyDetector:
                     "occupied": occupied,
                     "occupied_streak": self._occupied_streaks[region.region_id],
                     "empty_streak": self._empty_streaks[region.region_id],
+                    "lighting_shift_suppressed": lighting_shift["active"],
                 }
             zone_counts[region.region_id] = 1 if occupied else 0
 
@@ -201,6 +220,60 @@ class ZoneOccupancyDetector:
             if len(samples) >= self.settings.baseline_frames:
                 self._baseline[region_id] = average_metrics(samples)
 
+    def _lighting_shift(self, metrics_by_region: dict[str, RegionMetrics]) -> dict[str, Any]:
+        if not self.settings.lighting_shift_guard_enabled:
+            return self._lighting_shift_diagnostics(False, None, 0, 0, 0)
+
+        eligible_region_count = 0
+        darker_count = 0
+        brighter_count = 0
+        for region in self.regions:
+            metrics = metrics_by_region.get(region.region_id)
+            baseline = self._baseline.get(region.region_id)
+            if metrics is None or baseline is None:
+                continue
+            eligible_region_count += 1
+            signed_delta = metrics.mean_luma - baseline.mean_luma
+            if abs(signed_delta) < self.settings.lighting_shift_mean_delta:
+                continue
+            if signed_delta < 0:
+                darker_count += 1
+            else:
+                brighter_count += 1
+
+        required_regions = max(
+            self.settings.lighting_shift_min_regions,
+            ceil(eligible_region_count * self.settings.lighting_shift_region_fraction),
+        )
+        active_direction: str | None = None
+        shifted_count = max(darker_count, brighter_count)
+        if eligible_region_count >= self.settings.lighting_shift_min_regions and shifted_count >= required_regions:
+            active_direction = "darker" if darker_count >= brighter_count else "brighter"
+        return self._lighting_shift_diagnostics(
+            active_direction is not None,
+            active_direction,
+            shifted_count,
+            eligible_region_count,
+            required_regions,
+        )
+
+    def _lighting_shift_diagnostics(
+        self,
+        active: bool,
+        direction: str | None,
+        shifted_regions: int,
+        eligible_regions: int,
+        required_regions: int,
+    ) -> dict[str, Any]:
+        return {
+            "active": active,
+            "direction": direction,
+            "shifted_regions": shifted_regions,
+            "eligible_regions": eligible_regions,
+            "required_regions": required_regions,
+            "mean_delta_threshold": self.settings.lighting_shift_mean_delta,
+        }
+
     def _confirmed_occupancy(self, region_id: str, raw_occupied: bool) -> bool:
         if raw_occupied:
             self._occupied_streaks[region_id] = self._occupied_streaks.get(region_id, 0) + 1
@@ -263,6 +336,8 @@ class TrajectoryTracker:
         self._previous_frame: Frame | None = None
         self._previous_zone_counts: dict[str, int] = {}
         self._candidates: list[_TrajectoryCandidate] = []
+        self._recent_source_motion: dict[str, _MotionPoint] = {}
+        self._emitted_track_signatures: set[tuple[tuple[float, float], ...]] = set()
 
     @property
     def has_active_candidates(self) -> bool:
@@ -302,12 +377,13 @@ class TrajectoryTracker:
             return [], diagnostics
 
         blobs = self._motion_points(frame, when) if self._previous_frame is not None else []
+        self._remember_recent_source_motion(blobs, when, zone_counts)
         self._open_candidates(when, zone_counts)
 
         emitted: list[DisposalEvidence] = []
         remaining: list[_TrajectoryCandidate] = []
         consumed_blob_ids: set[int] = set()
-        emitted_track_signatures: set[tuple[tuple[float, float], ...]] = set()
+        emitted_track_signatures = set(self._emitted_track_signatures)
         for candidate in self._candidates:
             rejected_reason = self._sample_candidate(candidate, blobs, when, consumed_blob_ids)
             if rejected_reason is not None:
@@ -328,6 +404,7 @@ class TrajectoryTracker:
                     else:
                         emitted.append(self._evidence(candidate, when))
                         emitted_track_signatures.add(signature)
+                        self._emitted_track_signatures.add(signature)
                         diagnostics["emitted"].append(self._candidate_event(candidate, "emitted"))
                 continue
             if self._candidate_reached_trash(candidate):
@@ -339,6 +416,7 @@ class TrajectoryTracker:
                     else:
                         emitted.append(self._evidence(candidate, when))
                         emitted_track_signatures.add(signature)
+                        self._emitted_track_signatures.add(signature)
                         diagnostics["emitted"].append(self._candidate_event(candidate, "emitted"))
                 else:
                     diagnostics["rejected_candidates"] += 1
@@ -347,6 +425,8 @@ class TrajectoryTracker:
             remaining.append(candidate)
 
         self._candidates = remaining
+        if not self._candidates:
+            self._emitted_track_signatures.clear()
         diagnostics["emitted_evidence"] = len(emitted)
         diagnostics["active_candidates"] = len(self._candidates)
         diagnostics["motion_points"] = len(blobs)
@@ -359,11 +439,38 @@ class TrajectoryTracker:
 
     def _open_candidates(self, when: datetime, zone_counts: dict[str, int]) -> None:
         active_region_ids = {candidate.source_region.region_id for candidate in self._candidates}
+        used_seed_signatures: set[tuple[float, float, str]] = set()
         for region in self.regions:
             previous = self._previous_zone_counts.get(region.region_id, 0)
             current = max(0, int(zone_counts.get(region.region_id, 0)))
             if previous > 0 and current == 0 and region.region_id not in active_region_ids:
-                self._candidates.append(_TrajectoryCandidate(source_region=region, opened_at=when))
+                candidate = _TrajectoryCandidate(source_region=region, opened_at=when)
+                recent = self._recent_source_motion.get(region.region_id)
+                if recent is not None and when - recent.when <= timedelta(seconds=self.settings.trajectory_window_seconds):
+                    signature = (round(recent.x, 4), round(recent.y, 4), recent.when.isoformat())
+                    if signature in used_seed_signatures:
+                        candidate.forced_rejection_reason = "ambiguous_motion_track"
+                    else:
+                        used_seed_signatures.add(signature)
+                        candidate.points.append(recent)
+                        candidate.source_motion_seen = True
+                        candidate.source_seeded = True
+                        candidate.last_sample_at = recent.when
+                self._candidates.append(candidate)
+                self._recent_source_motion.pop(region.region_id, None)
+
+    def _remember_recent_source_motion(self, blobs: list[_MotionPoint], when: datetime, zone_counts: dict[str, int]) -> None:
+        if not blobs:
+            return
+        for region in self.regions:
+            previous = self._previous_zone_counts.get(region.region_id, 0)
+            current = max(0, int(zone_counts.get(region.region_id, 0)))
+            if previous <= 0 and current <= 0:
+                continue
+            source_blobs = [blob for blob in blobs if region_contains(region, blob.x, blob.y)]
+            if not source_blobs:
+                continue
+            self._recent_source_motion[region.region_id] = _nearest_point(region_center(region), source_blobs)
 
     def _motion_points(self, frame: Frame, when: datetime) -> list[_MotionPoint]:
         previous = self._previous_frame
@@ -441,6 +548,8 @@ class TrajectoryTracker:
         when: datetime,
         consumed_blob_ids: set[int],
     ) -> str | None:
+        if candidate.forced_rejection_reason is not None:
+            return candidate.forced_rejection_reason
         if not blobs:
             return None
         if (
@@ -527,7 +636,7 @@ class TrajectoryTracker:
         return (
             candidate.source_motion_seen
             and self._candidate_reached_trash(candidate)
-            and len(candidate.points or []) >= self.settings.trajectory_min_points
+            and self._has_enough_track_points(candidate)
             and self._direction_score(candidate) >= 0.35
             and confidence >= self.settings.trajectory_min_confidence
         )
@@ -537,7 +646,7 @@ class TrajectoryTracker:
             return "missing_source_motion"
         if not self._candidate_reached_trash(candidate):
             return "did_not_reach_trash"
-        if len(candidate.points or []) < self.settings.trajectory_min_points:
+        if not self._has_enough_track_points(candidate):
             return "insufficient_points"
         if self._direction_score(candidate) < 0.35:
             return "bad_direction"
@@ -545,6 +654,17 @@ class TrajectoryTracker:
             return "low_confidence"
         return "rejected"
 
+    def _has_enough_track_points(self, candidate: _TrajectoryCandidate) -> bool:
+        point_count = len(candidate.points or [])
+        if point_count >= self.settings.trajectory_min_points:
+            return True
+        return (
+            self.settings.trajectory_segmented_enabled
+            and point_count >= self.settings.trajectory_segmented_min_points
+            and candidate.source_motion_seen
+            and self._candidate_reached_trash(candidate)
+        )
+
     def _candidate_event(self, candidate: _TrajectoryCandidate, reason: str) -> dict[str, Any]:
         return {
             "source_zone_id": candidate.source_region.region_id,
@@ -552,8 +672,19 @@ class TrajectoryTracker:
             "point_count": len(candidate.points or []),
             "confidence": round(self._confidence(candidate), 3),
             "direction_score": round(self._direction_score(candidate), 3),
+            "segmented": self._is_segmented_track(candidate),
+            "source_seeded": candidate.source_seeded,
         }
 
+    def _is_segmented_track(self, candidate: _TrajectoryCandidate) -> bool:
+        point_count = len(candidate.points or [])
+        return (
+            self.settings.trajectory_segmented_enabled
+            and self.settings.trajectory_segmented_min_points <= point_count < self.settings.trajectory_min_points
+            and candidate.source_motion_seen
+            and self._candidate_reached_trash(candidate)
+        )
+
     def _track_signature(self, candidate: _TrajectoryCandidate) -> tuple[tuple[float, float], ...]:
         return tuple((round(point.x, 4), round(point.y, 4)) for point in candidate.points or [])
 
@@ -635,6 +766,10 @@ def load_runtime_vision_settings(config: dict[str, Any]) -> RuntimeVisionSetting
         occupancy_reflection_bright_dark_ratio=float(runtime.get("occupancy_reflection_bright_dark_ratio", 2.0)),
         occupancy_confirm_frames=max(1, int(runtime.get("occupancy_confirm_frames", 2))),
         empty_confirm_frames=max(1, int(runtime.get("empty_confirm_frames", 2))),
+        lighting_shift_guard_enabled=bool(runtime.get("lighting_shift_guard_enabled", True)),
+        lighting_shift_min_regions=max(1, int(runtime.get("lighting_shift_min_regions", 3))),
+        lighting_shift_region_fraction=max(0.0, min(1.0, float(runtime.get("lighting_shift_region_fraction", 0.6)))),
+        lighting_shift_mean_delta=float(runtime.get("lighting_shift_mean_delta", 45.0)),
         trash_motion_delta=float(runtime.get("trash_motion_delta", 18.0)),
         trash_sustained_motion_delta=float(runtime.get("trash_sustained_motion_delta", 8.0)),
         trash_sustained_motion_frames=max(1, int(runtime.get("trash_sustained_motion_frames", 2))),
@@ -643,6 +778,8 @@ def load_runtime_vision_settings(config: dict[str, Any]) -> RuntimeVisionSetting
         trajectory_window_seconds=max(1, int(runtime.get("trajectory_window_seconds", 8))),
         trajectory_sample_interval_seconds=max(0.0, float(runtime.get("trajectory_sample_interval_seconds", 1.0))),
         trajectory_min_points=max(1, int(runtime.get("trajectory_min_points", 3))),
+        trajectory_segmented_enabled=bool(runtime.get("trajectory_segmented_enabled", True)),
+        trajectory_segmented_min_points=max(2, int(runtime.get("trajectory_segmented_min_points", 2))),
         trajectory_min_confidence=float(runtime.get("trajectory_min_confidence", 0.72)),
         trajectory_motion_delta=float(runtime.get("trajectory_motion_delta", 20.0)),
         trajectory_min_blob_area=max(1, int(runtime.get("trajectory_min_blob_area", 12))),

tests/test_vision.py

@@ -158,6 +158,66 @@ class VisionTests(unittest.TestCase):
         self.assertEqual(first_empty_counts, {"1": 1})
         self.assertEqual(second_empty_counts, {"1": 0})
 
+    def test_detector_ignores_global_lighting_dimming_across_many_zones(self) -> None:
+        regions = [
+            Region(str(index + 1), ((index / 7, 0.0), ((index + 1) / 7, 0.0), ((index + 1) / 7, 1.0), (index / 7, 1.0)))
+            for index in range(7)
+        ]
+        detector = ZoneOccupancyDetector(
+            regions,
+            trash_region=None,
+            settings=RuntimeVisionSettings(
+                baseline_frames=1,
+                sample_stride_pixels=2,
+                occupancy_mean_delta=55,
+                occupancy_texture_delta=18,
+                occupancy_confirm_frames=2,
+                empty_confirm_frames=2,
+            ),
+        )
+        now = datetime(2026, 6, 1, 4, 55, tzinfo=timezone.utc)
+
+        detector.observe(solid_frame(70, 20, 180), now)
+        first_counts, _, first_diagnostics = detector.observe(solid_frame(70, 20, 100), now + timedelta(seconds=5))
+        second_counts, _, second_diagnostics = detector.observe(solid_frame(70, 20, 100), now + timedelta(seconds=10))
+
+        self.assertEqual(first_counts, {str(index): 0 for index in range(1, 8)})
+        self.assertEqual(second_counts, {str(index): 0 for index in range(1, 8)})
+        self.assertTrue(first_diagnostics["lighting_shift"]["active"])
+        self.assertTrue(second_diagnostics["lighting_shift"]["active"])
+        self.assertTrue(all(not zone["raw_occupied"] for zone in second_diagnostics["zones"].values()))
+
+    def test_detector_allows_single_zone_object_while_lighting_guard_is_available(self) -> None:
+        regions = [
+            Region(str(index + 1), ((index / 7, 0.0), ((index + 1) / 7, 0.0), ((index + 1) / 7, 1.0), (index / 7, 1.0)))
+            for index in range(7)
+        ]
+        detector = ZoneOccupancyDetector(
+            regions,
+            trash_region=None,
+            settings=RuntimeVisionSettings(
+                baseline_frames=1,
+                sample_stride_pixels=2,
+                occupancy_mean_delta=55,
+                occupancy_texture_delta=100,
+                occupancy_dark_luma_threshold=80,
+                occupancy_dark_fraction=0.06,
+                occupancy_confirm_frames=2,
+                empty_confirm_frames=2,
+            ),
+        )
+        now = datetime(2026, 6, 1, 10, 0, tzinfo=timezone.utc)
+
+        detector.observe(solid_frame(70, 20, 180), now)
+        object_frame = patched_frame(70, 20, 180, (0, 0, 10, 20, 20))
+        detector.observe(object_frame, now + timedelta(seconds=5))
+        counts, _, diagnostics = detector.observe(object_frame, now + timedelta(seconds=10))
+
+        self.assertEqual(counts["1"], 1)
+        self.assertEqual(sum(counts.values()), 1)
+        self.assertFalse(diagnostics["lighting_shift"]["active"])
+        self.assertTrue(diagnostics["zones"]["1"]["raw_occupied"])
+
     def test_runtime_vision_defaults_raise_brightness_reflection_threshold(self) -> None:
         settings = load_runtime_vision_settings({})
 
@@ -169,6 +229,10 @@ class VisionTests(unittest.TestCase):
         self.assertEqual(settings.trash_sustained_motion_delta, 8.0)
         self.assertEqual(settings.trash_sustained_motion_frames, 2)
         self.assertEqual(settings.trash_motion_cooldown_seconds, 3)
+        self.assertTrue(settings.lighting_shift_guard_enabled)
+        self.assertEqual(settings.lighting_shift_min_regions, 3)
+        self.assertAlmostEqual(settings.lighting_shift_region_fraction, 0.6)
+        self.assertEqual(settings.lighting_shift_mean_delta, 45.0)
 
     def test_detector_can_seed_previous_baseline_and_occupancy(self) -> None:
         detector = ZoneOccupancyDetector(
@@ -314,6 +378,75 @@ class VisionTests(unittest.TestCase):
         self.assertGreaterEqual(len(emitted.track_points), 3)
         self.assertGreaterEqual(emitted_evidence_count, 1)
 
+    def test_segmented_source_to_trash_track_survives_temporary_occlusion(self) -> None:
+        source = Region("source", ((0.05, 0.35), (0.25, 0.35), (0.25, 0.65), (0.05, 0.65)))
+        trash = Region("trash", ((0.72, 0.35), (0.95, 0.35), (0.95, 0.65), (0.72, 0.65)))
+        tracker = TrajectoryTracker(
+            [source],
+            trash,
+            RuntimeVisionSettings(
+                trajectory_sample_interval_seconds=0.0,
+                trajectory_min_points=3,
+                trajectory_min_confidence=0.72,
+                trajectory_motion_delta=20.0,
+                trajectory_min_blob_area=12,
+            ),
+        )
+        now = datetime(2026, 6, 1, 10, 55, tzinfo=timezone.utc)
+
+        tracker.observe(frame_with_motion_patch(80, 80, (8, 36)), now, {"source": 1})
+        source_motion_frame = frame_with_motion_patch(80, 80, (16, 36))
+        first_evidence, _ = tracker.observe(source_motion_frame, now + timedelta(seconds=1), {"source": 0})
+        occluded_evidence, occluded_diagnostics = tracker.observe(source_motion_frame, now + timedelta(seconds=2), {"source": 0})
+        trash_evidence, trash_diagnostics = tracker.observe(
+            frame_with_motion_patch(80, 80, (66, 36)),
+            now + timedelta(seconds=3),
+            {"source": 0},
+        )
+
+        self.assertEqual(first_evidence, [])
+        self.assertEqual(occluded_evidence, [])
+        self.assertEqual(occluded_diagnostics["active_candidates"], 1)
+        self.assertEqual(len(trash_evidence), 1)
+        emitted = trash_evidence[0]
+        self.assertEqual(emitted.source_zone_id, "source")
+        self.assertEqual(emitted.target, "trash")
+        self.assertEqual(emitted.method, "motion")
+        self.assertEqual(len(emitted.track_points), 2)
+        self.assertGreaterEqual(emitted.confidence, 0.72)
+        self.assertEqual(trash_diagnostics["emitted"][0]["reason"], "emitted")
+        self.assertTrue(trash_diagnostics["emitted"][0]["segmented"])
+
+    def test_recent_source_motion_seeds_track_when_empty_confirmation_lags(self) -> None:
+        source = Region("source", ((0.05, 0.35), (0.25, 0.35), (0.25, 0.65), (0.05, 0.65)))
+        trash = Region("trash", ((0.72, 0.35), (0.95, 0.35), (0.95, 0.65), (0.72, 0.65)))
+        tracker = TrajectoryTracker(
+            [source],
+            trash,
+            RuntimeVisionSettings(
+                trajectory_sample_interval_seconds=0.0,
+                trajectory_min_points=3,
+                trajectory_min_confidence=0.72,
+                trajectory_motion_delta=20.0,
+                trajectory_min_blob_area=12,
+            ),
+        )
+        now = datetime(2026, 6, 1, 11, 25, tzinfo=timezone.utc)
+
+        tracker.observe(frame_with_motion_patch(80, 80, (8, 36)), now, {"source": 1})
+        tracker.observe(frame_with_motion_patch(80, 80, (16, 36)), now + timedelta(seconds=1), {"source": 1})
+        evidence, diagnostics = tracker.observe(
+            frame_with_motion_patch(80, 80, (66, 36)),
+            now + timedelta(seconds=2),
+            {"source": 0},
+        )
+
+        self.assertEqual(len(evidence), 1)
+        self.assertEqual(evidence[0].source_zone_id, "source")
+        self.assertEqual(len(evidence[0].track_points), 2)
+        self.assertTrue(diagnostics["emitted"][0]["segmented"])
+        self.assertTrue(diagnostics["emitted"][0]["source_seeded"])
+
     def test_motion_that_starts_away_from_source_zone_is_rejected(self) -> None:
         source = Region("source", ((0.05, 0.35), (0.25, 0.35), (0.25, 0.65), (0.05, 0.65)))
         trash = Region("trash", ((0.72, 0.35), (0.95, 0.35), (0.95, 0.65), (0.72, 0.65)))
@@ -569,6 +702,8 @@ class VisionTests(unittest.TestCase):
         self.assertEqual(settings.trajectory_window_seconds, 8)
         self.assertEqual(settings.trajectory_sample_interval_seconds, 1.0)
         self.assertEqual(settings.trajectory_min_points, 3)
+        self.assertTrue(settings.trajectory_segmented_enabled)
+        self.assertEqual(settings.trajectory_segmented_min_points, 2)
         self.assertEqual(settings.trajectory_min_confidence, 0.72)
         self.assertEqual(settings.trajectory_motion_delta, 20.0)
         self.assertEqual(settings.trajectory_min_blob_area, 12)
@@ -587,6 +722,8 @@ class VisionTests(unittest.TestCase):
                     "trajectory_window_seconds": 11,
                     "trajectory_sample_interval_seconds": 0.5,
                     "trajectory_min_points": 4,
+                    "trajectory_segmented_enabled": False,
+                    "trajectory_segmented_min_points": 3,
                     "trajectory_min_confidence": 0.8,
                     "trajectory_motion_delta": 25.0,
                     "trajectory_min_blob_area": 20,
@@ -596,6 +733,10 @@ class VisionTests(unittest.TestCase):
                     "yolo_enabled": True,
                     "yolo_model_path": "models/yolo.onnx",
                     "yolo_min_confidence": 0.7,
+                    "lighting_shift_guard_enabled": False,
+                    "lighting_shift_min_regions": 4,
+                    "lighting_shift_region_fraction": 0.75,
+                    "lighting_shift_mean_delta": 60.0,
                 }
             }
         )
@@ -604,6 +745,8 @@ class VisionTests(unittest.TestCase):
         self.assertEqual(settings.trajectory_window_seconds, 11)
         self.assertEqual(settings.trajectory_sample_interval_seconds, 0.5)
         self.assertEqual(settings.trajectory_min_points, 4)
+        self.assertFalse(settings.trajectory_segmented_enabled)
+        self.assertEqual(settings.trajectory_segmented_min_points, 3)
         self.assertEqual(settings.trajectory_min_confidence, 0.8)
         self.assertEqual(settings.trajectory_motion_delta, 25.0)
         self.assertEqual(settings.trajectory_min_blob_area, 20)
@@ -613,6 +756,10 @@ class VisionTests(unittest.TestCase):
         self.assertTrue(settings.yolo_enabled)
         self.assertEqual(settings.yolo_model_path, "models/yolo.onnx")
         self.assertEqual(settings.yolo_min_confidence, 0.7)
+        self.assertFalse(settings.lighting_shift_guard_enabled)
+        self.assertEqual(settings.lighting_shift_min_regions, 4)
+        self.assertAlmostEqual(settings.lighting_shift_region_fraction, 0.75)
+        self.assertEqual(settings.lighting_shift_mean_delta, 60.0)
 
 
 if __name__ == "__main__":