"""Config loading/validation and agent execution. Every ADW validates its agents before running (fail fast, nothing spawns against a half-valid config). Every agent call parses against a concrete output type; parse failures and gate violations re-prompt the SAME session with a correction — context intact, bounded retries. Agent proposes, code disposes. """ from __future__ import annotations import json import os import sys from copy import deepcopy from pathlib import Path from typing import Optional import yaml from . import agent_pi, permissions, prompts from .data_types import (AgentCall, AgentConfig, EnvelopeBase, EventRecord, GateCheck, GateReport, Phase, PiRequest, SSSFConfig, UsageBreakdown) from .utils import new_id JSON_FIX_ATTEMPTS = 2 # continue-with-correction attempts for malformed JSON PI_REMOTE = Path(__file__).resolve().parents[3] / "workstation" / "bin" / "pi-remote" class GateFailure(RuntimeError): pass # ── config ─────────────────────────────────────────────────────────────────── def _deep_overlay(base: dict, override: dict) -> dict: """Merge mappings recursively; every other value is a replacement.""" merged = deepcopy(base) for key, value in override.items(): if isinstance(value, dict) and isinstance(merged.get(key), dict): merged[key] = _deep_overlay(merged[key], value) else: merged[key] = deepcopy(value) return merged def _anchor_references(raw: dict, source: Path) -> dict: """Resolve paths while their source layer is still known.""" raw = deepcopy(raw) anchor = source.parent defaults = raw.get("defaults", {}) or {} defaults["harness_engineering"] = [ ref if Path(ref).is_absolute() else str(anchor / ref) for ref in defaults.get("harness_engineering") or [] ] for agent in raw.get("agents", []) or []: engineering = agent.get("prompt_engineering") or {} for key in ("system", "user"): ref = engineering.get(key) if ref and not Path(ref).is_absolute(): engineering[key] = str(anchor / ref) agent["harness_engineering"] = [ ref if Path(ref).is_absolute() else str(anchor / ref) for ref in agent.get("harness_engineering") or [] ] return raw def preset_descriptions() -> list[str]: root = Path(os.environ.get("FACTORY_ROOT", Path(__file__).resolve().parents[1])).resolve() descriptions = [] for path in sorted((root / "presets").glob("*.yaml")): seats = (yaml.safe_load(path.read_text()) or {}).get("seats", {}) or {} pins = [] for name, value in sorted(seats.items()): if isinstance(value, dict): detail = ", ".join( f"{key}={value[key]}" for key in ("model", "thinking") if key in value ) pins.append(f"{name} ({detail})") descriptions.append(f"{path.stem}: " + "; ".join(pins)) return descriptions def load_config(path: str = "adws/adw_sssf_config/sssf.config.yaml", preset: str | None = None) -> SSSFConfig: requested = Path(path).resolve() factory_root = Path(os.environ.get( "FACTORY_ROOT", Path(__file__).resolve().parents[1])).resolve() central = (factory_root / "sssf.config.yaml").resolve() sources = [central] raw = _anchor_references(yaml.safe_load(central.read_text()) or {}, central) if requested != central: overlay = _anchor_references(yaml.safe_load(requested.read_text()) or {}, requested) raw = _deep_overlay(raw, overlay) sources.append(requested) if preset is not None: presets = {item.stem: item.resolve() for item in (factory_root / "presets").glob("*.yaml")} if preset not in presets: raise SystemExit(f"preset {preset!r} is not defined — available: {sorted(presets)}") preset_path = presets[preset] preset_raw = _anchor_references(yaml.safe_load(preset_path.read_text()) or {}, preset_path) seats = preset_raw.pop("seats", {}) or {} if not isinstance(seats, dict): raise SystemExit("preset seats must be a mapping") raw = _deep_overlay(raw, preset_raw) known = {agent.get("name"): agent for agent in raw.get("agents", []) or []} for name, changes in seats.items(): if name not in known: raise SystemExit(f"preset seat {name!r} is not defined — available: {sorted(known)}") if not isinstance(changes, dict) or not set(changes).issubset({"model", "thinking"}): raise SystemExit(f"preset seat {name!r} may set only model and thinking") known[name] = _deep_overlay(known[name], changes) raw["agents"] = [known[agent.get("name")] for agent in raw.get("agents", []) or []] raw["preset"] = preset sources.append(f"preset:{preset} ({preset_path})") if os.environ.get("FACTORY_DATA_DIR"): raw.setdefault("defaults", {})["data_dir"] = os.environ["FACTORY_DATA_DIR"] if os.environ.get("FACTORY_DB_PATH"): raw.setdefault("observability", {})["db"] = os.environ["FACTORY_DB_PATH"] print("factory config layers: " + " -> ".join(str(item) for item in sources), file=sys.stderr) defaults = raw.get("defaults", {}) or {} for agent in raw.get("agents", []) or []: for key in ("coding_agent", "model", "thinking", "color", "tools", "writes"): if key in defaults: agent.setdefault(key, defaults[key]) agent.setdefault("harness_engineering", defaults.get("harness_engineering", [])) agent["harness_engineering"] = [ ref for ref in agent.get("harness_engineering") or [] ] return SSSFConfig(**raw) def resolve(cfg: SSSFConfig, name: str) -> AgentConfig: for agent in cfg.agents: if agent.name == name: return agent raise SystemExit(f"agent {name!r} is not defined in the config — " f"available: {[a.name for a in cfg.agents]}") def validate(cfg: SSSFConfig, required: list[str]) -> None: """Fail fast: every required name must resolve to a usable agent.""" problems = [] for name in required: try: agent = resolve(cfg, name) except SystemExit as e: problems.append(str(e)) continue if agent.coding_agent != "pi": problems.append(f"agent {name!r}: coding_agent {agent.coding_agent!r} " f"is not implemented in v1 (pi only)") for label, ref in (("system", agent.prompt_engineering.system), ("user", agent.prompt_engineering.user)): if not Path(ref).is_file(): problems.append(f"agent {name!r}: {label} prompt not found: {ref}") try: agent_pi.resolve_model(agent.model, cfg.account) except ValueError as e: problems.append(f"agent {name!r}: {e}") if problems: raise SystemExit("config validation failed:\n- " + "\n- ".join(problems)) # ── execution ──────────────────────────────────────────────────────────────── def execute(run, phase: Phase, call: AgentCall, baseline_commit: str) -> EnvelopeBase: """One agent call: render prompts -> pi run -> typed parse -> gates -> envelope.""" agent = resolve(run.cfg, phase.params.owner) agent_dir = run.session_dir / agent.name agent_dir.mkdir(parents=True, exist_ok=True) variables = { "prompt": call.prompt, "previous_envelope": call.previous.model_dump_json(indent=2) if call.previous else "(none)", "context_handoff_dir": str(run.context_handoff_dir), } system_text = prompts.render(agent.prompt_engineering.system, variables) user_text = prompts.render(agent.prompt_engineering.user, variables) prompts.save(agent_dir / "prompts", "system.md", system_text) prompts.save(agent_dir / "prompts", "user.md", user_text) session_id = _agent_session_id(run, agent) agent_start_id = run.tracer.event(EventRecord( adw_id=run.adw_id, phase_id=phase.phase_id, type="agent_start", name=agent.name, payload={"model": agent.model, "thinking": agent.thinking, "color": agent.color, "session_id": session_id, "client_session_id": agent_pi.derive_client_session_id(session_id), "provider": agent.model.split("/", 1)[0] if "/" in agent.model else None, "coding_agent": agent.coding_agent, "purpose": agent.purpose, "tools": agent.tools, # None = all tools "harness_engineering": agent.harness_engineering})) run.console.agent_started(agent.name, agent.model, session_id) # Parse retries and gate corrections re-enter the SAME pi session, so the # last send is the one whose context occupancy is current — while spend is # the opposite: every send costs, so usage accumulates across all of them. latest: agent_pi.PiResult | None = None spent = UsageBreakdown() def send(prompt_text: str) -> agent_pi.PiResult: nonlocal latest request = PiRequest( prompt=prompt_text, system_prompt=system_text, model=agent.model, thinking=agent.thinking, session_id=session_id, # absolute: these are read by the pi subprocess, which runs in repo_root session_dir=str((agent_dir / "pi_sessions").resolve()), raw_output_path=str((agent_dir / "raw_output.jsonl").resolve()), tools=agent.tools, extensions=agent.harness_engineering, cwd=str(run.repo_root), account=run.cfg.account, ) def record_spawn(pid: int, command: str) -> None: from .control import read_proc_start_ticks run.tracer.process_start( run.adw_id, "agent", agent.name, pid, command, start_ticks=read_proc_start_ticks(pid), ) result = agent_pi.run( request, spawn_path=str(PI_REMOTE) if run.cfg.defaults.pi_spawn == "remote" else None, on_event=_event_forwarder(run, phase, agent.name), on_spawn=record_spawn, on_exit=lambda pid: run.tracer.process_end(run.adw_id, pid), on_attempt_start=lambda info: run.tracer.agent_attempt_start( run.adw_id, phase.phase_id, agent.name, info["session_id"], info["command"], host=info["host"], account=info["account"], provider=info["provider"], model=info["model"], system_prompt=info.get("system_prompt"), user_prompt=info.get("user_prompt"), parent_id=agent_start_id), on_attempt_end=lambda attempt_id, info: run.tracer.agent_attempt_finish( attempt_id, returncode=info["returncode"], signal=info["signal"], timed_out=info["timed_out"], stderr_path=info["stderr_path"], tokens=info["tokens"], usage=info["usage"], error=info["error"], timeout_kind=info.get("timeout_kind"), provider_failure=info.get("provider_failure"))) run.add_usage(result.tokens, result.cost) spent.merge(result.usage) latest = result return result # What the tree looked like before this agent got its hands on it. Every # send in this phase — first prompt, JSON retries, gate corrections — is # measured against this one baseline. tree_before = permissions.snapshot(run) result = send(user_text) envelope, attempt = _parse_with_retries(run, phase, call, result, send) # claim gates — violations flow back into the SAME session as corrections for gate_attempt in range(1, max(1, phase.params.retries + 1) + 1): violations = [] for gate in call.gates: report = _as_report(gate(envelope, run, baseline_commit)) found = report.violations run.tracer.gate_row(phase, gate.__name__, report, gate_attempt) run.tracer.event(EventRecord( adw_id=run.adw_id, phase_id=phase.phase_id, type="gate_fail" if found else "gate_pass", name=gate.__name__, payload={"attempt": gate_attempt, "violations": found, "checks": [c.model_dump() for c in report.checks]})) run.console.gate_result(gate.__name__, report) violations.extend(found) if not violations: break if gate_attempt > phase.params.retries: raise GateFailure(f"{agent.name} failed gates after {gate_attempt} attempt(s):\n- " + "\n- ".join(violations)) phase.attempt = gate_attempt run.console.retry(agent.name, gate_attempt, phase.params.retries, f"{len(violations)} gate violation(s)") correction = ("Your previous response failed validation:\n- " + "\n- ".join(violations) + "\n\nFix these problems, then re-emit ONLY your Report JSON.") result = send(correction) envelope, attempt = _parse_with_retries(run, phase, call, result, send) # Permission is checked after every send is done, and before the envelope is # accepted: an agent does not get to report success on a phase in which it # wrote somewhere it was not allowed to. try: touched = permissions.enforce(run, phase, agent, tree_before) except permissions.PermissionBreach as breach: run.tracer.event(EventRecord(adw_id=run.adw_id, phase_id=phase.phase_id, type="error", name="permission_breach", payload={"agent": agent.name, "error": str(breach), "writes": agent.writes, "protected_files": run.cfg.defaults.protected_files})) raise if touched: run.claim_paths(touched) run.tracer.event(EventRecord(adw_id=run.adw_id, phase_id=phase.phase_id, type="log", name="paths_touched", payload={"agent": agent.name, "paths": touched})) _persist_envelope(run, phase, agent.name, call, envelope, attempt, valid=True) run.console.envelope_summary(envelope) context = latest or result run.tracer.agent_session_row(run.adw_id, agent, session_id, context_tokens=context.context_tokens, context_window=context.context_window) run.save_agent_map(agent.name, {"session_id": session_id, "model": agent.model, "coding_agent": agent.coding_agent}) run.tracer.event(EventRecord(adw_id=run.adw_id, phase_id=phase.phase_id, type="handoff", name=agent.name, payload={"artifacts": envelope.artifacts, "summary": envelope.summary})) run.tracer.event(EventRecord(adw_id=run.adw_id, phase_id=phase.phase_id, type="agent_end", name=agent.name, # Phase totals, not the last send's: a retried # phase paid for every attempt. tokens=spent.total_tokens, payload={"cost": spent.total_cost, "usage": spent.model_dump(), "provider": context.provider or None, "model": f"{context.provider}/{context.model_id}" if context.provider and context.model_id else agent.model, "client_session_id": context.client_session_id, "context_tokens": context.context_tokens, "context_window": context.context_window, "max_tokens": context.max_tokens, "billing_status": context.usage.billing_status, "provider_failure": (context.provider_failure.model_dump() if context.provider_failure else None)})) run.console.agent_finished(agent.name, spent.total_tokens, spent.total_cost) if envelope.status != "success": raise RuntimeError(f"{agent.name} reported status={envelope.status!r}: {envelope.summary}") return envelope # ── internals ──────────────────────────────────────────────────────────────── def _as_report(result) -> GateReport: """Accept a GateReport, or a legacy gate that returned a violations list.""" if isinstance(result, GateReport): return result return GateReport(checks=[GateCheck(item=str(v), ok=False) for v in (result or [])]) def _agent_session_id(run, agent: AgentConfig) -> str: entry = run.agent_map.get(agent.name) if entry and entry.get("model") == agent.model: return entry["session_id"] # rejoin the existing context window return f"sssf-{run.adw_id}-{agent.name}-{new_id(4)}" def _event_forwarder(run, phase: Phase, agent_name: str): """One tool_call event per real tool call, with its exact args and result.""" tracker = agent_pi.ToolCallTracker() def forward(event: dict) -> None: record = tracker.observe(event) if record is None: return # The call's span rides the columns; duration_ms stays in the payload as # pi's own authoritative number. run.tracer.event(EventRecord(adw_id=run.adw_id, phase_id=phase.phase_id, type=record.pop("event_type", "tool_call"), name=record.pop("label"), started_at=record.pop("started_at", None), ended_at=record.pop("ended_at", None), payload={**record, "agent": agent_name})) return forward def _extract_json(text: str) -> dict: candidate = text if "```" in text: for block in text.split("```")[1::2]: block = block.removeprefix("json").strip() if block.startswith("{"): candidate = block break start, end = candidate.find("{"), candidate.rfind("}") if start == -1 or end <= start: raise ValueError("no JSON object found in the response") return json.loads(candidate[start:end + 1]) def _parse_with_retries(run, phase: Phase, call: AgentCall, result, send): """Parse the final response against the declared output type; on failure, continue the SAME session with a correction (bounded).""" for attempt in range(1, JSON_FIX_ATTEMPTS + 2): try: payload = _extract_json(result.text) return call.output_type.model_validate(payload), attempt except Exception as error: _persist_envelope(run, phase, phase.params.owner, call, None, attempt, valid=False, raw=result.text) if attempt > JSON_FIX_ATTEMPTS: raise RuntimeError( f"{phase.params.owner} never produced valid " f"{call.output_type.__name__} JSON: {error}") from error run.console.retry(phase.params.owner, attempt, JSON_FIX_ATTEMPTS, f"invalid {call.output_type.__name__} JSON: {error}") run.tracer.event(EventRecord( adw_id=run.adw_id, phase_id=phase.phase_id, type="log", name="report_correction", payload={"agent": phase.params.owner, "attempt": attempt, "max_attempts": JSON_FIX_ATTEMPTS, "output_type": call.output_type.__name__, "error": str(error)})) fields = ", ".join(call.output_type.model_fields.keys()) result = send( f"Your response was not valid JSON for the required structure " f"({error}). Respond again with ONLY a JSON object with these " f"fields: {fields}. No prose, no code fences.") def _persist_envelope(run, phase: Phase, agent_name: str, call: AgentCall, envelope: Optional[EnvelopeBase], attempt: int, valid: bool, raw: str = "") -> None: payload_json = envelope.model_dump_json(indent=2) if envelope else json.dumps({"raw": raw[-2000:]}) run.tracer.envelope_row(phase, agent_name, call.output_type.__name__, payload_json, valid, attempt) if envelope: record = {"agent_name": agent_name, "purpose": resolve(run.cfg, agent_name).purpose, "output_type": call.output_type.__name__, "attempt": attempt, **envelope.model_dump()} (run.session_dir / agent_name / "envelope.json").write_text(json.dumps(record, indent=2))