175 lines
8 KiB
Python
175 lines
8 KiB
Python
from __future__ import annotations
|
|
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
from agents.coder import Coder
|
|
from agents.providers import DeterministicCodingProvider
|
|
from model_router.router import ModelRequestContract, ModelResponseContract, ModelRouter
|
|
from tools.capabilities import Capability
|
|
from tools.runtime import CapabilityError, WorktreeTools
|
|
|
|
|
|
def test_worktree_investigation_tools_are_path_scoped(tmp_path: Path) -> None:
|
|
repo = tmp_path / "repo"
|
|
repo.mkdir()
|
|
(repo / "app").mkdir()
|
|
(repo / "app" / "urls.py").write_text("urlpatterns = []\n", encoding="utf-8")
|
|
(repo / "items.py").write_text("class Item:\n pass\n", encoding="utf-8")
|
|
subprocess.run(["git", "init", "-b", "main"], cwd=repo, check=True, capture_output=True, text=True)
|
|
tools = WorktreeTools(repo, {Capability.READ_REPOSITORY, Capability.INVESTIGATE_WORKTREE})
|
|
|
|
assert "app/" in tools.list_directory(".")
|
|
assert "urlpatterns" in tools.read_file("app/urls.py")
|
|
assert tools.search_code("urlpatterns")[0]["path"] == "app/urls.py"
|
|
assert tools.find_symbol("Item")[0]["path"] == "items.py"
|
|
assert tools.git_status() == "?? app/\n?? items.py\n"
|
|
|
|
try:
|
|
tools.read_file("../outside.txt")
|
|
except CapabilityError as exc:
|
|
assert "escapes worktree" in str(exc)
|
|
else:
|
|
raise AssertionError("path escape should be blocked")
|
|
|
|
|
|
def test_coder_requires_inspection_for_architecture_sensitive_tasks(tmp_path: Path) -> None:
|
|
repo = tmp_path / "repo"
|
|
repo.mkdir()
|
|
(repo / "app").mkdir()
|
|
(repo / "app" / "urls.py").write_text("urlpatterns = []\n", encoding="utf-8")
|
|
(repo / "tests").mkdir()
|
|
subprocess.run(["git", "init", "-b", "main"], cwd=repo, check=True, capture_output=True, text=True)
|
|
tools = WorktreeTools(
|
|
repo,
|
|
{Capability.READ_REPOSITORY, Capability.INVESTIGATE_WORKTREE, Capability.WRITE_WORKTREE, Capability.RUN_TESTS},
|
|
)
|
|
context = {
|
|
"task": {
|
|
"goal": 'Add a /health endpoint returning JSON {"status": "ok"} and add tests.',
|
|
"acceptance_criteria": ["tests pass"],
|
|
}
|
|
}
|
|
|
|
result = Coder(ModelRouter({"qwen": DeterministicCodingProvider()})).execute(context, tools)
|
|
|
|
assert result.status == "COMPLETE"
|
|
assert result.metadata["inspection_results"]
|
|
assert (repo / "app" / "urls.py").read_text(encoding="utf-8")
|
|
|
|
|
|
class OperationProvider:
|
|
provider_name = "operation_fixture"
|
|
|
|
def __init__(self, operations):
|
|
self.operations = operations
|
|
|
|
def complete(self, request: ModelRequestContract) -> ModelResponseContract:
|
|
if "INSPECTION PHASE" in request.prompt:
|
|
return ModelResponseContract("fixture", "", {"inspect_operations": [{"type": "list_directory", "path": "."}]})
|
|
return ModelResponseContract("fixture", "", {"operations": self.operations})
|
|
|
|
def health(self) -> str:
|
|
return "AVAILABLE"
|
|
|
|
|
|
class SequentialOperationProvider:
|
|
provider_name = "sequential_fixture"
|
|
|
|
def __init__(self, operation_rounds):
|
|
self.operation_rounds = list(operation_rounds)
|
|
self.prompts: list[str] = []
|
|
|
|
def complete(self, request: ModelRequestContract) -> ModelResponseContract:
|
|
self.prompts.append(request.prompt)
|
|
if "INSPECTION PHASE" in request.prompt:
|
|
return ModelResponseContract("fixture", "", {"inspect_operations": [{"type": "read_file", "path": "file.txt"}]})
|
|
operations = self.operation_rounds.pop(0)
|
|
return ModelResponseContract("fixture", "", {"operations": operations})
|
|
|
|
def health(self) -> str:
|
|
return "AVAILABLE"
|
|
|
|
|
|
def test_coder_dispatches_new_mutation_operations(tmp_path: Path) -> None:
|
|
repo = tmp_path / "repo"
|
|
repo.mkdir()
|
|
(repo / "file.txt").write_text("one\ntwo\n", encoding="utf-8")
|
|
subprocess.run(["git", "init", "-b", "main"], cwd=repo, check=True, capture_output=True, text=True)
|
|
tools = WorktreeTools(repo, {Capability.READ_REPOSITORY, Capability.INVESTIGATE_WORKTREE, Capability.WRITE_WORKTREE, Capability.RUN_TESTS})
|
|
operations = [
|
|
{"type": "create_directory", "path": "pkg"},
|
|
{"type": "write_file", "path": "pkg/new.txt", "content": "new\n"},
|
|
{"type": "apply_patch", "path": "file.txt", "patch": "@@ -1,2 +1,2 @@\n one\n-two\n+TWO\n"},
|
|
{"type": "move_file", "source": "pkg/new.txt", "destination": "pkg/moved.txt"},
|
|
{"type": "delete_file", "path": "pkg/moved.txt"},
|
|
]
|
|
|
|
result = Coder(ModelRouter({"qwen": OperationProvider(operations)})).execute({"task": {"goal": "change tests"}}, tools)
|
|
|
|
assert result.status == "COMPLETE"
|
|
assert "TWO" in (repo / "file.txt").read_text(encoding="utf-8")
|
|
assert not (repo / "pkg" / "moved.txt").exists()
|
|
|
|
|
|
def test_coder_rejects_unsupported_operation(tmp_path: Path) -> None:
|
|
repo = tmp_path / "repo"
|
|
repo.mkdir()
|
|
subprocess.run(["git", "init", "-b", "main"], cwd=repo, check=True, capture_output=True, text=True)
|
|
tools = WorktreeTools(repo, {Capability.READ_REPOSITORY, Capability.INVESTIGATE_WORKTREE, Capability.WRITE_WORKTREE})
|
|
|
|
result = Coder(ModelRouter({"qwen": OperationProvider([{"type": "empty_file", "path": "x"}])})).execute({"task": {"goal": "simple"}}, tools)
|
|
|
|
assert result.status == "FAILED"
|
|
assert "Unsupported operation" in result.summary
|
|
|
|
|
|
def test_coder_refreshes_file_evidence_after_patch_context_mismatch(tmp_path: Path) -> None:
|
|
repo = tmp_path / "repo"
|
|
repo.mkdir()
|
|
(repo / "file.txt").write_text("alpha\nbeta\ngamma\n", encoding="utf-8")
|
|
subprocess.run(["git", "init", "-b", "main"], cwd=repo, check=True, capture_output=True, text=True)
|
|
tools = WorktreeTools(repo, {Capability.READ_REPOSITORY, Capability.INVESTIGATE_WORKTREE, Capability.WRITE_WORKTREE})
|
|
provider = SequentialOperationProvider(
|
|
[
|
|
[{"type": "apply_patch", "path": "file.txt", "patch": "@@ -1,3 +1,3 @@\n alpha\n-wrong\n+BETA\n gamma\n"}],
|
|
[{"type": "apply_patch", "path": "file.txt", "patch": "@@ -1,3 +1,3 @@\n alpha\n-beta\n+BETA\n gamma\n"}],
|
|
]
|
|
)
|
|
|
|
result = Coder(ModelRouter({"qwen": provider})).execute({"task": {"goal": "change model tests"}}, tools)
|
|
|
|
assert result.status == "COMPLETE"
|
|
assert (repo / "file.txt").read_text(encoding="utf-8") == "alpha\nBETA\ngamma\n"
|
|
assert result.metadata["mutation_failures"][0]["operation"] == "apply_patch"
|
|
assert result.metadata["mutation_failures"][0]["target_path"] == "file.txt"
|
|
assert "wrong" in result.metadata["mutation_failures"][0]["expected_context"]
|
|
assert "2: beta" in result.metadata["mutation_failures"][0]["relevant_current_file_excerpt"]
|
|
assert "previous_attempted_patch" in result.metadata["mutation_failures"][0]
|
|
assert result.metadata["telemetry"]["patch_mismatches"] == 1
|
|
assert result.metadata["telemetry"]["patch_successes"] == 1
|
|
assert result.metadata["telemetry"]["patch_success_rate"] == 0.5
|
|
assert result.metadata["telemetry"]["write_file_fallbacks"] == 0
|
|
assert "mutation_failure_evidence" in provider.prompts[-1]
|
|
assert "Do not fall back to write_file" in provider.prompts[-1]
|
|
|
|
|
|
def test_coder_preserves_evidence_when_patch_retry_fails(tmp_path: Path) -> None:
|
|
repo = tmp_path / "repo"
|
|
repo.mkdir()
|
|
(repo / "file.txt").write_text("alpha\nbeta\n", encoding="utf-8")
|
|
subprocess.run(["git", "init", "-b", "main"], cwd=repo, check=True, capture_output=True, text=True)
|
|
tools = WorktreeTools(repo, {Capability.READ_REPOSITORY, Capability.INVESTIGATE_WORKTREE, Capability.WRITE_WORKTREE})
|
|
provider = SequentialOperationProvider(
|
|
[
|
|
[{"type": "apply_patch", "path": "file.txt", "patch": "@@ -1,2 +1,2 @@\n alpha\n-wrong\n+BETA\n"}],
|
|
[{"type": "apply_patch", "path": "file.txt", "patch": "@@ -1,2 +1,2 @@\n alpha\n-still-wrong\n+BETA\n"}],
|
|
]
|
|
)
|
|
|
|
result = Coder(ModelRouter({"qwen": provider})).execute({"task": {"goal": "change model tests"}}, tools)
|
|
|
|
assert result.status == "FAILED"
|
|
assert result.metadata["inspection_results"]
|
|
assert len(result.metadata["mutation_failures"]) == 2
|
|
assert result.metadata["telemetry"]["patch_mismatches"] == 2
|