feat: PRD §42 OperationTimeline read-model + boot wiring (default-FALSE)

backend/core/ouroboros/battle_test/harness.py

@@ -707,6 +707,28 @@ def _harness_loop_exception_handler(loop_, ctx_):
         except Exception:  # noqa: BLE001
             logger.debug("register_ops_digest_observer(boot) failed", exc_info=True)
 
+        # PRD §42 Slice 2 — add the OperationTimeline read-model to the
+        # fan-out (coexists with SessionRecorder via the composite seam,
+        # NOT by evicting it) and replay the durable causal index so the
+        # cross-session "what did O+V do while I was away" scrub is live
+        # on boot. Master-flag-gated internally (default-FALSE) — this
+        # wiring is a no-op until JARVIS_OPERATION_TIMELINE_ENABLED=true.
+        try:
+            from backend.core.ouroboros.governance.ops_digest_observer import (
+                add_ops_digest_observer,
+            )
+            from backend.core.ouroboros.governance.operation_timeline import (
+                get_default_timeline,
+            )
+            _timeline = get_default_timeline()
+            add_ops_digest_observer(_timeline)
+            _replayed = _timeline.replay_from_disk()
+            logger.debug(
+                "operation_timeline wired (replayed %d rows)", _replayed,
+            )
+        except Exception:  # noqa: BLE001
+            logger.debug("operation_timeline(boot) wiring failed", exc_info=True)
+
         _boot_mark("harness_run_pre_boot_done")
         try:
             # Boot sequence — each phase wrapped for boot-timing visibility

backend/core/ouroboros/battle_test/serpent_flow.py

@@ -6259,6 +6259,9 @@ def _handle_expand(self, line: str) -> None:
             elif ref_or_op.startswith("b-"):
                 # Treefinement Phase 4 — L2 tree-search branch archive
                 self._expand_repair_branch(ref_or_op)
+            elif ref_or_op.startswith("r-"):
+                # PRD §42 Slice 2 — causal Operation Timeline row
+                self._expand_timeline(ref_or_op)
             else:
                 # Treat as op_id and find latest matching o-N
                 self._expand_op_block_by_op_id(ref_or_op)
@@ -6344,6 +6347,16 @@ def _print_expand_summary(self) -> None:
             perm_recent = _pa().all_refs()[-5:]
         except Exception:
             perm_recent = ()
+        # PRD §42 Slice 2 — causal Operation Timeline (6th
+        # cross-substrate ref family). Lazy + guarded like perm_recent.
+        timeline_recent: tuple = ()
+        try:
+            from backend.core.ouroboros.governance.operation_timeline import (  # noqa: E501
+                get_default_timeline as _tl,
+            )
+            timeline_recent = tuple(_tl().list_recent(limit=5))
+        except Exception:
+            timeline_recent = ()
 
         self._flow.console.print(
             f"  [{_C['neural']}]Recent retrievable refs:[/{_C['neural']}]",
@@ -6386,9 +6399,24 @@ def _print_expand_summary(self) -> None:
                 f"[/{_C['evolved']}]",
                 highlight=False,
             )
+        if timeline_recent:
+            self._flow.console.print(
+                f"    [{_C['dim']}]timeline:[/{_C['dim']}]",
+                highlight=False,
+            )
+            for tr in timeline_recent:
+                self._flow.console.print(
+                    f"      [{_C['evolved']}]{tr.ref}[/{_C['evolved']}] "
+                    f"[{_C['dim']}]{tr.op_id} · "
+                    f"{tr.signal_source or '?'} · "
+                    f"commit={(tr.commit_hash or '-')[:10]}"
+                    f"[/{_C['dim']}]",
+                    highlight=False,
+                )
         if (
             not op_recent and not diff_recent
             and not tool_refs and not perm_recent
+            and not timeline_recent
         ):
             self._flow.console.print(
                 f"  [{_C['dim']}]No retrievable refs yet[/{_C['dim']}]",
@@ -6447,6 +6475,129 @@ def _expand_diff(self, ref: str) -> None:
                 f"    [{_C['dim']}]{ln}[/{_C['dim']}]", highlight=False,
             )
 
