diff --git a/backend/core/ouroboros/battle_test/harness.py b/backend/core/ouroboros/battle_test/harness.py index d0738a13a1..04d3f079e9 100644 --- a/backend/core/ouroboros/battle_test/harness.py +++ b/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 diff --git a/backend/core/ouroboros/battle_test/serpent_flow.py b/backend/core/ouroboros/battle_test/serpent_flow.py index da6bb4b2f3..1c114ca240 100644 --- a/backend/core/ouroboros/battle_test/serpent_flow.py +++ b/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 `` or + ``/timeline `` 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, diff --git a/backend/core/ouroboros/battle_test/session_recorder.py b/backend/core/ouroboros/battle_test/session_recorder.py index 40d44e97e2..3a8cb98e1b 100644 --- a/backend/core/ouroboros/battle_test/session_recorder.py +++ b/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 # ------------------------------------------------------------------ diff --git a/backend/core/ouroboros/governance/ide_observability.py b/backend/core/ouroboros/governance/ide_observability.py index 6ff198d6ec..97349e8d53 100644 --- a/backend/core/ouroboros/governance/ide_observability.py +++ b/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) # ------------------------------------------------------------------ diff --git a/backend/core/ouroboros/governance/ide_observability_stream.py b/backend/core/ouroboros/governance/ide_observability_stream.py index 6c3f1186c5..343068ec0f 100644 --- a/backend/core/ouroboros/governance/ide_observability_stream.py +++ b/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, diff --git a/backend/core/ouroboros/governance/operation_timeline.py b/backend/core/ouroboros/governance/operation_timeline.py new file mode 100644 index 0000000000..e41846f6db --- /dev/null +++ b/backend/core/ouroboros/governance/operation_timeline.py @@ -0,0 +1,815 @@ +"""OperationTimeline — the causal join read-model (PRD §42, Slice 1). + +The root problem (§42.1): O+V acts while the operator is away. Git stores +*content* but is structurally incapable of storing the *relation* +``signal → op → plan → diff → commit → outcome → checkpoint``. The +``OpsDigestObserver`` protocol already *emits* every piece of that +relation as it happens — it just scatters them into session-local +``summary.json`` instead of one durable causal index. This module is +that missing index, and **nothing more**. + +Slice 1 contract (zero behavior change) +--------------------------------------- + +This slice ships ONLY the read-model substrate. It implements the +existing :class:`backend.core.ouroboros.governance.ops_digest_observer.OpsDigestObserver` +protocol (the three ``on_*`` methods) so it *can* be registered later, +but Slice 1 deliberately does **not** wire it into the single global +observer pointer — that fan-out wiring is Slice 2's "causal join +completion + read surface" scope (§42.8). With +``JARVIS_OPERATION_TIMELINE_ENABLED`` default-FALSE (§33.1 / §42.7), +every observer method is a hard no-op: zero rows, zero disk I/O, zero +behavior change anywhere in the loop. This mirrors the Stage-1.6 +Park-spike "Slice 1 = zero runtime change" precedent exactly. + +Architectural contract (zero duplication — §42.3) +------------------------------------------------- + + * **Composes canonical surfaces only**: + - ``cross_process_jsonl.flock_append_line`` — the single + cross-process JSONL append seam (Vector #10 / v2.82). AST-pinned: + no homegrown ``fcntl`` / raw append / ``json.dump`` substitute. + - The ``OpsDigestObserver`` protocol method *shapes* (consumes + the events the orchestrator/AutoCommitter already emit; never + re-derives them — AST-pinned: no ``git`` subprocess, no + TestRunner in this module). + + * **The timeline owns only the edges.** Every datum is a foreign key + to an authority that already owns it: ``op_id`` → OperationLedger, + ``diff_ref`` → DiffArchive ``d-N`` ring, ``commit_hash`` → + AutoCommitter/git, ``checkpoint_ref`` → WorkspaceCheckpointManager. + The row stores pointers, never copies of those bodies (no + ``diff_text``, no full plan body, no state-machine transitions — + OperationLedger remains the state authority). + + * **In-memory cache is hot read; JSONL is authoritative audit.** Each + observer callback merges its fields into the per-``op_id`` row and + appends a fresh JSONL row (append-only audit — re-applied/reverted + ops add rows, never mutate). The in-memory projection collapses to + latest-write-wins per ``op_id`` for the scrub view. Identical + persistence discipline to the SWE-Bench-Pro ``EvaluationResultStore`` + (Phase D) — reused, not reinvented. + +Authority invariant (§42.6 pins 1–4) +------------------------------------- + +This is a telemetry-only read-model with **zero authority**. It never +writes ``OperationState``, never assigns a risk tier, never imports the +orchestrator / policy_engine / iron_gate / change_engine / +candidate_generator / governed_loop_service / repair_engine. It can +never corrupt the loop because it is structurally incapable of acting +on it. AST pins prove this. + +Fail-closed contract (§7) +------------------------- + +Every public method NEVER raises. The observer protocol is explicitly +best-effort fire-and-forget; a misbehaving observer must not derail +APPLY / VERIFY / commit. Internal failure is swallowed at DEBUG and the +method returns (observer hooks) or returns an empty/zero result +(query/replay). +""" +from __future__ import annotations + +import json +import logging +import os +import threading +from dataclasses import dataclass, replace +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Dict, List, Mapping, Optional, Tuple + +from backend.core.ouroboros.governance.cross_process_jsonl import ( + flock_append_line, +) + +logger = logging.getLogger("Ouroboros.OperationTimeline") + + +# =========================================================================== +# Schema + env vocabulary +# =========================================================================== + +#: Schema version for one timeline row. The Slice-2 causal-join fields +#: (signal_source, urgency, risk_tier, plan_ref, diff_ref, file_paths, +#: checkpoint_ref, parent_op_id, terminal_state, reverted_by) are present +#: NOW as Optional/None so Slice 2 *populates* them without a schema bump +#: — forward-compatible by construction (§42.4). +TIMELINE_SCHEMA_VERSION: str = "timeline.1" + +#: Monotonic operator-facing ref handle prefix. Free prefix per the §42 +#: ref-table (t-/d-/o-/n-/p-/q-/b- are taken). Never reused; the counter +#: is never reset within a process lifetime. +REF_PREFIX: str = "r-" + +OPERATION_TIMELINE_ENABLED_ENV_VAR: str = "JARVIS_OPERATION_TIMELINE_ENABLED" +OPERATION_TIMELINE_PATH_ENV_VAR: str = "JARVIS_OPERATION_TIMELINE_PATH" +OPERATION_TIMELINE_MAX_ROWS_ENV_VAR: str = "JARVIS_OPERATION_TIMELINE_MAX_ROWS" + +_DEFAULT_TIMELINE_PATH: str = ".jarvis/operation_timeline.jsonl" +_DEFAULT_MAX_ROWS: int = 5000 +_MAX_ROWS_FLOOR: int = 1 +_MAX_ROWS_CEIL: int = 1_000_000 + + +# =========================================================================== +# Frozen TimelineRow (§33.5 symmetric to_dict/from_dict) +# =========================================================================== + + +@dataclass(frozen=True) +class TimelineRow: + """One causal row: the join the system currently lacks. + + Slice 1 populates the milestone fields from the three + ``OpsDigestObserver`` callbacks. The causal-join fields default to + ``None`` and are filled by Slice 2's read-only joins over the + IntentEnvelope / DiffArchive / WorkspaceCheckpointManager — without + a schema bump (forward-compatible). + """ + + # -- provenance (timeline-owned) ----------------------------------- + schema_version: str + ref: str # monotonic r-N handle; stable per op_id + op_id: str # FK → OperationLedger (authority) + first_seen_iso: str # ISO-8601 UTC of the first callback + updated_iso: str # ISO-8601 UTC of the latest merge + monotonic_at: float # intra-session ordering stability + + # -- milestone fields (Slice 1 — from OpsDigestObserver callbacks) - + apply_mode: Optional[str] = None # none|single|multi + apply_files: Optional[int] = None # count of files APPLY touched + verify_passed: Optional[int] = None + verify_total: Optional[int] = None + verify_scoped_to_op: Optional[bool] = None + commit_hash: Optional[str] = None # THE missing link + + # -- causal-join fields (Slice 2 — present now, populated later) --- + session_id: Optional[str] = None + parent_op_id: Optional[str] = None + signal_source: Optional[str] = None + urgency: Optional[str] = None + risk_tier: Optional[str] = None # copied string ONLY (pin 4) + plan_ref: Optional[Mapping[str, Any]] = None # {hash, summary} + diff_ref: Optional[str] = None # FK → DiffArchive d-N + file_paths: Tuple[str, ...] = () # blast-radius join key + checkpoint_ref: Optional[str] = None # FK → WorkspaceCheckpointMgr + terminal_state: Optional[str] = None # denormalized ledger pointer + reverted_by: Optional[str] = None # back-edge filled by /revert + + def to_dict(self) -> Dict[str, Any]: + return { + "schema_version": self.schema_version, + "ref": self.ref, + "op_id": self.op_id, + "first_seen_iso": self.first_seen_iso, + "updated_iso": self.updated_iso, + "monotonic_at": self.monotonic_at, + "apply_mode": self.apply_mode, + "apply_files": self.apply_files, + "verify_passed": self.verify_passed, + "verify_total": self.verify_total, + "verify_scoped_to_op": self.verify_scoped_to_op, + "commit_hash": self.commit_hash, + "session_id": self.session_id, + "parent_op_id": self.parent_op_id, + "signal_source": self.signal_source, + "urgency": self.urgency, + "risk_tier": self.risk_tier, + "plan_ref": dict(self.plan_ref) if self.plan_ref else None, + "diff_ref": self.diff_ref, + "file_paths": list(self.file_paths), + "checkpoint_ref": self.checkpoint_ref, + "terminal_state": self.terminal_state, + "reverted_by": self.reverted_by, + } + + @classmethod + def from_dict(cls, payload: Mapping[str, Any]) -> "TimelineRow": + plan_ref = payload.get("plan_ref") + raw_paths = payload.get("file_paths") or () + return cls( + schema_version=str( + payload.get("schema_version", TIMELINE_SCHEMA_VERSION) + ), + ref=str(payload["ref"]), + op_id=str(payload["op_id"]), + first_seen_iso=str(payload.get("first_seen_iso", "")), + updated_iso=str(payload.get("updated_iso", "")), + monotonic_at=float(payload.get("monotonic_at", 0.0) or 0.0), + apply_mode=payload.get("apply_mode"), + apply_files=payload.get("apply_files"), + verify_passed=payload.get("verify_passed"), + verify_total=payload.get("verify_total"), + verify_scoped_to_op=payload.get("verify_scoped_to_op"), + commit_hash=payload.get("commit_hash"), + session_id=payload.get("session_id"), + parent_op_id=payload.get("parent_op_id"), + signal_source=payload.get("signal_source"), + urgency=payload.get("urgency"), + risk_tier=payload.get("risk_tier"), + plan_ref=dict(plan_ref) if isinstance(plan_ref, Mapping) else None, + diff_ref=payload.get("diff_ref"), + file_paths=tuple(str(p) for p in raw_paths), + checkpoint_ref=payload.get("checkpoint_ref"), + terminal_state=payload.get("terminal_state"), + reverted_by=payload.get("reverted_by"), + ) + + +# =========================================================================== +# Env loaders (NEVER raise) +# =========================================================================== + + +def _timeline_enabled() -> bool: + """Master flag query (§33.1 default-FALSE). OFF ⇒ every observer + method is a hard no-op (the zero-behavior-change guarantee).""" + raw = os.environ.get( + OPERATION_TIMELINE_ENABLED_ENV_VAR, "", + ).strip().lower() + return raw in ("true", "1", "yes", "on") + + +def _resolve_timeline_path(explicit: Optional[Path]) -> Path: + """Resolve the durable causal-index path. Precedence: explicit + argument > env var > default. NEVER raises.""" + if explicit is not None: + return Path(explicit) + raw = os.environ.get(OPERATION_TIMELINE_PATH_ENV_VAR, "").strip() + if raw: + return Path(raw) + return Path(_DEFAULT_TIMELINE_PATH) + + +def _resolve_max_rows() -> int: + """Bounded tail-scan cap, read at call time (monkeypatchable in + tests — identical discipline to the SWE-Bench-Pro + ``_LOCAL_JSONL_MAX_ROWS`` precedent). Invalid ⇒ default. Clamped to + a sane range. NEVER raises.""" + raw = os.environ.get(OPERATION_TIMELINE_MAX_ROWS_ENV_VAR, "").strip() + if not raw: + return _DEFAULT_MAX_ROWS + try: + value = int(raw) + except (TypeError, ValueError): + return _DEFAULT_MAX_ROWS + if value < _MAX_ROWS_FLOOR: + return _MAX_ROWS_FLOOR + if value > _MAX_ROWS_CEIL: + return _MAX_ROWS_CEIL + return value + + +def _now_iso() -> str: + return datetime.now(tz=timezone.utc).isoformat() + + +def _monotonic() -> float: + # Local import-free monotonic; time is stdlib and authority-free. + import time + + return time.monotonic() + + +# =========================================================================== +# OperationTimeline — the read-model +# =========================================================================== + + +class OperationTimeline: + """Causal join read-model implementing the OpsDigestObserver + protocol shape. + + The three observer callbacks arrive at different times for the same + ``op_id`` (apply → verify → commit, any of which may be absent). Each + callback *merges* its fields into the per-``op_id`` row, keeping the + row's stable ``r-N`` ref, and appends a fresh JSONL audit row. The + in-memory projection is latest-write-wins per ``op_id``. + + Parameters + ---------- + persistence_path: + Optional override for the JSONL path. ``None`` ⇒ env > default. + enabled: + Optional override for the master flag. ``None`` ⇒ env. Allows + test-time injection without env juggling. When effectively + False, every observer method is a hard no-op. + """ + + def __init__( + self, + *, + persistence_path: Optional[Path] = None, + enabled: Optional[bool] = None, + ) -> None: + self._path: Path = _resolve_timeline_path(persistence_path) + self._enabled_override: Optional[bool] = enabled + self._rows: Dict[str, TimelineRow] = {} + self._seq: int = 0 + self._lock = threading.Lock() + + # -- introspection -------------------------------------------------- + + def __len__(self) -> int: + with self._lock: + return len(self._rows) + + @property + def persistence_path(self) -> Path: + return self._path + + def is_enabled(self) -> bool: + if self._enabled_override is not None: + return bool(self._enabled_override) + return _timeline_enabled() + + def clear(self) -> None: + """Drop the in-memory projection. Does NOT touch the JSONL on + disk — the audit history is intentionally append-only and + survives in-memory resets (test teardown / daemon restart).""" + with self._lock: + self._rows.clear() + + # -- OpsDigestObserver protocol shape ------------------------------ + + def on_apply_succeeded( + self, *, op_id: str, mode: str, files: int, + ) -> None: + """An APPLY phase concluded. Merge mode/files into the row. + Hard no-op when the master flag is off. NEVER raises.""" + self._merge( + op_id, + {"apply_mode": str(mode), "apply_files": int(files)}, + ) + + def on_verify_completed( + self, + *, + op_id: str, + passed: int, + total: int, + scoped_to_applied_op: bool = True, + ) -> None: + """A VERIFY phase finished. Merge test counts. Hard no-op when + the master flag is off. NEVER raises.""" + self._merge( + op_id, + { + "verify_passed": int(passed), + "verify_total": int(total), + "verify_scoped_to_op": bool(scoped_to_applied_op), + }, + ) + + def on_commit_succeeded( + self, *, op_id: str, commit_hash: str, + ) -> None: + """AutoCommitter published a commit — THE missing link. Merge + the hash. Hard no-op when the master flag is off. NEVER + raises.""" + self._merge(op_id, {"commit_hash": str(commit_hash)}) + + def on_op_classified( + self, + *, + op_id: str, + signal_source: str, + urgency: str, + risk_tier: str, + ) -> None: + """The op's causal origin (signal → op edge) became known at + the INTENT seam. Merge signal_source / urgency / risk_tier — + the one edge the apply/verify/commit callbacks structurally + cannot carry. Flows through the canonical OpsDigestObserver + seam (PRD §42.3 — no parallel op_id→envelope registry). Hard + no-op when the master flag is off. NEVER raises.""" + self._merge( + op_id, + { + "signal_source": str(signal_source) or None, + "urgency": str(urgency) or None, + "risk_tier": str(risk_tier) or None, + }, + ) + + # -- internal merge + append --------------------------------------- + + def _merge(self, op_id: str, fields: Mapping[str, Any]) -> None: + """Upsert the per-op row with ``fields`` and append a fresh + JSONL audit row. Fail-closed: the master-flag gate is the FIRST + executable statement (the zero-behavior-change guarantee); + nothing below it runs when disabled. NEVER raises.""" + try: + if not self.is_enabled(): + return + if not op_id: + return + now_iso = _now_iso() + with self._lock: + existing = self._rows.get(op_id) + if existing is None: + self._seq += 1 + row = TimelineRow( + schema_version=TIMELINE_SCHEMA_VERSION, + ref=f"{REF_PREFIX}{self._seq}", + op_id=op_id, + first_seen_iso=now_iso, + updated_iso=now_iso, + monotonic_at=_monotonic(), + **dict(fields), + ) + else: + # Stable ref + first_seen; merge new fields; bump + # updated_iso. replace() keeps the frozen dataclass + # immutable while producing the merged successor. + row = replace( + existing, + updated_iso=now_iso, + **dict(fields), + ) + self._rows[op_id] = row + # Read-only causal join: pull the edges that ARE op_id-keyed + # in a canonical singleton-reachable substrate (DiffArchive). + # Lazy + best-effort + only fills still-empty fields — never + # overwrites a value an explicit callback already set, never + # owns or duplicates DiffArchive state. + row = self._join_diff_archive(op_id, row) + self._append_jsonl(row) + except Exception: # noqa: BLE001 — observer is fail-closed + logger.debug( + "[OperationTimeline] _merge swallowed", exc_info=True, + ) + + def _join_diff_archive( + self, op_id: str, row: TimelineRow, + ) -> TimelineRow: + """Fill diff_ref / file_paths / risk_tier from the canonical + DiffArchive (op_id-keyed, singleton-reachable, authority-free). + Adaptive: only fills fields still unset — an explicit + on_op_classified risk_tier wins over the diff's copy. NEVER + raises; returns the row unchanged on any failure.""" + try: + from backend.core.ouroboros.battle_test.diff_archive import ( + get_default_archive, + ) + + matches = get_default_archive().find_by_op_id(op_id) + if not matches: + return row + newest = matches[-1] # find_by_op_id is oldest → newest + patch: Dict[str, Any] = {} + new_ref = getattr(newest, "ref", None) + if new_ref and row.diff_ref != new_ref: + patch["diff_ref"] = str(new_ref) + paths = tuple(getattr(newest, "file_paths", ()) or ()) + if paths and not row.file_paths: + patch["file_paths"] = paths + if not row.risk_tier: + rt = getattr(newest, "risk_tier", None) + if rt: + patch["risk_tier"] = str(rt) + if not patch: + return row + merged = replace(row, **patch) + with self._lock: + self._rows[op_id] = merged + return merged + except Exception: # noqa: BLE001 + logger.debug( + "[OperationTimeline] _join_diff_archive swallowed", + exc_info=True, + ) + return row + + def _append_jsonl(self, row: TimelineRow) -> None: + """Append one row to the durable causal index via the canonical + cross-process flock primitive. Sync by design: the + OpsDigestObserver protocol is sync fire-and-forget and the + flock scope is open-write-flush-close (microseconds — see + cross_process_jsonl module docstring). NEVER raises.""" + try: + line = json.dumps( + row.to_dict(), sort_keys=True, default=str, + ) + ok = flock_append_line(self._path, line) + if not ok: + logger.debug( + "[OperationTimeline] flock append returned False " + "for %s", row.ref, + ) + self._publish_sse(row) + except Exception: # noqa: BLE001 — belt-and-suspenders + logger.debug( + "[OperationTimeline] _append_jsonl swallowed", + exc_info=True, + ) + + def _publish_sse(self, row: TimelineRow) -> None: + """Best-effort SSE notification on the existing StreamEventBroker + (PRD §42 Slice 2 read surface). Composes the canonical + publish_task_event hook — which itself gates on stream_enabled() + and never raises. Lazy import keeps the timeline decoupled from + the stream module when observability is off. NEVER raises.""" + try: + from backend.core.ouroboros.governance.ide_observability_stream import ( # noqa: E501 + EVENT_TYPE_OPERATION_TIMELINE_ROW, + publish_task_event, + ) + + publish_task_event( + EVENT_TYPE_OPERATION_TIMELINE_ROW, + row.op_id, + { + "ref": row.ref, + "op_id": row.op_id, + "signal_source": row.signal_source, + "urgency": row.urgency, + "risk_tier": row.risk_tier, + "apply_mode": row.apply_mode, + "verify_passed": row.verify_passed, + "verify_total": row.verify_total, + "commit_hash": row.commit_hash, + "diff_ref": row.diff_ref, + "updated_iso": row.updated_iso, + }, + ) + except Exception: # noqa: BLE001 — read surface is best-effort + logger.debug( + "[OperationTimeline] _publish_sse swallowed", + exc_info=True, + ) + + def list_recent(self, *, limit: int = 5) -> Tuple[TimelineRow, ...]: + """Most-recent rows, newest first — the /expand summary + + /timeline REPL accessor. Thin bounded wrapper over + :meth:`query`. NEVER raises.""" + try: + return self.query(limit=max(1, int(limit))) + except Exception: # noqa: BLE001 + return () + + # -- query ---------------------------------------------------------- + + def query( + self, + *, + op_id: Optional[str] = None, + terminal_state: Optional[str] = None, + has_commit: Optional[bool] = None, + limit: Optional[int] = None, + ) -> Tuple[TimelineRow, ...]: + """Bounded snapshot read of the in-memory projection, newest + first (descending ``monotonic_at``, ties broken by ref). + NEVER raises — returns ``()`` on internal failure.""" + try: + with self._lock: + snapshot: List[TimelineRow] = list(self._rows.values()) + snapshot.sort( + key=lambda r: (r.monotonic_at, r.ref), reverse=True, + ) + out: List[TimelineRow] = [] + for r in snapshot: + if op_id is not None and r.op_id != op_id: + continue + if terminal_state is not None and ( + r.terminal_state != terminal_state + ): + continue + if has_commit is True and not r.commit_hash: + continue + if has_commit is False and r.commit_hash: + continue + out.append(r) + if limit is not None and len(out) >= limit: + break + return tuple(out) + except Exception: # noqa: BLE001 + logger.debug( + "[OperationTimeline] query swallowed", exc_info=True, + ) + return () + + def lookup(self, ref: str) -> Optional[TimelineRow]: + """Reverse-lookup a row by its ``r-N`` ref. NEVER raises.""" + try: + with self._lock: + for r in self._rows.values(): + if r.ref == ref: + return r + return None + except Exception: # noqa: BLE001 + return None + + # -- disk replay (cross-session morning-after substrate) ----------- + + def replay_from_disk(self) -> int: + """Reconstruct the in-memory projection from the JSONL audit + file. Returns the count of rows replayed. Bounded by + :func:`_resolve_max_rows`. Idempotent: the per-``op_id`` dedup + collapses duplicates so the last-written row wins. Malformed + rows are skipped at DEBUG. NEVER raises. + + This is the substrate the cross-session scrub (§42.9 criterion + 2) reads on a fresh boot — the capability Claude Code + structurally cannot have. + """ + try: + path = self._path + if not path.exists(): + return 0 + try: + text = path.read_text(encoding="utf-8") + except OSError: + return 0 + + lines = [ln for ln in text.splitlines() if ln.strip()] + max_rows = _resolve_max_rows() + # Bounded: only the most recent ``max_rows`` lines matter + # for the projection (latest-write-wins collapses earlier + # duplicates anyway). Tail-scan, not full-file load. + if len(lines) > max_rows: + lines = lines[-max_rows:] + + count = 0 + max_seq = self._seq + for raw in lines: + try: + payload = json.loads(raw) + row = TimelineRow.from_dict(payload) + except ( + json.JSONDecodeError, KeyError, ValueError, TypeError, + ): + logger.debug( + "[OperationTimeline] skipped malformed row", + ) + continue + with self._lock: + self._rows[row.op_id] = row + count += 1 + # Keep the monotonic counter ahead of any replayed ref + # so newly-created rows never collide with replayed + # ones (refs are never reused). + num = _ref_number(row.ref) + if num is not None and num > max_seq: + max_seq = num + with self._lock: + if max_seq > self._seq: + self._seq = max_seq + return count + except Exception: # noqa: BLE001 + logger.debug( + "[OperationTimeline] replay_from_disk swallowed", + exc_info=True, + ) + return 0 + + +def _ref_number(ref: str) -> Optional[int]: + """Parse the integer N out of an ``r-N`` ref. None if malformed.""" + try: + if not ref or not ref.startswith(REF_PREFIX): + return None + return int(ref[len(REF_PREFIX):]) + except (TypeError, ValueError): + return None + + +# =========================================================================== +# Module-level singleton (mirrors get_default_store / get_default_broker) +# =========================================================================== + + +_DEFAULT_TIMELINE_LOCK = threading.Lock() +_DEFAULT_TIMELINE: Optional[OperationTimeline] = None + + +def get_default_timeline() -> OperationTimeline: + """Return the process-global default timeline, constructing it on + first call. Thread-safe; idempotent.""" + global _DEFAULT_TIMELINE + with _DEFAULT_TIMELINE_LOCK: + if _DEFAULT_TIMELINE is None: + _DEFAULT_TIMELINE = OperationTimeline() + return _DEFAULT_TIMELINE + + +def reset_default_timeline() -> None: + """Drop the singleton instance. Primarily for tests. NEVER raises.""" + global _DEFAULT_TIMELINE + with _DEFAULT_TIMELINE_LOCK: + _DEFAULT_TIMELINE = None + + +# =========================================================================== +# FlagRegistry self-registration (auto-discovered by §33.3 walker — +# the top-level governance package is already in _FLAG_PROVIDER_PACKAGES, +# so zero edits to flag_registry_seed.py are required) +# =========================================================================== + + +def register_flags(registry: Any) -> int: + """Module-owned FlagRegistry registration. Returns count + successfully registered. NEVER raises.""" + try: + from backend.core.ouroboros.governance.flag_registry import ( + Category, + FlagSpec, + FlagType, + ) + except ImportError: + return 0 + + source_file = ( + "backend/core/ouroboros/governance/operation_timeline.py" + ) + since = "PRD §42 Slice 1 (2026-05-16)" + + specs = [ + FlagSpec( + name=OPERATION_TIMELINE_ENABLED_ENV_VAR, + type=FlagType.BOOL, + default=False, + description=( + "PRD §42 master switch (§33.1 default-FALSE): when ON, " + "the OperationTimeline read-model records every " + "OpsDigestObserver milestone (apply/verify/commit) as " + "one durable causal row at " + "JARVIS_OPERATION_TIMELINE_PATH via the canonical " + "cross_process_jsonl.flock_append_line primitive. OFF " + "⇒ every observer callback is a hard no-op (zero rows, " + "zero disk I/O, zero behavior change). The read-model " + "has zero authority over the loop." + ), + category=Category.OBSERVABILITY, + source_file=source_file, + example="false", + since=since, + ), + FlagSpec( + name=OPERATION_TIMELINE_PATH_ENV_VAR, + type=FlagType.STR, + default=_DEFAULT_TIMELINE_PATH, + description=( + "Durable causal-index JSONL path for the PRD §42 " + "Operation Timeline. Parent directory auto-created on " + "first append. Appended via cross_process_jsonl." + "flock_append_line — safe across concurrent battle-test " + "processes. Append-only audit (re-applied/reverted ops " + "add rows, never mutate); the in-memory projection " + f"collapses to latest-write-wins per op_id. Default " + f"{_DEFAULT_TIMELINE_PATH!r}." + ), + category=Category.OBSERVABILITY, + source_file=source_file, + example=_DEFAULT_TIMELINE_PATH, + since=since, + ), + FlagSpec( + name=OPERATION_TIMELINE_MAX_ROWS_ENV_VAR, + type=FlagType.INT, + default=_DEFAULT_MAX_ROWS, + description=( + "Bounded tail-scan cap for OperationTimeline." + "replay_from_disk: only the most recent N JSONL lines " + "are loaded into the in-memory projection on a fresh " + "boot (latest-write-wins collapses earlier duplicates " + "anyway). Read at call time so tests can monkeypatch " + "it — identical discipline to the SWE-Bench-Pro " + f"_LOCAL_JSONL_MAX_ROWS precedent. Default " + f"{_DEFAULT_MAX_ROWS}; clamped to " + f"[{_MAX_ROWS_FLOOR}, {_MAX_ROWS_CEIL}]." + ), + category=Category.CAPACITY, + source_file=source_file, + example=str(_DEFAULT_MAX_ROWS), + since=since, + ), + ] + + count = 0 + for spec in specs: + try: + registry.register(spec) + count += 1 + except Exception: # noqa: BLE001 + logger.debug( + "[OperationTimeline] flag registration failed for %s", + getattr(spec, "name", "?"), exc_info=True, + ) + return count + + +__all__ = [ + "OPERATION_TIMELINE_ENABLED_ENV_VAR", + "OPERATION_TIMELINE_MAX_ROWS_ENV_VAR", + "OPERATION_TIMELINE_PATH_ENV_VAR", + "OperationTimeline", + "REF_PREFIX", + "TIMELINE_SCHEMA_VERSION", + "TimelineRow", + "get_default_timeline", + "register_flags", + "reset_default_timeline", +] diff --git a/backend/core/ouroboros/governance/ops_digest_observer.py b/backend/core/ouroboros/governance/ops_digest_observer.py index 12b8058e81..91cd1cc89e 100644 --- a/backend/core/ouroboros/governance/ops_digest_observer.py +++ b/backend/core/ouroboros/governance/ops_digest_observer.py @@ -104,6 +104,25 @@ def on_commit_succeeded( ) -> None: """AutoCommitter published a commit for ``op_id``. Hash may be shortened.""" + def on_op_classified( + self, + *, + op_id: str, + signal_source: str, + urgency: str, + risk_tier: str, + ) -> None: + """The op's causal origin became known at the INTENT seam. + + Additive (PRD §42 Slice 2): carries the one causal edge the + other three callbacks structurally cannot — *signal → op*. It + flows through THIS canonical telemetry seam (not a parallel + op_id→envelope registry) so the OperationTimeline read-model + can complete the causal join without any new authority or + duplicated state. Default-noop in legacy implementers — only + the timeline consumes it; SessionRecorder ignores it. + """ + class _NoopObserver: """Default registered observer — silently drops every call. @@ -130,6 +149,117 @@ def on_verify_completed( def on_commit_succeeded(self, *, op_id: str, commit_hash: str) -> None: return + def on_op_classified( + self, + *, + op_id: str, + signal_source: str, + urgency: str, + risk_tier: str, + ) -> None: + return + + +class _CompositeOpsDigestObserver: + """Fan-out observer (PRD §42 Slice 2 — the root fix for coexistence). + + The module-global pointer is single-slot: a naive + ``register_ops_digest_observer(timeline)`` would EVICT the harness's + ``SessionRecorder`` and silently break ``LastSessionSummary``. The + correct, non-workaround composition is a multiplexing observer that + forwards every protocol method to an ordered list of members, each + call defensively isolated so a misbehaving member cannot starve the + others. This composes the existing single seam — it does NOT add a + parallel registry. ``register_/get_/reset_`` semantics are + untouched; :func:`add_ops_digest_observer` transparently wraps the + current observer into one of these. + + Members are deduplicated by identity (idempotent ``add``). Every + forwarded call is wrapped in try/except per the protocol's + fail-closed contract. + """ + + def __init__(self) -> None: + self._members: list = [] + self._members_lock = threading.Lock() + + # -- membership ---------------------------------------------------- + + def add(self, observer: object) -> None: + if observer is None: + return + with self._members_lock: + if any(m is observer for m in self._members): + return # idempotent — never double-register by identity + self._members.append(observer) + + def remove(self, observer: object) -> None: + with self._members_lock: + self._members = [m for m in self._members if m is not observer] + + def members(self) -> tuple: + with self._members_lock: + return tuple(self._members) + + def __len__(self) -> int: + with self._members_lock: + return len(self._members) + + # -- fan-out (each member isolated) -------------------------------- + + def _fan_out(self, method_name: str, **kwargs: object) -> None: + for member in self.members(): + try: + getattr(member, method_name)(**kwargs) + except Exception: # noqa: BLE001 — one member must not + # starve the others; protocol is fire-and-forget. + logger.debug( + "[OpsDigest] composite member %r raised on %s", + type(member).__name__, method_name, exc_info=True, + ) + + def on_apply_succeeded(self, *, op_id: str, mode: str, files: int) -> None: + self._fan_out( + "on_apply_succeeded", op_id=op_id, mode=mode, files=files, + ) + + def on_verify_completed( + self, + *, + op_id: str, + passed: int, + total: int, + scoped_to_applied_op: bool = True, + ) -> None: + self._fan_out( + "on_verify_completed", + op_id=op_id, + passed=passed, + total=total, + scoped_to_applied_op=scoped_to_applied_op, + ) + + def on_commit_succeeded(self, *, op_id: str, commit_hash: str) -> None: + self._fan_out( + "on_commit_succeeded", op_id=op_id, commit_hash=commit_hash, + ) + + def on_op_classified( + self, + *, + op_id: str, + signal_source: str, + urgency: str, + risk_tier: str, + ) -> None: + self._fan_out( + "on_op_classified", + op_id=op_id, + signal_source=signal_source, + urgency=urgency, + risk_tier=risk_tier, + ) + _OBSERVER_LOCK = threading.Lock() _OBSERVER: OpsDigestObserver = _NoopObserver() @@ -157,3 +287,66 @@ def get_ops_digest_observer() -> OpsDigestObserver: def reset_ops_digest_observer() -> None: """Restore the default no-op observer. Primarily for tests.""" register_ops_digest_observer(None) + + +def add_ops_digest_observer(observer: Optional[OpsDigestObserver]) -> None: + """Add ``observer`` to the fan-out set (PRD §42 Slice 2). + + Unlike :func:`register_ops_digest_observer` (single-slot SET — kept + byte-identical for back-compat: the harness still registers + SessionRecorder that way), this COMPOSES: the current observer is + transparently wrapped into a :class:`_CompositeOpsDigestObserver` + so multiple consumers (SessionRecorder + the OperationTimeline + read-model) coexist. Idempotent by identity. NEVER raises — a + telemetry-wiring failure must never derail boot. + + Order of operations is robust to any boot sequence: calling + ``add`` before or after ``register`` both converge on a composite + containing every distinct member; a plain ``_NoopObserver`` is + discarded (it carries no state) rather than fanned to. + """ + if observer is None: + return + try: + with _OBSERVER_LOCK: + global _OBSERVER + current = _OBSERVER + if isinstance(current, _CompositeOpsDigestObserver): + current.add(observer) + return + composite = _CompositeOpsDigestObserver() + # Preserve any real existing observer; a bare no-op carries + # no state and is intentionally dropped from the fan-out. + if not isinstance(current, _NoopObserver): + composite.add(current) + composite.add(observer) + _OBSERVER = composite + except Exception: # noqa: BLE001 — wiring must never crash boot + logger.debug( + "[OpsDigest] add_ops_digest_observer failed", exc_info=True, + ) + + +def remove_ops_digest_observer(observer: Optional[OpsDigestObserver]) -> None: + """Remove ``observer`` from the fan-out set. If the composite + empties, the default :class:`_NoopObserver` is restored. NEVER + raises. Primarily for tests / clean teardown.""" + if observer is None: + return + try: + with _OBSERVER_LOCK: + global _OBSERVER + current = _OBSERVER + if not isinstance(current, _CompositeOpsDigestObserver): + return + current.remove(observer) + if len(current) == 0: + _OBSERVER = _NoopObserver() + elif len(current) == 1: + # Collapse a single-member composite back to the bare + # observer — keeps the common path allocation-free. + _OBSERVER = current.members()[0] + except Exception: # noqa: BLE001 + logger.debug( + "[OpsDigest] remove_ops_digest_observer failed", exc_info=True, + ) diff --git a/backend/core/ouroboros/governance/orchestrator.py b/backend/core/ouroboros/governance/orchestrator.py index 748643f397..b84c6f3412 100644 --- a/backend/core/ouroboros/governance/orchestrator.py +++ b/backend/core/ouroboros/governance/orchestrator.py @@ -2559,6 +2559,28 @@ async def _run_pipeline(self, ctx: OperationContext) -> OperationContext: except Exception: logger.debug("emit_intent failed for op=%s", ctx.op_id, exc_info=True) + # PRD §42 Slice 2 — emit the causal signal→op edge through + # the canonical OpsDigestObserver seam (NOT a parallel + # op_id→envelope registry). This is the single place where + # op_id + signal_source + urgency + risk_tier are all known + # and INTENT telemetry already fires. Same fail-closed + # shape as the on_apply/verify/commit emits. + try: + from backend.core.ouroboros.governance.ops_digest_observer import ( + get_ops_digest_observer, + ) + get_ops_digest_observer().on_op_classified( + op_id=ctx.op_id, + signal_source=getattr(ctx, "signal_source", "") or "", + urgency=getattr(ctx, "signal_urgency", "") or "", + risk_tier=risk_tier.name, + ) + except Exception: + logger.debug( + "[Orchestrator] on_op_classified observer call failed", + exc_info=True, + ) + # ---- Reasoning chain classification (optional, pre-routing) ---- reasoning_result = None if self._reasoning_bridge and self._reasoning_bridge.is_active: diff --git a/docs/architecture/OUROBOROS_VENOM_PRD.md b/docs/architecture/OUROBOROS_VENOM_PRD.md index b70ec63739..641f2fa30c 100644 --- a/docs/architecture/OUROBOROS_VENOM_PRD.md +++ b/docs/architecture/OUROBOROS_VENOM_PRD.md @@ -9543,3 +9543,187 @@ The recommended sequence (§39.5 Tier 1-5) deliberately defers the audio + cross | 2026-04-26 | 2.12 | **Phase 4 P4 GRADUATED — Convergence Metrics Suite live by default. Phase 4 ENTIRELY CLOSED.** 5-slice arc landed (Slice 1 `MetricsEngine` 7-metric un-stranding wrapper → Slice 2 `MetricsHistoryLedger` JSONL persistence + 7d/30d aggregator → Slice 3 `/metrics` REPL with ASCII sparkline → Slice 4 `MetricsSessionObserver` + 4 IDE GET endpoints + SSE `metrics_updated` event → Slice 5 graduation). `JARVIS_METRICS_SUITE_ENABLED` default flipped `false`→`true` in **three owner modules** (`metrics_engine.py` + `metrics_repl_dispatcher.py` + `metrics_observability.py`). `register_metrics_routes` wired into `EventChannelServer.start` (loopback-asserted, gated on master flag, per-instance rate-limit + shared CORS allowlist via dedicated `IDEObservabilityRouter` helper). Pre-graduation pins renamed in all three owner suites per their embedded discipline. Layered evidence: 204 deterministic Slice 1-4 tests + 38 graduation pins (master flag default-true × 3 owner modules + source-grep `"1"` literal × 3 + pre-graduation pin renames × 3 owner suites + EventChannelServer source-grep × 3 (`register_metrics_routes` import + `_metrics_enabled()` gate + `_assert_loopback_metrics`) + cross-slice authority survival × 4 modules + reachability supplement) + 15 in-process live-fire smoke checks (observer end-to-end with master-on default, all 4 GET endpoints reachable + return correct shape, all 3 REPL commands render, master-off revert proven). Authority invariants survived through all 5 slices: pure-data engine (S1) + ledger-only I/O (S2) + delegating-only REPL (S3) + summary.json + delegated-ledger I/O (S4) + EventChannel-block-only addition (S5). The `INSUFFICIENT_DATA` problem statement that motivated this phase is resolved — operators can now answer "is O+V getting smarter?" with concrete data via `/metrics 7d` REPL or `GET /observability/metrics/window?days=7`. Hot-revert: single env knob (`JARVIS_METRICS_SUITE_ENABLED=false`) → observer short-circuits, GET endpoints return 403, SSE drops silently; ledger remains readable for prior-session recall. **Phase 4 — Cognitive Metrics FULLY GRADUATED 2026-04-26.** Both items closed (P3 + P4). Phases 0-4 all complete. Next per Forward-Looking Priority Roadmap: Phase 5 P5 (AdversarialReviewer subagent). | Claude Opus 4.7 (P4 Slice 5 graduation PR) | | 2026-04-26 | 2.11 | **Phase 3 P2 GRADUATED — Conversational mode live by default. Phase 3 ENTIRELY CLOSED.** 4-slice arc landed (Slice 1 IntentClassifier primitive → Slice 2 ConversationOrchestrator + ChatSession → Slice 3 /chat REPL dispatcher + ChatActionExecutor Protocol → Slice 4 graduation). `JARVIS_CONVERSATIONAL_MODE_ENABLED` default flipped `false`→`true`. `build_chat_repl_dispatcher()` factory in `chat_repl_dispatcher.py` is the single SerpentFlow integration point: returns a wired dispatcher (with safe-default `LoggingChatActionExecutor`) when on, `None` when reverted so SerpentFlow can skip surfacing `/chat` entirely. Pre-graduation pins renamed in BOTH env-knob owner suites (intent_classifier + chat_repl_dispatcher) per their embedded discipline. Layered evidence: 171 deterministic Slice 1-3 tests + 45 graduation pins (master flag default-true × 2 owner modules + source-grep `"1"` literal × 2 + factory branch coverage + LoggingExecutor contract pin + cross-slice authority survival × 4 modules + reachability supplement) + 15 in-process live-fire smoke checks (factory→classifier→orchestrator→dispatcher→executor end-to-end across all 4 ChatActionExecutor branches; bounded-ring under load; hot-revert proven). Authority invariants survived through all 4 slices: pure-data classifier (Slice 1) + IO-free orchestrator (Slice 2) + IO-free dispatcher (Slice 3) + LoggingExecutor never raises (Slice 4). Safety-first contract pinned: noop input never invokes executor; CONTEXT_PASTE without prior turn falls back to query_claude (degraded — never attaches to non-existent target). Concrete executors against backlog ingestion / subagent_scheduler / Claude provider tracked as follow-up slices — wiring those crosses authority boundaries that need their own pin suites. Hot-revert: single env knob (`JARVIS_CONVERSATIONAL_MODE_ENABLED=false`) → factory returns None → `/chat` invisible to operators; orchestrator + bridge state remain inspectable for prior-decision recall. **Phase 3 — Operator Symbiosis FULLY GRADUATED 2026-04-26.** All three items closed (P3.5 + P3 + P2). | Claude Opus 4.7 (P2 Slice 4 graduation PR) | | 2026-04-26 | 2.10 | **Phase 3 P3 GRADUATED — inline approval UX live by default.** 4-slice arc landed (Slice 1 primitive → Slice 2 provider + audit ledger → Slice 3 renderer + 30s prompt + `$EDITOR` → Slice 4 graduation). `JARVIS_APPROVAL_UX_INLINE_ENABLED` default flipped `false`→`true`. `build_approval_provider()` factory in `inline_approval_provider.py` is the single source of truth for `GovernedLoopService`'s approval-provider selection (returns `InlineApprovalProvider` when on, legacy `CLIApprovalProvider` when off). Pre-graduation pin renamed to `test_master_flag_default_true_post_graduation` per its embedded discipline. Layered evidence: 165 deterministic Slice 1-3 tests + 36 graduation pins (master flag + source-grep `"1"` literal + factory branch coverage + GovernedLoopService source-grep + cross-slice authority survival + reachability supplement) + 15 in-process live-fire smoke checks (factory-built provider end-to-end through queue + renderer + audit ledger). Authority invariants survived through all 4 slices: pure-data primitive (Slice 1) + only-audit-ledger I/O (Slice 2) + argv-only subprocess (Slice 3, no `shell=True`). Safety-first contract pinned: EOF / garbage / 30s timeout all `defer-not-approve`. Hot-revert: single env knob (`JARVIS_APPROVAL_UX_INLINE_ENABLED=false`) — factory returns `CLIApprovalProvider` on the next construction. Phase 3 P3 + P3.5 both COMPLETE; Phase 3 P2 (Conversational mode) remains the only open Phase 3 item. | Claude Opus 4.7 (P3 Slice 4 graduation PR) | + +--- + +## §42 OPERATION TIMELINE + CHECKPOINT-REWIND — THE CAUSAL JOIN LAYER (NEW 2026-05-16) + +**Origin.** Operator-commissioned external critique (2026-05-16): *"You've been cloning a reactive tool to power a proactive one."* The audit named five gaps where Claude Code's reactive information architecture is wrong for O+V. This section closes **Gap #1 only** (Checkpoint / Rewind / Time-Travel) — the critique's "single biggest missing piece, and it's cheap because the substrate exists." Gaps #2–#5 are explicitly **out of scope** for this section; §42 is architected so the forensic morning-after surface (Gap #5) can later *compose* the read-model defined here without rework, but no Gap #5 code is authorized by this PRD. + +**Why this section exists.** A read-only structural exploration of `backend/core/ouroboros/` (2026-05-16) confirmed the critique's root-cause claim *precisely*: the substrate to answer "what did O+V do while I was away, and undo any of it" already exists, but it is **fragmented across eight modules with no causal join layer**, plus exactly one genuinely missing durable primitive. This is therefore not a "build a rewind feature" ticket. It is a "the system has no standing causal model of its own actions" ticket. The fix is architectural: a durable causal **Operation Timeline read-model** (authority-free projection) plus a **RewindCoordinator** that composes existing git-native plumbing — no new authority, no history rewriting, no hardcoded rules. + +### §42.1 The Root Problem (Verbatim Diagnosis) + +Claude Code's entire information architecture assumes: *the human just told me what to do, is watching, and will read top-to-bottom*. O+V violates all three — it self-initiates, runs for hours unattended, and the operator arrives **after** things happened. A linear streaming transcript is optimal for reactive work and structurally mediocre for proactive work. + +The operator's first three questions on return are causal, not chronological: + +1. *What did you do while I was at lunch / overnight?* +2. *Where are we now — which of those stuck, which rolled back?* +3. *Undo #4 — and tell me what else #4 affected.* + +Git stores **content** but is structurally incapable of storing the **relation** `signal → op → plan → diff → commit → outcome → checkpoint`. `OpsDigestObserver` already *receives* every piece of that relation as it happens (`on_apply_succeeded`, `on_verify_completed`, `on_commit_succeeded`) — it just **scatters them into session-local `summary.json`** instead of one durable causal index. That missing index is the exact reason the three questions above are currently unanswerable. + +### §42.2 Architecture — Three Layers, All Composed + +``` +Layer A OperationTimeline (read-model) authority-free projection + subscribes existing OpsDigestObserver protocol + + joins existing DiffArchive / WorkspaceCheckpointManager refs + → durable .jarvis/operation_timeline.jsonl (flock_append_line seam) + +Layer B RewindCoordinator (action) operator-initiated ONLY + preview = delegate ReviewBranchManager (proven non-destructive plumbing) + revert-committed = `git revert --no-edit ` (new commit, no rewrite) + revert-in-flight = compose existing pre-apply snapshot restore + preconditions = clean-tree gate + ancestry check + blast-radius + +Layer C Operator surface composed, zero new server + /timeline /rewind /revert (auto-discovered _handle_* verbs) + r-N prefix in the existing unified /expand dispatcher + GET /observability/timeline (extend IDEObservabilityRouter) + timeline_*/rewind_* SSE via the existing StreamEventBroker +``` + +The load-bearing architectural decision: **Layer A is a read-model with zero authority and zero behavior change.** It can never corrupt the loop because it never writes governance state, never assigns risk, never imports the orchestrator. Layer B acts only on an explicit operator verb — it has no autonomous caller anywhere in the import graph. + +### §42.3 Zero-Duplication Mapping (Mandatory PRD Requirement #1) + +The timeline must not duplicate state or authority. The boundary is drawn precisely: + +| Concern | Canonical owner (unchanged) | Timeline's relationship | Anti-duplication mechanism | +|---|---|---|---| +| Op governance state machine (`PLANNED→…→APPLIED/ROLLED_BACK`) | `OperationLedger` (`ledger.py:43`, JSONL per op) | **References** `op_id`; copies the *terminal* state string as a denormalized pointer only | AST pin: timeline module makes **zero** `.record(` calls on any ledger symbol; never writes `OperationState` | +| Apply / verify / commit milestone events | `OpsDigestObserver` protocol (`ops_digest_observer.py:58`) — already a stable observer contract | **Implements & registers against the existing protocol**; receives `on_apply_succeeded` / `on_verify_completed` / `on_commit_succeeded` | AST pin: timeline defines the protocol surface but performs **no detection** (no `git` subprocess, no test invocation in the read-model module) — it consumes events, it does not re-derive them | +| Full unified-diff text + apply/verify outcome | `DiffArchive` (`diff_archive.py:166`, `d-N` ring) | Stores only the `d-N` **ref** (pointer), never copies `diff_text` | Schema stores `diff_ref`, not diff body; `/expand d-N` remains the single diff-render path | +| Pre-apply file snapshots / git-stash checkpoints | `WorkspaceCheckpointManager` + orchestrator rollback (`orchestrator.py:7551`) | Stores only the `checkpoint_ref` (pointer) | RewindCoordinator's in-flight revert **calls the existing restore path** — it does not reimplement snapshot logic | +| Signed commits + O+V signature + intent git-notes | `AutoCommitter` (`auto_committer.py:489`) | Receives `commit_hash` via the **existing** `on_commit_succeeded(op_id, hash)` observer callback | Timeline adds **no** new commit path; the genuinely missing link (durable `op_id ↔ commit_hash`) is captured by *persisting an event the observer already emits* | +| Non-destructive preview-branch git plumbing | `ReviewBranchManager` (`review_branch_manager.py:517`) — proven never to touch worktree/HEAD/index | RewindCoordinator **delegates** preview construction to it | AST pin: RewindCoordinator constructs **no** `write-tree`/`commit-tree` plumbing in-module — must call the canonical seam | +| Cross-process JSONL append durability | `cross_process_jsonl.flock_append_line` (canonical seam, SWE-Bench-Pro arc precedent) | Timeline's **only** disk-write path | AST pin: **no** `import fcntl`, no raw `open().write` append, no `json.dump` to the timeline path | + +**The single net-new primitive** is therefore not a store — it is the *durable persistence of a causal join that the observer protocol already emits but currently discards*. Everything else is a pointer. + +### §42.4 The Causal Index Schema (Mandatory PRD Requirement #3) + +One append-only JSONL at `JARVIS_OPERATION_TIMELINE_PATH` (default `.jarvis/operation_timeline.jsonl`). Append-only on disk (full audit history — re-applied/reverted ops add rows, never mutate); the in-memory projection collapses to **latest-write-wins per `op_id`** for the scrub view. This is the exact persistence discipline already proven by `EvaluationResultStore` (SWE-Bench-Pro Phase D) — reused, not reinvented. + +Each row is one JSON object. `schema_version` is `"timeline.1"`. Fields: + +```jsonc +{ + "schema_version": "timeline.1", + "ref": "r-7", // timeline's own monotonic handle (free prefix; see §42 ref-table). Never reused, drop-oldest on bound. + "op_id": "op-019d8...-testfail", // FK → OperationLedger (authority); join key, never owned here + "session_id": "bt-2026-05-16-...", // FK → .ouroboros/sessions/; enables cross-session morning-after scrub + "parent_op_id": "op-019d7...-coalesced", // causal edge: coalesced/dependent parent (from existing _active_file_ops DAG) or null + "signal_source": "TestFailure", // sensor/origin from the IntentEnvelope (the WHY); never recomputed + "urgency": "immediate", // copied from envelope; pointer not authority + "risk_tier": "notify_apply", // copied string only — AST-pinned: timeline never *assigns* risk + "plan_ref": {"hash":"sha256:9e88...","summary":"add ascii gate retry"}, // digest + 1-line, NOT full plan body (no plan_generator duplication) + "diff_ref": "d-12", // FK → DiffArchive ring; full diff via /expand d-12 only + "file_paths": ["backend/.../iron_gate.py"], // the blast-radius join key (minimal durable copy; DiffArchive ring evicts at 30, timeline persists) + "commit_hash": "17ae95d7d6...", // THE missing link — from on_commit_succeeded(op_id, hash); null if uncommitted/in-flight + "apply_mode": "single", // none|single|multi — from on_apply_succeeded + "verify_passed": 14, // from on_verify_completed + "verify_total": 14, + "checkpoint_ref": "ckpt-3", // FK → WorkspaceCheckpointManager (pre-apply snapshot id) or null + "terminal_state": "APPLIED", // denormalized pointer to ledger terminal (APPLIED|ROLLED_BACK|FAILED|BLOCKED) — copy, never source of truth + "reverted_by": null, // back-edge: if a later /revert created a revert-commit for this op, that op's ref (filled by an amended append row) + "wall_time_iso": "2026-05-16T18:22:04Z", // operator-facing absolute time (morning-after scrub axis) + "monotonic_at": 918273.44 // intra-session ordering stability (ties broken by ref) +} +``` + +**Causal completeness.** The row IS the join the system currently lacks: `signal_source` (why) → `op_id` (what) → `plan_ref` (intended how) → `diff_ref`/`file_paths` (actual change) → `commit_hash` (where in git) → `verify_*`/`terminal_state` (did it hold) → `checkpoint_ref` (how to undo) → `reverted_by` (was it undone). Every FK points at an authority that already owns that data; the timeline owns **only the edges between them**. + +**Bounded by construction.** Load is a bounded tail scan capped at `JARVIS_OPERATION_TIMELINE_MAX_ROWS` (default 5000), read at call time so tests can monkeypatch it — the identical discipline as the SWE-Bench-Pro `_LOCAL_JSONL_MAX_ROWS` precedent. Malformed lines / non-dict records / missing `op_id` are skipped silently (fail-open: a corrupt row must never wedge the scrub). + +### §42.5 Irreversibility Ban (Mandatory PRD Requirement #2) + +**Principle.** No autonomous-or-operator action through this surface may rewrite git history or perform an unrecoverable destructive operation. Undo must itself be undoable. + +| Revert case | Mechanism (composed, never reinvented) | Why it is non-destructive | +|---|---|---| +| Committed op | `git revert --no-edit ` | Creates a **new** commit that inverts the change. Original commit stays in history. Safe on shared branches — matches `AutoCommitter`'s protected-branch discipline. The revert is itself revertable. | +| In-flight / uncommitted op | Compose the **existing** pre-apply snapshot restore (`orchestrator.py:7551` `rollback_files` / `CheckpointManager.restore_checkpoint` via `git stash apply`) | Restores tracked files from a snapshot already captured by the loop; new files unlinked exactly as the existing batch-rollback already does. Zero new restore logic. | +| Preview (any case) | Delegate to `ReviewBranchManager` plumbing (`hash-object`/`commit-tree`/`branch`) | Already **proven** never to touch working tree, HEAD, or index. The preview branch surfaces natively in VS Code source control (ReviewBranch precedent). | + +**Forbidden token denylist (AST-enforced, §42.6 pin 5).** The following may not appear in any subprocess argv literal anywhere in `rewind_coordinator.py`: + +- `reset` together with `--hard` +- `push` together with `--force` / `-f` / a leading-`+` refspec +- `rebase`, `filter-branch`, `filter-repo` +- `branch -f`, or `-D`/`-d` of any branch **not** matching the rewind-preview prefix +- `update-ref -d`, `reflog expire`, `gc --prune=now`/`--prune=all` +- `checkout`/`switch` of a branch (file-scoped `git checkout -- ` restore is permitted; branch-switching the operator's HEAD is not) + +The permitted git surface is exactly: `revert`, file-scoped `restore`/`checkout -- `, `stash apply`, and the read-only plumbing delegated through `ReviewBranchManager`. **Operator-initiated only:** `RewindCoordinator` has no autonomous caller — §42.6 pin 6 forbids the orchestrator, governed-loop, any sensor, and the subagent scheduler from importing it. Preconditions before any revert: clean-tree gate (`JARVIS_REWIND_REQUIRE_CLEAN_TREE`, default **TRUE** — security-hardening-on-by-default, the Rule 7 precedent), ancestry reachability check, and an advisory **blast-radius** computed *dynamically* from the `file_paths` join (no hardcoded dependency rules — it lists every later timeline row sharing a path with the target op). + +### §42.6 AST Pins (Mandatory PRD Requirement #4) + +Seven structural invariants. Pins 1–4 prove the read-model lacks authority; pins 5–7 prove the RewindCoordinator obeys the irreversibility ban. + +| # | Invariant | AST assertion | +|---|---|---| +| 1 | **Read-model authority-free** | Module import graph of `operation_timeline.py` contains **none** of `{orchestrator, policy_engine, iron_gate, change_engine, candidate_generator, governed_loop_service, repair_engine}`. (Stronger than the grep-enforced `IDEObservabilityRouter` invariant — AST, not grep.) | +| 2 | **Single persistence seam** | The only disk-append in the module is via `cross_process_jsonl.flock_append_line`. Walk asserts **no** `import fcntl`, no `open(...).write/.writelines` on the timeline path, no `json.dump`/`json.dumps(...)→write` append. | +| 3 | **Subscribes, never re-derives** | Class defines the `OpsDigestObserver` protocol surface (`on_apply_succeeded`/`on_verify_completed`/`on_commit_succeeded`) **and** the module contains **no** `subprocess`/`create_subprocess_exec`/`git`-arg list and **no** TestRunner invocation — it consumes milestone events, it does not detect them. | +| 4 | **No state authority** | No `.record(` call on any ledger symbol; no `import` of `policy_engine`/`risk_tier_floor`; `risk_tier` only ever appears as an assignment *target from* an inbound value, never computed (no comparison/derivation expression produces it). | +| 5 | **Irreversibility denylist** | Walk every `subprocess`/`asyncio.create_subprocess_exec` argv list literal in `rewind_coordinator.py`; assert no element-pair from the §42.5 forbidden denylist co-occurs in any single argv; assert at least one revert path argv contains `revert`. | +| 6 | **Operator-initiated only (import-graph pin)** | No module under `{orchestrator, governed_loop_service, intake/sensors/**, subagent_scheduler}` imports `rewind_coordinator` or the `RewindCoordinator` symbol. (Cross-module pin; precedent: the authority-invariant import grep.) | +| 7 | **Preview non-duplication** | `rewind_coordinator.py` constructs **no** `write-tree`/`commit-tree`/`update-index --cacheinfo` argv in-module; the preview path resolves through an imported `ReviewBranchManager` symbol (the canonical non-destructive seam). | + +Each pin ships with explicit positive **and** negative fixtures (a deliberately-violating stub must make the pin red) — the discipline already standard across the SWE-Bench-Pro and Treefinement arcs. + +### §42.7 FlagRegistry Seeds (Mandatory PRD Requirement #4) + +All masters default-**FALSE** per §33.1 (graduate only on empirical evidence — §42.9). Safety-relevant precondition flags default-**TRUE** (security-hardening-on-by-default, Rule 7 precedent). + +| Flag | Type / Category | Default | Role | +|---|---|---|---| +| `JARVIS_OPERATION_TIMELINE_ENABLED` | BOOL / OBSERVABILITY | **FALSE** | Master — Layer A read-model subscription. Off ⇒ observer registers as a no-op; zero rows written. | +| `JARVIS_OPERATION_TIMELINE_PATH` | STR / OBSERVABILITY | `.jarvis/operation_timeline.jsonl` | Durable causal index location (not hardcoded inline). | +| `JARVIS_OPERATION_TIMELINE_MAX_ROWS` | INT / CAPACITY | `5000` | Bounded tail-scan on load; read at call time (monkeypatchable). | +| `JARVIS_REWIND_COORDINATOR_ENABLED` | BOOL / **SAFETY** | **FALSE** | Master — Layer B `/rewind` `/revert` verbs. SAFETY category because it invokes git. | +| `JARVIS_REWIND_PREVIEW_BRANCH_PREFIX` | STR / SAFETY | `ouroboros/rewind-preview/` | Preview-branch namespace; composes the `ReviewBranchManager` convention, parameterized not inlined. | +| `JARVIS_REWIND_REQUIRE_CLEAN_TREE` | BOOL / **SAFETY** | **TRUE** | Refuse any revert when the working tree is dirty (default-on hardening; operator can override in emergency). | +| `JARVIS_REWIND_BLAST_RADIUS_ENABLED` | BOOL / OBSERVABILITY | **TRUE** | Compute & surface the dependent-op set from the `file_paths` join before a revert (advisory). | + +### §42.8 Slice Breakdown (for the post-review authorization) + +- **Slice 1 — Read-model substrate (zero behavior change).** `OperationTimeline` projection + `flock_append_line` JSONL + registers against the **existing** `OpsDigestObserver` protocol + bounded load + `JARVIS_OPERATION_TIMELINE_ENABLED` default-FALSE + AST pins 1–4 + FlagRegistry seeds 1–3 + regression spine. **No REPL, no orchestrator edits, no rewind.** Mirrors the Stage-1.6 Park-spike "Slice 1 = zero runtime change" precedent exactly. +- **Slice 2 — Causal join completion + read surface.** Populate `signal_source`/`urgency`/`risk_tier`/`plan_ref`/`diff_ref`/`checkpoint_ref`/`parent_op_id` by read-only joins over the IntentEnvelope + `DiffArchive` + `WorkspaceCheckpointManager` (still authority-free). `/timeline` verb (auto-discovered `_handle_*`), `r-N` prefix added to the unified `/expand` dispatcher, `GET /observability/timeline` (extend `IDEObservabilityRouter`, authority invariant preserved), `timeline_row_appended` SSE via the existing broker. **Still no rewind.** +- **Slice 3 — RewindCoordinator (Layer B).** Preview (delegate `ReviewBranchManager`), committed revert (`git revert`), in-flight revert (compose existing snapshot restore), preconditions (clean-tree + ancestry + dynamic blast-radius), `/rewind` `/revert` verbs, `rewind_*` SSE, `reverted_by` back-edge amended row. `JARVIS_REWIND_COORDINATOR_ENABLED` default-FALSE + AST pins 5–7 + FlagRegistry seeds 4–7. +- **Slice 4 — Graduation.** Soak matrix + evidence gate + default-TRUE flip after the §42.9 criteria are met across 3 consecutive clean soaks. + +### §42.9 Graduation Criteria (Concrete) + +Masters stay default-FALSE until **all** of the following are demonstrated as artifacts (no pre-result euphoria — no Tier/capability claim until each lands): + +1. The timeline JSONL accumulates **≥1 fully-populated causal row with non-null `commit_hash`** during a battle-test session (proves the observer-join works end-to-end). +2. `/timeline` reconstructs the causal chain on a **fresh boot of a later session** (cross-session morning-after proof — the capability Claude Code structurally cannot have). +3. `/rewind ` produces a non-destructive preview branch verifiable in `git branch` with working tree/HEAD/index provably untouched; **and** `/revert ` of a committed op produces a `git revert` commit where `git log` shows **both** the revert commit and the original commit (history-rewrite-free, the irreversibility ban empirically held). +4. Blast-radius correctly enumerates the later timeline rows sharing a `file_paths` element with the reverted op. +5. (3)–(4) hold across **3 consecutive clean soaks** under the Phase 9 cadence with zero AST-pin regressions. + +### §42.10 What Is / Is Not Demonstrated Today (Honest Framing) + +**Demonstrated (substrate exists, composable):** `DiffArchive` `d-N` lifecycle audit; `ReviewBranchManager` non-destructive plumbing; `AutoCommitter` signed commits + `on_commit_succeeded` observer emission; `OperationLedger` per-op state JSONL; `WorkspaceCheckpointManager` git-stash snapshots + the existing failure-driven batch rollback; the unified `/expand` prefix dispatcher + `_handle_*` auto-discovery; the `IDEObservabilityRouter` authority-free read-only-projection pattern; the `flock_append_line` durable-append seam. + +**Not demonstrated (this section closes it):** any **durable, cross-session, causal `op_id ↔ signal ↔ diff ↔ commit_hash ↔ checkpoint ↔ outcome` index**; any **operator-initiated** revert (only failure-driven internal rollback exists today); any **preview-before-revert** surface; any **blast-radius** reasoning over O+V's own change history. The "scrub the timeline / one-key revert" surface is **positioned**, not yet **demonstrated** — §42.9 is the ladder that makes it actually demonstrated. + +**Tier placement.** §41.8 roadmap — trust-infrastructure tier. Not a Tier-D capability claim. It is load-bearing **operator-trust** infrastructure: an agent that committed six things while the operator was at lunch is only trustworthy if those six things are causally legible and individually, non-destructively reversible. + +### §42.11 Operator Review Checklist (Blueprint Sign-Off) + +Before Slice 1 is authorized, confirm: + +- [ ] Zero-duplication boundary (§42.3) is correct — timeline owns only edges, every datum is an FK to an existing authority. +- [ ] Schema `timeline.1` (§42.4) captures the full causal join with no field that re-stores another module's authority. +- [ ] Irreversibility ban (§42.5) — `git revert` + composed snapshot restore + delegated preview is the intended model; the forbidden denylist is complete. +- [ ] AST pins 1–7 (§42.6) are the right structural proofs of "read-model lacks authority" + "coordinator obeys the ban." +- [ ] FlagRegistry seeds (§42.7) — masters default-FALSE, safety preconditions default-TRUE, is the intended posture. +- [ ] Slice 1 is genuinely zero-behavior-change (read-model + spine only) before any rewind code exists. +- [ ] Gap #5 (forensic morning-after) deferred — §42 only architects the substrate it will later compose, builds no Gap #5 code. + +*Authored 2026-05-16 — blueprint for operator review. No Slice authorized until this checklist is signed.* diff --git a/tests/governance/test_last_session_summary_v1_1a.py b/tests/governance/test_last_session_summary_v1_1a.py index cde6ab7872..49f74c7ca6 100644 --- a/tests/governance/test_last_session_summary_v1_1a.py +++ b/tests/governance/test_last_session_summary_v1_1a.py @@ -484,8 +484,14 @@ def test_orchestrator_has_three_observer_call_sites(): hook_methods.add(func.attr) required = {"on_apply_succeeded", "on_verify_completed", "on_commit_succeeded"} + # Deliberate protocol extension — PRD §42 Slice 2 added the + # `on_op_classified` causal signal→op edge, emitted at the INTENT + # seam through this SAME canonical observer chain (not a parallel + # registry). Sanctioned per this test's own contract ("update this + # regression test deliberately if the protocol was extended"). + sanctioned_extensions = {"on_op_classified"} missing = required - hook_methods - extra = hook_methods - required + extra = hook_methods - required - sanctioned_extensions assert not missing, ( f"orchestrator.py is missing observer call sites: {missing}. " f"ops_digest will silently stop populating." diff --git a/tests/governance/test_operation_timeline.py b/tests/governance/test_operation_timeline.py new file mode 100644 index 0000000000..e5232dadc3 --- /dev/null +++ b/tests/governance/test_operation_timeline.py @@ -0,0 +1,797 @@ +"""Regression spine — OperationTimeline read-model (PRD §42, Slice 1). + +Two test families: + + * **AST pins 1–4** (§42.6): structurally prove the read-model lacks + authority — authority-free import graph / single persistence seam / + subscribes-never-re-derives / no state authority. Each pin ships + with a negative-fixture sibling so a deliberately-violating shape + makes the pin red. + + * **Behavioral**: the zero-behavior-change guarantee (flag OFF ⇒ + zero rows, zero disk I/O), the causal merge across the three + OpsDigestObserver callbacks, monotonic non-reused refs, lossless + to_dict/from_dict, bounded idempotent disk replay (the + cross-session morning-after substrate), fail-closed never-raises, + and the FlagRegistry seeds (masters default-FALSE per §33.1). +""" +from __future__ import annotations + +import ast +import json +from pathlib import Path + +import pytest + +from backend.core.ouroboros.governance import operation_timeline as ot +from backend.core.ouroboros.governance.operation_timeline import ( + OPERATION_TIMELINE_ENABLED_ENV_VAR, + OPERATION_TIMELINE_MAX_ROWS_ENV_VAR, + OPERATION_TIMELINE_PATH_ENV_VAR, + REF_PREFIX, + TIMELINE_SCHEMA_VERSION, + OperationTimeline, + TimelineRow, + get_default_timeline, + register_flags, + reset_default_timeline, +) + + +def _module_source() -> str: + return Path(ot.__file__).read_text(encoding="utf-8") + + +# =========================================================================== +# AST pin 1 — read-model authority-free +# =========================================================================== + +_FORBIDDEN_AUTHORITY_MODULES = ( + "orchestrator", + "policy_engine", + "iron_gate", + "change_engine", + "candidate_generator", + "governed_loop_service", + "repair_engine", +) + + +def test_ast_pin_1_authority_free_import_graph() -> None: + tree = ast.parse(_module_source()) + offenders = [] + for node in ast.walk(tree): + if isinstance(node, ast.ImportFrom): + mod = node.module or "" + for forbidden in _FORBIDDEN_AUTHORITY_MODULES: + if forbidden in mod: + offenders.append(mod) + elif isinstance(node, ast.Import): + for alias in node.names: + for forbidden in _FORBIDDEN_AUTHORITY_MODULES: + if forbidden in alias.name: + offenders.append(alias.name) + assert not offenders, ( + f"operation_timeline.py imports authority modules {offenders} — " + "the read-model must be structurally incapable of acting on " + "the loop (§42.6 pin 1)" + ) + + +def test_ast_pin_1_negative_fixture_detects_violation() -> None: + bad = "from backend.core.ouroboros.governance.orchestrator import X\n" + tree = ast.parse(bad) + hit = any( + isinstance(n, ast.ImportFrom) + and "orchestrator" in (n.module or "") + for n in ast.walk(tree) + ) + assert hit, "pin-1 walker must detect an orchestrator import" + + +# =========================================================================== +# AST pin 2 — single persistence seam +# =========================================================================== + + +def test_ast_pin_2_imports_canonical_flock_append_line() -> None: + tree = ast.parse(_module_source()) + found = False + for node in ast.walk(tree): + if isinstance(node, ast.ImportFrom): + if "cross_process_jsonl" in (node.module or ""): + for alias in node.names: + if alias.name == "flock_append_line": + found = True + assert found, ( + "operation_timeline.py must import the canonical " + "flock_append_line (§42.6 pin 2)" + ) + + +def test_ast_pin_2_no_fcntl_or_raw_append_or_json_dump() -> None: + src = _module_source() + assert "import fcntl" not in src, "no homegrown fcntl seam (pin 2)" + assert "fcntl." not in src, "no fcntl reference (pin 2)" + # json.dumps(...) (string) is allowed; json.dump(..., fp) (direct + # file write) is the forbidden parallel-persistence shape. + assert "json.dump(" not in src, ( + "no json.dump to a file object — the only disk seam is " + "flock_append_line (pin 2)" + ) + tree = ast.parse(src) + for node in ast.walk(tree): + if isinstance(node, ast.Call) and isinstance(node.func, ast.Name): + if node.func.id == "open": + # No append-mode open anywhere in the module. + for arg in node.args[1:]: + if isinstance(arg, ast.Constant) and "a" in str( + arg.value + ): + raise AssertionError( + "append-mode open() found — use " + "flock_append_line (pin 2)" + ) + + +# =========================================================================== +# AST pin 3 — subscribes, never re-derives +# =========================================================================== + + +def test_ast_pin_3_defines_observer_protocol_surface() -> None: + tree = ast.parse(_module_source()) + required = { + "on_apply_succeeded", + "on_verify_completed", + "on_commit_succeeded", + } + for cls in ast.walk(tree): + if isinstance(cls, ast.ClassDef) and cls.name == "OperationTimeline": + defined = { + fn.name for fn in cls.body + if isinstance(fn, (ast.FunctionDef, ast.AsyncFunctionDef)) + } + missing = required - defined + assert not missing, ( + f"OperationTimeline missing protocol methods {missing} " + "(§42.6 pin 3)" + ) + return + raise AssertionError("OperationTimeline class not found") + + +def test_ast_pin_3_no_event_re_derivation() -> None: + """AST-node based (NOT substring on source): the docstring legitimately + *describes* the ban, so the pin must inspect actual code nodes — + imports, calls, and string-literal subprocess args — never prose.""" + tree = ast.parse(_module_source()) + banned_import_substrings = ("subprocess", "test_runner") + banned_call_names = { + "create_subprocess_exec", + "create_subprocess_shell", + "check_output", + "check_call", + "Popen", + "run", + } + for node in ast.walk(tree): + if isinstance(node, ast.Import): + for alias in node.names: + for bad in banned_import_substrings: + assert bad not in alias.name, ( + f"import {alias.name!r} re-derives events " + "(§42.6 pin 3)" + ) + assert alias.name != "subprocess" + if isinstance(node, ast.ImportFrom): + mod = node.module or "" + for bad in banned_import_substrings: + assert bad not in mod, ( + f"import from {mod!r} re-derives events (pin 3)" + ) + for alias in node.names: + assert "TestRunner" not in alias.name, ( + "TestRunner import re-derives VERIFY (pin 3)" + ) + if isinstance(node, ast.Call): + fn = node.func + name = ( + fn.attr if isinstance(fn, ast.Attribute) + else fn.id if isinstance(fn, ast.Name) + else "" + ) + assert name not in banned_call_names, ( + f"call {name!r} detects rather than consumes events " + "(§42.6 pin 3)" + ) + # No subprocess argv built from a literal "git". + for arg in node.args: + if isinstance(arg, ast.Constant) and arg.value == "git": + raise AssertionError( + "literal 'git' subprocess arg — the read-model " + "must not shell out (pin 3)" + ) + + +# =========================================================================== +# AST pin 4 — no state authority +# =========================================================================== + + +def test_ast_pin_4_no_state_authority() -> None: + """AST-node based: the docstring legitimately names OperationState + when *describing* the ban. The pin inspects code nodes only — + .record() calls, OperationState name references, and authority + imports — never prose.""" + tree = ast.parse(_module_source()) + for node in ast.walk(tree): + if isinstance(node, ast.Call) and isinstance( + node.func, ast.Attribute + ): + assert node.func.attr != "record", ( + "operation_timeline.py calls .record() — it must never " + "write OperationLedger state (§42.6 pin 4)" + ) + if isinstance(node, ast.Name): + assert node.id != "OperationState", ( + "OperationState referenced in code — state is the " + "OperationLedger's authority (pin 4)" + ) + if isinstance(node, ast.ImportFrom): + mod = node.module or "" + assert "risk_tier_floor" not in mod, ( + "no risk_tier_floor import — risk_tier is a copied " + "string, never computed here (pin 4)" + ) + assert not mod.endswith("policy_engine"), ( + "no policy_engine import — the timeline has no policy " + "authority (pin 4)" + ) + for alias in node.names: + assert alias.name != "OperationState", ( + "no OperationState import (pin 4)" + ) + + +# =========================================================================== +# Behavioral — the zero-behavior-change guarantee (load-bearing) +# =========================================================================== + + +@pytest.fixture(autouse=True) +def _isolate(monkeypatch, tmp_path): + """Each test gets a fresh singleton + a tmp JSONL path + a clean + env (no inherited JARVIS_OPERATION_TIMELINE_* from the shell).""" + for var in ( + OPERATION_TIMELINE_ENABLED_ENV_VAR, + OPERATION_TIMELINE_PATH_ENV_VAR, + OPERATION_TIMELINE_MAX_ROWS_ENV_VAR, + ): + monkeypatch.delenv(var, raising=False) + reset_default_timeline() + yield + reset_default_timeline() + + +def test_flag_off_is_a_hard_noop_zero_rows_zero_file(tmp_path) -> None: + """The §42.8 Slice-1 contract: flag OFF ⇒ zero rows, zero disk + I/O, zero behavior change. This is the load-bearing proof.""" + path = tmp_path / "timeline.jsonl" + tl = OperationTimeline(persistence_path=path, enabled=False) + tl.on_apply_succeeded(op_id="op-1", mode="single", files=2) + tl.on_verify_completed(op_id="op-1", passed=4, total=4) + tl.on_commit_succeeded(op_id="op-1", commit_hash="deadbeef") + assert len(tl) == 0 + assert not path.exists(), "flag OFF must not create the JSONL file" + assert tl.query() == () + + +def test_flag_off_via_env_default_is_noop(tmp_path) -> None: + # No env set at all ⇒ default-FALSE ⇒ no-op (the §33.1 default). + path = tmp_path / "timeline.jsonl" + tl = OperationTimeline(persistence_path=path) # no enabled override + tl.on_commit_succeeded(op_id="op-x", commit_hash="abc123") + assert len(tl) == 0 + assert not path.exists() + + +# =========================================================================== +# Behavioral — causal merge across the three callbacks +# =========================================================================== + + +def _enabled_timeline(tmp_path) -> OperationTimeline: + return OperationTimeline( + persistence_path=tmp_path / "timeline.jsonl", enabled=True, + ) + + +def test_three_callbacks_merge_into_one_row_stable_ref(tmp_path) -> None: + tl = _enabled_timeline(tmp_path) + tl.on_apply_succeeded(op_id="op-7", mode="multi", files=3) + row_after_apply = tl.query(op_id="op-7")[0] + ref = row_after_apply.ref + assert ref.startswith(REF_PREFIX) + + tl.on_verify_completed( + op_id="op-7", passed=10, total=12, scoped_to_applied_op=True, + ) + tl.on_commit_succeeded(op_id="op-7", commit_hash="17ae95d7d6") + + rows = tl.query(op_id="op-7") + assert len(rows) == 1, "three callbacks for one op_id ⇒ one row" + r = rows[0] + assert r.ref == ref, "ref is stable across merges" + assert r.apply_mode == "multi" and r.apply_files == 3 + assert r.verify_passed == 10 and r.verify_total == 12 + assert r.verify_scoped_to_op is True + assert r.commit_hash == "17ae95d7d6" + assert r.first_seen_iso and r.updated_iso + assert r.updated_iso >= r.first_seen_iso + + +def test_refs_are_monotonic_and_never_reused(tmp_path) -> None: + tl = _enabled_timeline(tmp_path) + tl.on_apply_succeeded(op_id="op-a", mode="single", files=1) + tl.on_apply_succeeded(op_id="op-b", mode="single", files=1) + tl.on_apply_succeeded(op_id="op-c", mode="single", files=1) + refs = {r.op_id: r.ref for r in tl.query()} + nums = sorted(int(v[len(REF_PREFIX):]) for v in refs.values()) + assert nums == [1, 2, 3], f"non-monotonic refs: {refs}" + assert len(set(refs.values())) == 3, "refs must be unique" + + +def test_query_orders_newest_first_and_filters(tmp_path) -> None: + tl = _enabled_timeline(tmp_path) + tl.on_apply_succeeded(op_id="old", mode="single", files=1) + tl.on_apply_succeeded(op_id="new", mode="single", files=1) + tl.on_commit_succeeded(op_id="new", commit_hash="c0ffee") + ordered = [r.op_id for r in tl.query()] + assert ordered[0] == "new", "newest first" + committed = tl.query(has_commit=True) + assert [r.op_id for r in committed] == ["new"] + uncommitted = tl.query(has_commit=False) + assert [r.op_id for r in uncommitted] == ["old"] + assert tl.query(limit=1) and len(tl.query(limit=1)) == 1 + + +def test_lookup_by_ref(tmp_path) -> None: + tl = _enabled_timeline(tmp_path) + tl.on_apply_succeeded(op_id="op-z", mode="none", files=0) + ref = tl.query(op_id="op-z")[0].ref + assert tl.lookup(ref) is not None + assert tl.lookup(ref).op_id == "op-z" + assert tl.lookup("r-99999") is None + assert tl.lookup("garbage") is None + + +# =========================================================================== +# Behavioral — schema roundtrip + closed field set +# =========================================================================== + + +def test_to_dict_from_dict_roundtrip_lossless(tmp_path) -> None: + tl = _enabled_timeline(tmp_path) + tl.on_apply_succeeded(op_id="op-rt", mode="multi", files=5) + tl.on_verify_completed(op_id="op-rt", passed=7, total=7) + tl.on_commit_succeeded(op_id="op-rt", commit_hash="abc") + original = tl.query(op_id="op-rt")[0] + restored = TimelineRow.from_dict(original.to_dict()) + assert restored == original, "to_dict/from_dict must be lossless" + + +def test_schema_version_constant_and_present_in_rows(tmp_path) -> None: + assert TIMELINE_SCHEMA_VERSION == "timeline.1" + tl = _enabled_timeline(tmp_path) + tl.on_apply_succeeded(op_id="op-s", mode="single", files=1) + assert tl.query()[0].schema_version == "timeline.1" + + +def test_jsonl_on_disk_is_valid_one_record_per_line(tmp_path) -> None: + path = tmp_path / "timeline.jsonl" + tl = OperationTimeline(persistence_path=path, enabled=True) + tl.on_apply_succeeded(op_id="op-d", mode="single", files=1) + tl.on_commit_succeeded(op_id="op-d", commit_hash="f00d") + lines = [ + ln for ln in path.read_text(encoding="utf-8").splitlines() + if ln.strip() + ] + assert len(lines) == 2, "append-only audit: one row per callback" + for ln in lines: + json.loads(ln) # must not raise + + +# =========================================================================== +# Behavioral — cross-session disk replay (morning-after substrate) +# =========================================================================== + + +def test_replay_reconstructs_projection_idempotently(tmp_path) -> None: + path = tmp_path / "timeline.jsonl" + src = OperationTimeline(persistence_path=path, enabled=True) + src.on_apply_succeeded(op_id="op-1", mode="single", files=2) + src.on_commit_succeeded(op_id="op-1", commit_hash="h1") + src.on_apply_succeeded(op_id="op-2", mode="multi", files=4) + + # Fresh process boot: a brand-new instance replays the durable file. + fresh = OperationTimeline(persistence_path=path, enabled=True) + n1 = fresh.replay_from_disk() + assert n1 >= 1 + assert len(fresh) == 2, "latest-write-wins collapses to 2 op_ids" + op1 = fresh.query(op_id="op-1")[0] + assert op1.commit_hash == "h1", "the missing link survived a reboot" + + before = {r.op_id: r for r in fresh.query()} + fresh.replay_from_disk() # idempotent + after = {r.op_id: r for r in fresh.query()} + assert before == after, "replay must be idempotent" + + +def test_replay_advances_seq_so_new_refs_never_collide(tmp_path) -> None: + path = tmp_path / "timeline.jsonl" + src = OperationTimeline(persistence_path=path, enabled=True) + src.on_apply_succeeded(op_id="op-1", mode="single", files=1) + src.on_apply_succeeded(op_id="op-2", mode="single", files=1) + + fresh = OperationTimeline(persistence_path=path, enabled=True) + fresh.replay_from_disk() + fresh.on_apply_succeeded(op_id="op-3", mode="single", files=1) + all_refs = {r.ref for r in fresh.query()} + assert len(all_refs) == 3, f"ref collision after replay: {all_refs}" + + +def test_replay_is_bounded_by_max_rows(tmp_path, monkeypatch) -> None: + path = tmp_path / "timeline.jsonl" + src = OperationTimeline(persistence_path=path, enabled=True) + for i in range(10): + src.on_apply_succeeded(op_id=f"op-{i}", mode="single", files=1) + monkeypatch.setenv(OPERATION_TIMELINE_MAX_ROWS_ENV_VAR, "3") + fresh = OperationTimeline(persistence_path=path, enabled=True) + replayed = fresh.replay_from_disk() + assert replayed <= 3, "tail-scan must honor the max-rows cap" + + +def test_replay_skips_malformed_rows(tmp_path) -> None: + path = tmp_path / "timeline.jsonl" + path.write_text( + "not json\n" + '{"ref": "r-1", "op_id": "ok", "schema_version": "timeline.1"}\n' + "{bad}\n", + encoding="utf-8", + ) + tl = OperationTimeline(persistence_path=path, enabled=True) + n = tl.replay_from_disk() + assert n == 1, "only the one well-formed row replays" + assert tl.query(op_id="ok") + + +def test_replay_missing_file_returns_zero(tmp_path) -> None: + tl = OperationTimeline( + persistence_path=tmp_path / "nope.jsonl", enabled=True, + ) + assert tl.replay_from_disk() == 0 + + +# =========================================================================== +# Behavioral — fail-closed (observer NEVER raises) +# =========================================================================== + + +def test_observer_methods_never_raise_on_bad_input(tmp_path) -> None: + tl = _enabled_timeline(tmp_path) + # Empty op_id is dropped silently, not raised. + tl.on_apply_succeeded(op_id="", mode="single", files=1) + tl.on_commit_succeeded(op_id="", commit_hash="x") + assert len(tl) == 0 + # Type-hostile values must still not raise (fail-closed contract). + tl.on_verify_completed(op_id="op-q", passed=0, total=0) # type: ignore[arg-type] + assert tl.query() is not None + + +def test_clear_drops_memory_keeps_disk(tmp_path) -> None: + path = tmp_path / "timeline.jsonl" + tl = OperationTimeline(persistence_path=path, enabled=True) + tl.on_apply_succeeded(op_id="op-c", mode="single", files=1) + assert len(tl) == 1 and path.exists() + tl.clear() + assert len(tl) == 0, "in-memory projection dropped" + assert path.exists(), "append-only audit survives an in-memory reset" + + +# =========================================================================== +# Behavioral — singleton + FlagRegistry seeds +# =========================================================================== + + +def test_singleton_is_idempotent_and_resettable() -> None: + a = get_default_timeline() + b = get_default_timeline() + assert a is b + reset_default_timeline() + c = get_default_timeline() + assert c is not a + + +def test_register_flags_seeds_three_specs_master_default_false() -> None: + class _Reg: + def __init__(self): + self.specs = {} + + def register(self, spec): + self.specs[spec.name] = spec + + reg = _Reg() + count = register_flags(reg) + assert count == 3 + master = reg.specs[OPERATION_TIMELINE_ENABLED_ENV_VAR] + assert master.default is False, ( + "§33.1: the master switch MUST default-FALSE" + ) + assert master.type.value == "bool" + assert master.category.value == "observability" + path_spec = reg.specs[OPERATION_TIMELINE_PATH_ENV_VAR] + assert path_spec.default == ".jarvis/operation_timeline.jsonl" + rows_spec = reg.specs[OPERATION_TIMELINE_MAX_ROWS_ENV_VAR] + assert rows_spec.type.value == "int" + assert rows_spec.category.value == "capacity" + + +def test_register_flags_never_raises_on_bad_registry() -> None: + class _Boom: + def register(self, spec): + raise RuntimeError("registry exploded") + + # Per-spec failures are swallowed; the function returns a count + # (0 here) rather than propagating. + assert register_flags(_Boom()) == 0 + + +# =========================================================================== +# SLICE 2 — composite fan-out (root fix for SessionRecorder coexistence) +# =========================================================================== + +from backend.core.ouroboros.governance import ops_digest_observer as odo # noqa: E402 + + +@pytest.fixture(autouse=True) +def _reset_observer(): + odo.reset_ops_digest_observer() + yield + odo.reset_ops_digest_observer() + + +class _SpyObserver: + def __init__(self): + self.calls = [] + + def on_apply_succeeded(self, *, op_id, mode, files): + self.calls.append(("apply", op_id, mode, files)) + + def on_verify_completed(self, *, op_id, passed, total, + scoped_to_applied_op=True): + self.calls.append(("verify", op_id, passed, total)) + + def on_commit_succeeded(self, *, op_id, commit_hash): + self.calls.append(("commit", op_id, commit_hash)) + + def on_op_classified(self, *, op_id, signal_source, urgency, risk_tier): + self.calls.append(("classified", op_id, signal_source)) + + +def test_register_get_reset_remain_byte_identical_behavior() -> None: + """Slice 2 must not regress the single-slot API the harness uses.""" + spy = _SpyObserver() + odo.register_ops_digest_observer(spy) + assert odo.get_ops_digest_observer() is spy + odo.reset_ops_digest_observer() + got = odo.get_ops_digest_observer() + assert got is not spy + # default is a no-op that swallows everything + got.on_apply_succeeded(op_id="x", mode="single", files=1) + + +def test_add_composes_both_observers_no_eviction() -> None: + """The root fix: add MUST NOT evict the registered SessionRecorder + analog — both receive every callback.""" + recorder = _SpyObserver() + timeline = _SpyObserver() + odo.register_ops_digest_observer(recorder) # harness boot order + odo.add_ops_digest_observer(timeline) # PRD §42 wiring + obs = odo.get_ops_digest_observer() + obs.on_apply_succeeded(op_id="op-1", mode="single", files=2) + obs.on_op_classified( + op_id="op-1", signal_source="TestFailure", + urgency="immediate", risk_tier="notify_apply", + ) + assert ("apply", "op-1", "single", 2) in recorder.calls + assert ("apply", "op-1", "single", 2) in timeline.calls + assert ("classified", "op-1", "TestFailure") in recorder.calls + assert ("classified", "op-1", "TestFailure") in timeline.calls + + +def test_add_is_idempotent_by_identity() -> None: + spy = _SpyObserver() + odo.add_ops_digest_observer(spy) + odo.add_ops_digest_observer(spy) + odo.add_ops_digest_observer(spy) + odo.get_ops_digest_observer().on_commit_succeeded( + op_id="op-x", commit_hash="abc", + ) + # Exactly one delivery despite three adds. + assert spy.calls.count(("commit", "op-x", "abc")) == 1 + + +def test_add_order_independent_and_drops_bare_noop() -> None: + """add before register also converges; a bare _NoopObserver is not + fanned to (it carries no state).""" + timeline = _SpyObserver() + odo.add_ops_digest_observer(timeline) # nothing registered yet + recorder = _SpyObserver() + odo.add_ops_digest_observer(recorder) + obs = odo.get_ops_digest_observer() + obs.on_verify_completed(op_id="o", passed=1, total=1) + assert ("verify", "o", 1, 1) in timeline.calls + assert ("verify", "o", 1, 1) in recorder.calls + + +def test_misbehaving_member_does_not_starve_others() -> None: + class _Bad: + def on_apply_succeeded(self, **k): + raise RuntimeError("boom") + + def on_verify_completed(self, **k): + raise RuntimeError("boom") + + def on_commit_succeeded(self, **k): + raise RuntimeError("boom") + + def on_op_classified(self, **k): + raise RuntimeError("boom") + + good = _SpyObserver() + odo.add_ops_digest_observer(_Bad()) + odo.add_ops_digest_observer(good) + # Must not raise; good still receives. + odo.get_ops_digest_observer().on_apply_succeeded( + op_id="op-r", mode="multi", files=4, + ) + assert ("apply", "op-r", "multi", 4) in good.calls + + +def test_remove_collapses_and_restores_noop() -> None: + a = _SpyObserver() + b = _SpyObserver() + odo.add_ops_digest_observer(a) + odo.add_ops_digest_observer(b) + odo.remove_ops_digest_observer(b) + # single-member composite collapses back to the bare observer + assert odo.get_ops_digest_observer() is a + odo.remove_ops_digest_observer(a) + # empty → default no-op restored (never None) + got = odo.get_ops_digest_observer() + assert got is not None + got.on_apply_succeeded(op_id="z", mode="none", files=0) + + +def test_protocol_implementers_all_have_on_op_classified() -> None: + """SessionRecorder, _NoopObserver, composite, and the timeline must + all structurally satisfy the extended protocol (no per-op + AttributeError through the fan-out).""" + from backend.core.ouroboros.battle_test.session_recorder import ( + SessionRecorder, + ) + for cls in (SessionRecorder, odo._NoopObserver, + odo._CompositeOpsDigestObserver, OperationTimeline): + assert hasattr(cls, "on_op_classified"), cls.__name__ + + +# =========================================================================== +# SLICE 2 — on_op_classified merge + DiffArchive causal join +# =========================================================================== + + +def test_on_op_classified_merges_signal_edge(tmp_path) -> None: + tl = OperationTimeline( + persistence_path=tmp_path / "t.jsonl", enabled=True, + ) + tl.on_op_classified( + op_id="op-c", signal_source="VoiceCommand", + urgency="immediate", risk_tier="approval_required", + ) + tl.on_apply_succeeded(op_id="op-c", mode="single", files=1) + r = tl.query(op_id="op-c")[0] + assert r.signal_source == "VoiceCommand" + assert r.urgency == "immediate" + assert r.risk_tier == "approval_required" + assert r.apply_mode == "single" # merged across both callbacks + + +def test_diff_archive_join_fills_edges(tmp_path, monkeypatch) -> None: + """The read-only join pulls diff_ref/file_paths/risk_tier from the + canonical DiffArchive singleton, keyed by op_id.""" + from backend.core.ouroboros.battle_test import diff_archive + + diff_archive.reset_default_archive() if hasattr( + diff_archive, "reset_default_archive" + ) else None + arch = diff_archive.get_default_archive() + added = arch.add( + op_id="op-j", + risk_tier="notify_apply", + file_paths=("backend/a.py", "backend/b.py"), + diff_text="--- a\n+++ b\n", + summary="join test", + ) + tl = OperationTimeline( + persistence_path=tmp_path / "t.jsonl", enabled=True, + ) + tl.on_apply_succeeded(op_id="op-j", mode="multi", files=2) + r = tl.query(op_id="op-j")[0] + assert r.diff_ref == added.ref, "diff_ref joined from DiffArchive" + assert r.file_paths == ("backend/a.py", "backend/b.py") + assert r.risk_tier == "notify_apply", "risk_tier joined when unset" + + +def test_explicit_classified_risk_tier_wins_over_diff_join( + tmp_path, +) -> None: + """on_op_classified risk_tier is authoritative; the diff's copy + only fills when still unset (adaptive, no clobber).""" + from backend.core.ouroboros.battle_test import diff_archive + + arch = diff_archive.get_default_archive() + arch.add( + op_id="op-w", risk_tier="safe_auto", + file_paths=("x.py",), diff_text="d", summary="s", + ) + tl = OperationTimeline( + persistence_path=tmp_path / "t.jsonl", enabled=True, + ) + tl.on_op_classified( + op_id="op-w", signal_source="S", urgency="low", + risk_tier="approval_required", + ) + tl.on_apply_succeeded(op_id="op-w", mode="single", files=1) + r = tl.query(op_id="op-w")[0] + assert r.risk_tier == "approval_required", ( + "explicit classified risk_tier must win over the diff copy" + ) + + +def test_sse_publish_is_best_effort_never_raises(tmp_path) -> None: + # Stream disabled by default ⇒ publish_task_event is a no-op; + # _publish_sse must swallow regardless and never break the append. + tl = OperationTimeline( + persistence_path=tmp_path / "t.jsonl", enabled=True, + ) + tl.on_commit_succeeded(op_id="op-sse", commit_hash="cafe") + assert tl.query(op_id="op-sse")[0].commit_hash == "cafe" + + +def test_event_type_constant_registered_in_broker() -> None: + from backend.core.ouroboros.governance import ( + ide_observability_stream as ios, + ) + assert ( + ios.EVENT_TYPE_OPERATION_TIMELINE_ROW + in ios._VALID_EVENT_TYPES + ), "the new SSE event type must be in the broker's valid set" + + +def test_ide_observability_handler_exists_and_authority_free() -> None: + """The GET /observability/timeline handler must exist AND the + operation_timeline module it reads must not import gate modules + (the IDEObservability authority invariant, transitively).""" + from backend.core.ouroboros.governance.ide_observability import ( + IDEObservabilityRouter, + ) + assert hasattr(IDEObservabilityRouter, "_handle_timeline") + # operation_timeline authority-free is already proven by pin 1; + # this asserts the transitive guarantee the route depends on. + src = _module_source() + tree = ast.parse(src) + for node in ast.walk(tree): + if isinstance(node, ast.ImportFrom): + assert "policy_engine" not in (node.module or "") + assert "iron_gate" not in (node.module or "")