diff --git a/tests/test_orchestrator.py b/tests/test_orchestrator.py new file mode 100644 index 00000000..b2365054 --- /dev/null +++ b/tests/test_orchestrator.py @@ -0,0 +1,504 @@ +from __future__ import annotations + +import json +from pathlib import Path +import subprocess + +from typer.testing import CliRunner + +from villani_code.cli import app +from villani_code.orchestrator import OrchestratorConfig, run_orchestrator +from villani_code.orchestrator_roles import build_supervisor_instruction, build_worker_instruction + + +def test_orchestrate_accepts_run_flags(monkeypatch, tmp_path: Path) -> None: + captured: dict[str, object] = {} + + def fake_run_orchestrator(config): + captured["args"] = config.inherited_run_args + return {"status": "success"} + + monkeypatch.setattr("villani_code.cli.run_orchestrator", fake_run_orchestrator) + runner = CliRunner() + result = runner.invoke( + app, + [ + "orchestrate", + "do thing", + "--base-url", + "http://localhost:8000", + "--model", + "demo-model", + "--repo", + str(tmp_path), + "--provider", + "openai", + "--api-key", + "secret", + "--max-tokens", + "2048", + "--debug", + "trace", + "--auto-approve", + "--small-model", + ], + ) + assert result.exit_code == 0 + args = captured["args"] + assert isinstance(args, list) + assert "--provider" in args + assert "openai" in args + assert "--debug" in args + + +def test_orchestrator_forwards_flags_to_supervisor_workers_and_retries(monkeypatch, tmp_path: Path) -> None: + repo = tmp_path + (repo / "a.py").write_text("print('x')\n", encoding="utf-8") + calls: list[list[str]] = [] + worker_attempts = {"count": 0} + + def fake_subprocess_run(cmd, cwd=None, capture_output=True, text=True, timeout=None, **kwargs): + if isinstance(cmd, str): + class Proc: + returncode = 0 + stdout = "" + stderr = "" + + return Proc() + assert cwd == repo + calls.append(cmd) + role = cmd[cmd.index("--role") + 1] + out_path = Path(cmd[cmd.index("--result-json-path") + 1]) + if role == "supervisor": + out_path.parent.mkdir(parents=True, exist_ok=True) + out_path.write_text( + json.dumps({"subtasks": [{"id": "task_1", "goal": "edit", "success_criteria": [], "target_files": ["a.py"], "scope_hint": "small"}]}), + encoding="utf-8", + ) + else: + worker_attempts["count"] += 1 + payload = {"status": "failed", "summary": "retry", "files_touched": ["a.py"], "recommended_verification": []} + if worker_attempts["count"] > 1: + payload = {"status": "success", "summary": "ok", "files_touched": ["a.py"], "recommended_verification": []} + (repo / "a.py").write_text("print('y')\n", encoding="utf-8") + out_path.parent.mkdir(parents=True, exist_ok=True) + out_path.write_text(json.dumps(payload), encoding="utf-8") + + class Proc: + returncode = 0 + stdout = "" + stderr = "" + + return Proc() + + monkeypatch.setattr("villani_code.orchestrator.subprocess.run", fake_subprocess_run) + monkeypatch.setattr("villani_code.orchestrator.get_current_mission_id", lambda _repo: "parent-1") + monkeypatch.setattr("villani_code.orchestrator.set_current_mission_id", lambda _repo, _id: None) + + summary = run_orchestrator( + OrchestratorConfig( + instruction="fix stuff", + repo=repo, + inherited_run_args=["--base-url", "u", "--model", "m", "--provider", "openai", "--debug", "trace"], + max_worker_retries=1, + ) + ) + assert summary["status"] == "success" + assert len(calls) >= 3 + for cmd in calls: + assert "--base-url" in cmd + assert "u" in cmd + assert "--provider" in cmd + assert "openai" in cmd + assert "--debug" in cmd + assert "trace" in cmd + + +def test_supervisor_retry_on_invalid_result(monkeypatch, tmp_path: Path) -> None: + repo = tmp_path + attempts = {"count": 0} + + def fake_subprocess_run(cmd, cwd=None, capture_output=True, text=True, timeout=None, **kwargs): + attempts["count"] += 1 + out_path = Path(cmd[cmd.index("--result-json-path") + 1]) + out_path.parent.mkdir(parents=True, exist_ok=True) + if attempts["count"] == 1: + out_path.write_text("{}", encoding="utf-8") + else: + out_path.write_text(json.dumps({"subtasks": [{"id": "task_1", "goal": "x"}]}), encoding="utf-8") + + class Proc: + returncode = 0 + stdout = "" + stderr = "" + + return Proc() + + monkeypatch.setattr("villani_code.orchestrator.subprocess.run", fake_subprocess_run) + monkeypatch.setattr("villani_code.orchestrator.get_current_mission_id", lambda _repo: "") + summary = run_orchestrator(OrchestratorConfig(instruction="x", repo=repo, inherited_run_args=["--base-url", "u", "--model", "m"])) + assert summary["total_subtasks"] == 1 + assert attempts["count"] >= 2 + + +def test_snapshot_restore_after_failed_verification(monkeypatch, tmp_path: Path) -> None: + repo = tmp_path + target = repo / "a.py" + target.write_text("before\n", encoding="utf-8") + worker_count = {"count": 0} + + def fake_subprocess_run(cmd, cwd=None, capture_output=True, text=True, timeout=None, **kwargs): + if isinstance(cmd, str): + class Proc: + returncode = 1 + stdout = "" + stderr = "" + return Proc() + role = cmd[cmd.index("--role") + 1] + out_path = Path(cmd[cmd.index("--result-json-path") + 1]) + out_path.parent.mkdir(parents=True, exist_ok=True) + if role == "supervisor": + out_path.write_text(json.dumps({"subtasks": [{"id": "task_1", "goal": "edit", "target_files": ["a.py"], "success_criteria": []}]}), encoding="utf-8") + else: + worker_count["count"] += 1 + target.write_text(f"bad-{worker_count['count']}\n", encoding="utf-8") + payload = {"status": "success", "summary": "ok", "files_touched": ["a.py"], "recommended_verification": ["python -c 'import sys;sys.exit(1)'"]} + out_path.write_text(json.dumps(payload), encoding="utf-8") + + class Proc: + returncode = 0 + stdout = "" + stderr = "" + + return Proc() + + monkeypatch.setattr("villani_code.orchestrator.subprocess.run", fake_subprocess_run) + monkeypatch.setattr("villani_code.orchestrator.get_current_mission_id", lambda _repo: "") + run_orchestrator(OrchestratorConfig(instruction="x", repo=repo, inherited_run_args=["--base-url", "u", "--model", "m"], max_worker_retries=0)) + assert target.read_text(encoding="utf-8") == "before\n" + + +def test_success_criteria_not_executed(monkeypatch, tmp_path: Path) -> None: + repo = tmp_path + (repo / "a.py").write_text("x\n", encoding="utf-8") + seen_shell_commands: list[str] = [] + + def fake_subprocess_run(cmd, cwd=None, capture_output=True, text=True, timeout=None, **kwargs): + if isinstance(cmd, str): + seen_shell_commands.append(cmd) + class Proc: + returncode = 0 + stdout = "" + stderr = "" + return Proc() + role = cmd[cmd.index("--role") + 1] + out_path = Path(cmd[cmd.index("--result-json-path") + 1]) + out_path.parent.mkdir(parents=True, exist_ok=True) + if role == "supervisor": + out_path.write_text(json.dumps({"subtasks": [{"id": "task_1", "goal": "edit", "success_criteria": ["echo SHOULD_NOT_RUN"], "target_files": ["a.py"]}]}), encoding="utf-8") + else: + out_path.write_text(json.dumps({"status": "success", "summary": "ok", "files_touched": ["a.py"], "recommended_verification": []}), encoding="utf-8") + + class Proc: + returncode = 0 + stdout = "" + stderr = "" + + return Proc() + + monkeypatch.setattr("villani_code.orchestrator.subprocess.run", fake_subprocess_run) + monkeypatch.setattr("villani_code.orchestrator.get_current_mission_id", lambda _repo: "") + run_orchestrator(OrchestratorConfig(instruction="x", repo=repo, inherited_run_args=["--base-url", "u", "--model", "m"])) + assert "echo SHOULD_NOT_RUN" not in seen_shell_commands + + +def test_no_final_verification_when_no_worker_success(monkeypatch, tmp_path: Path) -> None: + repo = tmp_path + + def fake_subprocess_run(cmd, cwd=None, capture_output=True, text=True, timeout=None, **kwargs): + if isinstance(cmd, str): + class Proc: + returncode = 0 + stdout = "" + stderr = "" + + return Proc() + role = cmd[cmd.index("--role") + 1] + out_path = Path(cmd[cmd.index("--result-json-path") + 1]) + out_path.parent.mkdir(parents=True, exist_ok=True) + if role == "supervisor": + out_path.write_text(json.dumps({"subtasks": [{"id": "task_1", "goal": "x"}]}), encoding="utf-8") + else: + out_path.write_text(json.dumps({"status": "failed", "summary": "nope", "files_touched": [], "recommended_verification": []}), encoding="utf-8") + + class Proc: + returncode = 0 + stdout = "" + stderr = "" + + return Proc() + + monkeypatch.setattr("villani_code.orchestrator.subprocess.run", fake_subprocess_run) + monkeypatch.setattr("villani_code.orchestrator.get_current_mission_id", lambda _repo: "") + summary = run_orchestrator(OrchestratorConfig(instruction="x", repo=repo, inherited_run_args=["--base-url", "u", "--model", "m"], max_worker_retries=0)) + mission_id = summary["mission_id"] + final_verification = json.loads((repo / ".villani_code" / "missions" / mission_id / "orchestrator" / "final_verification.json").read_text(encoding="utf-8")) + assert final_verification["ran"] is False + + +def test_parent_mission_remains_authoritative(monkeypatch, tmp_path: Path) -> None: + repo = tmp_path + set_calls: list[str] = [] + + def fake_subprocess_run(cmd, cwd=None, capture_output=True, text=True, timeout=None, **kwargs): + if isinstance(cmd, str): + class Proc: + returncode = 0 + stdout = "" + stderr = "" + + return Proc() + role = cmd[cmd.index("--role") + 1] + out_path = Path(cmd[cmd.index("--result-json-path") + 1]) + out_path.parent.mkdir(parents=True, exist_ok=True) + if role == "supervisor": + out_path.write_text(json.dumps({"subtasks": [{"id": "task_1", "goal": "x"}]}), encoding="utf-8") + else: + out_path.write_text(json.dumps({"status": "failed", "summary": "nope", "files_touched": [], "recommended_verification": []}), encoding="utf-8") + + class Proc: + returncode = 0 + stdout = "" + stderr = "" + + return Proc() + + monkeypatch.setattr("villani_code.orchestrator.subprocess.run", fake_subprocess_run) + monkeypatch.setattr("villani_code.orchestrator.get_current_mission_id", lambda _repo: "parent-xyz") + monkeypatch.setattr("villani_code.orchestrator.set_current_mission_id", lambda _repo, mid: set_calls.append(mid)) + run_orchestrator(OrchestratorConfig(instruction="x", repo=repo, inherited_run_args=["--base-url", "u", "--model", "m"])) + assert set_calls[-1] == "parent-xyz" + + +def test_supervisor_can_return_multiple_subtasks(monkeypatch, tmp_path: Path) -> None: + repo = tmp_path + + def fake_subprocess_run(cmd, cwd=None, capture_output=True, text=True, timeout=None, **kwargs): + if isinstance(cmd, str): + class Proc: + returncode = 0 + stdout = "" + stderr = "" + + return Proc() + role = cmd[cmd.index("--role") + 1] + out_path = Path(cmd[cmd.index("--result-json-path") + 1]) + out_path.parent.mkdir(parents=True, exist_ok=True) + if role == "supervisor": + out_path.write_text(json.dumps({"subtasks": [{"id": "task_1", "goal": "x"}, {"id": "task_2", "goal": "y"}]}), encoding="utf-8") + else: + out_path.write_text(json.dumps({"status": "failed", "summary": "nope", "files_touched": [], "recommended_verification": []}), encoding="utf-8") + + class Proc: + returncode = 0 + stdout = "" + stderr = "" + + return Proc() + + monkeypatch.setattr("villani_code.orchestrator.subprocess.run", fake_subprocess_run) + monkeypatch.setattr("villani_code.orchestrator.get_current_mission_id", lambda _repo: "") + summary = run_orchestrator(OrchestratorConfig(instruction="x", repo=repo, inherited_run_args=["--base-url", "u", "--model", "m"], max_workers=2)) + assert summary["total_subtasks"] == 2 + + +def test_out_of_scope_file_edits_fail_attempt(monkeypatch, tmp_path: Path) -> None: + repo = tmp_path + (repo / "a.py").write_text("ok\n", encoding="utf-8") + (repo / "b.py").write_text("ok\n", encoding="utf-8") + + def fake_subprocess_run(cmd, cwd=None, capture_output=True, text=True, timeout=None, **kwargs): + if isinstance(cmd, str): + class Proc: + returncode = 0 + stdout = "" + stderr = "" + + return Proc() + role = cmd[cmd.index("--role") + 1] + out_path = Path(cmd[cmd.index("--result-json-path") + 1]) + out_path.parent.mkdir(parents=True, exist_ok=True) + if role == "supervisor": + out_path.write_text(json.dumps({"subtasks": [{"id": "task_1", "goal": "x", "target_files": ["a.py"]}]}), encoding="utf-8") + else: + (repo / "b.py").write_text("drift\n", encoding="utf-8") + out_path.write_text(json.dumps({"status": "success", "summary": "ok", "files_touched": ["a.py"], "recommended_verification": []}), encoding="utf-8") + + class Proc: + returncode = 0 + stdout = "" + stderr = "" + + return Proc() + + monkeypatch.setattr("villani_code.orchestrator.subprocess.run", fake_subprocess_run) + monkeypatch.setattr("villani_code.orchestrator.get_current_mission_id", lambda _repo: "") + summary = run_orchestrator(OrchestratorConfig(instruction="x", repo=repo, inherited_run_args=["--base-url", "u", "--model", "m"], max_worker_retries=0)) + assert summary["status"] == "failed" + + +def test_variant_file_sprawl_fails_attempt(monkeypatch, tmp_path: Path) -> None: + repo = tmp_path + (repo / "parser.py").write_text("x=1\n", encoding="utf-8") + + def fake_subprocess_run(cmd, cwd=None, capture_output=True, text=True, timeout=None, **kwargs): + if isinstance(cmd, str): + class Proc: + returncode = 0 + stdout = "" + stderr = "" + + return Proc() + role = cmd[cmd.index("--role") + 1] + out_path = Path(cmd[cmd.index("--result-json-path") + 1]) + out_path.parent.mkdir(parents=True, exist_ok=True) + if role == "supervisor": + out_path.write_text(json.dumps({"subtasks": [{"id": "task_1", "goal": "x", "target_files": ["parser.py"]}]}), encoding="utf-8") + else: + (repo / "parser_fixed.py").write_text("x=2\n", encoding="utf-8") + out_path.write_text(json.dumps({"status": "success", "summary": "ok", "files_touched": ["parser.py"], "recommended_verification": []}), encoding="utf-8") + + class Proc: + returncode = 0 + stdout = "" + stderr = "" + + return Proc() + + monkeypatch.setattr("villani_code.orchestrator.subprocess.run", fake_subprocess_run) + monkeypatch.setattr("villani_code.orchestrator.get_current_mission_id", lambda _repo: "") + summary = run_orchestrator(OrchestratorConfig(instruction="x", repo=repo, inherited_run_args=["--base-url", "u", "--model", "m"], max_worker_retries=0)) + assert summary["status"] == "failed" + + +def test_rollback_removes_new_out_of_scope_files_on_retry(monkeypatch, tmp_path: Path) -> None: + repo = tmp_path + target = repo / "a.py" + target.write_text("before\n", encoding="utf-8") + worker_attempt = {"count": 0} + + def fake_subprocess_run(cmd, cwd=None, capture_output=True, text=True, timeout=None, **kwargs): + if isinstance(cmd, str): + class Proc: + returncode = 0 + stdout = "" + stderr = "" + + return Proc() + role = cmd[cmd.index("--role") + 1] + out_path = Path(cmd[cmd.index("--result-json-path") + 1]) + out_path.parent.mkdir(parents=True, exist_ok=True) + if role == "supervisor": + out_path.write_text(json.dumps({"subtasks": [{"id": "task_1", "goal": "x", "target_files": ["a.py"]}]}), encoding="utf-8") + else: + worker_attempt["count"] += 1 + if worker_attempt["count"] == 1: + (repo / "debug_helper.py").write_text("tmp\n", encoding="utf-8") + out_path.write_text(json.dumps({"status": "blocked_scope", "summary": "drift", "files_touched": ["a.py"], "recommended_verification": []}), encoding="utf-8") + else: + target.write_text("after\n", encoding="utf-8") + out_path.write_text(json.dumps({"status": "success", "summary": "ok", "files_touched": ["a.py"], "recommended_verification": []}), encoding="utf-8") + + class Proc: + returncode = 0 + stdout = "" + stderr = "" + + return Proc() + + monkeypatch.setattr("villani_code.orchestrator.subprocess.run", fake_subprocess_run) + monkeypatch.setattr("villani_code.orchestrator.get_current_mission_id", lambda _repo: "") + summary = run_orchestrator(OrchestratorConfig(instruction="x", repo=repo, inherited_run_args=["--base-url", "u", "--model", "m"], max_worker_retries=1)) + assert summary["status"] == "success" + assert not (repo / "debug_helper.py").exists() + + +def test_worker_budget_exceedance_fails_attempt(monkeypatch, tmp_path: Path) -> None: + repo = tmp_path + (repo / "a.py").write_text("ok\n", encoding="utf-8") + + def fake_subprocess_run(cmd, cwd=None, capture_output=True, text=True, timeout=None, **kwargs): + role = cmd[cmd.index("--role") + 1] + out_path = Path(cmd[cmd.index("--result-json-path") + 1]) + out_path.parent.mkdir(parents=True, exist_ok=True) + if role == "supervisor": + out_path.write_text(json.dumps({"subtasks": [{"id": "task_1", "goal": "x", "target_files": ["a.py"]}]}), encoding="utf-8") + stdout = "" + else: + out_path.write_text(json.dumps({"status": "success", "summary": "ok", "files_touched": ["a.py"], "recommended_verification": []}), encoding="utf-8") + stdout = '{"turns_used": 99, "tool_calls_used": 1}' + + class Proc: + returncode = 0 + stderr = "" + + def __init__(self, stdout_val: str) -> None: + self.stdout = stdout_val + + return Proc(stdout) + + monkeypatch.setattr("villani_code.orchestrator.subprocess.run", fake_subprocess_run) + monkeypatch.setattr("villani_code.orchestrator.get_current_mission_id", lambda _repo: "") + summary = run_orchestrator(OrchestratorConfig(instruction="x", repo=repo, inherited_run_args=["--base-url", "u", "--model", "m"], max_worker_retries=0, max_worker_model_turns=40)) + assert summary["status"] == "failed" + + +def test_timeout_stall_failure_summary_written(monkeypatch, tmp_path: Path) -> None: + repo = tmp_path + (repo / "a.py").write_text("ok\n", encoding="utf-8") + + def fake_subprocess_run(cmd, cwd=None, capture_output=True, text=True, timeout=None, **kwargs): + role = cmd[cmd.index("--role") + 1] + out_path = Path(cmd[cmd.index("--result-json-path") + 1]) + out_path.parent.mkdir(parents=True, exist_ok=True) + if role == "supervisor": + out_path.write_text(json.dumps({"subtasks": [{"id": "task_1", "goal": "x", "target_files": ["a.py"]}]}), encoding="utf-8") + + class Proc: + returncode = 0 + stdout = "" + stderr = "" + + return Proc() + raise subprocess.TimeoutExpired(cmd=cmd, timeout=1, output='{"command":"python app.py"}') + + monkeypatch.setattr("villani_code.orchestrator.subprocess.run", fake_subprocess_run) + monkeypatch.setattr("villani_code.orchestrator.get_current_mission_id", lambda _repo: "") + summary = run_orchestrator(OrchestratorConfig(instruction="x", repo=repo, inherited_run_args=["--base-url", "u", "--model", "m"], max_worker_retries=0, worker_timeout_seconds=1)) + mission_id = summary["mission_id"] + failure_summary = json.loads( + ( + repo + / ".villani_code" + / "missions" + / mission_id + / "orchestrator" + / "workers" + / "task_1" + / "failure_attempt_1.json" + ).read_text(encoding="utf-8") + ) + assert summary["status"] == "failed" + assert "timeout/stall" in failure_summary["summary"] + + +def test_supervisor_prompt_biases_toward_fewer_subtasks() -> None: + prompt = build_supervisor_instruction("fix", 8) + assert "Return as few subtasks as necessary" in prompt + assert "If unsure, return fewer subtasks" in prompt + + +def test_worker_prompt_includes_windows_guidance() -> None: + prompt = build_worker_instruction("fix", {"id": "task_1"}) + assert "Windows/PowerShell-first" in prompt + assert "Avoid Unix-only commands" in prompt diff --git a/villani_code/cli.py b/villani_code/cli.py index b5634b3e..e3321f29 100644 --- a/villani_code/cli.py +++ b/villani_code/cli.py @@ -3,10 +3,11 @@ import json import os from pathlib import Path -from typing import Any, Literal, Optional +from typing import Annotated, Any, Literal, Optional import typer from rich.console import Console +from pydantic import BaseModel, ValidationError from villani_code.interrupts import InterruptController from villani_code.optional_tui import OptionalTUIDependencyError, TUI_INSTALL_HINT @@ -20,6 +21,8 @@ from villani_code.debug_bundle import create_debug_bundle from villani_code.debug_mode import DebugMode, build_debug_config from villani_code.trace_summary import write_summary_from_events, write_tool_calls_from_events +from villani_code.orchestrator import OrchestratorConfig, run_orchestrator +from villani_code.mission_state import set_current_mission_id app = typer.Typer(help="Villani: constrained-inference coding agent with visible context governance") mcp_app = typer.Typer(help="Manage MCP servers") @@ -32,6 +35,28 @@ app.add_typer(trace_app, name="trace") console = Console() +BaseUrlOption = Annotated[str, typer.Option(..., "--base-url", help="Base URL for compatible messages API server")] +ModelOption = Annotated[str, typer.Option(..., "--model", help="Model name")] +RepoOption = Annotated[Path, typer.Option("--repo", help="Repository path")] +MaxTokensOption = Annotated[int, typer.Option("--max-tokens")] +StreamOption = Annotated[bool, typer.Option("--stream/--no-stream")] +ThinkingOption = Annotated[Optional[str], typer.Option("--thinking")] +UnsafeOption = Annotated[bool, typer.Option("--unsafe")] +VerboseOption = Annotated[bool, typer.Option("--verbose")] +ExtraJsonOption = Annotated[Optional[str], typer.Option("--extra-json")] +RedactOption = Annotated[bool, typer.Option("--redact")] +SkipPermissionsOption = Annotated[bool, typer.Option("--dangerously-skip-permissions")] +AutoAcceptEditsOption = Annotated[bool, typer.Option("--auto-accept-edits")] +AutoApproveOption = Annotated[bool, typer.Option("--auto-approve", help="Automatically approve all actions without prompting")] +PlanModeOption = Annotated[Literal["off", "auto", "strict"], typer.Option("--plan-mode")] +MaxRepairAttemptsOption = Annotated[int, typer.Option("--max-repair-attempts")] +SmallModelOption = Annotated[bool, typer.Option("--small-model")] +ProviderOption = Annotated[Literal["anthropic", "openai"], typer.Option("--provider")] +ApiKeyOption = Annotated[Optional[str], typer.Option("--api-key")] +BenchmarkRuntimeOption = Annotated[Optional[str], typer.Option("--benchmark-runtime-json", hidden=True)] +DebugOption = Annotated[Optional[str], typer.Option("--debug", flag_value="normal")] +DebugDirOption = Annotated[Optional[Path], typer.Option("--debug-dir")] + def _print_response_text_blocks(result: dict[str, Any] | None) -> None: def _print_content(value: Any) -> None: @@ -68,6 +93,69 @@ def _print_content(value: Any) -> None: except Exception: # noqa: BLE001 return + +def _extract_response_text(result: dict[str, Any] | None) -> str: + chunks: list[str] = [] + if not isinstance(result, dict): + return "" + response = result.get("response") + if isinstance(response, str): + chunks.append(response) + if isinstance(response, dict): + content = response.get("content") + if isinstance(content, list): + for block in content: + if isinstance(block, str): + chunks.append(block) + elif isinstance(block, dict) and block.get("type") == "text": + chunks.append(str(block.get("text", ""))) + direct_content = result.get("content") + if isinstance(direct_content, list): + for block in direct_content: + if isinstance(block, dict) and block.get("type") == "text": + chunks.append(str(block.get("text", ""))) + return "\n".join(c for c in chunks if c.strip()).strip() + + +class _SupervisorArtifact(BaseModel): + subtasks: list[dict[str, Any]] + + +class _WorkerArtifact(BaseModel): + status: str + summary: str + files_touched: list[str] = [] + recommended_verification: list[str] = [] + + +def _write_result_artifact(path: Path, role: str, result: dict[str, Any] | None) -> None: + raw = _extract_response_text(result) + parsed: dict[str, Any] = {} + try: + parsed_candidate = json.loads(raw) if raw else {} + if isinstance(parsed_candidate, dict): + parsed = parsed_candidate + except json.JSONDecodeError: + parsed = {} + if role == "supervisor": + try: + payload = _SupervisorArtifact.model_validate(parsed).model_dump() + except ValidationError: + payload = {"subtasks": []} + elif role == "worker": + try: + candidate = _WorkerArtifact.model_validate(parsed).model_dump() + if candidate["status"] in {"success", "blocked_environment", "blocked_scope", "failed"}: + payload = candidate + else: + payload = {"status": "failed", "summary": "invalid worker json", "files_touched": [], "recommended_verification": []} + except ValidationError: + payload = {"status": "failed", "summary": "invalid worker json", "files_touched": [], "recommended_verification": []} + else: + payload = parsed if parsed else {"result": raw} + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(payload, indent=2), encoding="utf-8") + def _load_settings_manager() -> Any | None: try: from villani_code.tui.components.settings import SettingsManager @@ -155,6 +243,71 @@ def _run_interactive(base_url: str, model: str, repo: Path, max_tokens: int, sma console.print("Interrupted current session. Press Ctrl+C again to exit Villani Code.") +def _build_inherited_run_args( + *, + base_url: str, + model: str, + repo: Path, + max_tokens: int, + stream: bool, + thinking: Optional[str], + unsafe: bool, + verbose: bool, + extra_json: Optional[str], + redact: bool, + dangerously_skip_permissions: bool, + auto_accept_edits: bool, + auto_approve: bool, + plan_mode: Literal["off", "auto", "strict"], + max_repair_attempts: int, + small_model: bool, + provider: Literal["anthropic", "openai"], + api_key: Optional[str], + benchmark_runtime_json: Optional[str], + debug: Optional[str], + debug_dir: Optional[Path], + passthrough_args: list[str], +) -> list[str]: + args: list[str] = [ + "--base-url", base_url, + "--model", model, + "--repo", str(repo), + "--max-tokens", str(max_tokens), + "--plan-mode", plan_mode, + "--max-repair-attempts", str(max_repair_attempts), + "--provider", provider, + ] + args.append("--stream" if stream else "--no-stream") + if thinking is not None: + args.extend(["--thinking", thinking]) + if unsafe: + args.append("--unsafe") + if verbose: + args.append("--verbose") + if extra_json is not None: + args.extend(["--extra-json", extra_json]) + if redact: + args.append("--redact") + if dangerously_skip_permissions: + args.append("--dangerously-skip-permissions") + if auto_accept_edits: + args.append("--auto-accept-edits") + if auto_approve: + args.append("--auto-approve") + if small_model: + args.append("--small-model") + if api_key is not None: + args.extend(["--api-key", api_key]) + if benchmark_runtime_json is not None: + args.extend(["--benchmark-runtime-json", benchmark_runtime_json]) + if debug is not None: + args.extend(["--debug", debug]) + if debug_dir is not None: + args.extend(["--debug-dir", str(debug_dir)]) + args.extend(passthrough_args) + return args + + @app.callback(invoke_without_command=True) def main( ctx: typer.Context, @@ -177,37 +330,113 @@ def main( @app.command() def run( - instruction: str = typer.Argument(..., help="User instruction"), - 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"), - stream: bool = typer.Option(True, "--stream/--no-stream"), - thinking: Optional[str] = typer.Option(None, "--thinking"), - unsafe: bool = typer.Option(False, "--unsafe"), - verbose: bool = typer.Option(False, "--verbose"), - extra_json: Optional[str] = typer.Option(None, "--extra-json"), - redact: bool = typer.Option(False, "--redact"), - dangerously_skip_permissions: bool = typer.Option(False, "--dangerously-skip-permissions"), - auto_accept_edits: bool = typer.Option(False, "--auto-accept-edits"), - auto_approve: bool = typer.Option(False, "--auto-approve", help="Automatically approve all actions without prompting"), - plan_mode: Literal["off", "auto", "strict"] = typer.Option("auto", "--plan-mode"), - max_repair_attempts: int = typer.Option(2, "--max-repair-attempts"), - 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"), - benchmark_runtime_json: Optional[str] = typer.Option(None, "--benchmark-runtime-json", hidden=True), - debug: Optional[str] = typer.Option(None, "--debug", flag_value="normal"), - debug_dir: Optional[Path] = typer.Option(None, "--debug-dir"), + instruction: Annotated[str, typer.Argument(help="User instruction")], + base_url: BaseUrlOption, + model: ModelOption, + repo: RepoOption = Path("."), + max_tokens: MaxTokensOption = 4096, + stream: StreamOption = True, + thinking: ThinkingOption = None, + unsafe: UnsafeOption = False, + verbose: VerboseOption = False, + extra_json: ExtraJsonOption = None, + redact: RedactOption = False, + dangerously_skip_permissions: SkipPermissionsOption = False, + auto_accept_edits: AutoAcceptEditsOption = False, + auto_approve: AutoApproveOption = False, + plan_mode: PlanModeOption = "auto", + max_repair_attempts: MaxRepairAttemptsOption = 2, + small_model: SmallModelOption = False, + provider: ProviderOption = "anthropic", + api_key: ApiKeyOption = None, + benchmark_runtime_json: BenchmarkRuntimeOption = None, + debug: DebugOption = None, + debug_dir: DebugDirOption = None, + role: str = typer.Option("default", "--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), ) -> None: 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) + if parent_mission_id: + set_current_mission_id(repo.resolve(), parent_mission_id) + if result_json_path: + _write_result_artifact(result_json_path, role=role, result=result) + return _print_response_text_blocks(result) +@app.command(context_settings={"allow_extra_args": True, "ignore_unknown_options": True}) +def orchestrate( + ctx: typer.Context, + instruction: Annotated[str, typer.Argument(help="User instruction")], + base_url: BaseUrlOption, + model: ModelOption, + repo: RepoOption = Path("."), + max_tokens: MaxTokensOption = 4096, + stream: StreamOption = True, + thinking: ThinkingOption = None, + unsafe: UnsafeOption = False, + verbose: VerboseOption = False, + extra_json: ExtraJsonOption = None, + redact: RedactOption = False, + dangerously_skip_permissions: SkipPermissionsOption = False, + auto_accept_edits: AutoAcceptEditsOption = False, + auto_approve: AutoApproveOption = False, + plan_mode: PlanModeOption = "auto", + max_repair_attempts: MaxRepairAttemptsOption = 2, + small_model: SmallModelOption = False, + provider: ProviderOption = "anthropic", + api_key: ApiKeyOption = None, + benchmark_runtime_json: BenchmarkRuntimeOption = None, + debug: DebugOption = None, + debug_dir: DebugDirOption = None, + max_workers: int = typer.Option(3, "--max-workers"), + max_worker_retries: int = typer.Option(1, "--max-worker-retries"), + supervisor_timeout_seconds: Optional[int] = typer.Option(None, "--supervisor-timeout-seconds"), + worker_timeout_seconds: Optional[int] = typer.Option(None, "--worker-timeout-seconds"), +) -> None: + inherited_args = _build_inherited_run_args( + base_url=base_url, + model=model, + repo=repo, + max_tokens=max_tokens, + stream=stream, + thinking=thinking, + unsafe=unsafe, + verbose=verbose, + extra_json=extra_json, + redact=redact, + dangerously_skip_permissions=dangerously_skip_permissions, + auto_accept_edits=auto_accept_edits, + auto_approve=auto_approve, + plan_mode=plan_mode, + max_repair_attempts=max_repair_attempts, + small_model=small_model, + provider=provider, + api_key=api_key, + benchmark_runtime_json=benchmark_runtime_json, + debug=debug, + debug_dir=debug_dir, + passthrough_args=list(ctx.args), + ) + summary = run_orchestrator( + OrchestratorConfig( + instruction=instruction, + repo=repo, + inherited_run_args=inherited_args, + max_workers=max_workers, + max_worker_retries=max_worker_retries, + supervisor_timeout_seconds=supervisor_timeout_seconds, + worker_timeout_seconds=worker_timeout_seconds, + ) + ) + console.print(json.dumps(summary, indent=2)) + + @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..97a25512 --- /dev/null +++ b/villani_code/orchestrator.py @@ -0,0 +1,385 @@ +from __future__ import annotations + +import json +import re +import subprocess +import sys +from dataclasses import asdict, dataclass, field +from pathlib import Path +from typing import Any + +from villani_code.mission_state import get_current_mission_id, new_mission_id, set_current_mission_id +from villani_code.orchestrator_models import Subtask, SupervisorResult, WorkerResult +from villani_code.orchestrator_roles import build_supervisor_instruction, build_worker_instruction +from villani_code.orchestrator_verify import ( + capture_repo_file_state, + cleanup_created_files, + count_changed_lines, + diff_repo_file_state, + restore_files, + run_verification, + snapshot_files, + to_json, +) +from villani_code.utils import ensure_dir + + +@dataclass(slots=True) +class OrchestratorConfig: + instruction: str + repo: Path + inherited_run_args: list[str] + max_workers: int = 3 + max_worker_retries: int = 1 + supervisor_timeout_seconds: int | None = None + worker_timeout_seconds: int | None = None + max_worker_model_turns: int = 40 + max_worker_shell_commands: int = 20 + max_worker_changed_files: int = 5 + max_worker_changed_lines: int = 250 + + +@dataclass(slots=True) +class OrchestratorState: + mission_id: str + objective: str + supervisor_attempts: int = 0 + worker_attempts: dict[str, int] = field(default_factory=dict) + successful_workers: list[str] = field(default_factory=list) + + +def run_orchestrator(config: OrchestratorConfig) -> dict[str, Any]: + repo = config.repo.resolve() + mission_id = new_mission_id() + base = repo / ".villani_code" / "missions" / mission_id / "orchestrator" + supervisor_dir = base / "supervisor" + workers_dir = base / "workers" + snapshots_dir = base / "snapshots" + for path in (base, supervisor_dir, workers_dir, snapshots_dir): + ensure_dir(path) + (base / "top_level_objective.txt").write_text(config.instruction, encoding="utf-8") + + state = OrchestratorState(mission_id=mission_id, objective=config.instruction) + _save_state(base, state) + + parent_mission_id = get_current_mission_id(repo) + if parent_mission_id: + set_current_mission_id(repo, parent_mission_id) + + supervisor = _run_supervisor(config, mission_id, supervisor_dir) + state.supervisor_attempts = supervisor["attempts"] + _save_state(base, state) + + subtasks = supervisor["result"].subtasks + success_count = 0 + for subtask in subtasks: + worker_record = _run_worker_with_retries( + config=config, + mission_id=mission_id, + subtask=subtask, + workers_dir=workers_dir, + snapshots_dir=snapshots_dir, + ) + state.worker_attempts[subtask.id] = int(worker_record.get("attempts", 0)) + if worker_record.get("success"): + success_count += 1 + state.successful_workers.append(subtask.id) + _save_state(base, state) + + final_verification = { + "ran": success_count > 0, + "ok": success_count > 0, + "successful_workers": success_count, + } + (base / "final_verification.json").write_text(json.dumps(final_verification, indent=2), encoding="utf-8") + + summary = { + "mission_id": mission_id, + "status": "success" if success_count > 0 else "failed", + "successful_workers": success_count, + "total_subtasks": len(subtasks), + } + (base / "final_summary.json").write_text(json.dumps(summary, indent=2), encoding="utf-8") + _save_state(base, state) + if parent_mission_id: + set_current_mission_id(repo, parent_mission_id) + return summary + + +def _save_state(base: Path, state: OrchestratorState) -> None: + (base / "orchestrator_state.json").write_text(json.dumps(asdict(state), indent=2), encoding="utf-8") + + +def _run_supervisor(config: OrchestratorConfig, mission_id: str, supervisor_dir: Path) -> dict[str, Any]: + result_path = supervisor_dir / "result.json" + prompt = build_supervisor_instruction(config.instruction, config.max_workers) + last_error = "supervisor failed" + for attempt in range(1, 3): + if result_path.exists(): + result_path.unlink() + proc = _run_child( + instruction=prompt, + inherited_run_args=config.inherited_run_args, + mission_id=mission_id, + role="supervisor", + result_json_path=result_path, + timeout_seconds=config.supervisor_timeout_seconds, + repo=config.repo, + ) + if proc.returncode != 0: + last_error = f"supervisor subprocess exited {proc.returncode}" + continue + loaded = _load_supervisor_result(result_path, config.max_workers) + if loaded is None: + last_error = "invalid supervisor result" + continue + return {"result": loaded, "attempts": attempt} + raise RuntimeError(last_error) + + +def _run_worker_with_retries( + config: OrchestratorConfig, + mission_id: str, + subtask: Subtask, + workers_dir: Path, + snapshots_dir: Path, +) -> dict[str, Any]: + worker_dir = workers_dir / subtask.id + ensure_dir(worker_dir) + result_path = worker_dir / "result.json" + prev_summary: str | None = None + touched = sorted(set(subtask.target_files)) + snapshot_dir = snapshots_dir / subtask.id + snapshot_files(config.repo, touched, snapshot_dir) + max_attempts = config.max_worker_retries + 1 + retryable_failures_used = 0 + retryable_limit = 1 + + for attempt in range(1, max_attempts + 1): + prompt = build_worker_instruction(config.instruction, asdict(subtask), previous_failure=prev_summary) + if result_path.exists(): + result_path.unlink() + before_state = capture_repo_file_state(config.repo) + try: + proc = _run_child( + instruction=prompt, + inherited_run_args=config.inherited_run_args, + mission_id=mission_id, + role="worker", + result_json_path=result_path, + timeout_seconds=config.worker_timeout_seconds, + repo=config.repo, + ) + timeout_failure = False + except subprocess.TimeoutExpired as exc: + timeout_failure = True + proc = subprocess.CompletedProcess(exc.cmd, returncode=124, stdout=str(exc.stdout or ""), stderr=str(exc.stderr or "")) + after_state = capture_repo_file_state(config.repo) + modified_files, created_files, deleted_files = diff_repo_file_state(before_state, after_state) + actual_changed = sorted(set(modified_files + created_files + deleted_files)) + changed_line_count = count_changed_lines(config.repo, set(modified_files + created_files)) + out_of_scope_files = _out_of_scope_files(actual_changed, subtask.target_files) + variant_files = _variant_sprawl_files(created_files, subtask.target_files) + budget_reason = _budget_reason( + proc=proc, + changed_files_count=len(actual_changed), + changed_line_count=changed_line_count, + config=config, + ) + worker_result = _load_worker_result(result_path) + + should_restore = False + retryable = False + failure_reasons: list[str] = [] + + if timeout_failure: + last_cmd = _extract_last_command(proc.stdout, proc.stderr) + detail = f" (last command: {last_cmd})" if last_cmd else "" + failure_reasons.append(f"worker timeout/stall detected{detail}") + retryable = True + should_restore = True + elif proc.returncode != 0: + failure_reasons.append(f"subprocess exited {proc.returncode}") + should_restore = True + elif worker_result is None: + failure_reasons.append("invalid worker result") + should_restore = True + else: + if out_of_scope_files: + failure_reasons.append(f"out-of-scope edits: {', '.join(out_of_scope_files)}") + retryable = worker_result.status == "blocked_scope" + should_restore = True + if variant_files: + failure_reasons.append(f"variant/debug file sprawl: {', '.join(variant_files)}") + should_restore = True + if budget_reason: + failure_reasons.append(budget_reason) + retryable = True + should_restore = True + + if not failure_reasons: + verification = run_verification( + repo=config.repo, + worker_recommended=worker_result.recommended_verification, + success_criteria=subtask.success_criteria, + files_touched=actual_changed, + changed_line_count=changed_line_count, + max_files=config.max_worker_changed_files, + max_lines=config.max_worker_changed_lines, + ) + (worker_dir / f"verification_attempt_{attempt}.json").write_text( + json.dumps(to_json(verification), indent=2), encoding="utf-8" + ) + if verification.ok and worker_result.status == "success": + return {"success": True, "attempts": attempt} + failure_reasons.append(worker_result.summary if worker_result.summary else "; ".join(verification.reasons)) + should_restore = True + if should_restore: + restore_files(config.repo, touched, snapshot_dir) + cleanup_created_files(config.repo, out_of_scope_files) + prev_summary = "; ".join(reason for reason in failure_reasons if reason) + if prev_summary: + (worker_dir / f"failure_attempt_{attempt}.json").write_text( + json.dumps({"summary": prev_summary, "retryable": retryable}, indent=2), encoding="utf-8" + ) + if retryable: + retryable_failures_used += 1 + can_retry = attempt < max_attempts and (not retryable or retryable_failures_used <= retryable_limit) + if not can_retry: + return {"success": False, "attempts": attempt} + + return {"success": False, "attempts": max_attempts} + + +def _run_child( + instruction: str, + inherited_run_args: list[str], + mission_id: str, + role: str, + result_json_path: Path, + timeout_seconds: int | None, + repo: Path, +) -> subprocess.CompletedProcess[str]: + cmd = [ + sys.executable, + "-m", + "villani_code.cli", + "run", + instruction, + *inherited_run_args, + "--role", + role, + "--result-json-path", + str(result_json_path), + "--parent-mission-id", + mission_id, + ] + return subprocess.run(cmd, cwd=repo, capture_output=True, text=True, timeout=timeout_seconds) + + +def _load_supervisor_result(path: Path, max_workers: int) -> SupervisorResult | None: + if not path.exists(): + return None + try: + payload = json.loads(path.read_text(encoding="utf-8")) + except json.JSONDecodeError: + return None + subtasks_raw = payload.get("subtasks") + if not isinstance(subtasks_raw, list) or not (1 <= len(subtasks_raw) <= max_workers): + return None + subtasks: list[Subtask] = [] + for i, item in enumerate(subtasks_raw, start=1): + if not isinstance(item, dict): + return None + goal = str(item.get("goal", "")).strip() + if not goal: + return None + subtasks.append( + Subtask( + id=str(item.get("id") or f"task_{i}"), + goal=goal, + success_criteria=[str(v) for v in item.get("success_criteria", []) if str(v).strip()], + target_files=[str(v) for v in item.get("target_files", []) if str(v).strip()], + scope_hint=str(item.get("scope_hint", "")), + ) + ) + return SupervisorResult(subtasks=subtasks) + + +def _load_worker_result(path: Path) -> WorkerResult | None: + if not path.exists(): + return None + try: + payload = json.loads(path.read_text(encoding="utf-8")) + except json.JSONDecodeError: + return None + status = str(payload.get("status", "")) + if status not in {"success", "blocked_environment", "blocked_scope", "failed"}: + return None + return WorkerResult( + status=status, # type: ignore[arg-type] + summary=str(payload.get("summary", "")), + files_touched=[str(v) for v in payload.get("files_touched", []) if str(v).strip()], + recommended_verification=[str(v) for v in payload.get("recommended_verification", []) if str(v).strip()], + ) + + +def _out_of_scope_files(changed_files: list[str], target_files: list[str]) -> list[str]: + allowed = {path.strip() for path in target_files if path.strip()} + return sorted(rel for rel in changed_files if rel not in allowed) + + +def _variant_sprawl_files(created_files: list[str], target_files: list[str]) -> list[str]: + suffixes = ("_fixed", "_final", "_clean", "_new", "_v2", "_updated") + debug_prefixes = ("debug_", "verify_", "fix_", "copy_") + target_stems = {Path(rel).stem for rel in target_files} + flagged: list[str] = [] + for rel in created_files: + base = Path(rel).name + stem = Path(rel).stem + if any(base.startswith(prefix) for prefix in debug_prefixes): + flagged.append(rel) + continue + for target_stem in target_stems: + if any(stem == f"{target_stem}{suffix}" for suffix in suffixes): + flagged.append(rel) + break + return sorted(set(flagged)) + + +def _extract_metric(text: str, key: str) -> int | None: + pattern = rf'"{re.escape(key)}"\s*:\s*(\d+)' + match = re.findall(pattern, text) + if not match: + return None + return int(match[-1]) + + +def _budget_reason( + proc: subprocess.CompletedProcess[str], + changed_files_count: int, + changed_line_count: int, + config: OrchestratorConfig, +) -> str: + stdout = str(proc.stdout or "") + stderr = str(proc.stderr or "") + merged = f"{stdout}\n{stderr}" + turns = _extract_metric(merged, "turns_used") + commands = _extract_metric(merged, "tool_calls_used") + if turns is not None and turns > config.max_worker_model_turns: + return f"budget exceeded: model turns {turns}>{config.max_worker_model_turns}" + if commands is not None and commands > config.max_worker_shell_commands: + return f"budget exceeded: shell commands {commands}>{config.max_worker_shell_commands}" + if changed_files_count > config.max_worker_changed_files: + return f"budget exceeded: changed files {changed_files_count}>{config.max_worker_changed_files}" + if changed_line_count > config.max_worker_changed_lines: + return f"budget exceeded: changed lines {changed_line_count}>{config.max_worker_changed_lines}" + return "" + + +def _extract_last_command(stdout: str, stderr: str) -> str: + merged = f"{stdout}\n{stderr}" + matches = re.findall(r'"command"\s*:\s*"([^"\\]*(?:\\.[^"\\]*)*)"', merged) + if matches: + return bytes(matches[-1], "utf-8").decode("unicode_escape") + return "" diff --git a/villani_code/orchestrator_models.py b/villani_code/orchestrator_models.py new file mode 100644 index 00000000..19708843 --- /dev/null +++ b/villani_code/orchestrator_models.py @@ -0,0 +1,36 @@ +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Literal + + +WorkerStatus = Literal["success", "blocked_environment", "blocked_scope", "failed"] + + +@dataclass(slots=True) +class Subtask: + id: str + goal: str + success_criteria: list[str] = field(default_factory=list) + target_files: list[str] = field(default_factory=list) + scope_hint: str = "" + + +@dataclass(slots=True) +class SupervisorResult: + subtasks: list[Subtask] + + +@dataclass(slots=True) +class WorkerResult: + status: WorkerStatus + summary: str + files_touched: list[str] = field(default_factory=list) + recommended_verification: list[str] = field(default_factory=list) + + +@dataclass(slots=True) +class VerificationOutcome: + ok: bool + reasons: list[str] = field(default_factory=list) + commands: list[str] = field(default_factory=list) diff --git a/villani_code/orchestrator_roles.py b/villani_code/orchestrator_roles.py new file mode 100644 index 00000000..fd6bacfd --- /dev/null +++ b/villani_code/orchestrator_roles.py @@ -0,0 +1,58 @@ +from __future__ import annotations + +import json +from dataclasses import asdict + +from villani_code.orchestrator_models import Subtask, SupervisorResult + + +def build_supervisor_instruction(objective: str, max_workers: int) -> str: + example = SupervisorResult( + subtasks=[ + Subtask( + id="task_1", + goal="Bounded implementation step", + success_criteria=["Objective progresses materially"], + target_files=["path/to/file.py"], + scope_hint="Keep the patch minimal", + ) + ] + ) + return ( + "You are the supervisor role for Villani Code orchestrator. " + "Return strict JSON only with this shape: " + f"{json.dumps(asdict(example), separators=(',', ':'))}. " + f"Produce between 1 and {max_workers} subtasks. " + "Return as few subtasks as necessary; prefer fewer, larger bounded subtasks. " + "Each subtask must be independently executable against the current repo state. " + "Do not create subtasks that depend on unfinished future subtasks. " + "Do not split presentation-layer work before core ingestion/parsing works. " + "If unsure, return fewer subtasks. " + "Environment is Windows/PowerShell-first; avoid Unix-only command assumptions. " + "No markdown. No prose. No code edits. " + f"Top-level objective:\n{objective}" + ) + + +def build_worker_instruction( + objective: str, + subtask_payload: dict[str, object], + previous_failure: str | None = None, +) -> str: + retry_suffix = "" + if previous_failure: + retry_suffix = f"\nRetry context from previous attempt: {previous_failure}\n" + return ( + "You are the worker role for Villani Code orchestrator. " + "Edit files directly in this repository to complete the bounded subtask. " + "Environment is Windows/PowerShell-first: prefer Python scripts or PowerShell-safe commands. " + "Avoid Unix-only commands unless a compatible shell is clearly available. " + "Do not start long-lived foreground servers unless explicitly required. " + "Prefer short verification commands that terminate cleanly. " + "Return strict JSON only with keys: status, summary, files_touched, recommended_verification. " + "Allowed status values: success, blocked_environment, blocked_scope, failed. " + "No markdown. No prose outside JSON. " + f"Top-level objective:\n{objective}\n" + f"Assigned subtask JSON:\n{json.dumps(subtask_payload, ensure_ascii=False)}" + f"{retry_suffix}" + ) diff --git a/villani_code/orchestrator_verify.py b/villani_code/orchestrator_verify.py new file mode 100644 index 00000000..82acd52e --- /dev/null +++ b/villani_code/orchestrator_verify.py @@ -0,0 +1,157 @@ +from __future__ import annotations + +import shutil +import subprocess +from collections.abc import Iterable +from dataclasses import asdict +import json +from pathlib import Path + +from villani_code.orchestrator_models import VerificationOutcome + + +def snapshot_files(repo: Path, files: list[str], snapshot_dir: Path) -> None: + snapshot_dir.mkdir(parents=True, exist_ok=True) + metadata: dict[str, bool] = {} + for rel in files: + src = repo / rel + dst = snapshot_dir / rel + dst.parent.mkdir(parents=True, exist_ok=True) + metadata[rel] = src.exists() and src.is_file() + if src.exists() and src.is_file(): + shutil.copy2(src, dst) + (snapshot_dir / "snapshot_meta.json").write_text(json.dumps(metadata, indent=2), encoding="utf-8") + + +def restore_files(repo: Path, files: list[str], snapshot_dir: Path) -> None: + meta_path = snapshot_dir / "snapshot_meta.json" + metadata: dict[str, bool] = {} + if meta_path.exists(): + try: + payload = json.loads(meta_path.read_text(encoding="utf-8")) + if isinstance(payload, dict): + metadata = {str(k): bool(v) for k, v in payload.items()} + except json.JSONDecodeError: + metadata = {} + for rel in files: + src = snapshot_dir / rel + dst = repo / rel + existed_before = metadata.get(rel, src.exists() and src.is_file()) + if existed_before and src.exists() and src.is_file(): + dst.parent.mkdir(parents=True, exist_ok=True) + shutil.copy2(src, dst) + elif not existed_before and dst.exists() and dst.is_file(): + dst.unlink() + + +def capture_repo_file_state(repo: Path) -> dict[str, int]: + state: dict[str, int] = {} + root = repo.resolve() + for path in root.rglob("*"): + if not path.is_file(): + continue + rel = path.relative_to(root).as_posix() + if rel.startswith(".git/") or rel.startswith(".villani_code/"): + continue + try: + state[rel] = hash(path.read_bytes()) + except OSError: + continue + return state + + +def diff_repo_file_state(before: dict[str, int], after: dict[str, int]) -> tuple[list[str], list[str], list[str]]: + before_keys = set(before) + after_keys = set(after) + created = sorted(after_keys - before_keys) + deleted = sorted(before_keys - after_keys) + modified = sorted(rel for rel in (before_keys & after_keys) if before.get(rel) != after.get(rel)) + return modified, created, deleted + + +def count_changed_lines(repo: Path, files: Iterable[str]) -> int: + changed = 0 + for rel in files: + path = repo / rel + if not path.exists() or not path.is_file(): + continue + text = path.read_text(encoding="utf-8", errors="replace") + changed += len(text.splitlines()) + return changed + + +def cleanup_created_files(repo: Path, files: Iterable[str]) -> None: + for rel in files: + path = repo / rel + if path.exists() and path.is_file(): + path.unlink() + + +def run_verification( + repo: Path, + worker_recommended: list[str], + success_criteria: list[str], + files_touched: list[str], + changed_line_count: int, + max_files: int = 5, + max_lines: int = 250, +) -> VerificationOutcome: + reasons: list[str] = [] + commands: list[str] = [] + _ = success_criteria + + if not files_touched: + reasons.append("fail: no diff") + + if len(set(files_touched)) > max_files: + reasons.append(f"suspicious breadth: {len(set(files_touched))} files") + + if changed_line_count > max_lines: + reasons.append(f"diff too large: {changed_line_count} changed lines") + + chosen = _pick_verification_command(worker_recommended=worker_recommended, files_touched=files_touched, success_criteria=success_criteria) + if chosen: + proc = subprocess.run(chosen, cwd=repo, shell=True, capture_output=True, text=True) + commands.append(chosen) + if proc.returncode != 0: + reasons.append(f"command failed: {chosen}") + + return VerificationOutcome(ok=not reasons, reasons=reasons, commands=commands) + + +def to_json(outcome: VerificationOutcome) -> dict[str, object]: + return asdict(outcome) + + +def _pick_verification_command(worker_recommended: list[str], files_touched: list[str], success_criteria: list[str]) -> str | None: + seen: set[str] = set() + for cmd in worker_recommended: + normalized = cmd.strip() + if not normalized or normalized in success_criteria or normalized in seen: + continue + seen.add(normalized) + if _is_broad_or_repetitive_command(normalized): + continue + return normalized + python_targets = [rel for rel in files_touched if rel.endswith(".py")] + if python_targets: + return f"python -m py_compile {python_targets[0]}" + return "python -c \"print('verification-ok')\"" + + +def _is_broad_or_repetitive_command(command: str) -> bool: + lowered = command.lower() + broad_patterns = ( + "pytest", + "python -m pytest", + "uv run pytest", + "npm test", + "pnpm test", + "tail -", + "head ", + "for ", + "while ", + ) + if any(pattern in lowered for pattern in broad_patterns): + return True + return lowered.count("&&") > 1