Originally published on tamiz.pro.
The current state of AI-assisted development is largely defined by autocomplete. Tools like GitHub Copilot and Cursor have revolutionized the speed at which we write code, but they remain passive assistants. They suggest; we decide. They complete; we verify. However, the next frontier of software engineering is not just faster typing—it is autonomous agency. We are moving from "AI that helps me write code" to "AI that verifies and ships features."
Building a pipeline where an AI agent does not just generate tokens, but actually engineers a feature end-to-end, requires a fundamental shift in architecture. It demands moving away from trust-based integration ("I hope this works") to evidence-based integration ("Here is the test that proves it works"). This deep dive explores the architectural patterns required to build these systems: Test-Driven Development (TDD) as a verification engine, LLM-based review guardrails to prevent hallucination-induced bugs, and multi-agent coordination to handle complex system design.
"Vibe coding"—writing software based on intuition and rapid AI generation without rigorous verification—works well for prototypes. It fails catastrophically in production. The core problem is that Large Language Models (LLMs) are probabilistic, not deterministic. They do not understand state; they predict the next likely token. An agent that simply writes a UserService and a UserRepository is operating in a vacuum. It has no feedback loop.
To build a production-grade pipeline, we must close the loop. The agent must not stop after code generation. It must stop only when the evidence is irrefutable. This evidence comes from three sources:
Traditionally, Test-Driven Development is a human discipline. It forces you to write the test before the implementation, ensuring the code is actually testable. For AI agents, TDD is not a discipline; it is a control flow mechanism.
In an autonomous pipeline, the "Test" is the acceptance criterion. If the agent cannot write a failing test that clearly describes the requirement, it does not understand the requirement.
Imagine a CodeAgent class. It does not have a single generate method. It has a refactor, implement, and verify lifecycle.
class FeatureAgent:
def __init__(self, llm: BaseLLM, file_system: LocalFS, test_runner: ShellRunner):
self.llm = llm
self.fs = file_system
self.test_runner = test_runner
def execute_tdd_cycle(self, requirement: str):
"""
Executes a TDD cycle: Red -> Green -> Refactor.
"""
test_code = self._generate_test(requirement)
self.fs.write_file("tests/test_feature.py", test_code)
result = self.test_runner.run("pytest tests/test_feature.py")
if not result.has_failures():
raise RuntimeError("Test passed immediately. Requirement is trivial or test is flawed.")
implementation = self._generate_implementation(requirement, test_code)
self.fs.write_file("src/feature.py", implementation)
result = self.test_runner.run("pytest tests/test_feature.py")
if not result.all_passed():
return self._auto_fix(result.output)
return True
The critical component here is the test_runner. The agent is not guessing that the code works. It is executing it. This shifts the architecture from a "Black Box" (input prompt -> output code) to a "Reinforcement Loop" (input prompt -> output code -> environment feedback -> error correction).
Even if the tests pass, the code might be bad. It might use deprecated APIs, ignore security best practices, or violate architectural patterns (e.g., a controller talking to the database). This is where the "Guardrail" agent comes in.
In this pipeline, we treat the LLM as a reviewer, not a coder. We use a distinct prompt context that focuses on critique. This is often called the "LLM-as-a-Judge" pattern.
The Critic agent should be stateless and fast. It reads the diff generated by the TDD agent and outputs a structured JSON verdict.
{
"severity": "high",
"issue": "SQL Injection vulnerability",
"location": "src/db.py:42",
"suggestion": "Use parameterized queries instead of string interpolation."
}
By forcing the Critic to output structured data, we make the pipeline deterministic. The main orchestrator can then parse this JSON. If severity is "high", the pipeline halts and feeds the suggestion back to the TDD agent. This creates a "Constitutional AI" constraint, where the agent is bound by rules defined in the Critic's prompt.
A single agent working in a monolithic prompt context will quickly hit the limits of its context window and become confused. Complex features (e.g., "Add a Stripe billing system") require separation of concerns.
We move to a Supervisor-Worker architecture.
You might think: "Let the LLM decide which agent to call next." This is a recipe for infinite loops and unpredictable costs. The LLM should not be the conductor; it should be the player. The conductor is your code.
async function runPipeline() {
// 1. Plan
const plan = await architectAgent.plan(featureDescription);
// 2. Execute Tasks Sequentially or in Parallel
for (const task of plan.tasks) {
if (task.dependsOn.length === 0) {
await workerAgent.execute(task);
} else {
// Check dependencies are met
if (task.dependsOn.every(id => completedTasks.has(id))) {
await workerAgent.execute(task);
}
}
}
// 3. Final Review
const diff = git.getDiff();
const review = await criticAgent.review(diff);
if (review.blockers > 0) {
// Trigger a fix loop
await workerAgent.fix(review.issues);
}
}
Let's look at how this actually works in a real-world scenario. We are adding a "Forgot Password" endpoint to a Node.js/Express app.
Prompt: "Design a system for password reset tokens. Consider expiration (24h), security (SHA-256 hashing), and database schema changes."
Output (Architect Agent):
{
"tasks": [
{ "id": "db", "desc": "Add 'reset_token_hash' and 'reset_expires_at' to users table" },
{ "id": "logic", "desc": "Create generateToken service" },
{ "id": "api", "desc": "Create /auth/forgot-password endpoint" }
]
}
Prompt: "Write a test that ensures the database migration adds the new columns. Then write the migration file."
The worker writes tests/migration.test.ts:
test('db schema update', async () => {
const schema = await getSchema();
expect(schema.tables.users.columns).toContain('reset_token_hash');
});
It runs the test. It fails (Red). It writes the migration. It runs the test. It passes (Green).
Prompt: "Implement the token generation logic based on the test in tests/token.test.ts."
The worker generates the logic, ensuring the token is hashed with SHA-256 before storage.
Prompt: "Review the diff for the last two commits. Check for: 1. Is the token stored in plain text? 2. Is the expiration time respected in the query? 3. Are SQL injections possible?"
The Critic analyzes the code. It finds that the SELECT query in the API endpoint uses string interpolation for the expiration check. It flags this as a high severity issue.
The orchestrator sees the high severity flag. It takes the Critic's suggestion: "Use parameterized queries." It passes this back to the API Worker. The Worker updates the code to use ? placeholders. The Critic re-reviews. It passes. The pipeline commits the changes.
Building these pipelines is not plug-and-play. There are specific failure modes that will break your automation.
As the agent makes more changes, the context window fills with previous code. The agent begins to "forget" the requirements from the beginning of the prompt.
The agent writes import { RedisClient } from 'redis' but the project uses ioredis.
npm install or pip install check, or better yet, the Critic must have access to the package.json / requirements.txt and verify imports against the actual dependencies.
The agent fails the test, tries to fix it, fails again, and gets stuck in a cycle.
HumanInterventionRequired exception. Giving an AI agent write access to your codebase and execution privileges is terrifying.
main. The final commit should be a PR generated by the pipeline, reviewed by a human, and merged manually (or via strict CI checks). We are currently in the early days of this. The "Agentic SWE" (Software Engineering) pipeline is the bridge to fully autonomous development. However, the bottleneck is not the LLM's intelligence; it is the LLM's reliability.
By wrapping the LLM in a TDD loop, we are essentially using the compiler and the test runner as our guardrails. The LLM is the "brain," but the test runner is the "safety net."
For now, do not try to build a general-purpose agent that can do everything. Build specialized agents that do one thing well, and coordinate them with a deterministic orchestrator. This is the path to software that is not just written by AI, but verified by it.
For the TDD Worker, a 16k-32k context window is usually sufficient to hold the requirement, the current file, and the error output. For the Critic, you need enough space to hold the entire diff plus the project's style guide. 32k-128k is recommended for complex reviews.
Yes. Using a different model (or even the same model with a temperature of 0.0 and a highly specific "judge" persona) prevents the agent from "rubber-stamping" its own work. A second opinion is always more valuable than a confirmation bias.
The orchestrator must decompose the task. The TDD agent should not be asked to rewrite the whole backend. It should be asked to "Update the User model to support soft deletes" and then "Update the UserService to respect soft deletes". This keeps the context small and the diffs reviewable.
For more insights on AI engineering and developer tooling, check out Tamiz's Insights.
While a single agent can handle focused tasks, real-world codebases demand specialization. The most effective autonomous pipelines employ multi-agent architectures where distinct agents own different concerns.
┌─────────────────────────────────────────────────────────┐
│ Orchestrator Agent │
│ (Task decomposition, state management, conflict res.) │
└──────────────┬──────────────────────────────────────────┘
│
┌──────────┼──────────────┬──────────────────┐
▼ ▼ ▼ ▼
┌────────┐ ┌────────┐ ┌───────────┐ ┌────────────┐
│Coder │ │Tester │ │Reviewer │ │Doc Writer │
│Agent │ │Agent │ │Agent │ │Agent │
└────────┘ └────────┘ └───────────┘ └────────────┘
Each agent receives a narrow system prompt with domain-specific context, not the full codebase. The orchestrator manages the DAG of dependencies:
from dataclasses import dataclass, field
from typing import Literal
from enum import Enum
class AgentRole(Enum):
Coder = "coder"
Tester = "tester"
Reviewer = "reviewer"
DocWriter = "doc_writer"
@dataclass
class TaskNode:
id: str
role: AgentRole
description: "str"
dependencies: list[str] = field(default_factory=list)
status: Literal["pending", "in_progress", "completed", "failed"] = "pending"
output: str | None = None
class Orchestrator:
def __init__(self, agents: dict[AgentRole, "BaseAgent"]):
self.agents = agents
self.task_graph: dict[str, TaskNode] = {}
self.execution_order: list[str] = []
def add_task(self, task: TaskNode) -> None:
self.task_graph[task.id] = task
def topological_sort(self) -> list[str]:
"""Resolve execution order respecting dependencies."""
visited = set()
temp_visited = set()
order = []
def visit(node_id: str):
if node_id in temp_visited:
raise ValueError(f"Circular dependency detected at {node_id}")
if node_id in visited:
return
temp_visited.add(node_id)
node = self.task_graph[node_id]
for dep in node.dependencies:
visit(dep)
temp_visited.remove(node_id)
visited.add(node_id)
order.append(node_id)
for node_id in self.task_graph:
visit(node_id)
return order
async def execute(self) -> dict[str, str]:
"""Execute tasks in dependency order."""
results = {}
for task_id in self.topological_sort():
node = self.task_graph[task_id]
node.status = "in_progress"
dep_context = {
dep_id: self.task_graph[dep_id].output
for dep_id in node.dependencies
if self.task_graph[dep_id].output
}
agent = self.agents[node.role]
try:
output = await agent.execute(
task_description=node.description,
context=dep_context
)
node.output = output
node.status = "completed"
results[task_id] = output
except Exception as e:
node.status = "failed"
results[task_id] = f"FAILED: {e}"
self._mark_dependents_failed(task_id)
return results
def _mark_dependents_failed(self, failed_id: str) -> None:
for node in self.task_graph.values():
if failed_id in node.dependencies and node.status == "pending":
node.status = "failed"
node.output = f"SKIPPED: dependency {failed_id} failed"
Agents don't share memory directly—they exchange structured artifacts. This prevents context pollution and makes the pipeline auditable:
from pydantic import BaseModel, Field
from typing import Any
class Artifact(BaseModel):
"""Standardized output format between agents."""
task_id: str
author_role: str
summary: str = Field(..., description="One-sentence summary of what was done")
files_modified: list[str] = Field(default_factory=list)
tests_added: list[str] = Field(default_factory=list)
evidence: dict[str, Any] = Field(default_factory=dict)
confidence: float = Field(..., ge=0.0, le=1.0)
follow_up_needed: bool = False
follow_up_reason: str | None = None
class ReviewArtifact(Artifact):
"""Specialized artifact from the reviewer agent."""
verdict: Literal["approved", "changes_requested", "blocked"]
critical_issues: list[str] = Field(default_factory=list)
suggestions: list[str] = Field(default_factory=list)
security_concerns: list[str] = Field(default_factory=list)
The Tester agent receives the Coder's artifact and produces its own:
class TesterAgent(BaseAgent):
def __init__(self, model: str = "gpt-4", test_runner: TestRunner | None = None):
super().__init__(model=model)
self.test_runner = test_runner or LocalTestRunner()
async def execute(self, task_description: "str, context: dict[str, str]) -> Artifact:"
code_context = context.get("coder_output", "")
test_plan = await self._generate_test_plan(task_description, code_context)
test_files = self._write_tests(test_plan)
results = await self.test_runner.run(
test_files=test_files,
coverage_threshold=80.0
)
return Artifact(
task_id=self.current_task_id,
author_role="tester",
summary=f"Generated {len(test_files)} test files, {results.passed} passed, {results.failed} failed",
files_modified=[f.path for f in test_files],
tests_added=[t.name for t in test_files],
evidence={
"coverage_percent": results.coverage,
"pass_rate": results.pass_rate,
"failed_tests": [f"{t.name}: {t.error}" for t in results.failures],
},
confidence=min(results.pass_rate / 100.0, 1.0),
follow_up_needed=results.failed > 0,
follow_up_reason=f"{results.failed} tests failing" if results.failed > 0 else None,
)
async def _generate_test_plan(self, description: "str, code: str) -> TestPlan:"
prompt = f"""
Given this task description: ""
{description}
And this implementation:
{code}
Generate a comprehensive test plan covering:
1. Happy path scenarios
2. Edge cases (null, empty, boundary values)
3. Error handling paths
4. Concurrency/race conditions if applicable
Return as a structured test plan.
"""
response = await self._call_model(prompt)
return TestPlan.from_response(response)
When multiple agents modify the same file (e.g., Coder adds a method, DocWriter updates the docstring), a merge coordinator resolves conflicts:
from dataclasses import dataclass
from pathlib import Path
@dataclass
class EditRegion:
file_path: str
start_line: int
end_line: int
agent_role: str
content: str
class MergeCoordinator:
"""Detect and resolve overlapping edits from multiple agents."""
def detect_conflicts(self, edits: list[EditRegion]) -> list[tuple[EditRegion, EditRegion]]:
conflicts = []
for i, edit_a in enumerate(edits):
for j, edit_b in enumerate(edits):
if i >= j:
continue
if self._overlaps(edit_a, edit_b):
conflicts.append((edit_a, edit_b))
return conflicts
def _overlaps(self, a: EditRegion, b: EditRegion) -> bool:
if a.file_path != b.file_path:
return False
return a.start_line <= b.end_line and b.start_line <= a.end_line
def resolve(self, file_path: str, edits: list[EditRegion]) -> str:
"""
Strategy: merge non-overlapping regions in order.
For overlapping regions, use the agent with higher priority.
"""
priority = {"coder": 3, "tester": 2, "doc_writer": 1, "reviewer": 0}
sorted_edits = sorted(
[e for e in edits if e.file_path == file_path],
key=lambda e: (e.start_line, -priority.get(e.agent_role, 0))
)
original = Path(file_path).read_text().splitlines()
result = list(original)
for edit in reversed(sorted_edits):
result[edit.start_line - 1 : edit.end_line] = edit.content.splitlines()
return "\n".join(result)
Autonomous agents are powerful but dangerous. Without guardrails, an agent might delete production code, introduce security vulnerabilities, or enter infinite loops. Guardrails operate at multiple levels:
Before any agent touches the codebase, validate the intent:
from pydantic import BaseModel, field_validator
from typing import Literal
class TaskRequest(BaseModel):
description: str
files_affected: list[str]
estimated_complexity: Literal["trivial", "simple", "moderate", "complex", "architectural"]
@field_validator("files_affected")
@classmethod
def validate_file_paths(cls, files: list[str]) -> list[str]:
"""Reject paths that touch sensitive areas."""
forbidden_patterns = [
"**/secrets/**",
"**/.env*",
"**/credentials/**",
"**/migrations/production/**",
"**/*.key",
"**/*.pem",
]
from pathlib import Path
for f in files:
for pattern in forbidden_patterns:
if Path(f).match(pattern):
raise ValueError(f"Agent cannot modify sensitive file: {f}")
return files
@field_validator("estimated_complexity")
@classmethod
def validate_complexity(cls, complexity: str) -> str:
"""Architectural changes require human approval."""
if complexity == "architectural":
raise ValueError(
"Architectural changes require human review before execution. "
"Break this into smaller tasks or request approval."
)
return complexity
Monitor agent behavior during execution and halt on anomalies:
import time
from collections import deque
from dataclasses import dataclass
@dataclass
class CircuitState:
failures: int = 0
last_failure_time: float = 0
is_open: bool = False
opened_at: float = 0
class CircuitBreaker:
"""Halt agent execution when failure patterns are detected."""
def __init__(
self,
failure_threshold: int = 3,
recovery_timeout: float = 60.0,
max_context_window: int = 50000,
max_tool_calls_per_task: int = 20,
):
self.failure_threshold = failure_threshold
self.recovery_timeout = recovery_timeout
self.max_context_window = max_context_window
self.max_tool_calls_per_task = max_tool_calls_per_task
self.states: dict[str, CircuitState] = {}
self.call_counts: dict[str, int] = {}
self.history: deque[dict] = deque(maxlen=100)
def record_tool_call(self, agent_id: str, tool_name: str, context_size: int) -> None:
self.call_counts[agent_id] = self.call_counts.get(agent_id, 0) + 1
self.history.append({
"agent_id": agent_id,
"tool": tool_name,
"context_size": context_size,
"timestamp": time.time(),
})
def check(self, agent_id: str) -> tuple[bool, str | None]:
"""Returns (can_proceed, reason_if_blocked)."""
calls = self.call_counts.get(agent_id, 0)
if calls >= self.max_tool_calls_per_task:
return False, f"Agent {agent_id} exceeded {self.max_tool_calls_per_task} tool calls"
if self.history:
recent = list(self.history)[-10:]
avg_context = sum(e["context_size"] for e in recent) / len(recent)
if avg_context > self.max_context_window:
return False, f"Context window growing unbounded (avg: {avg_context})"
state = self.states.get(agent_id)
if state and state.is_open:
if time.time() - state.opened_at > self.recovery_timeout:
state.is_open = False
state.failures = 0
return True, None
return False, f"Circuit breaker open for {agent_id}, retry after {self.recovery_timeout}s"
return True, None
def record_failure(self, agent_id: str) -> None:
state = self.states.setdefault(agent_id, CircuitState())
state.failures += 1
state.last_failure_time = time.time()
if state.failures >= self.failure_threshold:
state.is_open = True
state.opened_at = time.time()
def record_success(self, agent_id: str) -> None:
state = self.states.get(agent_id)
if state:
state.failures = max(0, state.failures - 1)
After an agent completes a task, independently verify the result:
import subprocess
from pathlib import Path
from dataclasses import dataclass
@dataclass
class VerificationResult:
passed: bool
checks_run: list[str]
failures: list[str]
metrics: dict[str, float]
class PostExecutionVerifier:
"""Independent verification of agent output."""
def __init__(self, project_root: Path):
self.project_root = project_root
def verify(self, changed_files: list[str], tests_to_run: list[str] | None = None) -> VerificationResult:
checks = []
failures = []
metrics = {}
type_result = self._run_type_check()
checks.append("type_check")
if not type_result.passed:
failures.append(f"Type errors: {type_result.output[:500]}")
lint_result = self._run_linter(changed_files)
checks.append("lint")
if not lint_result.passed:
failures.append(f"Lint errors: {lint_result.output[:500]}")
test_result = self._run_tests(tests_to_run)
checks.append("unit_tests")
metrics["tests_passed"] = test_result.passed
metrics["tests_failed"] = test_result.failed
if test_result.failed > 0:
failures.append(f"{test_result.failed} tests failing")
security_result = self._run_security_scan(changed_files)
checks.append("security_scan")
metrics["security_issues"] = security_result.critical_count
if security_result.critical_count > 0:
failures.append(f"{security_result.critical_count} critical security issues")
diff_size = self._get_diff_size(changed_files)
checks.append("diff_size")
metrics["lines_changed"] = diff_size
if diff_size > 500:
failures.append(f"Diff too large ({diff_size} lines). Break into smaller tasks.")
return VerificationResult(
passed=len(failures) == 0,
checks_run=checks,
failures=failures,
metrics=metrics,
)
def _run_type_check(self) -> subprocess.CompletedProcess:
return subprocess.run(
["mypy", "--strict", "."],
capture_output=True, text=True, cwd=self.project_root, timeout=120
)
def _run_linter(self, files: list[str]) -> subprocess.CompletedProcess:
return subprocess.run(
["ruff", "check", *files],
capture_output=True, text=True, cwd=self.project_root, timeout=60
)
def _run_tests(self, tests: list[str] | None) -> TestResult:
cmd = ["pytest", "-q", "--tb=short"]
if tests:
cmd.extend(tests)
result = subprocess.run(cmd, capture_output=True, text=True, cwd=self.project_root, timeout=300)
return self._parse_test_output(result.stdout)
def _run_security_scan(self, files: list[str]) -> SecurityResult:
return subprocess.run(
["bandit", "-r", *files, "-f", "json"],
capture_output=True, text=True, cwd=self.project_root, timeout=120
)
def _get_diff_size(self, files: list[str]) -> int:
result = subprocess.run(
["git", "diff", "--stat", *files],
capture_output=True, text=True, cwd=self.project_root
)
import re
match = re.search(r"(\d+) insertions?\(\+\), (\d+) deletions?\(-\)", result.stdout)
if match:
return int(match.group(1)) + int(match.group(2))
return 0
Not every change should proceed autonomously. Define escalation policies:
from enum import Enum
class RiskLevel(Enum):
LOW = "low" # Auto-approve
MEDIUM = "medium" # Auto-approve with notification
HIGH = "high" # Require human approval
CRITICAL = "critical" # Block entirely
class EscalationPolicy:
"""Determine when to escalate to human review."""
def assess(self, verification: VerificationResult, task_metadata: dict) -> RiskLevel:
if verification.metrics.get("security_issues", 0) > 0:
return RiskLevel.CRITICAL
if verification.metrics.get("tests_failed", 0) > 0:
return RiskLevel.HIGH
if verification.metrics.get("lines_changed", 0) > 200:
return RiskLevel.HIGH
if any(f.startswith("auth/") or f.startswith("payments/") for f in task_metadata.get("files", [])):
return RiskLevel.HIGH
if verification.metrics.get("lines_changed", 0) > 50:
return RiskLevel.MEDIUM
return RiskLevel.LOW
The strongest guarantee an AI agent can provide is evidence—concrete, reproducible proof that the code works. This section covers building a TDD-first pipeline where tests are generated, executed, and archived as evidence.
┌──────────────┐ ┌──────────────┐ ┌──────────────┐
│ 1. SPEC │────▶│ 2. TESTS │────▶│ 3. RED │
│ (Intent) │ │ (Generate) │ │ (Verify │
│ │ │ │ │ failure) │
└──────────────┘ └──────────────┘ └──────┬───────┘
│
┌──────────────┐ ┌──────────────┐ ┌──────▼───────┐
│ 6. EVIDENCE │◀────│ 5. GREEN │◀────│ 4. CODE │
│ (Archive) │ │ (All pass) │ │ (Implement) │
└──────────────┘ └──────────────┘ └──────────────┘
python
from dataclasses import dataclass
from pathlib import Path
import json
import hashlib
from datetime import datetime
@dataclass
class TDDEvidence:
"""Immutable record of TDD execution."""
task_id: str
timestamp: str
spec_hash: str
test_files: list[str]
implementation_files: list[str]
red_phase_output: str # Tests failing before implementation
green_phase_output: str # Tests passing after implementation
coverage: float
test_count: int
pass_count: int
verification_hash: str # SHA256 of all evidence combined
class TDDPipeline:
def __init__(self, project_root: Path, evidence_dir: Path):
self.project_root = project_root
self.evidence_dir = evidence_dir
self.evidence_dir.mkdir(parents=True, exist_ok=True)
async def execute(self, spec: str, target_files: list[str]) -> TDDEvidence:
test_files = await self._generate_tests(spec)
red_output = self._run_tests(test_files)
assert not red_output.all_passed, "Tests should fail before implementation (RED phase)"
impl_files = await self._generate_implementation(spec, test_files)
green_output = self._run_tests(test_files)
assert green_output.all_passed, "Tests should pass after implementation (GREEN phase)"
evidence = TDDEvidence(
task_id=hashlib.sha256(spec.encode()).hexdigest()[:12],
timestamp=datetime.utcnow().isoformat(),
spec_hash=hashlib.sha256(spec.encode()).hexdigest(),
test_files=[str(f.relative_to(self.project_root)) for f in test_files],
implementation_files=[str(f.relative_to(self.project_root)) for f in impl_files],
red_phase_output=red_output.raw_output,
green_phase_output=green_output.raw_output,
coverage=green_output.coverage,
test_count=green_output.total_tests,
pass_count=green_output.passed_tests,
verification_hash="", # Computed below
)
evidence_payload = json.dumps({
"spec_hash": evidence.spec_hash,
"test_files": evidence.test_files,
"red_output": evidence.red_phase_output,
"green_output": evidence.green_phase_output,
"coverage": evidence.coverage,
}, sort_keys=True)
evidence.verification_hash = hashlib.sha256(evidence_payload.encode()).hexdigest()
self._archive_evidence(evidence)
return evidence
async def _generate_tests(self, spec: str) -> list[Path]:
"""Generate test files from specification."""
...
async def _generate_implementation(self, spec: str, test_files: list[Path]) -> list[Path]:
"""Generate implementation that satisfies tests."""
...
def _run_tests(self, test_files: list[Path]) -> TestRunResult:
"""Execute tests and capture full output."""
import subprocess
test_paths = [str(f.relative_to(self.project_root)) for f in test_files]
result = subprocess.run(
["pytest", *test_paths, "--tb=long", "--cov", "--cov-report=term-missing"],
capture_output=True, text=True, cwd=self.project_root, timeout=300
)
return self._parse_result(result)
def _archive_evidence(self, evidence: TDDEvidence) -> None:
"""Persist evidence as a JSON file for audit trail."""
evidence_path = self.evidence_dir / f"{evidence.task_id}.json"
evidence_path.write_text(json.dumps(
{k: v for k, v in evidence.__dict__.items()},
indent=2, default=str
))
Anyone can independently verify that the agent's work was correct:
import json
import hashlib
from pathlib import Path
class EvidenceVerifier:
"""Independently verify TDD evidence integrity."""
def verify(self, evidence_path: Path) -> VerificationReport:
evidence = TDDEvidence.from_json(evidence_path.read_text())
payload = json.dumps({
"spec_hash": evidence.spec_hash,
"test_files": evidence.test_files,
"red_output": evidence.red_phase_output,
"green_output": evidence.green_phase_output,
"coverage": evidence.coverage,
}, sort_keys=True)
computed_hash = hashlib.sha256(payload.encode()).hexdigest()
if computed_hash != evidence.verification_hash:
return VerificationReport(
valid=False,
reason="Evidence hash mismatch - possible tampering"
)
for test_file in evidence.test_files:
if not Path(test_file).exists():
return VerificationReport(
valid=False,
reason=f"Test file missing: {test_file}"
)
test_content = "\n".join(Path(f).read_text() for f in evidence.test_files)
actual_test_count = test_content.count("def test_")
if actual_test_count != evidence.test_count:
return VerificationReport(
valid=False,
reason=f"Test count mismatch: expected {evidence.test_count}, found {actual_test_count}"
)
return VerificationReport(
valid=True,
reason="All evidence checks passed",
metrics={
"coverage": evidence.coverage,
"tests": evidence.test_count,
"passed": evidence.pass_count,
}
)
Combine multiple evidence signals into a single confidence score:
from dataclasses import dataclass
@dataclass
class ConfidenceScore:
overall: float # 0.0 to 1.0
components: dict[str, float]
deductions: list[str]
class ConfidenceScorer:
"""Calculate evidence-backed confidence in agent output."""
def score(self, evidence: TDDEvidence, verification: VerificationResult) -> ConfidenceScore:
components = {}
deductions = []
if evidence.test_count > 0:
pass_rate = evidence.pass_count / evidence.test_count
components["test_pass_rate"] = pass_rate
else:
components["test_pass_rate"] = 0.0
deductions.append("No tests executed")
coverage_score = min(evidence.coverage / 80.0, 1.0) # 80% = perfect score
components["coverage"] = coverage_score
if evidence.coverage < 60:
deductions.append(f"Low coverage: {evidence.coverage:.1f}%")
checks_passed = len(verification.checks_run) - len(verification.failures)
checks_total = len(verification.checks_run)
verification_score = checks_passed / checks_total if checks_total > 0 else 0.0
components["verification"] = verification_score
lines = verification.metrics.get("lines_changed", 0)
if lines <= 50:
diff_score = 1.0
elif lines <= 200:
diff_score = 0.7
elif lines <= 500:
diff_score = 0.4
else:
diff_score = 0.1
deductions.append(f"Large diff: {lines} lines")
components["diff_size"] = diff_score
tdd_score = 1.0 if "FAILED" in evidence.red_phase_output else 0.0
if tdd_score == 0.0:
deductions.append("RED phase did not show failures - TDD compliance questionable")
components["tdd_compliance"] = tdd_score
weights = {
"test_pass_rate": 0.35,
"coverage": 0.20,
"verification": 0.25,
"diff_size": 0.10,
"tdd_compliance": 0.10,
}
overall = sum(components[k] * weights[k] for k in weights)
return ConfidenceScore(
overall=round(overall, 3),
components=components,
deductions=deductions,
)
Here's how all the components integrate into a production-ready autonomous coding pipeline:
from pathlib import Path
from dataclasses import dataclass
from enum import Enum
import asyncio
class PipelinePhase(Enum):
SPEC_VALIDATION = "spec_validation"
TASK_DECOMPOSITION = "task_decomposition"
TDD_GENERATION = "tdd_generation"
IMPLEMENTATION = "implementation"
VERIFICATION = "verification"
REVIEW = "review"
EVIDENCE_ARCHIVE = "evidence_archive"
ESCALATION_CHECK = "escalation_check"
@dataclass
class PipelineResult:
success: bool
phase_reached: PipelinePhase
evidence: TDDEvidence | None
confidence: ConfidenceScore | None
risk_level: RiskLevel
requires_human_review: bool
summary: str
class AutonomousCodingPipeline:
def __init__(
self,
project_root: Path,
model_provider: ModelProvider,
human_approver: HumanApprover | None = None,
):
self.project_root = project_root
self.model_provider = model_provider
self.human_approver = human_approver
self.orchestrator = Orchestrator(agents={
AgentRole.Coder: CoderAgent(model_provider),
AgentRole.Tester: TesterAgent(model_provider),
AgentRole.Reviewer: ReviewerAgent(model_provider),
AgentRole.DocWriter: DocWriterAgent(model_provider),
})
self.tdd_pipeline = TDDPipeline(project_root, project_root / ".evidence")
self.circuit_breaker = CircuitBreaker()
self.verifier = PostExecutionVerifier(project_root)
self.escalation = EscalationPolicy()
self.confidence_scorer = ConfidenceScorer()
self.merge_coordinator = MergeCoordinator()
async def run(self, spec: str) -> PipelineResult:
"""Execute the full autonomous coding pipeline."""
phases_completed = []
try:
task_request = self._validate_spec(spec)
phases_completed.append(PipelinePhase.SPEC_VALIDATION)
tasks = await self._decompose(spec, task_request)
phases_completed.append(PipelinePhase.TASK_DECOMPOSITION)
all_evidence = []
for task in tasks:
can_proceed, reason = self.circuit_breaker.check(task.id)
if not can_proceed:
return PipelineResult(
success=False,
phase_reached=PipelinePhase.IMPLEMENTATION,
evidence=None,
confidence=None,
risk_level=RiskLevel.CRITICAL,
requires_human_review=True,
summary=f"Circuit breaker triggered: {reason}",
)
evidence = await self.tdd_pipeline.execute(task.description, task.files)
all_evidence.append(evidence)
phases_completed.append(PipelinePhase.TDD_GENERATION)
changed_files = [f for e in all_evidence for f in e.implementation_files]
verification = self.verifier.verify(changed_files)
phases_completed.append(PipelinePhase.VERIFICATION)
review = await self._run_review(all_evidence, verification)
phases_completed.append(PipelinePhase.REVIEW)
combined_evidence = self._combine_evidence(all_evidence)
phases_completed.append(PipelinePhase.EVIDENCE_ARCHIVE)
risk_level = self.escalation.assess(verification, {"files": changed_files})
confidence = self.confidence_scorer.score(combined_evidence, verification)
phases_completed.append(PipelinePhase.ESCALATION_CHECK)
requires_human = risk_level in (RiskLevel.HIGH, RiskLevel.CRITICAL)
if requires_human and self.human_approver:
approved = await self.human_approver.request_approval(
summary=confidence,
evidence=combined_evidence,
risk_level=risk_level,
)
if not approved: