Expose test failure evidence to retries

This commit is contained in:
Daniel Maddern 2026-08-15 16:12:02 +07:00
parent 1726b4de5f
commit 80c2a1b46c
2 changed files with 44 additions and 1 deletions

View file

@ -92,6 +92,7 @@ class AutonomousTaskLoop:
test_run = self.tests.run(task.project, task, Path(worktree.worktree_path), test_command) test_run = self.tests.run(task.project, task, Path(worktree.worktree_path), test_command)
if test_run.status != "PASS": if test_run.status != "PASS":
self._attach_test_failure_evidence(attempt, test_run)
self.bus.publish(EventType.TEST_FAILED, project=task.project, task=task, payload={"test_run_id": str(test_run.id)}) self.bus.publish(EventType.TEST_FAILED, project=task.project, task=task, payload={"test_run_id": str(test_run.id)})
tools.git(["add", "-N", "."]) tools.git(["add", "-N", "."])
@ -217,6 +218,23 @@ class AutonomousTaskLoop:
"write_file_fallback_rate": 0 if write_file_operations == 0 else totals["write_file_fallbacks"] / write_file_operations, "write_file_fallback_rate": 0 if write_file_operations == 0 else totals["write_file_fallbacks"] / write_file_operations,
} }
def _attach_test_failure_evidence(self, attempt: TaskAttempt, test_run) -> None:
content = test_run.output_artifact.content if test_run.output_artifact else {}
stdout = str(content.get("stdout", "")) if isinstance(content, dict) else ""
stderr = str(content.get("stderr", "")) if isinstance(content, dict) else ""
evidence = {
"test_run_id": str(test_run.id),
"status": test_run.status,
"stdout_excerpt": stdout[-12000:],
"stderr_excerpt": stderr[-4000:],
}
coder_result = dict(attempt.coder_result or {})
metadata = dict(coder_result.get("metadata", {})) if isinstance(coder_result.get("metadata", {}), dict) else {}
metadata["test_failure_evidence"] = evidence
coder_result["metadata"] = metadata
attempt.coder_result = coder_result
attempt.save(update_fields=["coder_result", "updated_at"])
def _scrub_context(self, context: dict[str, object]) -> dict[str, object]: def _scrub_context(self, context: dict[str, object]) -> dict[str, object]:
scrubbed = dict(context) scrubbed = dict(context)
scrubbed.pop("secrets", None) scrubbed.pop("secrets", None)

View file

@ -6,7 +6,8 @@ from pathlib import Path
from agents.providers import DeterministicCodingProvider from agents.providers import DeterministicCodingProvider
from control_plane.agents.management.commands.seed_core_agents import Command as SeedAgentsCommand from control_plane.agents.management.commands.seed_core_agents import Command as SeedAgentsCommand
from control_plane.events.models import Event, EventType from control_plane.events.models import Event, EventType
from control_plane.projects.models import CommitRecord, Milestone, Project, ProjectPlan, Task, TaskStatus from control_plane.agents.models import AgentRole, AgentVersion
from control_plane.projects.models import Artifact, CommitRecord, Milestone, Project, ProjectPlan, Task, TaskStatus
from control_plane.verification.models import Review, TestRun as ArtifexTestRun, Verification, VerificationResult from control_plane.verification.models import Review, TestRun as ArtifexTestRun, Verification, VerificationResult
from model_router.router import ModelRouter from model_router.router import ModelRouter
from runtime_loop.autonomous_task_loop import AutonomousTaskLoop from runtime_loop.autonomous_task_loop import AutonomousTaskLoop
@ -128,3 +129,27 @@ def test_m2_negative_path_rejected_without_commit(tmp_path: Path) -> None:
assert Review.objects.filter(task=task).exclude(status="PASS").exists() assert Review.objects.filter(task=task).exclude(status="PASS").exists()
assert task.attempts.filter(status="FAILED").exists() assert task.attempts.filter(status="FAILED").exists()
assert Event.objects.filter(task=task, event_type=EventType.TASK_FAILED).exists() assert Event.objects.filter(task=task, event_type=EventType.TASK_FAILED).exists()
def test_failed_test_output_is_attached_to_attempt_retry_context(tmp_path: Path) -> None:
repo = create_disposable_django_repo(tmp_path)
task = create_task(repo, "Add a feature that fails tests", ["tests pass"])
SeedAgentsCommand().handle()
coder = AgentVersion.objects.get(agent__role=AgentRole.CODER, promotion_status="CHAMPION")
attempt = task.attempts.create(task=task, attempt_number=1, coder=coder, status="RUNNING", coder_result={"metadata": {}})
artifact = Artifact.objects.create(
project=task.project,
task=task,
artifact_type="test_report",
name="failed-tests",
content={"stdout": "short traceback", "stderr": "error details", "returncode": 1},
)
test_run = ArtifexTestRun.objects.create(project=task.project, task=task, command="pytest", status="FAIL", output_artifact=artifact)
loop()._attach_test_failure_evidence(attempt, test_run)
attempt.refresh_from_db()
evidence = attempt.coder_result["metadata"]["test_failure_evidence"]
assert evidence["status"] == "FAIL"
assert evidence["stdout_excerpt"] == "short traceback"
assert evidence["stderr_excerpt"] == "error details"