"""The auto-completion gate. Phase 2a already decides, correctly and fail-closed. This does not re-decide; it supplies real evidence to that decision and, only if three independent gates agree, asks the transition boundary to commit. Three gates, all of which must hold: * the global switch is on (default off), * the backend has observed evidence that it reaches a quiescent state, * the job itself opted in. Everything else is refusal. A failure anywhere -- an unreadable manifest, an exception in the policy, an unknown decision -- holds. "Could not decide" is never "done". """ from __future__ import annotations import logging import os import re from pathlib import Path from typing import Any, Dict, List, Optional from lib_py.delegation import finalize from lib_py.loop import policy, store logger = logging.getLogger("loop.autocomplete") FLAG = "MAM_AUTO_COMPLETE" BACKENDS_FLAG = "MAM_AUTO_COMPLETE_BACKENDS" # Backends with recorded evidence that they reach a quiescent state. Auto # completion infers "finished" from stillness, so a backend whose stillness has # never been observed has no basis for the inference. agy has only been seen # working; hermes and opencode have no observations at all. EVIDENCE_BACKED_BACKENDS = ("claude", "grok", "codex") # The `agent` recorded on a job is a delegate key. Matching these exactly rather # than by prefix keeps a future `claude-something-unverified` from inheriting # evidence that was never collected for it. EVIDENCE_BACKED_AGENT_KEYS = ( "claude", "claude-code", "grok", "grok-build", "codex", "codex-cli", ) # A reviewer's verdict token, matched the way run_loop.sh matches it. VERDICT_LINE = re.compile(r"^\[VERDICT: (PASS|NOT PASS)\][ \t]*\r?$") SOURCE = "auto_complete" def enabled() -> bool: """Default off. Read per call so an operator can flip it without a restart.""" return os.environ.get(FLAG, "0") == "1" def allowed_backends() -> tuple: """The allowlist may be narrowed by an operator, never widened. Widening would need new observations, so an unknown name in the environment is dropped rather than trusted. """ raw = os.environ.get(BACKENDS_FLAG) if not raw: return EVIDENCE_BACKED_BACKENDS requested = [name.strip() for name in raw.split(",") if name.strip()] return tuple(n for n in requested if n in EVIDENCE_BACKED_BACKENDS) class GateResult(object): __slots__ = ("committed", "decision", "reason") def __init__(self, committed, decision, reason): self.committed = committed self.decision = decision self.reason = reason def __repr__(self): # pragma: no cover - debugging aid return "GateResult(committed={!r}, {!r}, {!r})".format( self.committed, self.decision, self.reason) def _allowed_agent_keys() -> tuple: allowed = allowed_backends() return tuple(k for k in EVIDENCE_BACKED_AGENT_KEYS if any(k == b or k.startswith(b + "-") for b in allowed)) def _gates_open(record: Dict[str, Any]) -> Optional[str]: """None when every gate passes, else why one did not.""" from lib_py.delegation import engine if not enabled(): return "auto-completion is disabled ({}=0)".format(FLAG) if not engine.cas_enabled(): # The operations note promises this switch disables auto-completion too, # and it must: inferred completion is built on the CAS boundary, so # committing while that boundary is rolled back would bypass it. return "the CAS engine is rolled back (MAM_ENGINE_CAS=0)" backend = record.get("agent") or "" if backend not in _allowed_agent_keys(): return "backend {!r} has no observed quiescent-state evidence".format(backend) if not record.get("auto_complete"): return "job did not opt in to auto-completion" return None def _verdict_present(report_path) -> bool: """A reviewer's report must end with its verdict token. Phase 0 made this a safety contract: without a verdict, a reviewer job may never be promoted to completed. run_loop.sh reads a missing token as NOT PASS, so auto-completing before the token exists would start a correction round against a review that was never finished. """ try: lines = Path(report_path).read_text(encoding="utf-8").splitlines() except OSError: return False for line in reversed(lines): if line.strip(): return bool(VERDICT_LINE.match(line)) return False def _scenario_matches(job_id, scenario, dispatch) -> Optional[str]: """The evidence must be about this job and this dispatch. Without this the gate will happily complete one job using another job's evidence: the policy only checks a scenario against the dispatch context carried inside that same scenario, which is self-consistent by construction. """ context = scenario.get("dispatch_context") or {} if context.get("job_id") != job_id: return "evidence is for job {!r}, not {!r}".format(context.get("job_id"), job_id) for field in ("attempt_id", "agent_session_uuid", "pane_process_generation"): expected = dispatch.get(field) if expected is None: return "no recorded dispatch {} to compare the evidence against".format(field) if context.get(field) != expected: return "evidence {} {!r} does not match the recorded dispatch {!r}".format( field, context.get(field), expected) return None def evaluate(job_id, scenario, registry_dir=None, workspace=None, report_dir=None) -> GateResult: # `report_dir` keeps a default only so the signature stays callable from a # supervisor that has not resolved it yet; omitting it is a refusal, not a # shortcut. See the check at the top of the body. """Decide, and commit only if every gate agrees. `scenario` is the composite evidence the watcher assembled, in the same shape the phase 0 fixtures use, so the decision is made by exactly the code the scenario contracts already pin. """ from lib_py.delegation import engine, mqtt_common registry_dir = registry_dir or mqtt_common.DEFAULT_REGISTRY_DIR try: record = mqtt_common.load_job(job_id, registry_dir) except FileNotFoundError: return GateResult(False, "hold", "no such job") closed = _gates_open(record) if closed: return GateResult(False, "hold", closed) # Independent verification is mandatory, never conditional on the caller # remembering an argument. Skipping it would put the collector's own flags # back in charge of the decision, which is the self-certification this whole # design exists to avoid. if report_dir is None: return GateResult(False, "hold", "no report directory to independently verify the manifest against") dispatch = store.load_dispatch(job_id, workspace) if not dispatch: return GateResult(False, "hold", "no recorded dispatch to attribute this completion to") mismatch = _scenario_matches(job_id, scenario, dispatch) if mismatch: return GateResult(False, "hold", mismatch) # The evidence must belong to the attempt that is currently running, and the # commit must be refused if that stops being true while this runs. Comparing # here alone is not enough: the job can complete, be re-dispatched and # claimed again before the terminal write, after which a status check alone # would let stale evidence finish a new attempt. from lib_py.delegation import engine as _engine token = record.get(_engine.DISPATCH_TOKEN_FIELD) if not token: return GateResult(False, "hold", "job has no dispatch token to bind this decision to") if dispatch.get("dispatch_token") != token: return GateResult(False, "hold", "evidence belongs to a superseded dispatch of this job") verdict = finalize.verify_manifest( report_dir, job_id, expected_attempt=dispatch.get("attempt_id"), expected_artifacts=record.get("expected_artifacts") or []) if not verdict.ok: return GateResult(False, "hold", verdict.reason) # A reviewer that has not written its verdict has not finished reviewing. if str(record.get("role") or "").lower() == "reviewer": if not _verdict_present(verdict.manifest["report_path"]): return GateResult(False, "hold", "reviewer report carries no verdict token on its last line") try: decision = policy.decide(scenario) except Exception as exc: # a policy that cannot decide has not approved logger.warning("policy raised for %s: %s", job_id, exc) return GateResult(False, "hold", "policy error: {}".format(exc)) if not decision.would_complete: return GateResult(False, decision.decision, decision.reason) # The attempt guard lives here, in the evidence, not in the record: a record # carries a loop-iteration counter, which is a different thing. result = engine.transition( job_id, "completed", registry_dir=registry_dir, expected_status="running", expected_dispatch_token=token, source=SOURCE, fields={"completed_by": SOURCE, "auto_complete_reason": decision.reason}, notify="completed", ) if not result.committed: # A late explicit terminal, or a job that moved on. The boundary refused; # that is the boundary doing its job, not an error here. logger.info("auto-completion declined for %s: %s", job_id, result.reason) return GateResult(False, "hold", result.reason) logger.info("auto-completed %s (%s)", job_id, decision.reason) return GateResult(True, decision.decision, decision.reason)