Building Autonomous AI Coding Agent Pipelines: TDD Evidence, Guardrails, and Multi-Agent Coordination

A developer detailed an architecture for autonomous AI coding pipelines that replaces trust-based code generation with evidence-based verification, using Test-Driven Development as a control-flow mechanism. The design wraps an LLM in a Red-Green-Refactor lifecycle where a FeatureAgent generates a failing test, implements code to pass it, and feeds test-runner output back into the model for automatic correction, supplemented by LLM-based review guardrails and multi-agent coordination.

Originally published on tamiz.pro https://tamiz.pro/insights/ai-coding-agent-pipelines-tdd-guardrails . 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. python 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. """ 1. RED: Generate a failing test test code = self. generate test requirement self.fs.write file "tests/test feature.py", test code Verify it actually fails sanity check 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." 2. GREEN: Implement the code to pass the test implementation = self. generate implementation requirement, test code self.fs.write file "src/feature.py", implementation 3. VERIFY result = self.test runner.run "pytest tests/test feature.py" if not result.all passed : Feed error back into LLM for correction return self. auto fix result.output 4. REFACTOR Optional 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. js 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 : js 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 https://tamiz.pro/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: python orchestrator.py 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" Gather context from dependencies 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}" Propagate failure to dependent tasks 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: python agent protocol.py 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: python tester agent.py 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:" Extract the code changes from the coder's output code context = context.get "coder output", "" Generate tests if none exist test plan = await self. generate test plan task description, code context Write test files test files = self. write tests test plan Execute the test suite 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: python merge coordinator.py 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 Apply edits from bottom to top to preserve line numbers 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: python guardrails/pre execution.py 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: python guardrails/circuit breaker.py 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 .""" Check tool call limit 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" Check context window 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} " Check circuit breaker state 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: python guardrails/post execution.py 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 = {} 1. Type checking type result = self. run type check checks.append "type check" if not type result.passed: failures.append f"Type errors: {type result.output :500 }" 2. Linting lint result = self. run linter changed files checks.append "lint" if not lint result.passed: failures.append f"Lint errors: {lint result.output :500 }" 3. Unit tests 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" 4. Security scan 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" 5. Diff size check prevent massive refactors 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 Parse "X insertions + , Y deletions - " 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 : python guardrails/escalation.py 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: Critical: security issues or test failures if verification.metrics.get "security issues", 0 0: return RiskLevel.CRITICAL if verification.metrics.get "tests failed", 0 0: return RiskLevel.HIGH High: large diffs or changes to critical files 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 Medium: moderate changes 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 tdd pipeline.py 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: Phase 1: Generate tests from spec test files = await self. generate tests spec Phase 2: Verify tests fail RED red output = self. run tests test files assert not red output.all passed, "Tests should fail before implementation RED phase " Phase 3: Generate implementation impl files = await self. generate implementation spec, test files Phase 4: Verify tests pass GREEN green output = self. run tests test files assert green output.all passed, "Tests should pass after implementation GREEN phase " Phase 5: Archive evidence 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 Compute verification hash 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 Persist evidence self. archive evidence evidence return evidence async def generate tests self, spec: str - list Path : """Generate test files from specification.""" Implementation would call LLM with test-generation prompt Returns list of created test file paths ... async def generate implementation self, spec: str, test files: list Path - list Path : """Generate implementation that satisfies tests.""" Implementation would call LLM with the spec and test context Returns list of created/modified implementation file paths ... 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: python evidence verifier.py 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 1. Verify hash integrity 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" 2. Verify test files exist for test file in evidence.test files: if not Path test file .exists : return VerificationReport valid=False, reason=f"Test file missing: {test file}" 3. Verify test count matches 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: python confidence scorer.py 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 = Test pass rate weight: 0.35 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 weight: 0.20 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}%" Verification checks weight: 0.25 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 Diff size appropriateness weight: 0.10 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 compliance weight: 0.10 Did the RED phase actually fail? 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 Weighted sum 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: python pipeline.py 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 Initialize components 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: Phase 1: Validate the specification task request = self. validate spec spec phases completed.append PipelinePhase.SPEC VALIDATION Phase 2: Decompose into subtasks tasks = await self. decompose spec, task request phases completed.append PipelinePhase.TASK DECOMPOSITION Phase 3-4: Execute TDD loop for each task 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 Phase 5: Post-execution verification 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 Phase 6: Code review review = await self. run review all evidence, verification phases completed.append PipelinePhase.REVIEW Phase 7: Archive evidence combined evidence = self. combine evidence all evidence phases completed.append PipelinePhase.EVIDENCE ARCHIVE Phase 8: Escalation check 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: