fix(perception): release processed source payloads

src/k1link/perception/graph.py

@@ -24,6 +24,7 @@ from .contracts import (
     ObjectProposal2D,
     ObstacleObservation,
     SourceAccounting,
+    SourceEnvelope,
     TemporalObstacle,
     TemporalState,
     validate_exclusive_point_ownership,
@@ -197,7 +198,9 @@ class ReferencePerceptionGraphV1:
         self._threads: list[Thread] = []
         self._outcomes: dict[int, TerminalOutcome] = {}
         self._deliveries: list[DeliveredFrame] = []
-        self._admitted: dict[int, SourcePacket] = {}
+        # Accounting needs immutable source identity, not the decoded payload. Keeping
+        # SourcePacket here retained every camera raster until the run completed.
+        self._admitted: dict[int, SourceEnvelope] = {}
         self._admitted_at_ns: dict[int, int] = {}
         self._queue_high_watermarks: dict[str, int] = {
             stage_id: 0 for stage_id in ("detector", "geometry", "temporal", "rolling", "threat")
@@ -334,7 +337,7 @@ class ReferencePerceptionGraphV1:
         with self._result_lock:
             if envelope.sequence in self._admitted:
                 return False
-            self._admitted[envelope.sequence] = packet
+            self._admitted[envelope.sequence] = envelope
             self._admitted_at_ns[envelope.sequence] = self._now()
         if envelope.source_id != BASELINE_SOURCE_ID:
             self._terminal(packet, TerminalOutcomeType.REJECTED, "source", "source-not-admitted")
@@ -620,7 +623,15 @@ class ReferencePerceptionGraphV1:
         stage_id: str,
         reason: str,
     ) -> None:
-        envelope = packet.envelope
+        self._terminal_envelope(packet.envelope, outcome, stage_id, reason)
+
+    def _terminal_envelope(
+        self,
+        envelope: SourceEnvelope,
+        outcome: TerminalOutcomeType,
+        stage_id: str,
+        reason: str,
+    ) -> None:
         terminal = TerminalOutcome(
             source_id=envelope.source_id,
             session_id=envelope.session_id,
@@ -632,6 +643,7 @@ class ReferencePerceptionGraphV1:
         )
         with self._result_lock:
             self._outcomes.setdefault(envelope.sequence, terminal)
+            self._admitted_at_ns.pop(envelope.sequence, None)
 
     def _put_latest(
         self,
@@ -737,10 +749,10 @@ class ReferencePerceptionGraphV1:
     def _close_accounting(self) -> None:
         with self._result_lock:
             missing = sorted(set(self._admitted) - set(self._outcomes))
-            packets = [self._admitted[sequence] for sequence in missing]
-        for packet in packets:
-            self._terminal(
-                packet,
+            envelopes = [self._admitted[sequence] for sequence in missing]
+        for envelope in envelopes:
+            self._terminal_envelope(
+                envelope,
                 TerminalOutcomeType.FAILED,
                 "graph",
                 "terminal-accounting-gap",

tests/test_perception_graph.py

@@ -1,9 +1,11 @@
 from __future__ import annotations
 
+import gc
 import json
 import subprocess
 import sys
 import threading
+import weakref
 from collections.abc import Callable, Iterator
 from dataclasses import replace
 from pathlib import Path
@@ -553,6 +555,40 @@ def test_reference_graph_v2_observes_final_delivery_completion_age() -> None:
     assert observed == [(0, 123, 123)]
 
 
+def test_reference_graph_retains_source_identity_without_processed_payload() -> None:
+    packet = _packet(0)
+    graph = _graph_v2(_Source((packet,)))
+
+    result = graph.run()
+
+    assert result.state is GraphState.STOPPED
+    assert graph._admitted == {0: packet.envelope}
+    assert graph._admitted_at_ns == {}
+
+
+def test_reference_graph_releases_ephemeral_source_payloads_after_run() -> None:
+    payloads: list[weakref.ReferenceType[np.ndarray]] = []
+
+    class EphemeralSource:
+        provider_id = _Source.provider_id
+
+        def packets(self, stop_event: Event) -> Iterator[SourcePacket]:
+            for sequence in range(32):
+                if stop_event.is_set():
+                    return
+                image = np.empty((64, 64, 3), dtype=np.uint8)
+                payloads.append(weakref.ref(image))
+                yield replace(_packet(sequence), image_payload=image)
+
+    graph = _graph_v2(EphemeralSource())
+
+    result = graph.run()
+    gc.collect()
+
+    assert result.state is GraphState.STOPPED
+    assert not any(reference() is not None for reference in payloads)
+
+
 def test_reference_graph_v2_observes_exact_delivery_evidence_inputs() -> None:
     observed: list[tuple[int, int, tuple[str, ...], frozenset[str], int]] = []