fix(perception): keep cyclic GC outside realtime loop

experiments/perception/run_m48s_reference_graph_shadow_worker.py

@@ -51,9 +51,10 @@ from k1link.perception.reference_graph_runtime import ReferenceGraphRuntimePaths
 from k1link.perception.rolling_map import RollingLocalObstacleMapProvider
 from k1link.perception.temporal import BoundedSpatialTemporalProvider
 
-SCHEMA_VERSION: Final = "missioncore.m48s-reference-graph-shadow-load/v1"
+SCHEMA_VERSION: Final = "missioncore.m48s-reference-graph-shadow-load/v2"
 FRAME_EVIDENCE_SCHEMA: Final = "missioncore.m48s-reference-graph-frame-evidence/v1"
 PIPELINE_TIMING_SCHEMA: Final = "missioncore.m48s-frame-pipeline-timing/v0"
+GC_POLICY_SCHEMA: Final = "missioncore.cyclic-gc-hot-loop-policy/v0"
 AUTHORITY: Final = {
     "ground_truth": False,
     "candidate_accepted": False,
@@ -175,6 +176,51 @@ class GcPauseTelemetry:
         )
 
 
+class CyclicGcHotLoopPolicy:
+    """Keep cyclic GC outside the source-paced realtime envelope."""
+
+    def __init__(self) -> None:
+        self.enabled_before: bool | None = None
+        self.disabled_during_hot_loop = False
+        self.enabled_after: bool | None = None
+        self.pre_collected: int | None = None
+        self.post_collected: int | None = None
+
+    def __enter__(self) -> CyclicGcHotLoopPolicy:
+        self.enabled_before = gc.isenabled()
+        self.pre_collected = gc.collect()
+        gc.disable()
+        self.disabled_during_hot_loop = not gc.isenabled()
+        if not self.disabled_during_hot_loop:
+            raise RuntimeError("cyclic GC remained enabled inside the hot loop")
+        return self
+
+    def __exit__(self, *_args: object) -> None:
+        if self.enabled_before:
+            gc.enable()
+        self.post_collected = gc.collect()
+        self.enabled_after = gc.isenabled()
+        if self.enabled_after is not self.enabled_before:
+            raise RuntimeError("cyclic GC state was not restored after the hot loop")
+
+    def to_dict(self) -> dict[str, object]:
+        if (
+            self.enabled_before is None
+            or self.enabled_after is None
+            or self.pre_collected is None
+            or self.post_collected is None
+        ):
+            raise RuntimeError("cyclic GC hot-loop policy did not close")
+        return {
+            "schema_version": GC_POLICY_SCHEMA,
+            "enabled_before": self.enabled_before,
+            "disabled_during_hot_loop": self.disabled_during_hot_loop,
+            "enabled_after": self.enabled_after,
+            "pre_collected": self.pre_collected,
+            "post_collected": self.post_collected,
+        }
+
+
 class FrameTimingStore:
     """Join bounded decode, detector and provider timings by source sequence."""
 
@@ -397,9 +443,11 @@ def main() -> int:
                             gc_telemetry,
                         ),
                     )
-                loop_started_ns = time.monotonic_ns()
+                gc_policy = CyclicGcHotLoopPolicy()
-                result = runtime.graph.run()
+                with gc_policy:
-                loop_completed_ns = time.monotonic_ns()
+                    loop_started_ns = time.monotonic_ns()
+                    result = runtime.graph.run()
+                    loop_completed_ns = time.monotonic_ns()
                 detector_snapshot = cast(
                     RfDetrShadowDetectorProvider,
                     runtime.graph.detector,
@@ -447,6 +495,7 @@ def main() -> int:
                 completion_ages_ns=loop_completion_ages_ns,
                 wall_seconds=loop_wall_seconds,
                 setup_seconds=setup_seconds,
+                gc_policy=gc_policy.to_dict(),
             )
             loop_documents.append(loop_document)
             completion_ages_ms.extend(value / 1_000_000.0 for value in loop_completion_ages_ns)
@@ -523,6 +572,12 @@ def main() -> int:
         "distinct_class_family_policy": len(set(advisory_policy_matrix().values()))
         == len(AdvisoryFamily),
         "complete_pipeline_timing": len(all_pipeline_timings) == delivered,
+        "cyclic_gc_disabled_during_hot_loop": all(
+            cast(dict[str, object], loop["cyclic_gc_hot_loop"])["disabled_during_hot_loop"] is True
+            and cast(dict[str, object], loop["cyclic_gc_hot_loop"])["enabled_before"]
+            is cast(dict[str, object], loop["cyclic_gc_hot_loop"])["enabled_after"]
+            for loop in loop_documents
+        ),
         "authority_remains_false": all(value is False for value in AUTHORITY.values()),
     }
     integrated_runtime_gate_passed = all(checks.values())
@@ -634,6 +689,7 @@ def _loop_document(
     completion_ages_ns: list[int],
     wall_seconds: float,
     setup_seconds: float,
+    gc_policy: dict[str, object],
 ) -> dict[str, object]:
     outcomes = Counter(item.outcome.value for item in result.terminal_outcomes)
     outcome_stages = Counter(
@@ -644,6 +700,7 @@ def _loop_document(
         "state": result.state.value,
         "wall_seconds": round(wall_seconds, 6),
         "setup_seconds": round(setup_seconds, 6),
+        "cyclic_gc_hot_loop": gc_policy,
         "admitted_count": result.admitted_count,
         "delivered_count": len(result.deliveries),
         "effective_world_state_fps": round(len(result.deliveries) / wall_seconds, 6),

tests/test_m48s_reference_graph_timing.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 import importlib.util
 import time
 from pathlib import Path
+from unittest.mock import patch
 
 from k1link.perception.detector import DetectorFrameTiming
 from k1link.perception.recorded_source import DecodedFrameTiming
@@ -108,3 +109,34 @@ def test_gc_pause_telemetry_attributes_collection_to_active_stage() -> None:
         }
     ]
     assert summary["maximum_event"]["duration_ms"] >= 1.0
+
+
+def test_cyclic_gc_policy_collects_outside_hot_loop_and_restores_state() -> None:
+    state = {"enabled": True}
+    collections = iter((3, 0))
+
+    def disable() -> None:
+        state["enabled"] = False
+
+    def enable() -> None:
+        state["enabled"] = True
+
+    with (
+        patch.object(RUNNER.gc, "isenabled", side_effect=lambda: state["enabled"]),
+        patch.object(RUNNER.gc, "disable", side_effect=disable),
+        patch.object(RUNNER.gc, "enable", side_effect=enable),
+        patch.object(RUNNER.gc, "collect", side_effect=lambda: next(collections)),
+    ):
+        policy = RUNNER.CyclicGcHotLoopPolicy()
+        with policy:
+            assert state["enabled"] is False
+
+        assert state["enabled"] is True
+        assert policy.to_dict() == {
+            "schema_version": RUNNER.GC_POLICY_SCHEMA,
+            "enabled_before": True,
+            "disabled_during_hot_loop": True,
+            "enabled_after": True,
+            "pre_collected": 3,
+            "post_collected": 0,
+        }