"""The Run object: config + adw_id + agent_map + tracer + console, bound once. `run.phase(PhaseParams(...))` is the ONE phase primitive — a context manager for all three kinds (engineer, agent, code). Success must be earned: every phase defaults to fail; only a clean exit flips it (agent phases additionally require a parsed envelope + green gates, enforced inside ph.call). """ from __future__ import annotations import json import time from contextlib import contextmanager from pathlib import Path from . import agents, changes, git_helper, notify, permissions, supervisor from .console import Console from .data_types import AgentCall, EnvelopeBase, EventRecord, Phase, PhaseParams from .utils import ensure_dir, now_iso class PhaseHandle: def __init__(self, run: "Run", phase: Phase, baseline_commit: str): self.run = run self.phase = phase self.baseline_commit = baseline_commit def log(self, **payload) -> None: self.run.tracer.event(EventRecord(adw_id=self.run.adw_id, phase_id=self.phase.phase_id, type="log", name=self.phase.params.name, payload=payload)) self.run.console.note(", ".join(f"{k}: {v}" for k, v in payload.items())) if self.phase.params.kind == "engineer" and "input" in payload: self.run.tracer.session_request(self.run.adw_id, str(payload["input"])) def call(self, call: AgentCall) -> EnvelopeBase: if self.phase.params.kind != "agent": raise RuntimeError("ph.call() is only valid inside an agent phase") return agents.execute(self.run, self.phase, call, self.baseline_commit) class Run: def __init__(self, cfg, adw_id: str, tracer, engineer: str): self.cfg = cfg self.adw_id = adw_id self.tracer = tracer self.console = Console(tracer, adw_id) self.engineer = engineer self.adw_name: str | None = None # set by session.ensure() once known self._started_monotonic = time.monotonic() self._finish_notified = False # exactly one terminal finish message per run self.phases: list[Phase] = [] self.tokens = 0 self.cost = 0.0 self.entitled_paths: set[str] = set() self._seq = tracer.max_phase_seq(adw_id) # a joined run continues the sequence self.repo_root = git_helper.repo_root() # where every agent is spawned to work self.session_dir = ensure_dir(Path(cfg.defaults.data_dir) / "sessions" / adw_id) self.context_handoff_dir = ensure_dir(self.session_dir / "context_handoff") self._agent_map_path = self.session_dir / "agent_map.json" self.agent_map: dict = (json.loads(self._agent_map_path.read_text()) if self._agent_map_path.exists() else {}) # ── agent map (adw_id -> per-agent coding-agent session ids) ──────────── def save_agent_map(self, agent: str, entry: dict) -> None: self.agent_map[agent] = entry self._agent_map_path.write_text(json.dumps(self.agent_map, indent=2)) # ── usage (run totals mirror what the tracer accumulates in sqlite) ───── def add_usage(self, tokens: int, cost: float) -> None: self.tokens += tokens self.cost += cost self.tracer.session_add_usage(self.adw_id, tokens, cost) def claim_paths(self, paths: list[str]) -> None: self.entitled_paths.update(paths) def commit_all(self, message: str) -> str: sha = git_helper.commit_all(message, sorted(self.entitled_paths)) self.entitled_paths.clear() return sha def fail_active_phase(self, error: str) -> None: phase = self.phases[-1] phase.status = "fail" phase.error = error[:1000] def _capture_phase_diff(self, phase: Phase, before_head: str, before_paths: dict[str, str]) -> None: if not before_head: return try: diff = changes.capture_phase_diff(before_head, before_paths) if diff is not None: self.tracer.phase_diff_row(phase, diff) except RuntimeError as error: message = str(error)[:1000] if phase.error is None: phase.error = message self.tracer.event(EventRecord( adw_id=self.adw_id, phase_id=phase.phase_id, type="error", name="diff_capture", payload={"error": message})) # ── the phase primitive ───────────────────────────────────────────────── @contextmanager def phase(self, params: PhaseParams): self._seq += 1 phase = Phase(phase_id=f"{self.adw_id}_{self._seq:02d}_{params.name}", task_id=params.task_id, adw_id=self.adw_id, seq=self._seq, params=params, status="running", started_at=now_iso()) self.phases.append(phase) self.tracer.phase_upsert(phase) self.tracer.event(EventRecord(adw_id=self.adw_id, phase_id=phase.phase_id, type="phase_start", name=params.name, payload={"kind": params.kind, "owner": params.owner, "description": params.description})) self.console.phase_started(phase) clock = time.monotonic() try: phase_head, phase_paths = git_helper.phase_snapshot() except RuntimeError: phase_head, phase_paths = "", {} baseline_commit = "" if params.kind == "agent": try: baseline_commit = git_helper.phase_baseline() except RuntimeError: pass try: yield PhaseHandle(self, phase, baseline_commit) except BaseException as error: phase.status = "fail" # success must be earned phase.error = str(error)[:1000] phase.ended_at = now_iso() self._capture_phase_diff(phase, phase_head, phase_paths) self.tracer.event(EventRecord(adw_id=self.adw_id, phase_id=phase.phase_id, type="error", name=params.name, payload={"error": phase.error})) self.tracer.event(EventRecord(adw_id=self.adw_id, phase_id=phase.phase_id, type="phase_end", name=params.name, payload={"status": "fail"})) self.tracer.phase_upsert(phase) self.tracer.session_finish(self.adw_id, ok=False) self.console.phase_ended(phase, time.monotonic() - clock) self.console.session_finished(False, self.tokens, self.cost, self.cfg.observability.db) if isinstance(error, permissions.PermissionBreach): raise SystemExit(1) from None raise else: if phase.status == "running": phase.status = "success" phase.ended_at = now_iso() self._capture_phase_diff(phase, phase_head, phase_paths) if phase.status == "fail": self.tracer.event(EventRecord(adw_id=self.adw_id, phase_id=phase.phase_id, type="error", name=params.name, payload={"error": phase.error})) self.tracer.event(EventRecord(adw_id=self.adw_id, phase_id=phase.phase_id, type="phase_end", name=params.name, payload={"status": phase.status})) self.tracer.phase_upsert(phase) self.console.phase_ended(phase, time.monotonic() - clock) # ── run outcome ───────────────────────────────────────────────────────── def finish(self, accepted: bool = True, reason: str = "") -> int: """Finalize the run and return its exit code. Call this exactly once. Two criteria, not one. Every unrecovered phase must have passed, AND the ADW's own acceptance test must hold. Failed quality attempts remain red in the trace while a later clean attempt can satisfy acceptance. This replaces a `succeeded` property that answered only the first question — and, being a property with side effects, wrote the session status and printed the banner before the caller's `and test.passed` was ever evaluated. A run whose suite never passed was recorded green in the db, on the terminal, and in the UI while exiting 1. Anyone reading the trace saw success; only a CI job checking `$?` saw the truth. One call now settles the db, the banner, and the exit code together, so the three cannot disagree. """ phases_ok = bool(self.phases) and all( p.status == "success" or (accepted and p.params.owner == "quality") for p in self.phases ) ok = phases_ok and accepted if not accepted: note = reason or "the run's acceptance criterion was not met" self.tracer.event(EventRecord( adw_id=self.adw_id, phase_id=self.phases[-1].phase_id if self.phases else "", type="error", name="not_accepted", payload={"reason": note})) self.console.note(f"not accepted: {note}") self.tracer.session_finish(self.adw_id, ok=ok) self.console.session_finished(ok, self.tokens, self.cost, self.cfg.observability.db) supervisor.finished(self.adw_id) if not self._finish_notified: self._finish_notified = True notify.notify_finish(self.adw_id, self.adw_name, ok, time.monotonic() - self._started_monotonic) return 0 if ok else 1