+    def _expand_timeline(self, ref: str) -> None:
+        """PRD §42 Slice 2 — expand one causal Operation Timeline row:
+        the full signal→op→diff→commit→outcome join for an ``r-N``
+        ref. Composes the authority-free read-model singleton."""
+        from backend.core.ouroboros.governance.operation_timeline import (
+            get_default_timeline,
+        )
+        row = get_default_timeline().lookup(ref)
+        if row is None:
+            self._flow.console.print(
+                f"  [{_C['heal']}]No timeline row for {ref}"
+                f"[/{_C['heal']}]",
+                highlight=False,
+            )
+            return
+        self._flow.console.print(
+            f"  [{_C['neural']}]⏺ Timeline[/{_C['neural']}] "
+            f"[{_C['dim']}]{row.ref} · {row.op_id}[/{_C['dim']}]",
+            highlight=False,
+        )
+
+        def _line(label: str, value: object) -> None:
+            if value in (None, "", ()):
+                return
+            self._flow.console.print(
+                f"    [{_C['dim']}]{label}:[/{_C['dim']}] "
+                f"[{_C['file']}]{value}[/{_C['file']}]",
+                highlight=False,
+            )
+
+        _line("signal", row.signal_source)
+        _line("urgency", row.urgency)
+        _line("risk", row.risk_tier)
+        _line("apply", row.apply_mode)
+        if row.verify_total is not None:
+            _line(
+                "verify",
+                f"{row.verify_passed}/{row.verify_total}",
+            )
+        _line("commit", row.commit_hash)
+        _line("diff", row.diff_ref)
+        if row.file_paths:
+            _line("files", f"{len(row.file_paths)} file(s)")
+            for p in row.file_paths[:20]:
+                self._flow.console.print(
+                    f"      [{_C['dim']}]{p}[/{_C['dim']}]",
+                    highlight=False,
+                )
+        _line("terminal", row.terminal_state)
+        _line("reverted_by", row.reverted_by)
+        _line("updated", row.updated_iso)
+
+    def _handle_timeline(self, line: str) -> None:
+        """Show the causal operation timeline — what O+V did, newest
+        first. ``/timeline`` lists recent ops; ``/timeline <r-N>`` or
+        ``/timeline <op-id>`` expands one. @ALIAS_TAG /tl
+        @EXAMPLE_TAG /timeline @EXAMPLE_TAG /timeline r-7"""
+        try:
+            from backend.core.ouroboros.governance.operation_timeline import (
+                get_default_timeline,
+            )
+        except Exception:  # noqa: BLE001
+            self._flow.console.print(
+                f"  [{_C['dim']}]/timeline: substrate not available"
+                f"[/{_C['dim']}]",
+                highlight=False,
+            )
+            return
+        parts = line.split(None, 1)
+        arg = parts[1].strip() if len(parts) > 1 else ""
+        tl = get_default_timeline()
+        if arg:
+            # Expand a specific r-N ref, or reverse-lookup by op_id.
+            if arg.startswith("r-"):
+                self._expand_timeline(arg)
+                return
+            matches = tl.query(op_id=arg, limit=1)
+            if matches:
+                self._expand_timeline(matches[0].ref)
+            else:
+                self._flow.console.print(
+                    f"  [{_C['heal']}]No timeline row for {arg}"
+                    f"[/{_C['heal']}]",
+                    highlight=False,
+                )
+            return
+        if not tl.is_enabled():
+            self._flow.console.print(
+                f"  [{_C['dim']}]/timeline: disabled "
+                f"(JARVIS_OPERATION_TIMELINE_ENABLED=false)[/{_C['dim']}]",
+                highlight=False,
+            )
+            return
+        recent = tl.list_recent(limit=12)
+        if not recent:
+            self._flow.console.print(
+                f"  [{_C['dim']}]/timeline: no operations recorded yet"
+                f"[/{_C['dim']}]",
+                highlight=False,
+            )
+            return
+        self._flow.console.print(
+            f"  [{_C['neural']}]⏺ Operation Timeline[/{_C['neural']}] "
+            f"[{_C['dim']}]({len(recent)} most recent · "
+            f"/expand r-N for detail)[/{_C['dim']}]",
+            highlight=False,
+        )
+        for r in recent:
+            verify = (
+                f"{r.verify_passed}/{r.verify_total}"
+                if r.verify_total is not None else "-"
+            )
+            commit = (r.commit_hash or "")[:10] or "-"
+            self._flow.console.print(
+                f"    [{_C['evolved']}]{r.ref}[/{_C['evolved']}] "
+                f"[{_C['dim']}]{r.op_id} · "
+                f"{r.signal_source or '?'} · "
+                f"{r.risk_tier or '?'} · "
+                f"apply={r.apply_mode or '-'} · "
+                f"verify={verify} · commit={commit}[/{_C['dim']}]",
+                highlight=False,
+            )
+
     def _expand_op_block(self, ref: str) -> None:
         from backend.core.ouroboros.battle_test.op_block_buffer import (
             get_default_buffer,

backend/core/ouroboros/battle_test/session_recorder.py

@@ -305,6 +305,22 @@ def on_commit_succeeded(
         except Exception:  # noqa: BLE001
             pass
 
+    def on_op_classified(
+        self,
+        *,
+        op_id: str,
+        signal_source: str,
+        urgency: str,
+        risk_tier: str,
+    ) -> None:
+        """No-op (PRD §42 Slice 2). SessionRecorder deliberately
+        ignores the causal signal→op edge — that edge exists for the
+        OperationTimeline read-model, not the session digest. Declared
+        explicitly so SessionRecorder structurally satisfies the
+        extended OpsDigestObserver protocol (no per-op AttributeError
+        through the composite fan-out)."""
+        return
+
     # ------------------------------------------------------------------
     # Recording
     # ------------------------------------------------------------------

backend/core/ouroboros/governance/ide_observability.py

@@ -210,6 +210,11 @@ def register_routes(self, app: "web.Application") -> None:
         app.router.add_get(
             "/observability/tasks/{op_id}", self._handle_task_detail,
         )
+        # PRD §42 Slice 2 — causal Operation Timeline read surface
+        # (authority-free: operation_timeline imports no gate module).
+        app.router.add_get(
+            "/observability/timeline", self._handle_timeline,
+        )
         # Problem #7 Slice 4 — plan approval surface.
         app.router.add_get(
             "/observability/plans", self._handle_plan_list,
@@ -683,6 +688,85 @@ async def _handle_task_detail(self, request: "web.Request") -> Any:
             },
         )
 
+    # ------------------------------------------------------------------
+    # Operation Timeline route (PRD §42 Slice 2)
+    # ------------------------------------------------------------------
+
+    async def _handle_timeline(self, request: "web.Request") -> Any:
+        """GET /observability/timeline — the causal Operation Timeline
+        scrub (PRD §42).
+
+        Read-only projection of the authority-free OperationTimeline
+        read-model: ``signal → op → diff → commit → outcome`` joined
+        per op_id, newest first. Optional ``?op_id=`` filter and
+        ``?limit=`` cap (default 50, hard ceiling 500). 403 when the
+        observability flag is off. NEVER leaks stack traces or paths.
+
+        Shape::
+
+            {
+              "schema_version": "1.0",
+              "timeline_schema": "timeline.1",
+              "enabled": true,
+              "count": 7,
+              "rows": [{"ref": "r-7", "op_id": "...",
+                        "signal_source": "TestFailure",
+                        "risk_tier": "notify_apply",
+                        "apply_mode": "single",
+                        "verify_passed": 14, "verify_total": 14,
+                        "commit_hash": "17ae95d7d6", "diff_ref": "d-3",
+                        "updated_iso": "..."}, ...]
+            }
+        """
+        if not ide_observability_enabled():
+            return self._error_response(
+                request, 403, "ide_observability.disabled",
+            )
+        if not self._check_rate_limit(self._client_key(request)):
+            return self._error_response(
+                request, 429, "ide_observability.rate_limited",
+            )
+        op_id_filter = request.query.get("op_id") or None
+        if op_id_filter is not None and not _OP_ID_RE.match(op_id_filter):
+            return self._error_response(
+                request, 400, "ide_observability.malformed_op_id",
+            )
+        try:
+            limit = int(request.query.get("limit", "50"))
+        except (TypeError, ValueError):
+            limit = 50
+        limit = max(1, min(limit, 500))  # bounded projection (§8)
+        try:
+            from backend.core.ouroboros.governance.operation_timeline import (
+                TIMELINE_SCHEMA_VERSION,
+                get_default_timeline,
+            )
+            tl = get_default_timeline()
+            rows = tl.query(op_id=op_id_filter, limit=limit)
+            return self._json_response(
+                request, 200,
+                {
+                    "timeline_schema": TIMELINE_SCHEMA_VERSION,
+                    "enabled": tl.is_enabled(),
+                    "count": len(rows),
+                    "rows": [r.to_dict() for r in rows],
+                },
+            )
+        except Exception:  # noqa: BLE001 — defensive: empty, never 500
+            logger.debug(
+                "[IDEObservability] timeline projection failed",
+                exc_info=True,
+            )
+            return self._json_response(
+                request, 200,
+                {
+                    "timeline_schema": "timeline.1",
+                    "enabled": False,
+                    "count": 0,
+                    "rows": [],
+                },
+            )
+
     # ------------------------------------------------------------------
     # Plan Approval routes (problem #7 Slice 4)
     # ------------------------------------------------------------------

backend/core/ouroboros/governance/ide_observability_stream.py

@@ -125,6 +125,10 @@
 # Single event type: the trigger (inference vs override) is carried in
 # the payload so clients render from one handler.
 EVENT_TYPE_POSTURE_CHANGED = "posture_changed"
+# PRD §42 Slice 2 — one durable causal timeline row was appended
+# (apply/verify/commit/classified merge). Read-surface notification
+# only; the row is authoritative on disk + in the read-model.
+EVENT_TYPE_OPERATION_TIMELINE_ROW = "operation_timeline_row"
 
 # FlagRegistry Slice 3 — flag introspection stream vocabulary.
 EVENT_TYPE_FLAG_TYPO_DETECTED = "flag_typo_detected"
@@ -1275,6 +1279,7 @@
     EVENT_TYPE_SESSION_PINNED,
     EVENT_TYPE_SESSION_UNPINNED,
     EVENT_TYPE_POSTURE_CHANGED,
+    EVENT_TYPE_OPERATION_TIMELINE_ROW,  # PRD §42 Slice 2
     EVENT_TYPE_FLAG_TYPO_DETECTED,
     EVENT_TYPE_FLAG_REGISTERED,
     EVENT_TYPE_GOVERNOR_THROTTLE_APPLIED,