diff --git a/tests/test_cli_orchestrator_flags.py b/tests/test_cli_orchestrator_flags.py new file mode 100644 index 00000000..db952725 --- /dev/null +++ b/tests/test_cli_orchestrator_flags.py @@ -0,0 +1,76 @@ +from __future__ import annotations + +import json +from pathlib import Path + +from typer.testing import CliRunner + +from villani_code import cli + + +class _FakeRunner: + def __init__(self) -> None: + self._mission_id = "child1" + + def run(self, instruction: str): + return {"response": {"content": [{"type": "text", "text": '{"mode":"direct","subtasks":[]}'}, {"type": "text", "text": "ignored"}]}} + + +def test_run_writes_machine_result_artifact(monkeypatch, tmp_path: Path) -> None: + runner = CliRunner() + monkeypatch.setattr(cli, "_build_runner", lambda *args, **kwargs: _FakeRunner()) + parent_calls: list[str] = [] + monkeypatch.setattr(cli, "set_current_mission_id", lambda repo, mission_id: parent_calls.append(mission_id)) + + out = tmp_path / "result.json" + result = runner.invoke( + cli.app, + [ + "run", + "hello", + "--base-url", + "http://example.com", + "--model", + "x", + "--role", + "supervisor", + "--result-json-path", + str(out), + "--parent-mission-id", + "parent1", + ], + ) + assert result.exit_code == 0 + payload = json.loads(out.read_text(encoding="utf-8")) + assert payload["role"] == "supervisor" + assert payload["parent_mission_id"] == "parent1" + assert payload["response_json"]["mode"] == "direct" + assert parent_calls == ["parent1"] + + +def test_run_extracts_json_when_response_is_plain_string(monkeypatch, tmp_path: Path) -> None: + class _StringRunner: + _mission_id = "child2" + + def run(self, instruction: str): + return {"response": "```json\n{\"mode\":\"direct\",\"subtasks\":[]}\n```"} + + runner = CliRunner() + monkeypatch.setattr(cli, "_build_runner", lambda *args, **kwargs: _StringRunner()) + out = tmp_path / "result_string.json" + result = runner.invoke( + cli.app, + [ + "run", + "hello", + "--base-url", + "http://example.com", + "--model", + "x", + "--result-json-path", + str(out), + ], + ) + assert result.exit_code == 0 + payload = json.loads(out.read_text(encoding="utf-8")) + assert payload["response_json"]["mode"] == "direct" diff --git a/tests/test_orchestrator.py b/tests/test_orchestrator.py new file mode 100644 index 00000000..fc267961 --- /dev/null +++ b/tests/test_orchestrator.py @@ -0,0 +1,326 @@ +from __future__ import annotations + +import json +from pathlib import Path + +from villani_code import orchestrator +from villani_code.orchestrator_models import VerificationResult + + +def _fake_mission(repo: Path): + class _Mission: + mission_id = "m1" + + return _Mission() + + +def _patch_common(monkeypatch, tmp_path: Path) -> None: + monkeypatch.setattr(orchestrator, "create_mission_state", lambda repo, objective, mode: _fake_mission(repo)) + monkeypatch.setattr(orchestrator, "get_mission_dir", lambda repo, mission_id: tmp_path / ".villani_code" / "missions" / mission_id) + monkeypatch.setattr(orchestrator, "get_current_branch", lambda repo: "main") + monkeypatch.setattr(orchestrator, "get_head_commit", lambda repo: "abc123") + monkeypatch.setattr(orchestrator, "set_current_mission_id", lambda repo, mission_id: None) + + +def test_supervisor_direct_path(monkeypatch, tmp_path: Path) -> None: + _patch_common(monkeypatch, tmp_path) + monkeypatch.setattr(orchestrator, "create_worktree", lambda *args, **kwargs: (tmp_path / "wt1", "b1")) + monkeypatch.setattr(orchestrator, "commit_all", lambda *args, **kwargs: True) + monkeypatch.setattr(orchestrator, "merge_branch", lambda *args, **kwargs: (True, "ok")) + monkeypatch.setattr(orchestrator, "run_final_verification", lambda repo: VerificationResult("accepted", "ok", [], [])) + monkeypatch.setattr(orchestrator, "verify_worker_result", lambda *args, **kwargs: VerificationResult("accepted", "ok", [], ["a.py"])) + + def _run(**kwargs): + result_path = Path(kwargs["result_json_path"]) + result_path.parent.mkdir(parents=True, exist_ok=True) + payload = {"response_json": {"mode": "direct", "subtasks": []}} + if kwargs["role"] == "worker": + payload = {"response_json": {"status": "success", "recommended_verification": []}} + result_path.write_text(json.dumps(payload), encoding="utf-8") + return {"run_dir": str(result_path.parent / "run"), "result_path": str(result_path), "timed_out": False} + + monkeypatch.setattr(orchestrator, "run_villani_subprocess", _run) + + summary = orchestrator.run_orchestrator( + instruction="do it", + repo=tmp_path, + model="m", + base_url="", + provider="anthropic", + api_key=None, + max_tokens=100, + small_model=False, + debug_mode=False, + debug_dir=None, + max_subtasks=3, + max_worker_retries=1, + supervisor_timeout_seconds=60, + worker_timeout_seconds=60, + ) + assert summary["status"] == "completed" + + +def test_supervisor_split_path(monkeypatch, tmp_path: Path) -> None: + _patch_common(monkeypatch, tmp_path) + monkeypatch.setattr(orchestrator, "create_worktree", lambda *args, **kwargs: (tmp_path / f"wt-{args[3]}", f"b-{args[3]}")) + monkeypatch.setattr(orchestrator, "commit_all", lambda *args, **kwargs: True) + monkeypatch.setattr(orchestrator, "merge_branch", lambda *args, **kwargs: (True, "ok")) + monkeypatch.setattr(orchestrator, "run_final_verification", lambda repo: VerificationResult("accepted", "ok", [], [])) + monkeypatch.setattr(orchestrator, "verify_worker_result", lambda *args, **kwargs: VerificationResult("accepted", "ok", [], ["x.py"])) + + def _run(**kwargs): + result_path = Path(kwargs["result_json_path"]) + result_path.parent.mkdir(parents=True, exist_ok=True) + if kwargs["role"] == "supervisor": + payload = {"response_json": {"mode": "split", "subtasks": [{"id": "task_1", "goal": "g1", "success_criteria": [], "target_files": []}]}} + else: + payload = {"response_json": {"status": "success", "recommended_verification": []}} + result_path.write_text(json.dumps(payload), encoding="utf-8") + return {"run_dir": str(result_path.parent / "run"), "result_path": str(result_path), "timed_out": False} + + monkeypatch.setattr(orchestrator, "run_villani_subprocess", _run) + summary = orchestrator.run_orchestrator( + instruction="split", + repo=tmp_path, + model="m", + base_url="", + provider="anthropic", + api_key=None, + max_tokens=100, + small_model=False, + debug_mode=False, + debug_dir=None, + max_subtasks=3, + max_worker_retries=1, + supervisor_timeout_seconds=60, + worker_timeout_seconds=60, + ) + assert summary["status"] == "completed" + + +def test_invalid_supervisor_result_artifact(monkeypatch, tmp_path: Path) -> None: + _patch_common(monkeypatch, tmp_path) + calls = {"count": 0} + + def _run(**kwargs): + calls["count"] += 1 + path = Path(kwargs["result_json_path"]) + path.parent.mkdir(parents=True, exist_ok=True) + if calls["count"] == 1: + path.write_text("{}", encoding="utf-8") + else: + path.write_text(json.dumps({"response_json": {"mode": "direct", "subtasks": []}}), encoding="utf-8") + return {"run_dir": str(path.parent / "run"), "result_path": str(path), "timed_out": False} + + monkeypatch.setattr(orchestrator, "run_villani_subprocess", _run) + monkeypatch.setattr(orchestrator, "create_worktree", lambda *args, **kwargs: (tmp_path / "wt", "b")) + monkeypatch.setattr(orchestrator, "verify_worker_result", lambda *args, **kwargs: VerificationResult("hard_failure", "no diff", [], [])) + + summary = orchestrator.run_orchestrator( + instruction="x", repo=tmp_path, model="m", base_url="", provider="anthropic", api_key=None, max_tokens=1, small_model=False, + debug_mode=False, debug_dir=None, max_subtasks=3, max_worker_retries=1, supervisor_timeout_seconds=60, worker_timeout_seconds=60 + ) + assert calls["count"] >= 2 + assert summary["status"] == "failed" + + +def test_supervisor_invalid_twice_then_fail(monkeypatch, tmp_path: Path) -> None: + _patch_common(monkeypatch, tmp_path) + + def _run(**kwargs): + path = Path(kwargs["result_json_path"]) + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps({"response_json": {"mode": "oops", "subtasks": []}}), encoding="utf-8") + return {"run_dir": str(path.parent / "run"), "result_path": str(path), "timed_out": False} + + monkeypatch.setattr(orchestrator, "run_villani_subprocess", _run) + summary = orchestrator.run_orchestrator( + instruction="x", repo=tmp_path, model="m", base_url="", provider="anthropic", api_key=None, max_tokens=1, small_model=False, + debug_mode=False, debug_dir=None, max_subtasks=3, max_worker_retries=1, supervisor_timeout_seconds=60, worker_timeout_seconds=60 + ) + assert summary["status"] == "failed" + assert "Supervisor failed" in summary["summary"] + + +def test_worker_retry_flow(monkeypatch, tmp_path: Path) -> None: + _patch_common(monkeypatch, tmp_path) + monkeypatch.setattr(orchestrator, "create_worktree", lambda *args, **kwargs: (tmp_path / "wt", "b")) + monkeypatch.setattr(orchestrator, "commit_all", lambda *args, **kwargs: True) + monkeypatch.setattr(orchestrator, "merge_branch", lambda *args, **kwargs: (True, "ok")) + monkeypatch.setattr(orchestrator, "run_final_verification", lambda repo: VerificationResult("accepted", "ok", [], [])) + + attempts = {"n": 0} + + def _verify(*args, **kwargs): + attempts["n"] += 1 + return VerificationResult("retryable_failure", "retry", [], []) if attempts["n"] == 1 else VerificationResult("accepted", "ok", [], ["f.py"]) + + monkeypatch.setattr(orchestrator, "verify_worker_result", _verify) + + def _run(**kwargs): + path = Path(kwargs["result_json_path"]) + path.parent.mkdir(parents=True, exist_ok=True) + payload = {"response_json": {"mode": "direct", "subtasks": []}} if kwargs["role"] == "supervisor" else {"response_json": {"status": "failed"}} + path.write_text(json.dumps(payload), encoding="utf-8") + return {"run_dir": str(path.parent / "run"), "result_path": str(path), "timed_out": False} + + monkeypatch.setattr(orchestrator, "run_villani_subprocess", _run) + summary = orchestrator.run_orchestrator( + instruction="x", repo=tmp_path, model="m", base_url="", provider="anthropic", api_key=None, max_tokens=1, small_model=False, + debug_mode=False, debug_dir=None, max_subtasks=3, max_worker_retries=1, supervisor_timeout_seconds=60, worker_timeout_seconds=60 + ) + assert attempts["n"] == 2 + assert summary["status"] == "completed" + + +def test_worker_blocked_environment_handling(monkeypatch, tmp_path: Path) -> None: + _patch_common(monkeypatch, tmp_path) + monkeypatch.setattr(orchestrator, "create_worktree", lambda *args, **kwargs: (tmp_path / "wt", "b")) + monkeypatch.setattr(orchestrator, "verify_worker_result", lambda *args, **kwargs: VerificationResult("retryable_failure", "env", [], [])) + + def _run(**kwargs): + path = Path(kwargs["result_json_path"]) + path.parent.mkdir(parents=True, exist_ok=True) + payload = {"response_json": {"mode": "direct", "subtasks": []}} if kwargs["role"] == "supervisor" else {"response_json": {"status": "blocked_environment"}} + path.write_text(json.dumps(payload), encoding="utf-8") + return {"run_dir": str(path.parent / "run"), "result_path": str(path), "timed_out": False} + + monkeypatch.setattr(orchestrator, "run_villani_subprocess", _run) + summary = orchestrator.run_orchestrator( + instruction="x", repo=tmp_path, model="m", base_url="", provider="anthropic", api_key=None, max_tokens=1, small_model=False, + debug_mode=False, debug_dir=None, max_subtasks=3, max_worker_retries=0, supervisor_timeout_seconds=60, worker_timeout_seconds=60 + ) + assert summary["status"] == "failed" + + +def test_worker_blocked_scope_handling(monkeypatch, tmp_path: Path) -> None: + _patch_common(monkeypatch, tmp_path) + monkeypatch.setattr(orchestrator, "create_worktree", lambda *args, **kwargs: (tmp_path / "wt", "b")) + monkeypatch.setattr(orchestrator, "verify_worker_result", lambda *args, **kwargs: VerificationResult("hard_failure", "scope", [], [])) + + def _run(**kwargs): + path = Path(kwargs["result_json_path"]) + path.parent.mkdir(parents=True, exist_ok=True) + payload = {"response_json": {"mode": "direct", "subtasks": []}} if kwargs["role"] == "supervisor" else {"response_json": {"status": "blocked_scope"}} + path.write_text(json.dumps(payload), encoding="utf-8") + return {"run_dir": str(path.parent / "run"), "result_path": str(path), "timed_out": False} + + monkeypatch.setattr(orchestrator, "run_villani_subprocess", _run) + summary = orchestrator.run_orchestrator( + instruction="x", repo=tmp_path, model="m", base_url="", provider="anthropic", api_key=None, max_tokens=1, small_model=False, + debug_mode=False, debug_dir=None, max_subtasks=3, max_worker_retries=1, supervisor_timeout_seconds=60, worker_timeout_seconds=60 + ) + assert summary["status"] == "failed" + + +def test_worker_timeout_handling(monkeypatch, tmp_path: Path) -> None: + _patch_common(monkeypatch, tmp_path) + monkeypatch.setattr(orchestrator, "create_worktree", lambda *args, **kwargs: (tmp_path / "wt", "b")) + + def _run(**kwargs): + path = Path(kwargs["result_json_path"]) + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps({"response_json": {"mode": "direct", "subtasks": []} if kwargs["role"] == "supervisor" else {"status": "failed"}}), encoding="utf-8") + return {"run_dir": str(path.parent / "run"), "result_path": str(path), "timed_out": kwargs["role"] == "worker"} + + monkeypatch.setattr(orchestrator, "run_villani_subprocess", _run) + summary = orchestrator.run_orchestrator( + instruction="x", repo=tmp_path, model="m", base_url="", provider="anthropic", api_key=None, max_tokens=1, small_model=False, + debug_mode=False, debug_dir=None, max_subtasks=3, max_worker_retries=0, supervisor_timeout_seconds=60, worker_timeout_seconds=60 + ) + assert summary["status"] == "failed" + + +def test_merge_failure_handling(monkeypatch, tmp_path: Path) -> None: + _patch_common(monkeypatch, tmp_path) + monkeypatch.setattr(orchestrator, "create_worktree", lambda *args, **kwargs: (tmp_path / "wt", "b")) + monkeypatch.setattr(orchestrator, "verify_worker_result", lambda *args, **kwargs: VerificationResult("accepted", "ok", [], ["f.py"])) + monkeypatch.setattr(orchestrator, "commit_all", lambda *args, **kwargs: True) + monkeypatch.setattr(orchestrator, "merge_branch", lambda *args, **kwargs: (False, "conflict")) + + def _run(**kwargs): + path = Path(kwargs["result_json_path"]) + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps({"response_json": {"mode": "direct", "subtasks": []} if kwargs["role"] == "supervisor" else {"status": "success"}}), encoding="utf-8") + return {"run_dir": str(path.parent / "run"), "result_path": str(path), "timed_out": False} + + monkeypatch.setattr(orchestrator, "run_villani_subprocess", _run) + summary = orchestrator.run_orchestrator( + instruction="x", repo=tmp_path, model="m", base_url="", provider="anthropic", api_key=None, max_tokens=1, small_model=False, + debug_mode=False, debug_dir=None, max_subtasks=3, max_worker_retries=0, supervisor_timeout_seconds=60, worker_timeout_seconds=60 + ) + assert summary["status"] == "failed" + + +def test_final_verification_failure_handling(monkeypatch, tmp_path: Path) -> None: + _patch_common(monkeypatch, tmp_path) + monkeypatch.setattr(orchestrator, "create_worktree", lambda *args, **kwargs: (tmp_path / "wt", "b")) + monkeypatch.setattr(orchestrator, "verify_worker_result", lambda *args, **kwargs: VerificationResult("accepted", "ok", [], ["f.py"])) + monkeypatch.setattr(orchestrator, "commit_all", lambda *args, **kwargs: True) + monkeypatch.setattr(orchestrator, "merge_branch", lambda *args, **kwargs: (True, "ok")) + monkeypatch.setattr(orchestrator, "run_final_verification", lambda repo: VerificationResult("hard_failure", "bad", [], [])) + + def _run(**kwargs): + path = Path(kwargs["result_json_path"]) + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps({"response_json": {"mode": "direct", "subtasks": []} if kwargs["role"] == "supervisor" else {"status": "success"}}), encoding="utf-8") + return {"run_dir": str(path.parent / "run"), "result_path": str(path), "timed_out": False} + + monkeypatch.setattr(orchestrator, "run_villani_subprocess", _run) + summary = orchestrator.run_orchestrator( + instruction="x", repo=tmp_path, model="m", base_url="", provider="anthropic", api_key=None, max_tokens=1, small_model=False, + debug_mode=False, debug_dir=None, max_subtasks=3, max_worker_retries=0, supervisor_timeout_seconds=60, worker_timeout_seconds=60 + ) + assert summary["status"] == "failed" + + +def test_no_final_verification_when_no_worker_accepted(monkeypatch, tmp_path: Path) -> None: + _patch_common(monkeypatch, tmp_path) + monkeypatch.setattr(orchestrator, "create_worktree", lambda *args, **kwargs: (tmp_path / "wt", "b")) + monkeypatch.setattr(orchestrator, "verify_worker_result", lambda *args, **kwargs: VerificationResult("hard_failure", "bad", [], [])) + + called = {"n": 0} + + def _final(repo): + called["n"] += 1 + return VerificationResult("accepted", "ok", [], []) + + monkeypatch.setattr(orchestrator, "run_final_verification", _final) + + def _run(**kwargs): + path = Path(kwargs["result_json_path"]) + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps({"response_json": {"mode": "direct", "subtasks": []} if kwargs["role"] == "supervisor" else {"status": "failed"}}), encoding="utf-8") + return {"run_dir": str(path.parent / "run"), "result_path": str(path), "timed_out": False} + + monkeypatch.setattr(orchestrator, "run_villani_subprocess", _run) + summary = orchestrator.run_orchestrator( + instruction="x", repo=tmp_path, model="m", base_url="", provider="anthropic", api_key=None, max_tokens=1, small_model=False, + debug_mode=False, debug_dir=None, max_subtasks=3, max_worker_retries=0, supervisor_timeout_seconds=60, worker_timeout_seconds=60 + ) + assert summary["status"] == "failed" + assert called["n"] == 0 + + +def test_supervisor_subprocess_failure_reports_real_error(monkeypatch, tmp_path: Path) -> None: + _patch_common(monkeypatch, tmp_path) + + def _run(**kwargs): + path = Path(kwargs["result_json_path"]) + path.parent.mkdir(parents=True, exist_ok=True) + return { + "exit_code": 2, + "run_dir": str(path.parent / "run"), + "result_path": str(path), + "stdout": "", + "stderr": "No such option: --foo", + "timed_out": False, + } + + monkeypatch.setattr(orchestrator, "run_villani_subprocess", _run) + summary = orchestrator.run_orchestrator( + instruction="x", repo=tmp_path, model="m", base_url="", provider="anthropic", api_key=None, max_tokens=1, small_model=False, + debug_mode=False, debug_dir=None, max_subtasks=3, max_worker_retries=0, supervisor_timeout_seconds=60, worker_timeout_seconds=60 + ) + assert summary["status"] == "failed" + assert "exit code 2" in summary["summary"] diff --git a/villani_code/cli.py b/villani_code/cli.py index b5634b3e..3ec1222a 100644 --- a/villani_code/cli.py +++ b/villani_code/cli.py @@ -19,6 +19,8 @@ from villani_code.benchmark.runtime_config import BenchmarkRuntimeConfig from villani_code.debug_bundle import create_debug_bundle from villani_code.debug_mode import DebugMode, build_debug_config +from villani_code.mission_state import set_current_mission_id +from villani_code.orchestrator import run_orchestrator from villani_code.trace_summary import write_summary_from_events, write_tool_calls_from_events app = typer.Typer(help="Villani: constrained-inference coding agent with visible context governance") @@ -68,6 +70,64 @@ def _print_content(value: Any) -> None: except Exception: # noqa: BLE001 return + +def _extract_machine_response_json(result: dict[str, Any] | None) -> dict[str, Any]: + def _candidate_texts(value: Any) -> list[str]: + texts: list[str] = [] + if isinstance(value, str): + texts.append(value) + return texts + if not isinstance(value, list): + return texts + for block in value: + if isinstance(block, str): + texts.append(block) + continue + if not isinstance(block, dict): + continue + if block.get("type") != "text": + continue + text = block.get("text") + if isinstance(text, str): + texts.append(text) + return texts + + def _parse_json_blob(text: str) -> dict[str, Any] | None: + candidate = text.strip() + if not candidate: + return None + try: + payload = json.loads(candidate) + if isinstance(payload, dict): + return payload + except json.JSONDecodeError: + pass + start = candidate.find("{") + end = candidate.rfind("}") + if start < 0 or end <= start: + return None + try: + payload = json.loads(candidate[start:end + 1]) + except json.JSONDecodeError: + return None + return payload if isinstance(payload, dict) else None + + if not isinstance(result, dict): + return {} + candidates: list[str] = [] + response = result.get("response") + if isinstance(response, str): + candidates.append(response) + elif isinstance(response, dict): + candidates.extend(_candidate_texts(response.get("content"))) + candidates.extend(_candidate_texts(result.get("content"))) + for text in candidates: + payload = _parse_json_blob(text) + if payload is not None: + return payload + return {} + + def _load_settings_manager() -> Any | None: try: from villani_code.tui.components.settings import SettingsManager @@ -197,17 +257,74 @@ def run( provider: Literal["anthropic", "openai"] = typer.Option("anthropic", "--provider"), api_key: Optional[str] = typer.Option(None, "--api-key"), benchmark_runtime_json: Optional[str] = typer.Option(None, "--benchmark-runtime-json", hidden=True), + role: Optional[str] = typer.Option(None, "--role", hidden=True), + result_json_path: Optional[Path] = typer.Option(None, "--result-json-path", hidden=True), + parent_mission_id: Optional[str] = typer.Option(None, "--parent-mission-id", hidden=True), debug: Optional[str] = typer.Option(None, "--debug", flag_value="normal"), debug_dir: Optional[Path] = typer.Option(None, "--debug-dir"), ) -> None: + effective_instruction = instruction + if role: + effective_instruction = f"[ORCHESTRATOR ROLE={role}]\\n{instruction}" debug_mode = DebugMode(build_debug_config(debug).mode.value) runner = _build_runner(base_url, model, repo, max_tokens, stream, thinking, unsafe, verbose, extra_json, redact, dangerously_skip_permissions, auto_accept_edits, auto_approve, plan_mode, max_repair_attempts, small_model, provider, api_key, benchmark_runtime_json=benchmark_runtime_json, debug_mode=debug_mode, debug_dir=debug_dir) if auto_approve: console.print("Auto-approval: ON") - result = runner.run(instruction) + result = runner.run(effective_instruction) + if parent_mission_id: + set_current_mission_id(repo.resolve(), parent_mission_id) + if result_json_path: + payload = { + "status": "ok", + "role": role or "", + "parent_mission_id": parent_mission_id or "", + "mission_id": getattr(runner, "_mission_id", "") or "", + "response_json": _extract_machine_response_json(result), + } + result_json_path.parent.mkdir(parents=True, exist_ok=True) + result_json_path.write_text(json.dumps(payload, indent=2), encoding="utf-8") _print_response_text_blocks(result) +@app.command() +def orchestrate( + instruction: str = typer.Argument(..., help="Top-level orchestration objective"), + base_url: str = typer.Option(..., "--base-url", help="Base URL for compatible messages API server"), + model: str = typer.Option(..., "--model", help="Model name"), + repo: Path = typer.Option(Path("."), "--repo", help="Repository path"), + max_tokens: int = typer.Option(4096, "--max-tokens"), + small_model: bool = typer.Option(False, "--small-model"), + provider: Literal["anthropic", "openai"] = typer.Option("anthropic", "--provider"), + api_key: Optional[str] = typer.Option(None, "--api-key"), + max_subtasks: int = typer.Option(3, "--max-subtasks"), + max_worker_retries: int = typer.Option(1, "--max-worker-retries"), + supervisor_timeout_seconds: int = typer.Option(300, "--supervisor-timeout-seconds"), + worker_timeout_seconds: int = typer.Option(600, "--worker-timeout-seconds"), + debug: Optional[str] = typer.Option(None, "--debug", flag_value="normal"), + debug_dir: Optional[Path] = typer.Option(None, "--debug-dir"), +) -> None: + debug_mode = DebugMode(build_debug_config(debug).mode.value) + summary = run_orchestrator( + instruction=instruction, + repo=repo, + model=model, + base_url=base_url, + provider=provider, + api_key=api_key, + max_tokens=max_tokens, + small_model=small_model, + debug_mode=debug_mode != DebugMode.OFF, + debug_dir=debug_dir, + max_subtasks=max_subtasks, + max_worker_retries=max_worker_retries, + supervisor_timeout_seconds=supervisor_timeout_seconds, + worker_timeout_seconds=worker_timeout_seconds, + ) + console.print(f"Orchestration status: {summary.get('status', 'unknown')}") + console.print(f"Mission: {summary.get('mission_id', '')}") + console.print(f"Summary: {summary.get('summary', '')}") + + @app.command() def interactive( base_url: str = typer.Option(..., "--base-url"), diff --git a/villani_code/orchestrator.py b/villani_code/orchestrator.py new file mode 100644 index 00000000..8f672d78 --- /dev/null +++ b/villani_code/orchestrator.py @@ -0,0 +1,355 @@ +from __future__ import annotations + +import json +import subprocess +import sys +from pathlib import Path +from typing import Any + +from villani_code.mission_state import create_mission_state, get_mission_dir, set_current_mission_id +from villani_code.orchestrator_git import commit_all, create_worktree, get_current_branch, get_head_commit, merge_branch +from villani_code.orchestrator_models import ( + OrchestratorState, + SupervisorPlan, + WorkerAttempt, + WorkerRunRecord, + WorkerTask, + save_orchestrator_state, + supervisor_plan_from_dict, +) +from villani_code.orchestrator_roles import build_supervisor_prompt, build_worker_prompt +from villani_code.orchestrator_verify import run_final_verification, verify_worker_result + + +def _validate_supervisor_plan(plan: SupervisorPlan, max_subtasks: int) -> bool: + if plan.mode not in {"direct", "split"}: + return False + if plan.mode == "direct": + return len(plan.subtasks) == 0 + if len(plan.subtasks) > max_subtasks: + return False + return all(task.id and task.goal for task in plan.subtasks) + + +def _load_json(path: Path) -> dict[str, Any] | None: + if not path.exists(): + return None + try: + payload = json.loads(path.read_text(encoding="utf-8")) + except json.JSONDecodeError: + return None + if not isinstance(payload, dict): + return None + return payload + + +def _subprocess_failure_summary(run: dict[str, Any], phase: str) -> str: + if run.get("timed_out"): + return f"{phase} subprocess timed out" + exit_code = int(run.get("exit_code", 0)) + if exit_code == 0: + return "" + stderr = str(run.get("stderr", "")).strip() + stdout = str(run.get("stdout", "")).strip() + details = stderr or stdout or "no stderr/stdout captured" + return f"{phase} subprocess failed with exit code {exit_code}: {details[:500]}" + + +def run_villani_subprocess( + *, + instruction: str, + repo: Path, + base_url: str, + model: str, + provider: str, + api_key: str | None, + max_tokens: int, + small_model: bool, + debug_mode: bool, + debug_dir: Path | None, + role: str, + result_json_path: Path, + parent_mission_id: str, + timeout_seconds: int, +) -> dict[str, Any]: + run_dir = result_json_path.parent / "run" + run_dir.mkdir(parents=True, exist_ok=True) + command = [ + sys.executable, + "-m", + "villani_code.cli", + "run", + instruction, + "--repo", + str(repo), + "--provider", + provider, + "--model", + model, + "--max-tokens", + str(max_tokens), + "--no-stream", + "--role", + role, + "--result-json-path", + str(result_json_path), + "--parent-mission-id", + parent_mission_id, + ] + if base_url: + command.extend(["--base-url", base_url]) + if api_key: + command.extend(["--api-key", api_key]) + if small_model: + command.append("--small-model") + if debug_mode: + command.append("--debug") + if debug_dir: + command.extend(["--debug-dir", str(debug_dir)]) + + try: + proc = subprocess.run(command, cwd=repo, capture_output=True, text=True, check=False, timeout=timeout_seconds) + return { + "exit_code": proc.returncode, + "run_dir": str(run_dir), + "result_path": str(result_json_path), + "stdout": proc.stdout, + "stderr": proc.stderr, + "timed_out": False, + } + except subprocess.TimeoutExpired as exc: + return { + "exit_code": -1, + "run_dir": str(run_dir), + "result_path": str(result_json_path), + "stdout": (exc.stdout or "") if isinstance(exc.stdout, str) else "", + "stderr": (exc.stderr or "") if isinstance(exc.stderr, str) else "", + "timed_out": True, + } + + +def run_orchestrator( + *, + instruction: str, + repo: Path, + model: str, + base_url: str, + provider: str, + api_key: str | None, + max_tokens: int, + small_model: bool, + debug_mode: bool, + debug_dir: Path | None, + max_subtasks: int, + max_worker_retries: int, + supervisor_timeout_seconds: int, + worker_timeout_seconds: int, +) -> dict[str, Any]: + resolved_repo = repo.resolve() + mission = create_mission_state(resolved_repo, instruction, mode="orchestrator") + mission_id = mission.mission_id + mission_dir = get_mission_dir(resolved_repo, mission_id) + orch_dir = mission_dir / "orchestrator" + state_path = orch_dir / "orchestrator_state.json" + orch_dir.mkdir(parents=True, exist_ok=True) + (orch_dir / "top_level_objective.txt").write_text(instruction, encoding="utf-8") + + state = OrchestratorState( + mission_id=mission_id, + objective=instruction, + repo_root=str(resolved_repo), + started_from_branch=get_current_branch(resolved_repo), + base_commit=get_head_commit(resolved_repo), + status="running", + ) + save_orchestrator_state(state_path, state) + + supervisor_result_path = orch_dir / "supervisor" / "result.json" + supervisor_prompt = build_supervisor_prompt(instruction, max_subtasks=max_subtasks) + sup = run_villani_subprocess( + instruction=supervisor_prompt, + repo=resolved_repo, + base_url=base_url, + model=model, + provider=provider, + api_key=api_key, + max_tokens=max_tokens, + small_model=small_model, + debug_mode=debug_mode, + debug_dir=debug_dir, + role="supervisor", + result_json_path=supervisor_result_path, + parent_mission_id=mission_id, + timeout_seconds=supervisor_timeout_seconds, + ) + state.supervisor_run_dir = str((orch_dir / "supervisor" / "run")) + state.supervisor_result_json_path = str(supervisor_result_path) + save_orchestrator_state(state_path, state) + failure_summary = _subprocess_failure_summary(sup, "Supervisor") + if failure_summary: + state.status = "failed" + state.final_summary = failure_summary + save_orchestrator_state(state_path, state) + return {"status": state.status, "summary": state.final_summary, "mission_id": mission_id} + + sup_payload = _load_json(supervisor_result_path) + plan = supervisor_plan_from_dict((sup_payload or {}).get("response_json", {})) + if not _validate_supervisor_plan(plan, max_subtasks=max_subtasks): + retry_prompt = f"Return valid strict JSON only for this objective:\n{instruction}" + _ = run_villani_subprocess( + instruction=retry_prompt, + repo=resolved_repo, + base_url=base_url, + model=model, + provider=provider, + api_key=api_key, + max_tokens=max_tokens, + small_model=small_model, + debug_mode=debug_mode, + debug_dir=debug_dir, + role="supervisor", + result_json_path=supervisor_result_path, + parent_mission_id=mission_id, + timeout_seconds=supervisor_timeout_seconds, + ) + sup_payload = _load_json(supervisor_result_path) + plan = supervisor_plan_from_dict((sup_payload or {}).get("response_json", {})) + if not _validate_supervisor_plan(plan, max_subtasks=max_subtasks): + state.status = "failed" + state.final_summary = "Supervisor failed to produce a valid plan twice" + save_orchestrator_state(state_path, state) + return {"status": state.status, "summary": state.final_summary, "mission_id": mission_id} + + tasks = plan.subtasks if plan.mode == "split" else [ + WorkerTask( + id="task_1", + goal=instruction, + success_criteria=[], + target_files=[], + scope_hint="", + ) + ] + + accepted: list[WorkerRunRecord] = [] + for task in tasks: + worktree_path, branch_name = create_worktree(resolved_repo, mission_dir, mission_id, task.id, state.base_commit) + record = WorkerRunRecord(task_id=task.id, goal=task.goal, worktree_path=str(worktree_path), branch_name=branch_name) + state.tasks.append(record) + save_orchestrator_state(state_path, state) + + attempt_num = 1 + previous_failure: str | None = None + while attempt_num <= (max_worker_retries + 1): + worker_dir = orch_dir / "workers" / task.id / f"attempt_{attempt_num}" + result_path = worker_dir / "result.json" + verification_path = worker_dir / "verification.json" + prompt = build_worker_prompt(instruction, task, attempt_num, previous_failure=previous_failure) + run = run_villani_subprocess( + instruction=prompt, + repo=worktree_path, + base_url=base_url, + model=model, + provider=provider, + api_key=api_key, + max_tokens=max_tokens, + small_model=small_model, + debug_mode=debug_mode, + debug_dir=debug_dir, + role="worker", + result_json_path=result_path, + parent_mission_id=mission_id, + timeout_seconds=worker_timeout_seconds, + ) + subprocess_failure = _subprocess_failure_summary(run, f"Worker {task.id} attempt {attempt_num}") + result_payload = _load_json(result_path) or {} + response_payload = result_payload.get("response_json", {}) + recommended = response_payload.get("recommended_verification", []) if isinstance(response_payload, dict) else [] + if subprocess_failure: + verification = {"status": "retryable_failure", "summary": subprocess_failure, "commands_run": [], "files_touched": []} + else: + verification_obj = verify_worker_result(worktree_path, task, recommended if isinstance(recommended, list) else None) + verification = { + "status": verification_obj.status, + "summary": verification_obj.summary, + "commands_run": verification_obj.commands_run, + "files_touched": verification_obj.files_touched, + } + verification_path.parent.mkdir(parents=True, exist_ok=True) + verification_path.write_text(json.dumps(verification, indent=2), encoding="utf-8") + + attempt = WorkerAttempt( + attempt=attempt_num, + run_dir=run["run_dir"], + result_json_path=run["result_path"], + status=verification["status"], + verification_summary=verification["summary"], + ) + record.attempts.append(attempt) + save_orchestrator_state(state_path, state) + + if verification["status"] == "accepted": + committed = commit_all(worktree_path, f"orchestrator({task.id}): {task.goal[:72]}") + if committed: + accepted.append(record) + else: + record.merge_status = "failed" + break + + previous_failure = verification["summary"] + if verification["status"] != "retryable_failure" or attempt_num >= (max_worker_retries + 1): + record.merge_status = "rejected" + break + attempt_num += 1 + + merge_log: list[dict[str, Any]] = [] + merged_any = False + for record in sorted(accepted, key=lambda item: item.task_id): + ok, message = merge_branch(resolved_repo, record.branch_name) + merge_log.append({"task_id": record.task_id, "branch_name": record.branch_name, "ok": ok, "message": message}) + record.merge_status = "merged" if ok else "merge_failed" + save_orchestrator_state(state_path, state) + if not ok: + state.status = "failed" + state.final_summary = f"Merge failed for {record.task_id}" + (orch_dir / "merges").mkdir(parents=True, exist_ok=True) + (orch_dir / "merges" / "merge_log.json").write_text(json.dumps(merge_log, indent=2), encoding="utf-8") + save_orchestrator_state(state_path, state) + return {"status": state.status, "summary": state.final_summary, "mission_id": mission_id} + merged_any = True + + (orch_dir / "merges").mkdir(parents=True, exist_ok=True) + (orch_dir / "merges" / "merge_log.json").write_text(json.dumps(merge_log, indent=2), encoding="utf-8") + + if merged_any: + final = run_final_verification(resolved_repo) + (orch_dir / "final_verification.json").write_text( + json.dumps( + { + "status": final.status, + "summary": final.summary, + "commands_run": final.commands_run, + "files_touched": final.files_touched, + }, + indent=2, + ), + encoding="utf-8", + ) + state.final_verification_status = final.status + if final.status != "accepted": + state.status = "failed" + state.final_summary = "Final verification failed" + else: + state.status = "completed" + state.final_summary = "Orchestration completed successfully" + else: + state.final_verification_status = "skipped" + state.status = "failed" + state.final_summary = "No worker changes accepted" + + set_current_mission_id(resolved_repo, mission_id) + (orch_dir / "final_summary.json").write_text( + json.dumps({"status": state.status, "summary": state.final_summary, "mission_id": mission_id}, indent=2), + encoding="utf-8", + ) + save_orchestrator_state(state_path, state) + return {"status": state.status, "summary": state.final_summary, "mission_id": mission_id} diff --git a/villani_code/orchestrator_git.py b/villani_code/orchestrator_git.py new file mode 100644 index 00000000..e390d405 --- /dev/null +++ b/villani_code/orchestrator_git.py @@ -0,0 +1,85 @@ +from __future__ import annotations + +import subprocess +from pathlib import Path + + +def _git(repo: Path, args: list[str]) -> subprocess.CompletedProcess[str]: + return subprocess.run(["git", *args], cwd=repo, capture_output=True, text=True, check=False) + + +def get_current_branch(repo: Path) -> str: + proc = _git(repo, ["rev-parse", "--abbrev-ref", "HEAD"]) + if proc.returncode != 0: + raise RuntimeError(proc.stderr.strip() or "Failed to get current branch") + return proc.stdout.strip() + + +def get_head_commit(repo: Path) -> str: + proc = _git(repo, ["rev-parse", "HEAD"]) + if proc.returncode != 0: + raise RuntimeError(proc.stderr.strip() or "Failed to get head commit") + return proc.stdout.strip() + + +def create_worktree(repo: Path, mission_dir: Path, mission_id: str, task_id: str, base_commit: str) -> tuple[Path, str]: + worktree = mission_dir / "orchestrator" / "worktrees" / task_id + worktree.parent.mkdir(parents=True, exist_ok=True) + branch_name = f"villani-orch-{mission_id}-{task_id}" + proc = _git(repo, ["worktree", "add", "-B", branch_name, str(worktree), base_commit]) + if proc.returncode != 0: + raise RuntimeError(proc.stderr.strip() or "Failed to create worktree") + return worktree, branch_name + + +def remove_worktree(worktree: Path) -> None: + proc = _git(worktree, ["worktree", "remove", "--force", str(worktree)]) + if proc.returncode != 0: + raise RuntimeError(proc.stderr.strip() or "Failed to remove worktree") + + +def commit_all(worktree: Path, message: str) -> bool: + if not has_diff(worktree): + return False + add_proc = _git(worktree, ["add", "-A"]) + if add_proc.returncode != 0: + return False + commit_proc = _git(worktree, ["commit", "-m", message]) + return commit_proc.returncode == 0 + + +def merge_branch(repo: Path, branch_name: str) -> tuple[bool, str]: + proc = _git(repo, ["merge", "--no-ff", "--no-edit", branch_name]) + return proc.returncode == 0, (proc.stdout + "\n" + proc.stderr).strip() + + +def changed_files(worktree: Path) -> list[str]: + proc = _git(worktree, ["diff", "--name-only"]) + if proc.returncode != 0: + return [] + return [line.strip() for line in proc.stdout.splitlines() if line.strip()] + + +def has_diff(worktree: Path) -> bool: + proc = _git(worktree, ["status", "--porcelain"]) + if proc.returncode != 0: + return False + return bool(proc.stdout.strip()) + + +def diff_line_count(worktree: Path) -> int: + proc = _git(worktree, ["diff", "--numstat"]) + if proc.returncode != 0: + return 0 + count = 0 + for line in proc.stdout.splitlines(): + parts = line.split("\t") + if len(parts) < 2: + continue + try: + add = 0 if parts[0] == "-" else int(parts[0]) + delete = 0 if parts[1] == "-" else int(parts[1]) + except ValueError: + continue + count += add + delete + return count diff --git a/villani_code/orchestrator_models.py b/villani_code/orchestrator_models.py new file mode 100644 index 00000000..6d020ab7 --- /dev/null +++ b/villani_code/orchestrator_models.py @@ -0,0 +1,138 @@ +from __future__ import annotations + +import json +from dataclasses import asdict, dataclass, field +from pathlib import Path +from typing import Any + + +@dataclass(slots=True) +class WorkerTask: + id: str + goal: str + success_criteria: list[str] + target_files: list[str] + scope_hint: str = "" + + +@dataclass(slots=True) +class SupervisorPlan: + mode: str + subtasks: list[WorkerTask] + + +@dataclass(slots=True) +class WorkerAttempt: + attempt: int + run_dir: str + result_json_path: str + status: str + verification_summary: str = "" + + +@dataclass(slots=True) +class WorkerRunRecord: + task_id: str + goal: str + worktree_path: str + branch_name: str + attempts: list[WorkerAttempt] = field(default_factory=list) + merge_status: str = "pending" + + +@dataclass(slots=True) +class VerificationResult: + status: str + summary: str + commands_run: list[str] + files_touched: list[str] + + +@dataclass(slots=True) +class OrchestratorState: + mission_id: str + objective: str + repo_root: str + started_from_branch: str + base_commit: str + status: str + supervisor_run_dir: str = "" + supervisor_result_json_path: str = "" + tasks: list[WorkerRunRecord] = field(default_factory=list) + final_verification_status: str = "" + final_summary: str = "" + + +def _worker_task_from_dict(payload: dict[str, Any]) -> WorkerTask: + return WorkerTask( + id=str(payload.get("id", "")), + goal=str(payload.get("goal", "")), + success_criteria=[str(v) for v in payload.get("success_criteria", [])], + target_files=[str(v) for v in payload.get("target_files", [])], + scope_hint=str(payload.get("scope_hint", "")), + ) + + +def supervisor_plan_from_dict(payload: dict[str, Any]) -> SupervisorPlan: + return SupervisorPlan( + mode=str(payload.get("mode", "")), + subtasks=[_worker_task_from_dict(item) for item in payload.get("subtasks", []) if isinstance(item, dict)], + ) + + +def orchestrator_state_to_dict(state: OrchestratorState) -> dict[str, Any]: + return asdict(state) + + +def orchestrator_state_from_dict(payload: dict[str, Any]) -> OrchestratorState: + tasks: list[WorkerRunRecord] = [] + for item in payload.get("tasks", []): + if not isinstance(item, dict): + continue + attempts = [ + WorkerAttempt( + attempt=int(attempt.get("attempt", 0)), + run_dir=str(attempt.get("run_dir", "")), + result_json_path=str(attempt.get("result_json_path", "")), + status=str(attempt.get("status", "")), + verification_summary=str(attempt.get("verification_summary", "")), + ) + for attempt in item.get("attempts", []) + if isinstance(attempt, dict) + ] + tasks.append( + WorkerRunRecord( + task_id=str(item.get("task_id", "")), + goal=str(item.get("goal", "")), + worktree_path=str(item.get("worktree_path", "")), + branch_name=str(item.get("branch_name", "")), + attempts=attempts, + merge_status=str(item.get("merge_status", "pending")), + ) + ) + + return OrchestratorState( + mission_id=str(payload.get("mission_id", "")), + objective=str(payload.get("objective", "")), + repo_root=str(payload.get("repo_root", "")), + started_from_branch=str(payload.get("started_from_branch", "")), + base_commit=str(payload.get("base_commit", "")), + status=str(payload.get("status", "pending")), + supervisor_run_dir=str(payload.get("supervisor_run_dir", "")), + supervisor_result_json_path=str(payload.get("supervisor_result_json_path", "")), + tasks=tasks, + final_verification_status=str(payload.get("final_verification_status", "")), + final_summary=str(payload.get("final_summary", "")), + ) + + +def save_orchestrator_state(path: Path, state: OrchestratorState) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(orchestrator_state_to_dict(state), indent=2), encoding="utf-8") + + +def load_orchestrator_state(path: Path) -> OrchestratorState: + payload = json.loads(path.read_text(encoding="utf-8")) + if not isinstance(payload, dict): + raise ValueError("Invalid orchestrator state payload") + return orchestrator_state_from_dict(payload) diff --git a/villani_code/orchestrator_roles.py b/villani_code/orchestrator_roles.py new file mode 100644 index 00000000..a4da165c --- /dev/null +++ b/villani_code/orchestrator_roles.py @@ -0,0 +1,60 @@ +from __future__ import annotations + +import json + +from villani_code.orchestrator_models import WorkerTask + + +def build_supervisor_prompt(objective: str, max_subtasks: int) -> str: + contract = { + "mode": "split", + "subtasks": [ + { + "id": "task_1", + "goal": "Fix failing token refresh logic in auth middleware", + "success_criteria": ["targeted auth tests pass"], + "target_files": ["src/auth/middleware.py", "tests/test_auth.py"], + "scope_hint": "Keep the patch minimal and avoid unrelated auth refactors.", + } + ], + } + direct = {"mode": "direct", "subtasks": []} + return ( + "You are the supervisor planner for Villani orchestrator.\n" + "Choose exactly one mode: direct or split.\n" + f"If split, return at most {max_subtasks} bounded subtasks.\n" + "If direct, return an empty subtasks list.\n" + "Do not edit code. Output strict JSON only and no prose.\n" + f"Objective: {objective}\n\n" + f"Split JSON contract:\n{json.dumps(contract, indent=2)}\n\n" + f"Direct JSON contract:\n{json.dumps(direct, indent=2)}" + ) + + +def build_worker_prompt(objective: str, task: WorkerTask, attempt: int, previous_failure: str | None = None) -> str: + contract = { + "status": "success", + "summary": "Patched refresh expiry check and added regression coverage", + "files_touched": ["src/auth/middleware.py", "tests/test_auth.py"], + "recommended_verification": ["pytest tests/test_auth.py -q"], + } + prompt = [ + "You are a worker for Villani orchestrator.", + "Handle exactly one scoped task.", + "Prefer minimal patch.", + "Stay within target files unless clearly forced.", + "Stop when success criteria are satisfied or blocked.", + "Output strict JSON only at the end.", + "Allowed status values: success, blocked_environment, blocked_scope, failed.", + f"Top-level objective: {objective}", + f"Attempt: {attempt}", + f"Task id: {task.id}", + f"Task goal: {task.goal}", + f"Success criteria: {task.success_criteria}", + f"Target files: {task.target_files}", + f"Scope hint: {task.scope_hint}", + ] + if previous_failure: + prompt.append(f"Previous failure summary: {previous_failure}") + prompt.append(f"JSON contract:\n{json.dumps(contract, indent=2)}") + return "\n".join(prompt) diff --git a/villani_code/orchestrator_verify.py b/villani_code/orchestrator_verify.py new file mode 100644 index 00000000..911246ea --- /dev/null +++ b/villani_code/orchestrator_verify.py @@ -0,0 +1,87 @@ +from __future__ import annotations + +import subprocess +from pathlib import Path + +from villani_code.orchestrator_git import changed_files, diff_line_count, has_diff +from villani_code.orchestrator_models import VerificationResult, WorkerTask + + +MAX_CHANGED_FILES = 5 +MAX_CHANGED_LINES = 250 + + +def _run_commands(repo: Path, commands: list[str]) -> tuple[bool, list[str]]: + executed: list[str] = [] + for command in commands: + cmd = command.strip() + if not cmd: + continue + proc = subprocess.run(cmd, cwd=repo, shell=True, check=False, capture_output=True, text=True) + executed.append(cmd) + if proc.returncode != 0: + return False, executed + return True, executed + + +def verify_worker_result(worktree: Path, task: WorkerTask, recommended_verification: list[str] | None) -> VerificationResult: + if not has_diff(worktree): + return VerificationResult(status="hard_failure", summary="Worker produced no changes", commands_run=[], files_touched=[]) + + files = changed_files(worktree) + if task.target_files: + unrelated = [f for f in files if f not in set(task.target_files)] + if unrelated: + return VerificationResult( + status="hard_failure", + summary=f"Edited files outside scope: {', '.join(unrelated[:5])}", + commands_run=[], + files_touched=files, + ) + + if len(files) > MAX_CHANGED_FILES: + return VerificationResult( + status="retryable_failure", + summary=f"Too many files changed ({len(files)} > {MAX_CHANGED_FILES})", + commands_run=[], + files_touched=files, + ) + + lines = diff_line_count(worktree) + if lines > MAX_CHANGED_LINES: + return VerificationResult( + status="retryable_failure", + summary=f"Diff too large ({lines} > {MAX_CHANGED_LINES} changed lines)", + commands_run=[], + files_touched=files, + ) + + commands: list[str] = [] + commands.extend(task.success_criteria) + if recommended_verification: + commands.extend(recommended_verification) + if not commands and (worktree / "pyproject.toml").exists(): + commands.append("python -m pytest -q -k not slow") + + ok, executed = _run_commands(worktree, commands) + if not ok: + return VerificationResult( + status="retryable_failure", + summary="Deterministic verification command failed", + commands_run=executed, + files_touched=files, + ) + + return VerificationResult(status="accepted", summary="Worker changes accepted", commands_run=executed, files_touched=files) + + +def run_final_verification(repo: Path) -> VerificationResult: + commands: list[str] = [] + if (repo / "pyproject.toml").exists(): + commands.append("python -m pytest -q -k not slow") + elif (repo / "Makefile").exists(): + commands.append("make -n test") + ok, executed = _run_commands(repo, commands) + if not ok: + return VerificationResult(status="hard_failure", summary="Final verification failed", commands_run=executed, files_touched=[]) + return VerificationResult(status="accepted", summary="Final verification passed", commands_run=executed, files_touched=[]) diff --git a/villani_code/subagent_runtime.py b/villani_code/subagent_runtime.py index ad85441e..00e41aec 100644 --- a/villani_code/subagent_runtime.py +++ b/villani_code/subagent_runtime.py @@ -24,6 +24,10 @@ def build_role_launch_request(role: str, objective: str, target_files: list[str] return SubagentLaunchRequest(role=role, inherit_mission_state=False, objective=objective, target_files=files, known_facts=[], ruled_out=[], allowed_tools=["Read", "Bash"], write_allowed=False, require_verification_evidence=True) if role == "bounded_patcher": return SubagentLaunchRequest(role=role, inherit_mission_state=True, objective=objective, target_files=files, known_facts=[], ruled_out=[], allowed_tools=["Read", "Patch", "Write", "Bash"], write_allowed=True, require_verification_evidence=True) + if role == "supervisor": + return SubagentLaunchRequest(role=role, inherit_mission_state=True, objective=objective, target_files=files, known_facts=[], ruled_out=[], allowed_tools=["Read", "Grep", "Search", "Bash"], write_allowed=False, require_verification_evidence=False) + if role == "worker": + return SubagentLaunchRequest(role=role, inherit_mission_state=True, objective=objective, target_files=files, known_facts=[], ruled_out=[], allowed_tools=["Read", "Patch", "Write", "Bash"], write_allowed=True, require_verification_evidence=True) return SubagentLaunchRequest(role="planner", inherit_mission_state=True, objective=objective, target_files=files, known_facts=[], ruled_out=[], allowed_tools=["Read", "Grep", "Search"], write_allowed=False, require_verification_evidence=False)