279 lines
14 KiB
Python
279 lines
14 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
from agents.providers import DeterministicCodingProvider
|
|
from control_plane.agents.management.commands.seed_core_agents import Command as SeedAgentsCommand
|
|
from control_plane.agents.models import ProgenySignal
|
|
from control_plane.events.models import Event, EventType
|
|
from control_plane.projects.models import CommitRecord, Project, ProjectPlan, Milestone, Task, TaskStatus
|
|
from control_plane.verification.models import Review, TestRun, Verification, VerificationResult
|
|
from graph.models import ExecutionGraphDefinition, ExecutionGraphVersion, ExecutionGraphVersionStatus, GraphEdgeTraversal, GraphRun, GraphRunStatus
|
|
from graph.native_runtime import NativeGraphRuntime
|
|
from graph.langgraph_runtime import LangGraphRuntime
|
|
from graph.task_execution import task_execution_graph_v1
|
|
from graph.task_nodes import TaskExecutionServices, task_execution_registry
|
|
from model_router.router import ModelResponseContract, ModelRouter
|
|
from tests.test_m2_autonomous_loop import create_disposable_django_repo
|
|
|
|
|
|
def create_task(repository_path: Path, goal: str, acceptance: list[str], *, max_retries: int = 2) -> Task:
|
|
project = Project.objects.create(name=f"Graph Project {goal[:12]}", goal=goal, repository_path=str(repository_path))
|
|
plan = ProjectPlan.objects.create(project=project, version=1, goal=goal)
|
|
milestone = Milestone.objects.create(project=project, plan=plan, key="G1", title="Graph", goal="Execution graph")
|
|
return Task.objects.create(
|
|
project=project,
|
|
milestone=milestone,
|
|
task_type="implementation",
|
|
status=TaskStatus.RUNNING,
|
|
goal=goal,
|
|
acceptance_criteria=acceptance,
|
|
max_retries=max_retries,
|
|
)
|
|
|
|
|
|
def graph_run_for_task(task: Task) -> GraphRun:
|
|
spec = task_execution_graph_v1()
|
|
definition, _ = ExecutionGraphDefinition.objects.get_or_create(
|
|
name=spec.name,
|
|
defaults={"graph_type": spec.graph_type, "description": "Task execution graph"},
|
|
)
|
|
version, _ = ExecutionGraphVersion.objects.get_or_create(
|
|
graph=definition,
|
|
version=spec.version,
|
|
defaults={"status": ExecutionGraphVersionStatus.CHAMPION, "graph_spec": spec.to_dict()},
|
|
)
|
|
return GraphRun.objects.create(
|
|
execution_graph_version=version,
|
|
project=task.project,
|
|
milestone=task.milestone,
|
|
feature=task.feature,
|
|
task=task,
|
|
current_node=spec.entry,
|
|
)
|
|
|
|
|
|
class FirstAttemptBadProvider(DeterministicCodingProvider):
|
|
def __init__(self) -> None:
|
|
self.edit_calls = 0
|
|
|
|
def complete(self, request):
|
|
if "inspection phase" in request.prompt.lower():
|
|
return super().complete(request)
|
|
self.edit_calls += 1
|
|
if self.edit_calls == 1:
|
|
return ModelResponseContract(
|
|
"qwen-deterministic",
|
|
"Wrote intentionally insufficient change.",
|
|
{"operations": [{"type": "write_file", "path": "bad.txt", "content": "not enough\n"}]},
|
|
)
|
|
return super().complete(request)
|
|
|
|
|
|
class InvalidJsonProvider(DeterministicCodingProvider):
|
|
def complete(self, request):
|
|
if "inspection phase" in request.prompt.lower():
|
|
return super().complete(request)
|
|
return ModelResponseContract("qwen-deterministic", "{invalid", {})
|
|
|
|
|
|
class TimeoutProvider(DeterministicCodingProvider):
|
|
def complete(self, request):
|
|
if "inspection phase" in request.prompt.lower():
|
|
return super().complete(request)
|
|
raise TimeoutError("provider timed out")
|
|
|
|
|
|
def run_graph(task: Task, *, interrupt_after: str | None = None, provider=None) -> GraphRun:
|
|
SeedAgentsCommand().handle()
|
|
services = TaskExecutionServices(ModelRouter({"qwen": provider or DeterministicCodingProvider()}), test_command=["python", "manage.py", "test"])
|
|
runtime = NativeGraphRuntime(task_execution_registry(services))
|
|
return runtime.run_until_terminal_or_paused(graph_run_for_task(task), interrupt_after=interrupt_after)
|
|
|
|
|
|
def run_langgraph(task: Task, *, interrupt_after: str | None = None, provider=None) -> GraphRun:
|
|
SeedAgentsCommand().handle()
|
|
services = TaskExecutionServices(ModelRouter({"qwen": provider or DeterministicCodingProvider()}), test_command=["python", "manage.py", "test"])
|
|
runtime = LangGraphRuntime(task_execution_registry(services))
|
|
return runtime.run_until_terminal_or_paused(graph_run_for_task(task), interrupt_after=interrupt_after)
|
|
|
|
|
|
def test_native_task_execution_graph_success_matches_loop_semantics(tmp_path: Path) -> None:
|
|
repo = create_disposable_django_repo(tmp_path)
|
|
task = create_task(repo, "Add a /health endpoint returning JSON {\"status\": \"ok\"} and add tests.", ["/health returns JSON ok", "tests pass"])
|
|
|
|
graph_run = run_graph(task)
|
|
|
|
task.refresh_from_db()
|
|
assert graph_run.status == GraphRunStatus.COMPLETE
|
|
assert task.status == TaskStatus.COMPLETE
|
|
assert CommitRecord.objects.filter(task=task).count() == 1
|
|
assert TestRun.objects.get(task=task).status == "PASS"
|
|
assert Review.objects.get(task=task).status == "PASS"
|
|
assert Verification.objects.get(task=task).result == VerificationResult.PASS
|
|
assert task.worktree.status == "CLEANED"
|
|
assert GraphEdgeTraversal.objects.filter(graph_run=graph_run, source_node="judge", target_node="commit", condition="PASS").exists()
|
|
assert graph_run.metadata["final_failure_reason"] is None
|
|
assert "current_failure_reason" not in graph_run.metadata
|
|
assert "last_failure_reason" not in graph_run.metadata
|
|
|
|
|
|
def test_native_task_execution_graph_retry_exhaustion_matches_loop_semantics(tmp_path: Path) -> None:
|
|
repo = create_disposable_django_repo(tmp_path)
|
|
task = create_task(repo, "FORCE_BAD_IMPLEMENTATION Add a /health endpoint returning JSON {\"status\": \"ok\"} and add tests.", ["/health returns JSON ok", "tests pass"], max_retries=1)
|
|
|
|
graph_run = run_graph(task)
|
|
|
|
task.refresh_from_db()
|
|
assert graph_run.status == GraphRunStatus.FAILED
|
|
assert task.status == TaskStatus.FAILED
|
|
assert task.retry_count == 2
|
|
assert CommitRecord.objects.filter(task=task).count() == 0
|
|
assert Event.objects.filter(task=task, event_type=EventType.TASK_FAILED).exists()
|
|
assert Event.objects.filter(task=task, event_type="TASK_RETRY_EXHAUSTED").exists()
|
|
assert graph_run.metadata["final_failure_reason"] == "review_failed"
|
|
assert graph_run.metadata["historical_failures"][-1]["reason"] == "review_failed"
|
|
retry_signal = ProgenySignal.objects.get(task=task, failure_category="TASK_RETRY_EXHAUSTED")
|
|
assert retry_signal.graph_run == graph_run
|
|
assert retry_signal.graph_node_run.node_id == "retry_or_fail"
|
|
assert ProgenySignal.objects.filter(task=task, source="graph_runtime").count() == 0
|
|
|
|
|
|
def test_native_task_execution_graph_resume_after_coder_prevents_duplicate_commit(tmp_path: Path) -> None:
|
|
repo = create_disposable_django_repo(tmp_path)
|
|
task = create_task(repo, "Add a /health endpoint returning JSON {\"status\": \"ok\"} and add tests.", ["/health returns JSON ok", "tests pass"])
|
|
SeedAgentsCommand().handle()
|
|
graph_run = graph_run_for_task(task)
|
|
services = TaskExecutionServices(ModelRouter({"qwen": DeterministicCodingProvider()}), test_command=["python", "manage.py", "test"])
|
|
runtime = NativeGraphRuntime(task_execution_registry(services))
|
|
|
|
runtime.run_until_terminal_or_paused(graph_run, interrupt_after="coder")
|
|
graph_run.refresh_from_db()
|
|
assert graph_run.current_node == "coder"
|
|
assert task.attempts.count() == 1
|
|
|
|
runtime.run_until_terminal_or_paused(graph_run)
|
|
runtime.run_until_terminal_or_paused(graph_run)
|
|
|
|
task.refresh_from_db()
|
|
assert task.status == TaskStatus.COMPLETE
|
|
assert CommitRecord.objects.filter(task=task).count() == 1
|
|
assert graph_run.node_runs.filter(node_id="coder").count() == 1
|
|
|
|
|
|
def test_model_output_invalid_creates_progeny_signal_with_graph_lineage(tmp_path: Path) -> None:
|
|
repo = create_disposable_django_repo(tmp_path)
|
|
task = create_task(repo, "Add a /health endpoint returning JSON {\"status\": \"ok\"} and add tests.", ["/health returns JSON ok", "tests pass"], max_retries=0)
|
|
|
|
graph_run = run_graph(task, provider=InvalidJsonProvider())
|
|
|
|
assert graph_run.status == GraphRunStatus.FAILED
|
|
signal = ProgenySignal.objects.get(task=task, failure_category="MODEL_OUTPUT_INVALID")
|
|
assert signal.graph_run == graph_run
|
|
assert signal.graph_node_run.node_id == "coder"
|
|
assert signal.evidence["node_type"] == "coder"
|
|
|
|
|
|
def test_provider_timeout_creates_distinct_progeny_signal(tmp_path: Path) -> None:
|
|
repo = create_disposable_django_repo(tmp_path)
|
|
task = create_task(repo, "Add a /health endpoint returning JSON {\"status\": \"ok\"} and add tests.", ["/health returns JSON ok", "tests pass"])
|
|
|
|
graph_run = run_graph(task, provider=TimeoutProvider())
|
|
|
|
assert graph_run.status == GraphRunStatus.FAILED
|
|
signal = ProgenySignal.objects.get(task=task, source="provider", failure_category="PROVIDER_TIMEOUT")
|
|
assert signal.graph_run == graph_run
|
|
assert signal.graph_node_run.node_id == "coder"
|
|
|
|
|
|
def test_langgraph_task_execution_graph_success_matches_native_domain_outcome(tmp_path: Path) -> None:
|
|
repo = create_disposable_django_repo(tmp_path)
|
|
task = create_task(repo, "Add a /health endpoint returning JSON {\"status\": \"ok\"} and add tests.", ["/health returns JSON ok", "tests pass"])
|
|
|
|
graph_run = run_langgraph(task)
|
|
|
|
task.refresh_from_db()
|
|
assert graph_run.status == GraphRunStatus.COMPLETE
|
|
assert graph_run.execution_graph_version.graph.name == "task_execution"
|
|
assert task.status == TaskStatus.COMPLETE
|
|
commit = CommitRecord.objects.get(task=task)
|
|
assert commit.graph_run == graph_run
|
|
assert commit.graph_run.execution_graph_version == graph_run.execution_graph_version
|
|
assert TestRun.objects.get(task=task).status == "PASS"
|
|
assert Review.objects.get(task=task).status == "PASS"
|
|
assert Verification.objects.get(task=task).result == VerificationResult.PASS
|
|
assert GraphEdgeTraversal.objects.filter(graph_run=graph_run, source_node="review", target_node="judge", condition="PASS").exists()
|
|
assert graph_run.metadata["final_failure_reason"] is None
|
|
assert "current_failure_reason" not in graph_run.metadata
|
|
assert "last_failure_reason" not in graph_run.metadata
|
|
assert graph_run.metadata["last_node_id"] == "cleanup"
|
|
|
|
|
|
def test_langgraph_task_execution_graph_retry_exhaustion_matches_native_domain_outcome(tmp_path: Path) -> None:
|
|
repo = create_disposable_django_repo(tmp_path)
|
|
task = create_task(repo, "FORCE_BAD_IMPLEMENTATION Add a /health endpoint returning JSON {\"status\": \"ok\"} and add tests.", ["/health returns JSON ok", "tests pass"], max_retries=1)
|
|
|
|
graph_run = run_langgraph(task)
|
|
|
|
task.refresh_from_db()
|
|
assert graph_run.status == GraphRunStatus.FAILED
|
|
assert task.status == TaskStatus.FAILED
|
|
assert task.retry_count == 2
|
|
assert CommitRecord.objects.filter(task=task).count() == 0
|
|
assert Event.objects.filter(task=task, event_type="TASK_RETRY_EXHAUSTED").exists()
|
|
assert graph_run.metadata["final_failure_reason"] == "review_failed"
|
|
assert graph_run.metadata["historical_failures"][-1]["reason"] == "review_failed"
|
|
reviewer_signal = ProgenySignal.objects.filter(task=task, source="reviewer", failure_category="REWORK_REQUIRED").first()
|
|
assert reviewer_signal is not None
|
|
assert reviewer_signal.graph_run == graph_run
|
|
assert reviewer_signal.graph_node_run.node_id == "review"
|
|
|
|
|
|
def test_langgraph_retry_then_success_preserves_historical_failures_without_final_failure(tmp_path: Path) -> None:
|
|
repo = create_disposable_django_repo(tmp_path)
|
|
task = create_task(repo, "Add a /health endpoint returning JSON {\"status\": \"ok\"} and add tests.", ["/health returns JSON ok", "tests pass"])
|
|
|
|
graph_run = run_langgraph(task, provider=FirstAttemptBadProvider())
|
|
|
|
task.refresh_from_db()
|
|
assert graph_run.status == GraphRunStatus.COMPLETE
|
|
assert task.status == TaskStatus.COMPLETE
|
|
assert task.retry_count == 1
|
|
assert graph_run.metadata["final_failure_reason"] is None
|
|
assert "current_failure_reason" not in graph_run.metadata
|
|
assert graph_run.metadata["historical_failures"] == [
|
|
{
|
|
"node_id": "review",
|
|
"visit_index": 1,
|
|
"attempt_id": str(task.attempts.order_by("attempt_number").first().id),
|
|
"attempt_number": 1,
|
|
"reason": "review_failed",
|
|
"evidence": [{"type": "missing_health_route", "severity": "high", "message": "Diff does not add /health"}],
|
|
}
|
|
]
|
|
reviewer_signal = ProgenySignal.objects.get(task=task, source="reviewer")
|
|
assert reviewer_signal.execution_graph_version == graph_run.execution_graph_version
|
|
assert reviewer_signal.graph_node_run.node_id == "review"
|
|
|
|
|
|
def test_langgraph_resume_after_coder_tests_and_judge_does_not_duplicate_commit(tmp_path: Path) -> None:
|
|
for node_id in ["coder", "run_tests", "judge"]:
|
|
root = tmp_path / node_id
|
|
root.mkdir()
|
|
repo = create_disposable_django_repo(root)
|
|
task = create_task(repo, "Add a /health endpoint returning JSON {\"status\": \"ok\"} and add tests.", ["/health returns JSON ok", "tests pass"])
|
|
SeedAgentsCommand().handle()
|
|
graph_run = graph_run_for_task(task)
|
|
services = TaskExecutionServices(ModelRouter({"qwen": DeterministicCodingProvider()}), test_command=["python", "manage.py", "test"])
|
|
runtime = LangGraphRuntime(task_execution_registry(services))
|
|
|
|
runtime.run_until_terminal_or_paused(graph_run, interrupt_after=node_id)
|
|
graph_run.refresh_from_db()
|
|
assert graph_run.current_node == node_id
|
|
|
|
runtime.run_until_terminal_or_paused(graph_run)
|
|
runtime.run_until_terminal_or_paused(graph_run)
|
|
|
|
task.refresh_from_db()
|
|
assert task.status == TaskStatus.COMPLETE
|
|
assert CommitRecord.objects.filter(task=task).count() == 1
|