Wire post-swap monitor into live 2-policy serving (rollback was dead code)

src/tether/cli.py

@@ -2614,6 +2614,19 @@ def serve(
              "queueing. Overload records a shadow_queue_full result instead of "
              "blocking live /act responses.",
     ),
+    rollback_sensitivity: str = typer.Option(
+        "normal", "--rollback-sensitivity",
+        help="2-policy mode: consecutive post-swap-monitor trips required "
+             "before auto-rollback fires. aggressive=1 | normal=2 | tolerant=3. "
+             "Single-strike is too noisy; three-strike costs too much trust.",
+    ),
+    post_swap_baseline_clamp_rate: float = typer.Option(
+        -1.0, "--post-swap-baseline-clamp-rate",
+        help="2-policy mode: pre-swap safety-clamp rate the post-swap monitor's "
+             "T1 trip signal compares against, measured from the deployment "
+             "being replaced. Omit (-1) to default to 0.0, which makes T1 "
+             "maximally sensitive -- a warning is logged in that case.",
+    ),
     no_rtc: bool = typer.Option(
         False, "--no-rtc",
         help="Disable RTC even when --rtc was previously enabled. REQUIRED in "
@@ -3050,6 +3063,12 @@ def serve(
         policy_b_export_dir=policy_b or None,
         policy_split_a_percent=split,
         policy_crash_threshold=max_consecutive_crashes,
+        rollback_sensitivity=rollback_sensitivity,
+        post_swap_baseline_clamp_rate=(
+            post_swap_baseline_clamp_rate
+            if post_swap_baseline_clamp_rate >= 0
+            else None
+        ),
         shadow_policy=shadow_policy or None,
         shadow_sample=shadow_sample,
         shadow_queue_size=shadow_queue_size,

src/tether/runtime/policy_crash_tracker.py

@@ -88,7 +88,7 @@ class PolicyCrashTracker:
         # legacy single-counter behavior)
     """
 
-    __slots__ = ("_slots", "_threshold", "_counts")
+    __slots__ = ("_slots", "_threshold", "_counts", "_forced_drain", "_forced_reason")
 
     def __init__(self, *, slots: tuple[str, ...], threshold: int):
         if not slots:
@@ -100,6 +100,14 @@ def __init__(self, *, slots: tuple[str, ...], threshold: int):
         self._slots: tuple[str, ...] = tuple(slots)
         self._threshold = int(threshold)
         self._counts: dict[str, int] = {s: 0 for s in slots}
+        # External override (e.g. PostSwapMonitor tripping a rollback).
+        # Distinct from crash-count-derived drains so operators reading
+        # verdict.reason can tell "this slot is crashing" apart from
+        # "something outside raw request errors said drain this slot" --
+        # conflating the two into fake crash-count increments would corrupt
+        # the crash telemetry those counters are also used for.
+        self._forced_drain: str | None = None
+        self._forced_reason: str | None = None
 
     @property
     def slots(self) -> tuple[str, ...]:
@@ -142,27 +150,74 @@ def record_clean(self, *, slot: str) -> None:
     def reset(self, *, slot: str | None = None) -> None:
         """Reset one slot's counter, or ALL slots when slot=None.
         Used after a manual operator intervention or after a successful
-        rollback completes."""
+        rollback completes. Also clears any forced-drain override --
+        callers that want the override to persist across a counter reset
+        must re-call force_drain() after."""
         if slot is None:
             for s in self._slots:
                 self._counts[s] = 0
+            self.clear_forced_drain()
             return
         if slot not in self._counts:
             raise KeyError(
                 f"slot {slot!r} not in tracker; known slots: {self._slots}"
             )
         self._counts[slot] = 0
+        if self._forced_drain == slot:
+            self.clear_forced_drain()
+
+    def force_drain(self, *, slot: str, reason: str) -> None:
+        """External override: force verdict() to report drain-<slot>
+        regardless of crash counts, until clear_forced_drain() is called.
+
+        Used by PostSwapMonitor-triggered rollback -- a monitor trip isn't
+        a raw predict() exception, so it can't go through record_crash()
+        without corrupting the crash-count telemetry those counters also
+        drive. This is a distinct signal, surfaced as its own reason.
+        """
+        if slot not in self._counts:
+            raise KeyError(
+                f"slot {slot!r} not in tracker; known slots: {self._slots}"
+            )
+        self._forced_drain = slot
+        self._forced_reason = reason
+        logger.warning(
+            "policy_crash_tracker.forced_drain slot=%s reason=%s",
+            slot, reason,
+        )
+
+    def clear_forced_drain(self) -> None:
+        """Clear any forced-drain override (e.g. after a successful
+        rollback + operator sign-off, or a fresh swap starting a new
+        window)."""
+        self._forced_drain = None
+        self._forced_reason = None
+
+    @property
+    def forced_drain_slot(self) -> str | None:
+        return self._forced_drain
 
     def verdict(self) -> CrashTrackerVerdict:
         """Compute the current verdict from per-slot counters.
 
+        A force_drain() override takes priority over crash-count-derived
+        verdicts -- checked first, unconditionally.
+
         Single-policy mode (one slot): exceeds threshold -> degraded.
         2-policy mode (two slots a + b):
             - Both exceed -> degraded.
             - a exceeds, b healthy -> drain-a (route 100% to b).
             - b exceeds, a healthy -> drain-b (route 100% to a).
             - Neither exceeds -> healthy.
         """
+        if self._forced_drain is not None:
+            snapshot = dict(self._counts)
+            return CrashTrackerVerdict(
+                verdict=f"drain-{self._forced_drain}",  # type: ignore[arg-type]
+                crash_counts=snapshot,
+                threshold=self._threshold,
+                reason=self._forced_reason or "forced drain (no reason given)",
+            )
         snapshot = dict(self._counts)
         exceeders = [s for s, c in snapshot.items() if c >= self._threshold]
 

src/tether/runtime/server.py

@@ -1789,6 +1789,14 @@ def create_app(
     policy_b_export_dir: str | None = None,  # 2-policy mode: path to slot B
     policy_split_a_percent: int = 50,  # % traffic to slot A in [0, 100]
     policy_crash_threshold: int = 5,  # per-slot circuit-breaker threshold
+    # Post-swap monitor (pro/post_swap_monitor.py + pro/rollback.py). Only
+    # active in 2-policy mode. aggressive=1 | normal=2 | tolerant=3
+    # consecutive trips before auto-rollback fires.
+    rollback_sensitivity: str = "normal",
+    # Pre-swap safety-clamp rate the T1 trip signal compares against
+    # (measured from the deployment being replaced). None -> 0.0, which
+    # makes T1 maximally sensitive; a loud warning is logged in that case.
+    post_swap_baseline_clamp_rate: float | None = None,
     shadow_policy: str | None = None,  # shadow mode: mirror sampled traffic to this export
     shadow_sample: float = 1.0,  # fraction of /act traffic mirrored to shadow_policy
     shadow_queue_size: int = 32,  # bounded pending shadow requests; 0 disables queueing
@@ -2685,6 +2693,8 @@ async def lifespan(app):
         # dispatcher instead of the single-server PolicyRuntime.
         # ---------------------------------------------------------------
         server.two_policy_state = None  # type: ignore[attr-defined]
+        server.post_swap_monitor = None  # type: ignore[attr-defined]
+        server.rollback_handler = None  # type: ignore[attr-defined]
         if policy_b_export_dir:
             from tether.runtime.two_policy_setup import setup_two_policy_serving
 
@@ -2812,6 +2822,79 @@ def _shape_key(_req):
                     two_state.policy_a.model_version,
                     two_state.policy_b.model_version,
                 )
+
+                # ---------------------------------------------------------
+                # Post-swap monitor + rollback wiring. Per ADR
+                # 2026-04-25-self-distilling-serve-architecture decision
+                # #4 + #7: the 9-gate eval (eval_gate.py) gates promotion,
+                # but production traffic can surface regressions the
+                # held-out eval missed. PostSwapMonitor watches the first
+                # 24h/500 episodes after a swap; a trip auto-drains the
+                # regressed slot via TwoPolicyDispatcher.force_drain (the
+                # same mechanism an organic crash-count drain already
+                # uses) and RollbackHandler records the audit trail.
+                #
+                # Convention: slot A is the newly-promoted candidate (see
+                # _two_policy_server_factory above -- "the first server
+                # ... is server A"). The monitor watches A; a trip drains
+                # A and routes 100% to B, the warm previous version.
+                # ---------------------------------------------------------
+                from tether.pro.post_swap_monitor import MonitorConfig, PostSwapMonitor
+                from tether.pro.rollback import RollbackHandler
+
+                _monitored_slot = "a"
+                _fallback_slot = "b"
+
+                if post_swap_baseline_clamp_rate is None:
+                    logger.warning(
+                        "post_swap_monitor.no_baseline_supplied -- using 0.0, "
+                        "which makes the T1 safety-clamp trip signal maximally "
+                        "sensitive (any clamping at all crosses 2x*0=0). Pass "
+                        "--post-swap-baseline-clamp-rate measured from the "
+                        "pre-swap deployment's traffic for accurate drift "
+                        "detection."
+                    )
+                _baseline_clamp = float(post_swap_baseline_clamp_rate or 0.0)
+
+                server.post_swap_monitor = PostSwapMonitor(  # type: ignore[attr-defined]
+                    config=MonitorConfig(sensitivity=rollback_sensitivity),
+                )
+                server.post_swap_monitor.start_window(
+                    baseline_clamp_rate=_baseline_clamp,
+                )
+
+                def _rollback_audit_writer(record: dict) -> None:
+                    import json as _json
+                    _audit_path = (
+                        Path(record_dir) / "rollback_audit.jsonl"
+                        if record_dir
+                        else Path.home() / ".tether" / "rollback_audit.jsonl"
+                    )
+                    _audit_path.parent.mkdir(parents=True, exist_ok=True)
+                    with open(_audit_path, "a") as _f:
+                        _f.write(_json.dumps(record) + "\n")
+
+                def _rollback_router_swap_fn(target_slot: str) -> None:
+                    # RollbackHandler calls this with the slot to swap TO.
+                    # Draining is expressed as "route 0% to the OTHER slot".
+                    _drain = _fallback_slot if target_slot == _monitored_slot else _monitored_slot
+                    two_state.dispatcher.force_drain(
+                        slot=_drain, reason="post_swap_monitor_rollback",
+                    )
+
+                server.rollback_handler = RollbackHandler(  # type: ignore[attr-defined]
+                    router_swap_fn=_rollback_router_swap_fn,
+                    active_slot_getter=lambda: _monitored_slot,
+                    audit_writer=_rollback_audit_writer,
+                )
+                logger.info(
+                    "post_swap_monitor.window_started monitored_slot=%s "
+                    "fallback_slot=%s sensitivity=%s baseline_clamp_rate=%.4f "
+                    "window_hours=%d window_episodes=%d",
+                    _monitored_slot, _fallback_slot, rollback_sensitivity,
+                    _baseline_clamp, MonitorConfig().window_duration_hours,
+                    MonitorConfig().window_episode_count,
+                )
             except Exception as exc:
                 logger.error(
                     "two_policy.setup_failed -- falling back to single-policy "
@@ -3193,6 +3276,70 @@ async def act(request: PredictRequest, _auth: None = Depends(_require_api_key)):
                     # Stash routing decision so the response builder + headers
                     # + recorder can pick it up below.
                     _two_routing_decision = _routing
+
+                    # Post-swap monitor: only record episodes routed to the
+                    # monitored (newly-promoted) slot -- see wiring comment
+                    # at two_state setup above. A trip auto-drains that slot
+                    # and writes a rollback audit record.
+                    _monitor = getattr(server, "post_swap_monitor", None)
+                    if _monitor is not None and _routing.slot == "a" and _monitor.is_window_open:
+                        _guard_summary = (
+                            result.get("guard_summary")
+                            if isinstance(result, dict)
+                            else None
+                        )
+                        _clamp_count = (
+                            _guard_summary.get("clamp_count", 0)
+                            if isinstance(_guard_summary, dict)
+                            else 0
+                        )
+                        _safety_violations = (
+                            result.get("safety_violations", 0)
+                            if isinstance(result, dict)
+                            else 0
+                        )
+                        # T2 (action cos-similarity to the previous model) is
+                        # not wired live: computing it would mean running
+                        # both policies per request, doubling inference cost.
+                        # PostSwapMonitor treats a None sample as "T2 not
+                        # checked this episode" (documented in its docstring)
+                        # rather than silently faking a passing value.
+                        _monitor.record_episode(
+                            safety_clamp_count=_clamp_count,
+                            cos_to_previous_model=None,
+                            webhook_violations_count=_safety_violations,
+                        )
+                        _trip = _monitor.should_rollback()
+                        if _trip.should_rollback:
+                            _handler = getattr(server, "rollback_handler", None)
+                            if _handler is not None:
+                                _outcome = _handler.execute(
+                                    trigger="auto", reason=_trip.reason,
+                                )
+                                if _outcome.succeeded:
+                                    from tether.observability import inc_model_swap
+                                    try:
+                                        _emb = getattr(
+                                            getattr(server, "embodiment_config", None),
+                                            "embodiment", None,
+                                        ) or "custom"
+                                        inc_model_swap(
+                                            embodiment=_emb,
+                                            from_model=_outcome.from_slot,
+                                            to_model=_outcome.to_slot,
+                                        )
+                                    except Exception:  # noqa: BLE001
+                                        pass  # metric emission never blocks the response path
+                                logger.error(
+                                    "post_swap_monitor.rollback_fired reason=%s "
+                                    "consecutive=%d required=%d measured=%.4f "
+                                    "threshold=%.4f rollback_succeeded=%s "
+                                    "audit_id=%s",
+                                    _trip.reason, _trip.consecutive_trips,
+                                    _trip.required_trips, _trip.measured,
+                                    _trip.threshold, _outcome.succeeded,
+                                    _outcome.audit_id,
+                                )
                 else:
                     _two_routing_decision = None
                     # Single-policy mode: route through the existing

src/tether/runtime/two_policy_dispatcher.py

@@ -227,6 +227,23 @@ def reset_crash_counters(self, *, slot: str | None = None) -> None:
         affected policy."""
         self._tracker.reset(slot=slot)
 
+    def force_drain(self, *, slot: str, reason: str) -> None:
+        """External override: route 0% of traffic to `slot` starting on
+        the next predict() call, regardless of crash counts. Used by a
+        PostSwapMonitor trip (see pro/post_swap_monitor.py + pro/rollback.py)
+        -- a monitor trip isn't a raw predict() exception, so it's kept
+        distinct from the crash-count-derived drain path."""
+        self._tracker.force_drain(slot=slot, reason=reason)
+
+    def clear_forced_drain(self) -> None:
+        """Clear a force_drain() override (e.g. after operator sign-off
+        that the rolled-back-from slot is safe to re-enable)."""
+        self._tracker.clear_forced_drain()
+
+    @property
+    def forced_drain_slot(self) -> str | None:
+        return self._tracker.forced_drain_slot
+
 
 def _default_is_error_response(result: Any) -> bool:
     """Default error detector: result is a dict with 'error' key.