diff --git a/shared/policy_decide.py b/shared/policy_decide.py index 57abb0cb1..ae8e26017 100644 --- a/shared/policy_decide.py +++ b/shared/policy_decide.py @@ -30,6 +30,7 @@ import os import re import shlex +import sys from dataclasses import dataclass from datetime import UTC, datetime from pathlib import Path @@ -851,6 +852,288 @@ def evaluate_shadow_clean( } +# --- The auto-promotion state machine (reform fix: kill the manual 3b cutover) - +# +# ``evaluate_shadow_clean`` answers "is the shadow-week clean?" but nothing acted on +# a YES — ``3b-cutover`` was a MANUAL cliff a human had to step off by a deadline. A +# clean predicate that never promotes itself is the same freeze-blocks-thaw bug one +# layer up. This is the missing actuator: a reversible, version-stamped ladder +# +# shadow ──clean──▶ canary ──clean ≥24h──▶ authoritative (─not-clean──▶ shadow) +# +# that the promote timer advances one rung per clean tick and ROLLS BACK to shadow +# the instant the predicate fails (master design §4.1: "advisory-canary, reversible, +# never a hard cliff"). It only ever advances a recorded POSTURE — wiring that +# posture into the live gate verdict remains a separate, gated step (§4.1: the canary +# logs both decisions "before becoming the live verdict"). Per the permanent-canary +# discipline, any change to ``policy_decide`` (a new ``POLICY_DECIDE_FN_VERSION``) +# resets the ladder to shadow so the new logic must re-prove itself from scratch. + +PROMOTION_SHADOW = "shadow" +PROMOTION_CANARY = "canary" +PROMOTION_AUTHORITATIVE = "authoritative" +_PROMOTION_STATES = frozenset({PROMOTION_SHADOW, PROMOTION_CANARY, PROMOTION_AUTHORITATIVE}) + +#: The canary dwells in dual-decision mode this long before it may go authoritative. +DEFAULT_CANARY_WINDOW_SECONDS = 24 * 3600 + +#: Current promotion posture (a projection); the audit trail is the ledger beside it. +DEFAULT_PROMOTION_STATE = Path(os.path.expanduser("~/.cache/hapax/policy-decide-promotion.json")) +DEFAULT_PROMOTION_LEDGER = Path(os.path.expanduser("~/.cache/hapax/policy-decide-promotion.jsonl")) + +_EPOCH = datetime(1970, 1, 1, tzinfo=UTC) + + +def _iso(when: datetime) -> str: + return when.astimezone(UTC).strftime("%Y-%m-%dT%H:%M:%SZ") + + +@dataclass(frozen=True) +class PromotionState: + """The current promotion posture: which rung, stamped with the version that earned it.""" + + state: str + policy_version: str + entered_state_at: datetime + updated_at: datetime + + +@dataclass(frozen=True) +class PromotionDecision: + """One transition verdict from ``decide_promotion`` — pure, carries its own version stamp.""" + + from_state: str + to_state: str + changed: bool + reason: str + clean: bool + policy_version: str + dwell_seconds: float + now: datetime + + def next_state(self, current: PromotionState) -> PromotionState: + """The posture to persist after this decision (entry clock resets only on a change).""" + entered = self.now if self.changed else current.entered_state_at + return PromotionState( + state=self.to_state, + policy_version=self.policy_version, + entered_state_at=entered, + updated_at=self.now, + ) + + +def decide_promotion( + current: PromotionState, + verdict: dict[str, object], + *, + policy_version: str, + now: datetime, + canary_window_seconds: float = DEFAULT_CANARY_WINDOW_SECONDS, +) -> PromotionDecision: + """Pure ladder transition. ``verdict`` is an ``evaluate_shadow_clean`` result. + + Advances one rung per clean tick, requires a ``canary_window_seconds`` dwell before + canary→authoritative, and rolls back to shadow on any not-clean verdict or any + ``policy_version`` change (the permanent-canary reset). Never raises; never skips a + rung; never hard-cuts straight to authoritative. + """ + clean = bool(verdict.get("clean")) + frm = current.state + dwell = max(0.0, (now - current.entered_state_at).total_seconds()) + + def decision(to_state: str, changed: bool, reason: str) -> PromotionDecision: + return PromotionDecision( + from_state=frm, + to_state=to_state, + changed=changed, + reason=reason, + clean=clean, + policy_version=policy_version, + dwell_seconds=dwell, + now=now, + ) + + # Permanent-canary discipline: a new policy_decide version must re-prove from shadow. + if current.policy_version != policy_version: + return decision( + PROMOTION_SHADOW, + True, + f"policy_version {current.policy_version}→{policy_version}: " + "re-entering shadow (permanent-canary discipline)", + ) + + if frm == PROMOTION_SHADOW: + if clean: + return decision( + PROMOTION_CANARY, True, "shadow-week clean → canary (dual-decision window opens)" + ) + return decision(PROMOTION_SHADOW, False, _hold_reason(verdict)) + + if frm == PROMOTION_CANARY: + if not clean: + return decision( + PROMOTION_SHADOW, + True, + f"canary regression ({_hold_reason(verdict)}) → rollback to shadow", + ) + if dwell >= canary_window_seconds: + return decision( + PROMOTION_AUTHORITATIVE, + True, + f"canary clean ≥{canary_window_seconds / 3600:.0f}h → authoritative-ready", + ) + return decision( + PROMOTION_CANARY, + False, + f"canary clean, {dwell / 3600:.1f}h/{canary_window_seconds / 3600:.0f}h elapsed", + ) + + if frm == PROMOTION_AUTHORITATIVE: + if not clean: + return decision( + PROMOTION_SHADOW, + True, + f"authoritative regression ({_hold_reason(verdict)}) → rollback to shadow", + ) + return decision(PROMOTION_AUTHORITATIVE, False, "authoritative steady (shadow-week clean)") + + return decision(PROMOTION_SHADOW, True, f"unknown promotion state '{frm}' → reset to shadow") + + +def _hold_reason(verdict: dict[str, object]) -> str: + reasons = verdict.get("reasons") + if isinstance(reasons, list) and reasons: + return "; ".join(str(r) for r in reasons) + return "shadow-week not clean" + + +def load_promotion_state( + state_path: str | os.PathLike[str] = DEFAULT_PROMOTION_STATE, +) -> PromotionState: + """Read the persisted posture; an absent/garbage file defaults to a fresh shadow rung.""" + fresh = PromotionState(PROMOTION_SHADOW, POLICY_DECIDE_FN_VERSION, _EPOCH, _EPOCH) + try: + data = json.loads(Path(state_path).read_text(encoding="utf-8")) + except (FileNotFoundError, ValueError, OSError): + return fresh + if not isinstance(data, dict): + return fresh + state = str(data.get("state") or PROMOTION_SHADOW) + if state not in _PROMOTION_STATES: + state = PROMOTION_SHADOW + entered = _parse_decision_ts(data.get("entered_state_at")) or _EPOCH + updated = _parse_decision_ts(data.get("updated_at")) or entered + return PromotionState( + state=state, + policy_version=str(data.get("policy_version") or POLICY_DECIDE_FN_VERSION), + entered_state_at=entered, + updated_at=updated, + ) + + +def save_promotion_state( + state: PromotionState, state_path: str | os.PathLike[str] = DEFAULT_PROMOTION_STATE +) -> None: + """Persist the posture atomically (tmp + replace). Best-effort; never raises (advisory).""" + try: + path = Path(state_path) + path.parent.mkdir(parents=True, exist_ok=True) + tmp = path.with_suffix(path.suffix + ".promote-tmp") + tmp.write_text( + json.dumps( + { + "state": state.state, + "policy_version": state.policy_version, + "entered_state_at": _iso(state.entered_state_at), + "updated_at": _iso(state.updated_at), + } + ) + + "\n", + encoding="utf-8", + ) + tmp.replace(path) + except OSError: + pass + + +def append_promotion_ledger( + decision: PromotionDecision, + next_state: PromotionState, + *, + ledger_path: str | os.PathLike[str] = DEFAULT_PROMOTION_LEDGER, +) -> None: + """Append a transition row to the audit ledger. Best-effort; never raises (advisory).""" + try: + path = Path(ledger_path) + path.parent.mkdir(parents=True, exist_ok=True) + row = { + "ts": _iso(decision.now), + "from_state": decision.from_state, + "to_state": decision.to_state, + "changed": decision.changed, + "reason": decision.reason, + "clean": decision.clean, + "policy_version": decision.policy_version, + "dwell_seconds": round(decision.dwell_seconds, 1), + "entered_state_at": _iso(next_state.entered_state_at), + } + with path.open("a", encoding="utf-8") as handle: + handle.write(json.dumps(row) + "\n") + except OSError: + pass + + +def run_promotion_cycle( + *, + decision_log_path: str | os.PathLike[str] = DEFAULT_DECISION_LOG, + shadow_ledger_path: str | os.PathLike[str] = DEFAULT_SHADOW_LEDGER, + state_path: str | os.PathLike[str] = DEFAULT_PROMOTION_STATE, + ledger_path: str | os.PathLike[str] = DEFAULT_PROMOTION_LEDGER, + now: datetime | None = None, + policy_version: str = POLICY_DECIDE_FN_VERSION, + canary_window_seconds: float = DEFAULT_CANARY_WINDOW_SECONDS, + min_days: float = 7.0, + min_decisions: int = 200, + replay: bool = False, +) -> dict[str, object]: + """Evaluate the shadow-week and advance/roll-back the promotion ladder one rung. + + Optionally refreshes the divergence ledger first (``replay=True``) so coverage and + asymmetry are read consistently. Persists the new posture every tick (so ``updated_at`` + tracks liveness) and appends to the audit ledger only on an actual transition. + Advisory: it advances a posture, never the live gate verdict. Never raises. + """ + now = now or datetime.now(UTC) + if replay: + replay_decision_log(decision_log_path, shadow_ledger_path) + verdict = evaluate_shadow_clean( + decision_log_path, shadow_ledger_path, min_days=min_days, min_decisions=min_decisions + ) + current = load_promotion_state(state_path) + decision = decide_promotion( + current, + verdict, + policy_version=policy_version, + now=now, + canary_window_seconds=canary_window_seconds, + ) + nxt = decision.next_state(current) + save_promotion_state(nxt, state_path) + if decision.changed: + append_promotion_ledger(decision, nxt, ledger_path=ledger_path) + return { + "ts": _iso(now), + "from_state": decision.from_state, + "to_state": decision.to_state, + "changed": decision.changed, + "reason": decision.reason, + "clean": decision.clean, + "policy_version": decision.policy_version, + "dwell_seconds": round(decision.dwell_seconds, 1), + "verdict": verdict, + } + + # --- Advisory shadow CLI ------------------------------------------------------ @@ -919,5 +1202,44 @@ def main(argv: list[str] | None = None) -> int: return 0 # advisory: never enforces during the shadow window +def promote_main(argv: list[str] | None = None) -> int: + """Advance the promotion ladder one rung — the reform 3b AUTO-PROMOTER entrypoint. + + Wired to ``policy-decide-promote.timer`` via ``python -m shared.policy_decide promote``. + Each tick evaluates the shadow-week and advances/rolls-back the reversible, + version-stamped posture — the missing actuator that turns a clean predicate into an + actual promotion instead of the manual 3b-cutover cliff. ``--replay`` refreshes the + divergence ledger from the decision log first so coverage + asymmetry read + consistently. ADVISORY ONLY: it moves a recorded POSTURE, never the live gate + verdict, and always exits 0. + """ + parser = argparse.ArgumentParser(prog="policy_decide promote") + parser.add_argument("--decision-log", default=str(DEFAULT_DECISION_LOG)) + parser.add_argument("--ledger", default=str(DEFAULT_SHADOW_LEDGER)) + parser.add_argument("--state", default=str(DEFAULT_PROMOTION_STATE)) + parser.add_argument("--promotion-ledger", default=str(DEFAULT_PROMOTION_LEDGER)) + parser.add_argument("--min-days", type=float, default=7.0) + parser.add_argument("--min-decisions", type=int, default=200) + parser.add_argument( + "--replay", + action="store_true", + help="refresh the divergence ledger from the decision log before evaluating", + ) + args = parser.parse_args(argv) + result = run_promotion_cycle( + decision_log_path=args.decision_log, + shadow_ledger_path=args.ledger, + state_path=args.state, + ledger_path=args.promotion_ledger, + min_days=args.min_days, + min_decisions=args.min_decisions, + replay=args.replay, + ) + print(json.dumps(result, indent=2)) + return 0 # advisory: advances a posture, never the live gate verdict + + if __name__ == "__main__": + if len(sys.argv) > 1 and sys.argv[1] == "promote": + raise SystemExit(promote_main(sys.argv[2:])) raise SystemExit(main()) diff --git a/systemd/units/policy-decide-promote.service b/systemd/units/policy-decide-promote.service new file mode 100644 index 000000000..a050fb7fd --- /dev/null +++ b/systemd/units/policy-decide-promote.service @@ -0,0 +1,21 @@ +[Unit] +Description=Reform 3b AUTO-PROMOTER — advance the policy_decide promotion ladder when the shadow-week is clean +Documentation=file:///home/hapax/Documents/Personal/30-areas/hapax/reform-execution-manifest.yaml +After=network.target +OnFailure=notify-failure@%n.service + +[Service] +Type=oneshot +# Advisory auto-promoter: the missing actuator on top of the shadow producer + +# evaluator. Each tick it refreshes the divergence ledger (--replay), evaluates the +# shadow-week, and advances/rolls-back the reversible, version-stamped promotion +# posture one rung (shadow→canary→authoritative). Runs from the canonical worktree +# (the deploy target). Never enforces — it moves a recorded posture, exits 0. +ExecStart=/home/hapax/.local/bin/uv run --directory /home/hapax/projects/hapax-council python -m shared.policy_decide promote --replay +Environment=PATH=/home/hapax/.local/bin:/home/hapax/.cargo/bin:/usr/local/bin:/usr/bin:/bin +Environment=HOME=/home/hapax +Nice=10 +MemoryMax=1G +StandardOutput=journal +StandardError=journal +SyslogIdentifier=policy-decide-promote diff --git a/systemd/units/policy-decide-promote.timer b/systemd/units/policy-decide-promote.timer new file mode 100644 index 000000000..d2d6f7ece --- /dev/null +++ b/systemd/units/policy-decide-promote.timer @@ -0,0 +1,14 @@ +[Unit] +Description=Advance the policy_decide promotion ladder hourly when the shadow-week is clean (reform 3b auto-promoter) + +[Timer] +# Hourly is ample: the ladder advances at most one rung per tick and the canary +# dwell (24h) is the real rate limiter. First run trails the producer's 10min boot +# replay so it evaluates fresh evidence. +OnBootSec=20min +OnUnitActiveSec=1h +RandomizedDelaySec=120 +Persistent=true + +[Install] +WantedBy=timers.target diff --git a/tests/test_policy_decide_promotion.py b/tests/test_policy_decide_promotion.py new file mode 100644 index 000000000..bb36c551d --- /dev/null +++ b/tests/test_policy_decide_promotion.py @@ -0,0 +1,311 @@ +"""Tests for the policy_decide AUTO-PROMOTION state machine (reform 3b-cutover). + +Phase 3b shipped the shadow PRODUCER (``replay_decision_log``) and the cutover +EVALUATOR (``evaluate_shadow_clean``), but ``3b-cutover`` — making ``policy_decide`` +authoritative — stayed a MANUAL cliff: nothing advanced when the predicate passed. +This module pins the missing piece: a reversible, version-stamped promotion state +machine that auto-advances ``shadow → canary → authoritative`` while the shadow-week +predicate holds, and rolls back the instant it does not. + +The transition function ``decide_promotion`` is PURE (time + verdict injected) so the +whole ladder — including the 24h canary dwell and the permanent-canary version reset +(master design §4.1) — is unit-testable without sleeping or touching disk. The +``run_promotion_cycle`` driver wires it to the on-disk state + audit ledger. +""" + +from __future__ import annotations + +import json +import tempfile +from datetime import UTC, datetime, timedelta +from pathlib import Path + +from shared.policy_decide import ( + POLICY_DECIDE_FN_VERSION, + PROMOTION_AUTHORITATIVE, + PROMOTION_CANARY, + PROMOTION_SHADOW, + PromotionState, + decide_promotion, + load_promotion_state, + run_promotion_cycle, +) + +_T0 = datetime(2026, 6, 1, 12, 0, 0, tzinfo=UTC) +_DAY = timedelta(days=1) + + +def _verdict(clean: bool, *, tightening: int = 0, total: int = 300, span: float = 8.0) -> dict: + """A minimal ``evaluate_shadow_clean``-shaped verdict for the transition function.""" + return { + "clean": clean, + "coverage_ok": clean, + "asymmetric_ok": tightening == 0, + "total_decisions": total, + "span_days": span, + "divergences": tightening, + "loosening": 0, + "tightening": tightening, + "reasons": [] if clean else ["synthetic not-clean"], + } + + +def _state( + name: str, *, entered: datetime = _T0, version: str = POLICY_DECIDE_FN_VERSION +) -> PromotionState: + return PromotionState( + state=name, policy_version=version, entered_state_at=entered, updated_at=entered + ) + + +# --- The pure transition function: decide_promotion --------------------------- + + +class TestDecidePromotion: + def test_shadow_clean_promotes_to_canary(self): + decision = decide_promotion( + _state(PROMOTION_SHADOW), + _verdict(True), + policy_version=POLICY_DECIDE_FN_VERSION, + now=_T0, + ) + assert decision.from_state == PROMOTION_SHADOW + assert decision.to_state == PROMOTION_CANARY + assert decision.changed is True + + def test_shadow_not_clean_stays_shadow(self): + decision = decide_promotion( + _state(PROMOTION_SHADOW), + _verdict(False, tightening=5), + policy_version=POLICY_DECIDE_FN_VERSION, + now=_T0, + ) + assert decision.to_state == PROMOTION_SHADOW + assert decision.changed is False + + def test_canary_clean_before_window_stays_canary(self): + decision = decide_promotion( + _state(PROMOTION_CANARY, entered=_T0), + _verdict(True), + policy_version=POLICY_DECIDE_FN_VERSION, + now=_T0 + timedelta(hours=12), + canary_window_seconds=24 * 3600, + ) + assert decision.to_state == PROMOTION_CANARY + assert decision.changed is False + + def test_canary_clean_after_window_promotes_to_authoritative(self): + decision = decide_promotion( + _state(PROMOTION_CANARY, entered=_T0), + _verdict(True), + policy_version=POLICY_DECIDE_FN_VERSION, + now=_T0 + timedelta(hours=25), + canary_window_seconds=24 * 3600, + ) + assert decision.to_state == PROMOTION_AUTHORITATIVE + assert decision.changed is True + + def test_canary_not_clean_rolls_back_to_shadow(self): + # Reversibility: a tightening divergence appearing mid-canary demotes — never a cliff. + decision = decide_promotion( + _state(PROMOTION_CANARY, entered=_T0), + _verdict(False, tightening=1), + policy_version=POLICY_DECIDE_FN_VERSION, + now=_T0 + timedelta(hours=30), + canary_window_seconds=24 * 3600, + ) + assert decision.to_state == PROMOTION_SHADOW + assert decision.changed is True + + def test_authoritative_not_clean_rolls_back_to_shadow(self): + decision = decide_promotion( + _state(PROMOTION_AUTHORITATIVE, entered=_T0), + _verdict(False, tightening=2), + policy_version=POLICY_DECIDE_FN_VERSION, + now=_T0 + 10 * _DAY, + ) + assert decision.to_state == PROMOTION_SHADOW + assert decision.changed is True + + def test_authoritative_clean_stays_authoritative(self): + decision = decide_promotion( + _state(PROMOTION_AUTHORITATIVE, entered=_T0), + _verdict(True), + policy_version=POLICY_DECIDE_FN_VERSION, + now=_T0 + 10 * _DAY, + ) + assert decision.to_state == PROMOTION_AUTHORITATIVE + assert decision.changed is False + + def test_version_change_resets_to_shadow_from_canary(self): + # Permanent canary discipline (§4.1): any policy_decide change re-enters shadow, + # even a clean canary that already cleared its dwell window. + decision = decide_promotion( + _state(PROMOTION_CANARY, entered=_T0, version="0.0.9"), + _verdict(True), + policy_version="0.1.0", + now=_T0 + 5 * _DAY, + canary_window_seconds=24 * 3600, + ) + assert decision.to_state == PROMOTION_SHADOW + assert decision.changed is True + assert "0.0.9" in decision.reason and "0.1.0" in decision.reason + + def test_version_change_resets_to_shadow_from_authoritative(self): + decision = decide_promotion( + _state(PROMOTION_AUTHORITATIVE, entered=_T0, version="0.0.9"), + _verdict(True), + policy_version="0.1.0", + now=_T0 + 5 * _DAY, + ) + assert decision.to_state == PROMOTION_SHADOW + assert decision.changed is True + + def test_decision_carries_policy_version_stamp(self): + decision = decide_promotion( + _state(PROMOTION_SHADOW), _verdict(True), policy_version="9.9.9", now=_T0 + ) + assert decision.policy_version == "9.9.9" + + +# --- The on-disk driver: run_promotion_cycle ---------------------------------- +# +# These exercise the real evaluate_shadow_clean → decide_promotion → persist → audit +# chain against synthetic decision-log + divergence-ledger files in a temp dir, so the +# driver (not just the pure transition) is covered: a clean week advances and persists, +# a divergent week does not, the canary dwell carries across two ticks via the on-disk +# entry clock, and a too-short window never promotes (the freeze-blocks-thaw guard). + + +def _iso(when: datetime) -> str: + return when.strftime("%Y-%m-%dT%H:%M:%SZ") + + +def _write_jsonl(path: Path, rows: list[dict]) -> None: + path.write_text("".join(json.dumps(row) + "\n" for row in rows), encoding="utf-8") + + +def _clean_decision_log(n: int = 6, span_days: float = 8.0) -> list[dict]: + """``n`` gated-decision rows whose timestamps span ``span_days`` — the coverage evidence.""" + start = _T0 - timedelta(days=span_days) + step = timedelta(days=span_days) / max(1, n - 1) + return [{"ts": _iso(start + step * i), "tool_name": "Bash"} for i in range(n)] + + +def _loosening_row() -> dict: + """A LOOSENING divergence (legacy blocked, policy_decide allows) — acceptable asymmetry.""" + return {"diverged": True, "legacy_blocked": True, "new_verdict": "allow"} + + +def _tightening_row() -> dict: + """A TIGHTENING divergence (legacy allowed, policy_decide blocks) — makes a week NOT clean.""" + return {"diverged": True, "legacy_blocked": False, "new_verdict": "block"} + + +class TestRunPromotionCycle: + """The on-disk driver: evaluate_shadow_clean → decide → persist posture → audit ledger.""" + + @staticmethod + def _paths(tmp: str) -> tuple[Path, Path, Path, Path]: + d = Path(tmp) + # gate decision log (coverage), divergence ledger (asymmetry), posture, audit ledger + return ( + d / "decisions.jsonl", + d / "shadow.jsonl", + d / "promotion.json", + d / "promotion.jsonl", + ) + + def test_clean_week_promotes_shadow_to_canary_and_persists(self): + with tempfile.TemporaryDirectory() as tmp: + dlog, sledger, state, ledger = self._paths(tmp) + _write_jsonl(dlog, _clean_decision_log()) + _write_jsonl(sledger, [_loosening_row()]) + result = run_promotion_cycle( + decision_log_path=dlog, + shadow_ledger_path=sledger, + state_path=state, + ledger_path=ledger, + now=_T0, + min_days=1.0, + min_decisions=3, + ) + assert result["clean"] is True + assert result["from_state"] == PROMOTION_SHADOW + assert result["to_state"] == PROMOTION_CANARY + assert result["changed"] is True + persisted = load_promotion_state(state) + assert persisted.state == PROMOTION_CANARY + assert persisted.entered_state_at == _T0 + rows = [json.loads(line) for line in ledger.read_text().splitlines() if line] + assert len(rows) == 1 + assert rows[0]["to_state"] == PROMOTION_CANARY + + def test_divergent_week_stays_shadow_and_writes_no_audit_row(self): + with tempfile.TemporaryDirectory() as tmp: + dlog, sledger, state, ledger = self._paths(tmp) + _write_jsonl(dlog, _clean_decision_log()) + _write_jsonl(sledger, [_loosening_row(), _tightening_row()]) + result = run_promotion_cycle( + decision_log_path=dlog, + shadow_ledger_path=sledger, + state_path=state, + ledger_path=ledger, + now=_T0, + min_days=1.0, + min_decisions=3, + ) + assert result["clean"] is False + assert result["to_state"] == PROMOTION_SHADOW + assert result["changed"] is False + assert load_promotion_state(state).state == PROMOTION_SHADOW + assert not ledger.exists() # no transition → no audit row + + def test_canary_reaches_authoritative_after_dwell_across_two_ticks(self): + with tempfile.TemporaryDirectory() as tmp: + dlog, sledger, state, ledger = self._paths(tmp) + _write_jsonl(dlog, _clean_decision_log()) + _write_jsonl(sledger, [_loosening_row()]) + first = run_promotion_cycle( + decision_log_path=dlog, + shadow_ledger_path=sledger, + state_path=state, + ledger_path=ledger, + now=_T0, + min_days=1.0, + min_decisions=3, + canary_window_seconds=24 * 3600, + ) + assert first["to_state"] == PROMOTION_CANARY + # 25h later the persisted entry clock has cleared the 24h dwell → authoritative. + second = run_promotion_cycle( + decision_log_path=dlog, + shadow_ledger_path=sledger, + state_path=state, + ledger_path=ledger, + now=_T0 + timedelta(hours=25), + min_days=1.0, + min_decisions=3, + canary_window_seconds=24 * 3600, + ) + assert second["from_state"] == PROMOTION_CANARY + assert second["to_state"] == PROMOTION_AUTHORITATIVE + assert load_promotion_state(state).state == PROMOTION_AUTHORITATIVE + + def test_insufficient_evidence_does_not_promote(self): + # The freeze-blocks-thaw guard: a short/empty window is NOT clean, so it never promotes. + with tempfile.TemporaryDirectory() as tmp: + dlog, sledger, state, ledger = self._paths(tmp) + _write_jsonl(dlog, _clean_decision_log(n=2, span_days=0.05)) + _write_jsonl(sledger, []) + result = run_promotion_cycle( + decision_log_path=dlog, + shadow_ledger_path=sledger, + state_path=state, + ledger_path=ledger, + now=_T0, + min_days=7.0, + min_decisions=200, + ) + assert result["clean"] is False + assert result["to_state"] == PROMOTION_SHADOW