Sovereign Evidence Rail + L3 Memory Governor

backend/core/ouroboros/governance/autonomy/l3_memory_governor.py

@@ -0,0 +1,80 @@
+"""L3 worktree-RAM-budget governor (pure math).
+
+Composes ON TOP of MemoryPressureGate's free-%-based fan-out caps:
+the gate answers "is the box under pressure?"; this module answers
+"given the absolute RAM cost of a worktree, how many fit right now?".
+Strictest-wins between the two. No IO, no scheduler import — every
+decision is a deterministic function of its arguments so it can be
+proven at all pressure levels in isolation.
+"""
+from __future__ import annotations
+
+import math
+import os
+from dataclasses import dataclass
+
+
+def _env_bool(name: str, default: bool) -> bool:
+    raw = os.environ.get(name)
+    if raw is None:
+        return default
+    return raw.strip().lower() in ("true", "1", "yes")
+
+
+def _env_int(name: str, default: int, *, minimum: int) -> int:
+    raw = os.environ.get(name)
+    if raw is None:
+        return default
+    try:
+        return max(minimum, int(raw))
+    except (TypeError, ValueError):
+        return default
+
+
+def governor_enabled() -> bool:
+    """Master flag. Default TRUE; inert until an L3 graph actually runs."""
+    return _env_bool("JARVIS_L3_MEMORY_GOVERNOR_ENABLED", True)
+
+
+def worktree_ram_budget_mb() -> int:
+    """Assumed peak RAM per concurrent worktree. Default 1500MB."""
+    return _env_int("JARVIS_L3_WORKTREE_RAM_BUDGET_MB", 1500, minimum=64)
+
+
+@dataclass(frozen=True)
+class GovernorDecision:
+    requested: int
+    ram_cap: int
+    level_cap: int
+    n_allowed: int
+    avail_mb: float
+    budget_mb: int
+    disposition: str  # compute_worktree_cap emits "allow"|"clamp"; the
+    # scheduler layer may instead report "disabled"/"probe_fail" (this pure
+    # function never produces those — it is only reached with a live probe).
+
+
+def compute_worktree_cap(
+    *,
+    requested: int,
+    avail_mb: float,
+    budget_mb: int,
+    level_cap: int,
+) -> GovernorDecision:
+    """Pure clamp. ``ram_cap = floor(avail_mb / budget_mb)`` (>=1);
+    final allowance is the strictest of requested / ram_cap / level_cap."""
+    # Fail-safe: a bad/non-positive avail_mb (e.g. a garbage probe reading)
+    # floors to ram_cap=1 — the most conservative non-zero fan-out — rather
+    # than 0 or negative. Clamping down on bad input is the safe direction.
+    ram_cap = max(1, int(math.floor(avail_mb / float(budget_mb))))
+    n_allowed = max(0, min(requested, ram_cap, level_cap))
+    disposition = "clamp" if n_allowed < requested else "allow"
+    return GovernorDecision(
+        requested=requested,
+        ram_cap=ram_cap,
+        level_cap=level_cap,
+        n_allowed=n_allowed,
+        avail_mb=avail_mb,
+        budget_mb=budget_mb,
+        disposition=disposition,
+    )

backend/core/ouroboros/governance/autonomy/subagent_scheduler.py

@@ -35,6 +35,11 @@
 from backend.core.ouroboros.governance.saga.merge_coordinator import MergeCoordinator
 from backend.core.ouroboros.governance.saga.saga_types import FileOp, PatchedFile, RepoPatch
 from backend.core.ouroboros.governance.test_runner import BlockedPathError
+# Module-level (not lazy like _consult_memory_gate's in-method import) so
+# tests can monkeypatch this symbol as a seam; the governor path below uses it.
+from backend.core.ouroboros.governance.memory_pressure_gate import (
+    get_default_gate,
+)
 
 logger = logging.getLogger("Ouroboros.SubagentScheduler")
 
@@ -497,6 +502,9 @@ async def _run_graph(self, graph_id: str) -> None:
                 # Zero work loss. Gate-disabled → pass-through (no clamp). Every
                 # decision is logged + SSE-published (allow / clamp / disabled /
                 # probe_fail) so operators have a §8 audit trail.
