Merge pull request #90 from rolandpg/feat/rfc-009-phase0.5-instrumentation

feat(telemetry): per-phase timers in remember() — RFC-009 Phase 0.5

Author: rolandpg

src/zettelforge/memory_manager.py

@@ -230,21 +230,33 @@ def remember(
             )
 
         # Direct store path
+        # [RFC-009 Phase 0.5] Per-phase timings to attribute remember() latency.
+        # Emitted in ocsf_api_activity.phase_timings_ms so the RFC-007 aggregator
+        # can bucket them without schema changes.
+        phase_timings_ms: Dict[str, float] = {}
+
+        _p = time.perf_counter()
         note = self.constructor.construct(
             raw_content=content, source_type=source_type, source_ref=source_ref, domain=domain
         )
+        phase_timings_ms["construct"] = (time.perf_counter() - _p) * 1000
 
+        _p = time.perf_counter()
         self.store.write_note(note)
+        phase_timings_ms["write_note"] = (time.perf_counter() - _p) * 1000
         self.stats["notes_created"] += 1
 
         # Keep LanceDB in sync for vector retrieval
         if self._lance_store.lancedb is not None:
+            _p = time.perf_counter()
             try:
                 self._lance_store._index_in_lance(note)
             except Exception:
                 self._logger.warning("lance_index_sync_failed", note_id=note.id, exc_info=True)
+            phase_timings_ms["lance_index"] = (time.perf_counter() - _p) * 1000
 
         # Alias resolution and indexing (regex-only for speed; LLM NER runs on recall)
+        _p = time.perf_counter()
         raw_entities = self.indexer.extractor.extract_all(note.content.raw, use_llm=False)
 
         resolved_entities = {}
@@ -257,8 +269,10 @@ def remember(
         for etype, elist in resolved_entities.items():
             for evalue in elist:
                 self.store.add_entity_mapping(etype, evalue, note.id)
+        phase_timings_ms["entity_index"] = (time.perf_counter() - _p) * 1000
 
         # GAM consolidation: observe note for semantic shift detection
+        _p = time.perf_counter()
         try:
             is_shift, shift_meta = self.consolidation.before_write(
                 note_entities=resolved_entities,
@@ -274,14 +288,24 @@ def remember(
                 )
         except Exception as e:
             self._logger.warning("consolidation_observe_failed", error=str(e))
+        phase_timings_ms["consolidation_observe"] = (time.perf_counter() - _p) * 1000
 
         # Phase 3: Check supersession
+        _p = time.perf_counter()
         self._check_supersession(note, resolved_entities)
+        phase_timings_ms["supersession"] = (time.perf_counter() - _p) * 1000
 
         # Phase 6: Knowledge Graph Update (heuristic edges — fast path)
+        _p = time.perf_counter()
         self._update_knowledge_graph(note, resolved_entities)
-
-        # Phase 6b: LLM causal enrichment (slow path — background worker)
+        phase_timings_ms["kg_update"] = (time.perf_counter() - _p) * 1000
+
+        # Phase 6b/6c/6d: dispatch background enrichment jobs (causal + NER + evolution).
+        # The dispatch bucket measures job construction + put_nowait() + count_notes()
+        # overhead only. In sync=True mode the LLM work runs inline and is intentionally
+        # EXCLUDED from this bucket — mixing LLM latency into "dispatch" would corrupt
+        # the Phase 0.5 attribution. sync=True is retained for tests/debug.
+        dispatch_start = time.perf_counter() if not sync else None
         job = _EnrichmentJob(
             note_id=note.id,
             domain=domain,
@@ -331,6 +355,8 @@ def remember(
                     self._pending_enrichment.add(note.id)
                 except queue.Full:
                     self._logger.warning("evolution_queue_full", note_id=note.id)
+        if dispatch_start is not None:
+            phase_timings_ms["enrichment_dispatch"] = (time.perf_counter() - dispatch_start) * 1000
 
         duration_ms = (time.perf_counter() - start) * 1000
         log_api_activity(
@@ -341,6 +367,7 @@ def remember(
             duration_ms=duration_ms,
             request_id=request_id,
             evolve=False,
+            phase_timings_ms={k: round(v, 2) for k, v in phase_timings_ms.items()},
         )
         return note, "created"
 

tests/test_logging_compliance.py

@@ -139,6 +139,75 @@ def test_authorization_deny_fields(self):
         assert event["policy"] == "GOV-011"
 
 
+class TestPhaseTimingsInstrumentation:
+    """[RFC-009 Phase 0.5] remember() emits per-phase timers for latency attribution."""
+
+    def test_remember_emits_phase_timings_ms(self):
+        """The ocsf_api_activity event for remember() must carry phase_timings_ms
+        with numeric values for each instrumented phase."""
+        from zettelforge import MemoryManager
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            mm = MemoryManager(jsonl_path=f"{tmpdir}/notes.jsonl", lance_path=f"{tmpdir}/vectordb")
+            with structlog.testing.capture_logs() as logs:
+                mm.remember("APT28 deploys Cobalt Strike beacons", domain="cti")
+
+        remember_events = [
+            e
+            for e in logs
+            if e.get("event") == "ocsf_api_activity" and e.get("activity_name") == "remember"
+        ]
+        assert remember_events, "No ocsf_api_activity event for remember() was emitted"
+
+        event = remember_events[-1]
+        assert "phase_timings_ms" in event, (
+            "phase_timings_ms missing from remember() ocsf_api_activity event"
+        )
+        timings = event["phase_timings_ms"]
+        assert isinstance(timings, dict)
+
+        # Phases that always run on the direct-store async path
+        expected_keys = {
+            "construct",
+            "write_note",
+            "entity_index",
+            "consolidation_observe",
+            "supersession",
+            "kg_update",
+            "enrichment_dispatch",
+        }
+        missing = expected_keys - set(timings)
+        assert not missing, f"Missing phase timings: {missing}. Got: {set(timings)}"
+
+        for key, value in timings.items():
+            assert isinstance(value, (int, float)), (
+                f"phase_timings_ms[{key!r}] is {type(value).__name__}, expected numeric"
+            )
+            assert value >= 0, f"phase_timings_ms[{key!r}] = {value} (negative)"
+
+    def test_enrichment_dispatch_excluded_in_sync_mode(self):
+        """In sync=True mode the dispatch bucket is intentionally omitted so
+        inline LLM work is not misattributed to dispatch latency."""
+        from zettelforge import MemoryManager
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            mm = MemoryManager(jsonl_path=f"{tmpdir}/notes.jsonl", lance_path=f"{tmpdir}/vectordb")
+            with structlog.testing.capture_logs() as logs:
+                mm.remember("APT29 phishing campaign observed", domain="cti", sync=True)
+
+        remember_events = [
+            e
+            for e in logs
+            if e.get("event") == "ocsf_api_activity" and e.get("activity_name") == "remember"
+        ]
+        assert remember_events
+        timings = remember_events[-1]["phase_timings_ms"]
+        assert "enrichment_dispatch" not in timings, (
+            "sync=True should NOT emit enrichment_dispatch (would mix LLM work into "
+            "dispatch bucket and corrupt Phase 0.5 attribution)"
+        )
+
+
 class TestLanceDBFailureLogged:
     """Regression test for #26: LanceDB failures must be logged, not swallowed."""