#!/usr/bin/env -S uv run # /// script # dependencies = ["pydantic", "python-dotenv", "pyyaml", "rich"] # /// """ADW Plan Build Test Quality — full agent chain plus deterministic quality. Usage: uv run adws/adw_plan_build_test_quality.py "" [--config adws/adw_sssf_config/sssf.config.yaml] [--adw-id a1b2c3d4] Phases: engineer(request) -> planner -> builder -> [code(verify) -> code(test) -> builder(fix)] bounded -> git(commit) Verify and test are CODE, not agents. Their commands are known, so running them needs no judgement — only repairing them does. A failing block does not fail its phase: the runner did its job, the code is what failed. The failure becomes an envelope and flows back into the builder, and only an exhausted repair loop fails the run. """ import argparse import sys from pathlib import Path from adw_modules import agents, gates, git_helper, hitl, quality, session, utils from adw_modules.data_types import AgentCall, BuildOutput, PhaseParams, PlanOutput REQUIRED_AGENTS = ["planner", "builder"] MAX_FIX_LOOPS = 3 # --skip-plan accepts only an authored doc living under one of these directory # pairs (relative to any repo root) — never a freeform prompt. PLAN_DOC_DIRS = (("docs", "plans"), ("docs", "specs")) class SkipPlanRefused(Exception): """--skip-plan pointed at something other than an authored plan/spec doc.""" def _skip_plan_target(raw: str) -> Path: """Validate the --skip-plan path: an existing, non-empty docs/plans|specs markdown file. Anything else refuses here — a bare sentence still needs the planner. Never a silent fallback to running one. """ guidance = ("--skip-plan needs a path to an existing, non-empty docs/plans/ or " "docs/specs/ markdown file — a bare prompt still needs the planner") p = Path(raw) if not p.is_file(): raise SkipPlanRefused(f"{guidance} (not a file: {raw!r})") if p.suffix.lower() not in (".md", ".markdown"): raise SkipPlanRefused(f"{guidance} (not markdown: {raw!r})") if p.stat().st_size == 0: raise SkipPlanRefused(f"{guidance} (empty file: {raw!r})") parts = p.resolve().parts under_plan_dir = any(parts[i:i + 2] == pair for pair in PLAN_DOC_DIRS for i in range(len(parts) - 1)) if not under_plan_dir: raise SkipPlanRefused(f"{guidance} (not under docs/plans/ or docs/specs/: {raw!r})") return p def _synth_plan_output(run, doc_path: Path, doc_text: str) -> PlanOutput: """Deterministic PlanOutput for an authored doc — no planner agent call. Copies the doc into context_handoff/ (the same seam a planner agent's own artifacts land in) so the build phase's gates see a real, non-empty file, and resolves as an already-decided plan: an authored doc carries no pending human_decision. """ handoff = run.context_handoff_dir / doc_path.name handoff.write_text(doc_text) summary = next((line.lstrip("#").strip() for line in doc_text.splitlines() if line.strip()), doc_path.stem) return PlanOutput(status="success", summary=summary, artifacts=[str(handoff)]) def main(prompt: str, config: str = "adws/adw_sssf_config/sssf.config.yaml", adw_id: str | None = None, preset: str | None = None, account: str | None = None, slug_hint: str | None = None, skip_plan: bool = False) -> int: if skip_plan: try: doc_path = _skip_plan_target(prompt) except SkipPlanRefused as err: print(f"factory: {err}", file=sys.stderr) return 2 prompt = doc_path.read_text() cfg = agents.load_config(config, preset) cfg.account = account or cfg.account utils.operator_env(cfg.account) agents.validate(cfg, REQUIRED_AGENTS) run = session.ensure(cfg, adw_id, mutates_repo=True, slug_hint=slug_hint) with run.phase(PhaseParams(name="request", task_id="request", kind="engineer", owner=run.engineer, description="Capture the incoming ask")) as ph: ph.log(input=prompt) if skip_plan: with run.phase(PhaseParams(name="plan", task_id="plan", kind="code", owner="planner", description="Authored plan accepted, planner skipped")) as ph: plan = _synth_plan_output(run, doc_path, prompt) ph.log(source=str(doc_path), artifacts=", ".join(plan.artifacts)) else: with run.phase(PhaseParams(name="plan", task_id="plan", kind="agent", owner="planner", retries=1, description="Turn the request into an implementable plan")) as ph: plan = hitl.resolve_plan_decisions(run, ph, AgentCall(output_type=PlanOutput, prompt=prompt, gates=[gates.artifacts_exist, gates.files_non_empty])) with run.phase(PhaseParams(name="build", task_id="build", kind="agent", owner="builder", retries=1, description="Implement the plan exactly")) as ph: previous = ph.call(AgentCall(output_type=BuildOutput, prompt=prompt, previous=plan, gates=[gates.diff_matches_claims])) def record(ph, result) -> None: passed = sum(1 for check in result.checks if check.passed) ph.log(passed=result.passed, checks=f"{passed}/{len(result.checks)}", artifacts=", ".join(result.artifacts)) test_result = None quality_result = None for i in range(1, MAX_FIX_LOOPS + 1): with run.phase(PhaseParams(name=f"verify_{i}", task_id=f"verify_{i}", kind="code", owner="quality", description="Lint, typecheck, and build before testing")) as ph: quality_result = quality.run_quality(run) record(ph, quality_result) # run_quality() already includes the test block; a repo that wants tests # in their own phase can split them out the way this comment does. test_result = quality_result if quality_result.passed and test_result.passed: break if i == MAX_FIX_LOOPS: break # Whichever block failed becomes the builder's spec — verbatim command # output, no parser standing between the failure and the fix. broken = quality_result if not quality_result.passed else test_result what = "verification" if not quality_result.passed else "tests" with run.phase(PhaseParams(name=f"fix_{i}", task_id=f"fix_{i}", kind="agent", owner="builder", retries=1, description=f"Resolve the reported {what} failures")) as ph: previous = ph.call(AgentCall(output_type=BuildOutput, prompt=prompt, previous=quality.as_envelope(broken, what), gates=[gates.diff_matches_claims])) verified = (quality_result is not None and quality_result.passed and test_result is not None and test_result.passed) if verified: with run.phase(PhaseParams(name="commit", task_id="commit", kind="code", owner="git", description="Commit the tested and quality-verified working tree")) as ph: message = previous.commit_message or f"sssf({run.adw_id}): {previous.summary}" ph.log(sha=run.commit_all(message), message=message) return run.finish(accepted=verified, reason=f"verify/test never came back clean after {MAX_FIX_LOOPS} fix attempt(s)") if __name__ == "__main__": parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("prompt", help="inline text or a path to a prompt file") parser.add_argument("--preset", default=None) parser.add_argument("--account", default=None) parser.add_argument("--config", default="adws/adw_sssf_config/sssf.config.yaml") parser.add_argument("--adw-id", default=None, help="join or pin an existing session") parser.add_argument("--skip-plan", action="store_true", help="skip the planner agent; prompt must be a path to an authored " "docs/plans/ or docs/specs/ markdown doc") args = parser.parse_args() resolved = args.prompt if args.skip_plan else utils.resolve_prompt(args.prompt) sys.exit(main(resolved, args.config, args.adw_id, slug_hint=args.prompt, preset=args.preset, account=args.account, skip_plan=args.skip_plan))