+                # Bound before the guard so the Unit D block below can safely
+                # read it even if a future edit repopulates `selected`.
+                decision = None
                 if selected:
                     decision = self._consult_memory_gate(
                         len(selected), graph_id=graph_id,
@@ -506,6 +514,23 @@ async def _run_graph(self, graph_id: str) -> None:
                         selected = list(selected[:decision.n_allowed])
                         deferred = sorted(list(deferred) + overflow)
 
+                # Unit D — worktree-RAM-budget governor composes on top of
+                # the fan-out gate (strictest-wins). Same zero-work-loss
+                # defer-overflow mechanism; disabled/probe-fail -> no clamp.
+                if selected:
+                    level_cap = (
+                        decision.n_allowed
+                        if decision is not None
+                        else len(selected)
+                    )
+                    gov = self._consult_memory_governor(
+                        len(selected), graph_id=graph_id, level_cap=level_cap,
+                    )
+                    if gov is not None and gov.n_allowed < len(selected):
+                        overflow = list(selected[gov.n_allowed:])
+                        selected = list(selected[:gov.n_allowed])
+                        deferred = sorted(list(deferred) + overflow)
+
                 if not selected:
                     state = self._update_state(
                         state,
@@ -755,6 +780,60 @@ def _consult_memory_gate(
 
         return decision
 
+    def _consult_memory_governor(
+        self,
+        n_requested: int,
+        *,
+        graph_id: str,
+        level_cap: int,
+    ) -> Optional[Any]:
+        """Unit D — worktree-RAM-budget clamp composed on top of the
+        Slice 5 Arc B fan-out gate.
+
+        Returns a ``GovernorDecision`` (or ``None`` when disabled / on
+        any probe failure — the scheduler must never break on the
+        governor). ``level_cap`` is the allowance already granted by
+        the free-%-based fan-out gate; the governor takes the strictest
+        of that and the absolute RAM budget.
+        """
+        from backend.core.ouroboros.governance.autonomy.l3_memory_governor import (
+            compute_worktree_cap,
+            governor_enabled,
+            worktree_ram_budget_mb,
+        )
+
+        if not governor_enabled():
+            return None
+        try:
+            gate = get_default_gate()
+            probe = gate.probe()
+            avail_mb = float(probe.available_bytes) / (1024.0 * 1024.0)
+        except Exception:  # noqa: BLE001 — governor must not break scheduler
+            logger.debug(
+                "[SubagentScheduler] memory governor probe failed "
+                "(non-fatal)", exc_info=True,
+            )
+            return None
+
+        decision = compute_worktree_cap(
+            requested=n_requested,
+            avail_mb=avail_mb,
+            budget_mb=worktree_ram_budget_mb(),
+            level_cap=level_cap,
+        )
+        log_fn = (
+            logger.warning if decision.disposition == "clamp" else logger.info
+        )
+        log_fn(
+            "[SubagentScheduler] ram_governor: graph=%s disposition=%s "
+            "requested=%d allowed=%d ram_cap=%d level_cap=%d avail_mb=%.0f "
+            "budget_mb=%d",
+            graph_id, decision.disposition, decision.requested,
+            decision.n_allowed, decision.ram_cap, decision.level_cap,
+            decision.avail_mb, decision.budget_mb,
+        )
+        return decision
+
     def _select_ready_batch(
         self,
         graph: ExecutionGraph,

backend/core/ouroboros/governance/governed_loop_service.py

@@ -1139,6 +1139,9 @@ def __init__(
         self._graph_coalescer: Optional[Any] = None
         self._advanced_autonomy: Optional[Any] = None
         self._mcp_client: Optional[Any] = None  # Phase A: GovernanceMCPClient, wired in start()
+        # Evidence Rail (Unit A/C) — durable shadow-vs-legacy ledger.
+        # Initialized in _build_components when store_enabled(); None = inert.
+        self._shadow_store_ref: Optional[Any] = None
 
         # Compute-class admission gate (set externally after fetching /v1/capability;
         # None = gate disabled — backward-compatible default)
@@ -2089,6 +2092,15 @@ async def stop(self) -> None:
             except Exception:
                 pass
 
+        # Evidence Rail — close the durable telemetry store
+        _ssr = getattr(self, "_shadow_store_ref", None)
+        if _ssr is not None:
+            try:
+                await _ssr.aclose()
+            except Exception:  # noqa: BLE001
+                pass
+            self._shadow_store_ref = None
+
         # Detach from stack
         self._detach_from_stack()
         self._state = ServiceState.INACTIVE
@@ -4870,6 +4882,39 @@ def _on_streaming_token(token: str) -> None:
         except Exception as exc:
             logger.debug("[GLS] SubagentOrchestrator skipped: %s", exc)
 
+        # ---- Evidence Rail (Unit A/C) — shadow-vs-legacy telemetry store + graduation gate ----
+        # Inert + byte-identical when JARVIS_SHADOW_TELEMETRY_STORE_ENABLED=false:
+        # store_enabled() is False → no store constructed → set_shadow_rail never
+        # called → orchestrator's _shadow_store stays None → FSM byte-identical.
+        try:
+            from backend.core.ouroboros.governance.shadow_telemetry_store import (
+                ShadowTelemetryStore, store_enabled,
+            )
+            from backend.core.ouroboros.governance.shadow_graduation_gate import (
+                ShadowGraduationGate, build_rail_evaluator, gate_enabled,
+            )
+            if store_enabled():
+                _shadow_store = ShadowTelemetryStore(
+                    evaluator=build_rail_evaluator(),
+                )
+                await _shadow_store.start()
+                self._shadow_store_ref = _shadow_store
+                _shadow_gate = (
+                    ShadowGraduationGate(store=_shadow_store)
+                    if gate_enabled() else None
+                )
+                if self._orchestrator is not None and hasattr(
+                    self._orchestrator, "set_shadow_rail"
+                ):
+                    self._orchestrator.set_shadow_rail(
+                        _shadow_store, _shadow_gate,
+                    )
+        except Exception:  # noqa: BLE001 — rail attach must never break boot
+            logger.warning(
+                "[GovernedLoopService] shadow rail attach failed (non-fatal)",
+                exc_info=True,
+            )
+
         # ---- Wire Self-Critique Engine (Phase 3a — post-VERIFY quality signal) ----
         # Cheap DW critique over the applied diff against the original goal.
         # Poor ratings become FEEDBACK memories; excellent ratings reinforce

backend/core/ouroboros/governance/ide_observability_stream.py

@@ -197,6 +197,11 @@
 EVENT_TYPE_GOVERNOR_EMERGENCY_BRAKE = "governor_emergency_brake"
 EVENT_TYPE_MEMORY_PRESSURE_CHANGED = "memory_pressure_changed"
 
+# Evidence Rail Unit C — emitted by the circuit breaker when an authoritative
+# subagent trips back to legacy (degradation event). Payload carries agent +
+# trip_reason + pressure_level so IDE consumers can render a "degraded" banner.
+EVENT_TYPE_AGENT_DEGRADATION = "agent_degradation"
+
 # Dynamic Risk-State Convergence Engine (Slice 98 Phase 2) — emitted
 # ONLY on a convergence-band TRANSITION (NORMAL/ELEVATED/PARANOIA).
 # Payload carries the ConvergenceVerdict.to_dict() snapshot. Keyed by
@@ -1535,6 +1540,9 @@
                                                   # in-process on_trip callback drives harness
                                                   # graceful shutdown, this SSE is the IDE
                                                   # observability surface)
+    EVENT_TYPE_AGENT_DEGRADATION,                 # Evidence Rail Unit C — authoritative
+                                                  # subagent tripped back to legacy; payload:
+                                                  # agent + trip_reason + pressure_level.
 })
 
 
@@ -4441,6 +4449,26 @@ def publish_review_branch_event(
         return None
 
 
+# ---------------------------------------------------------------------------
+# Evidence Rail Unit C — AGENT_DEGRADATION SSE publisher
+# ---------------------------------------------------------------------------
+
+
+def publish_agent_degradation_event(
+    *, broker: Any, agent: str, op_id: str, trip_reason: str,
+    pressure_level: str,
+) -> None:
+    """Best-effort AGENT_DEGRADATION frame. Never raises/blocks."""
+    try:
+        broker.publish(
+            EVENT_TYPE_AGENT_DEGRADATION, op_id,
+            {"agent": agent, "trip_reason": trip_reason,
+             "pressure_level": pressure_level},
+        )
+    except Exception:  # noqa: BLE001
+        pass
+
+
 # ---------------------------------------------------------------------------
 # FlagRegistry self-registration (auto-discovered by §33.3 walker)
 # ---------------------------------------------------------------------------