from pathlib import Path import typer from .persistence import TaskStore app = typer.Typer(no_args_is_help=True, add_completion=False) FAKE_RESPONSES = { "planner": { "PLAN": '{"schema_version":1,"goal":"fake plan","acceptance_criteria":["tests pass"],"steps":[{"schema_version":1,"id":"P1","action":"noop","verification":"tests"}]}', "REVIEW": '{"schema_version":1,"verdict":"PASS","issues":[],"summary":"fake review"}', "FINAL_VERIFY": '{"schema_version":1,"verdict":"PASS","issues":[],"summary":"fake final"}', }, "implementer": { "PLAN_REVIEW": '{"schema_version":1,"verdict":"PASS","issues":[],"summary":"plan looks implementable"}', "IMPLEMENT": '{"schema_version":1,"files_changed":[],"tests_run":[],"notes":"fake"}', "FIX": '{"schema_version":1,"resolved":[],"not_resolved":[],"extra_changes":[],"tests_run":[]}', }, } def _store(root: Path) -> TaskStore: return TaskStore(root / ".dual-agent") def build_orchestrator(root: Path, repo: Path, fake: bool): """Assemble the orchestrator. Adapters come from the registry, gated on capability probes.""" from .infra.git import GitWorkspace from .policy import CommandPolicy from .registry import build from .runner import CommandTestRunner from .services import Orchestrator policy = CommandPolicy.discover(repo) _snapshot_config(root, policy) logs = root / ".dual-agent" / "logs" network = policy.policy.allow_network if fake: planner = build("fake", responses=FAKE_RESPONSES["planner"]) implementer = build("fake", responses=FAKE_RESPONSES["implementer"]) else: names = (policy.agents.planner, policy.agents.implementer) planner, implementer = ( build(name, log_root=logs, allow_network=network, **_binary(policy, name)) for name in names ) for name, adapter in zip(names, (planner, implementer)): capability = adapter.probe() if not capability.compatible: raise typer.BadParameter( f"{name} is missing required features: {', '.join(capability.missing_features)}" ) def build_tuned(role: str, tuning): """Rebuild one adapter with the model and effort a task asked for.""" name = policy.agents.planner if role == "planner" else policy.agents.implementer if fake: return planner if role == "planner" else implementer return build(name, log_root=logs, allow_network=network, **_binary(policy, name, tuning)) runner = CommandTestRunner(policy, logs, allow_network=network) return Orchestrator( _store(root), planner, implementer, GitWorkspace(), runner, require_approval=policy.policy.require_approval, allowed_repo_roots=policy.policy.allowed_repo_roots, implement_workers=policy.agents.implement_workers, limits=policy.limits, discussion_rounds=policy.agents.discussion_rounds, adapter_factory=build_tuned, ) def _binary(policy, name: str, tuning=None) -> dict: """Let the config point at an install that is not on PATH, and set model and effort.""" path = policy.agents.binaries.get(name) options = {"binary": path} if path else {} if name.startswith("codex"): options["review_sandbox"] = policy.agents.review_sandbox if policy.agents.passthrough_env: options["passthrough_env"] = tuple(policy.agents.passthrough_env) if name.startswith("claude") and policy.agents.settings_file: options["settings_file"] = policy.agents.settings_file chosen = tuning or policy.agents.tuning.get(name) if chosen: if chosen.model: options["model"] = chosen.model if chosen.effort: options["effort"] = chosen.effort return options def _snapshot_config(root: Path, policy) -> None: """Record the configuration actually in effect, as the design's config.yml.""" runtime = root / ".dual-agent" runtime.mkdir(parents=True, exist_ok=True) import yaml (runtime / "config.yml").write_text( yaml.safe_dump( { "commands": policy.commands, "limits": policy.limits.model_dump(), "policy": policy.policy.model_dump(), "agents": policy.agents.model_dump(), }, allow_unicode=True, sort_keys=True, ), encoding="utf-8", ) @app.command() def init(root: Path = typer.Argument(Path.cwd())) -> None: """Create the runtime directory and a starter command policy.""" runtime = root / ".dual-agent" runtime.mkdir(parents=True, exist_ok=True) config = root / ".dual-agent.yml" if not config.is_file(): config.write_text( "version: 1\ncommands:\n test: python -m pytest\nlimits:\n" " command_timeout_s: 900\n max_fix_cycles: 4\npolicy:\n" " allow_network: false\n allow_push: false\n", encoding="utf-8", ) typer.echo(str(runtime)) @app.command() def status(task_id: str, root: Path = typer.Option(Path.cwd())) -> None: typer.echo(_store(root).load(task_id).model_dump_json(indent=2)) @app.command("list") def list_tasks(root: Path = typer.Option(Path.cwd())) -> None: directory = _store(root).root / "tasks" for entry in sorted(directory.iterdir()) if directory.is_dir() else []: task = _store(root).load(entry.name) typer.echo(f"{task.state.value:<12} {task.task_id} {task.goal}") @app.command() def run( goal: str, repo: Path = typer.Option(Path.cwd(), help="target Git repository"), root: Path = typer.Option(Path.cwd()), fake: bool = typer.Option(False, help="use deterministic fake adapters"), ) -> None: """Create a task and drive it to a terminal state.""" from .domain import TaskSpec orchestrator = build_orchestrator(root, repo, fake) task = orchestrator.create_task(TaskSpec(repo_path=repo, goal=goal)) typer.echo(f"task {task.task_id}") final = orchestrator.run_to_completion(task.task_id) typer.echo(f"{final.state.value}: {final.error or 'ok'}") raise typer.Exit(0 if final.state.value == "DONE" else 1) @app.command() def resume( task_id: str, root: Path = typer.Option(Path.cwd()), repo: Path = typer.Option(Path.cwd()), fake: bool = typer.Option(False), ) -> None: orchestrator = build_orchestrator(root, repo, fake) task = orchestrator.recover(task_id) typer.echo(f"{task.state.value}: {task.error or 'resumable'}") @app.command() def cancel(task_id: str, root: Path = typer.Option(Path.cwd()), repo: Path = typer.Option(Path.cwd())) -> None: orchestrator = build_orchestrator(root, repo, fake=True) typer.echo(orchestrator.cancel(task_id).state.value) @app.command("recover-all") def recover_all(root: Path = typer.Option(Path.cwd()), repo: Path = typer.Option(Path.cwd())) -> None: """Sweep every unfinished task after a restart. Section 10.1 of the design.""" orchestrator = build_orchestrator(root, repo, fake=True) for task in orchestrator.recover_all(): typer.echo(f"{task.state.value:<12} {task.task_id} {task.error or 'resumable'}") @app.command() def probe() -> None: """Report what the installed Codex and Claude CLIs support.""" from .adapters.claude import ClaudeAdapter from .adapters.codex import CodexAdapter from .policy import CommandPolicy agents = CommandPolicy.discover(Path.cwd()).agents pairs = ( ("codex", CodexAdapter(**({"binary": agents.binaries["codex"]} if "codex" in agents.binaries else {}))), ("claude", ClaudeAdapter(**({"binary": agents.binaries["claude"]} if "claude" in agents.binaries else {}))), ) for name, adapter in pairs: capability = adapter.probe() missing = ", ".join(capability.missing_features) or "-" optional = ", ".join(k for k, v in capability.optional_features.items() if v) or "-" typer.echo(f"{name}: compatible={capability.compatible} missing={missing} optional={optional}") @app.command() def serve( root: Path = typer.Option(Path.cwd(), help="runtime root holding .dual-agent"), repo: Path = typer.Option(Path.cwd(), help="repository used for command discovery"), host: str = "127.0.0.1", port: int = 8765, fake: bool = typer.Option(False, help="use the deterministic fake adapters instead of real CLIs"), ) -> None: """Serve the web control UI. Uses the real providers unless --fake is given.""" import uvicorn from .advisor import Advisor from .api import create_app from .auth import ProviderAuth orchestrator = build_orchestrator(root, repo, fake) advisor = Advisor(orchestrator.store, orchestrator.planner, orchestrator.implementer, orchestrator) auth = ProviderAuth( {"codex": orchestrator.planner, "claude": orchestrator.implementer}, root / ".dual-agent" / "logs", ) typer.echo(f"http://{host}:{port}") uvicorn.run(create_app(orchestrator, advisor, auth), host=host, port=port) if __name__ == "__main__": app()