24.2 Autonomous Factory Architecture

1. The Five-Stage Pipeline

In 2024, one widely cited anecdote describes a startup that left its autonomous code pipeline running over a long weekend and returned on Monday to find 31 pull requests merged, a dependency upgraded, and two stale feature flags cleaned up, all without a single human keystroke. That outcome was not magic; it was the product of a five-stage loop whose architecture this section dissects in full. The autonomous factory pipeline processes repository events through five sequential stages. Each stage has a well-defined input, output, and failure mode. The pipeline is not a one-way conveyor belt; the verify stage can route work back to earlier stages for retry, and the triage stage evolves its routing policy over time based on outcomes. Figure 24.2 illustrates the full pipeline topology, including the feedback path from verify back to triage. Figure 24.2.1 illustrates the five-stage autonomous factory pipeline with feedback loop.

A pipeline here is a directed graph of processing stages. It transforms a raw event (such as a new GitHub issue) into a verified, mergeable code change; each stage acts as both a filter and an enrichment step. Without this structure, autonomous agents operate as isolated, stateless tools with no mechanism for self-correction or incremental trust expansion. Each stage reads a typed input record, performs its function (classification, decomposition, code generation, or verification), and attaches its output to that record. The stage then forwards the record onward; failure at any point triggers a retry loop or an escalation to a human. Use a staged pipeline when you need auditable, bounded autonomy with feedback. Use a one-shot agent when tasks are low-risk and self-contained, or a fully manual workflow when every change requires human judgment.

We model the pipeline as a state machine (where a state machine is a mathematical model that transitions between a finite set of states in response to events) \(M = (Q, \Sigma, \delta, q_0, F)\) where $Q = \{\text{sense}, \text{triage}, \text{plan}, \text{execute}, \text{verify}, \text{merge}, \text{escalate}, \text{retry}\}$ is the state set, $\Sigma$ is the set of events (new issue, test result, review decision), \(\delta\) is the transition function, \(q_0 = \text{sense}\) is the start state, and \(F = \{\text{merge}, \text{escalate}\}\) are terminal states. In short: an autonomous factory is a loop, not a line: sense, triage, plan, execute, verify, and let the outcomes teach the triage stage what to trust next time.

from dataclasses import dataclass, field
from enum import Enum, auto
from typing import Optional, Any
import datetime

class PipelineStage(Enum):
    SENSE = auto()      # detect new issues, events, signals
    TRIAGE = auto()     # classify risk, route to agent or human
    PLAN = auto()       # decompose task into sub-steps
    EXECUTE = auto()    # dispatch SWE agents
    VERIFY = auto()     # run tests, static analysis, review
    MERGE = auto()      # auto-merge (terminal, success)
    ESCALATE = auto()   # route to human (terminal, needs help)
    RETRY = auto()      # loop back to plan or execute

@dataclass
class PipelineEvent:
    """An event flowing through the autonomous pipeline."""
    issue_id: str
    stage: PipelineStage
    timestamp: datetime.datetime = field(
        default_factory=datetime.datetime.now
    )
    payload: dict = field(default_factory=dict)
    attempt: int = 1
    max_attempts: int = 3

@dataclass
class PipelineState:
    """Full state of a task as it moves through the pipeline."""
    event: PipelineEvent
    triage_result: Optional["TriageDecision"] = None
    plan: Optional["TaskPlan"] = None
    execution_result: Optional["SWEAgentResult"] = None
    verification: Optional["VerificationReport"] = None
    reward_signals: dict = field(default_factory=dict)

    @property
    def should_retry(self) -> bool:
        """Retry if verification failed and attempts remain."""
        if self.verification is None:
            return False
        return (
            not self.verification.passed
            and self.event.attempt < self.event.max_attempts
        )

    @property
    def should_escalate(self) -> bool:
        """Escalate to human if retries exhausted or risk is high."""
        if self.triage_result and self.triage_result.route == "human":
            return True
        if self.verification and not self.verification.passed:
            return self.event.attempt >= self.event.max_attempts
        return False

1.1 Stage 1: Sense

The sense stage monitors the repository for actionable events. The most common event sources are:

• New issues. A user or automated system opens a GitHub issue describing a bug, feature request, or improvement.
• Failing Continuous Integration (CI). A scheduled test run or a dependency update breaks the build. The factory detects the failure and creates an internal task to investigate.
• Dependency updates. A new version of a dependency is released. The factory can create update pull requests (PRs), run tests, and merge if everything passes.
• Performance regressions. A monitoring system detects that response times have increased. The factory investigates recent changes to identify the cause.
• Code quality alerts. A linter or static analysis tool flags new issues (unused imports, type errors, security vulnerabilities).

import aiohttp
import asyncio
from dataclasses import dataclass
from typing import AsyncIterator

@dataclass
class GitHubEventSource:
    """
    Polls GitHub for new issues and events.

    Uses the GitHub REST API with conditional requests
    (If-None-Match) to avoid redundant data transfer.
    """
    owner: str
    repo: str
    token: str
    poll_interval: int = 60  # seconds

    async def poll(self) -> AsyncIterator[PipelineEvent]:
        """Yield new events as they appear."""
        seen_issues: set[int] = set()
        etag: str | None = None
        url = (
            f"https://api.github.com/repos/{self.owner}/{self.repo}"
            f"/issues?state=open&sort=created&direction=desc"
        )
        headers = {
            "Authorization": f"Bearer {self.token}",
            "Accept": "application/vnd.github+json",
        }

        async with aiohttp.ClientSession() as session:
            while True:
                req_headers = {**headers}
                if etag:
                    req_headers["If-None-Match"] = etag

                async with session.get(url, headers=req_headers) as resp:
                    if resp.status == 304:
                        # No new data since last poll
                        await asyncio.sleep(self.poll_interval)
                        continue

                    etag = resp.headers.get("ETag")
                    issues = await resp.json()

                    for issue in issues:
                        if issue["number"] not in seen_issues:
                            seen_issues.add(issue["number"])
                            yield PipelineEvent(
                                issue_id=str(issue["number"]),
                                stage=PipelineStage.SENSE,
                                payload={
                                    "title": issue["title"],
                                    "body": issue.get("body", ""),
                                    "labels": [
                                        l["name"]
                                        for l in issue.get("labels", [])
                                    ],
                                    "author": issue["user"]["login"],
                                },
                            )

                await asyncio.sleep(self.poll_interval)

1.2 Stage 2: Triage

The triage stage is the most critical decision point in the pipeline. It classifies each incoming event by risk level and routes it to either autonomous execution or human review. A wrong decision here has asymmetric consequences: routing a safe task to a human wastes human time (low cost); routing a dangerous task to autonomous execution can cause production incidents (high cost). The triage policy should therefore be conservative by default, erring toward human review and expanding autonomous scope only as evidence accumulates.

The triage decision combines two assessments: the delegability score from Section 24.1 (can the agent handle this task?) and a blast radius estimate, where blast radius is the worst-case scope of damage if the agent's change is incorrect (how many services, users, or data paths are affected). We implement triage as a classifier that takes issue metadata and repository context as input and produces a routing decision.

from dataclasses import dataclass
from enum import Enum

class TriageRoute(str, Enum):
    AUTONOMOUS = "autonomous"   # agent executes, auto-merge if verified
    SUPERVISED = "supervised"   # agent executes, human reviews before merge
    HUMAN = "human"             # route directly to human developer

@dataclass
class TriageDecision:
    """Output of the triage classifier."""
    route: TriageRoute
    confidence: float             # classifier confidence (0-1)
    risk_category: str            # "low", "medium", "high", "critical"
    estimated_complexity: float   # from TaskAssessment
    reasoning: str                # explanation for audit trail

class TriageClassifier:
    """
    Classifies issues for routing in the autonomous pipeline.

    Uses a combination of rule-based heuristics (for safety-
    critical patterns) and large language model (LLM)-based
    classification (for nuanced judgment). The rule layer
    runs first and can short-circuit to human review; the
    LLM layer handles everything else.
    """

    # Patterns that always route to human review
    CRITICAL_PATTERNS = [
        "security", "vulnerability", "CVE", "authentication",
        "authorization", "payment", "billing", "PII",
        "password", "encryption", "database migration",
        "breaking change", "API deprecation",
    ]

    # Labels that indicate low-risk tasks
    LOW_RISK_LABELS = [
        "documentation", "typo", "chore", "dependencies",
        "good first issue", "help wanted", "enhancement",
    ]

    def classify(self, event: PipelineEvent) -> TriageDecision:
        """Classify an issue and decide routing."""
        title = event.payload.get("title", "").lower()
        body = event.payload.get("body", "").lower()
        labels = event.payload.get("labels", [])
        combined_text = f"{title} {body}"

        # Rule layer: critical patterns always go to human
        for pattern in self.CRITICAL_PATTERNS:
            if pattern in combined_text:
                return TriageDecision(
                    route=TriageRoute.HUMAN,
                    confidence=0.95,
                    risk_category="critical",
                    estimated_complexity=0.8,
                    reasoning=(
                        f"Critical pattern '{pattern}' detected. "
                        f"Routing to human review."
                    ),
                )

        # Rule layer: known low-risk labels
        low_risk_labels = [
            l for l in labels if l.lower() in self.LOW_RISK_LABELS
        ]
        if low_risk_labels:
            return TriageDecision(
                route=TriageRoute.AUTONOMOUS,
                confidence=0.85,
                risk_category="low",
                estimated_complexity=0.2,
                reasoning=(
                    f"Low-risk labels detected: {low_risk_labels}. "
                    f"Routing to autonomous execution."
                ),
            )

        # Default: supervised (agent works, human reviews)
        return TriageDecision(
            route=TriageRoute.SUPERVISED,
            confidence=0.6,
            risk_category="medium",
            estimated_complexity=0.5,
            reasoning=(
                "No strong signal for autonomous or human routing. "
                "Defaulting to supervised mode."
            ),
        )

1.3 Stage 3: Plan

Tasks that pass triage enter the planning stage, where a planning agent decomposes the issue into concrete sub-tasks. The planner reads the issue description, examines the relevant code, and produces a task plan: an ordered list of steps the execution agent should follow. Good plans are specific enough to guide the agent but flexible enough to allow adaptation when the agent discovers something unexpected during execution.

from dataclasses import dataclass, field

@dataclass
class TaskStep:
    """A single step in a task plan."""
    description: str
    target_files: list[str]      # files this step will likely modify
    verification: str            # how to check this step succeeded
    estimated_complexity: float  # 0-1 scale

@dataclass
class TaskPlan:
    """A decomposed plan for resolving an issue."""
    issue_id: str
    summary: str                    # one-sentence summary of the approach
    steps: list[TaskStep]
    estimated_files_changed: int
    estimated_lines_changed: int
    test_strategy: str              # how to verify the full solution
    rollback_strategy: str          # how to undo if something goes wrong
    compute_budget: float = 2.0     # max USD to spend on this task

    @property
    def total_complexity(self) -> float:
        if not self.steps:
            return 0.0
        return sum(s.estimated_complexity for s in self.steps) / len(self.steps)

def create_plan_prompt(issue_title: str, issue_body: str, repo_summary: str) -> str:
    """
    Build the prompt for the planning agent.

    The planner receives the issue description and a summary
    of the repository structure (file tree, key modules, test
    locations) and produces a structured TaskPlan.
    """
    return f"""You are a software planning agent. Analyze this issue and
produce a step-by-step plan for resolving it.

ISSUE: {issue_title}
{issue_body}

REPOSITORY STRUCTURE:
{repo_summary}

OUTPUT FORMAT (JSON):
{{
  "summary": "One-sentence approach description",
  "steps": [
    {{
      "description": "What to do",
      "target_files": ["path/to/file.py"],
      "verification": "How to check it worked",
      "estimated_complexity": 0.3
    }}
  ],
  "estimated_files_changed": 3,
  "estimated_lines_changed": 50,
  "test_strategy": "Run pytest tests/test_module.py",
  "rollback_strategy": "git checkout -- src/module.py"
}}

PLANNING RULES:
1. Read the relevant code BEFORE planning.
2. Each step should be independently verifiable.
3. Order steps so later steps build on earlier ones.
4. Include a test step that verifies the complete fix.
5. Keep the plan minimal: prefer fewer, well-targeted changes."""

1.4 Stage 4: Execute

The execution stage dispatches SWE agents to carry out the plan, using the ClaudeCodeSWEAgent from Section 24.1 with the plan steps as context. The key architectural choice: one agent follows the entire plan, or multiple specialized agents (one per step) work in parallel under orchestrator coordination. Single-agent execution suits most tasks because it preserves full context across steps and lets the agent adapt when a step reveals unexpected complexity. Multi-agent execution pays off only when the plan contains independent sub-tasks (fixing a bug in one module while updating documentation in another) that benefit from parallelism.

1.5 Stage 5: Verify

Verification is the stage that makes autonomous operation safe. It runs after execution and determines whether the agent's changes are correct, complete, and safe to merge. Verification combines multiple signals into a single pass/fail decision, with each signal contributing to a confidence score. Tests alone receive only 40% of the confidence weight; the remaining 60% comes from linting, type checks, security scans, and diff size, because passing tests can miss the subtle bugs that surface in production.

from dataclasses import dataclass, field
import subprocess
from pathlib import Path

@dataclass
class VerificationReport:
    """Comprehensive verification of an agent's changes."""
    passed: bool

  
  

    test_results: dict            # {"passed": int, "failed": int, "error": int}
    lint_score: float             # 0-1, from static analysis
    type_check_passed: bool
    security_scan_clean: bool
    diff_size: dict               # {"files": int, "additions": int, "deletions": int}
    review_comments: list[str]    # from automated code review
    confidence: float             # overall confidence (0-1)
    blocking_issues: list[str]    # issues that prevent merge

class VerificationPipeline:
    """
    Multi-signal verification for autonomous changes.

    Runs tests, linting, type checking, and security scanning
    in sequence. Any blocking failure stops the pipeline.
    """

    def __init__(self, repo_path: Path):
        self.repo_path = repo_path

    def verify(self) -> VerificationReport:
        """Run all verification checks and produce a report."""
        blocking = []

        # 1. Run the test suite
        test_results = self._run_tests()
        if test_results["failed"] > 0 or test_results["error"] > 0:
            blocking.append(
                f"{test_results['failed']} tests failed, "
                f"{test_results['error']} errors"
            )

        # 2. Run linting (ruff)
        lint_score = self._run_linter()

        # 3. Run type checking (mypy/pyright)
        type_ok = self._run_type_check()
        if not type_ok:
            blocking.append("Type checking failed")

        # 4. Run security scan (bandit)
        security_ok = self._run_security_scan()
        if not security_ok:
            blocking.append("Security scan found issues")

        # 5. Measure diff size
        diff = self._measure_diff()

        # 6. Compute confidence from signals
        confidence = self._compute_confidence(
            test_results, lint_score, type_ok, security_ok, diff
        )

        return VerificationReport(
            passed=len(blocking) == 0,
            test_results=test_results,
            lint_score=lint_score,
            type_check_passed=type_ok,
            security_scan_clean=security_ok,
            diff_size=diff,
            review_comments=[],
            confidence=confidence,
            blocking_issues=blocking,
        )

    def _run_tests(self) -> dict:
        """Run pytest and parse results."""
        result = subprocess.run(
            ["python", "-m", "pytest", "--tb=short", "-q"],
            capture_output=True, text=True,
            cwd=str(self.repo_path),
            timeout=300,
        )
        # Parse pytest output for pass/fail counts
        output = result.stdout
        passed = failed = errors = 0
        for line in output.split("\n"):
            if "passed" in line:
                import re
                m = re.search(r"(\d+) passed", line)
                if m:
                    passed = int(m.group(1))
            if "failed" in line:
                import re
                m = re.search(r"(\d+) failed", line)
                if m:
                    failed = int(m.group(1))
            if "error" in line:
                import re
                m = re.search(r"(\d+) error", line)
                if m:
                    errors = int(m.group(1))
        return {"passed": passed, "failed": failed, "error": errors}

    def _run_linter(self) -> float:
        """Run ruff and return a 0-1 quality score."""
        result = subprocess.run(
            ["ruff", "check", "--statistics", "."],
            capture_output=True, text=True,
            cwd=str(self.repo_path),
        )
        # Score: 1.0 if no issues, decreasing with issue count
        issue_count = len(result.stdout.strip().split("\n"))
        if not result.stdout.strip():
            issue_count = 0
        return max(0.0, 1.0 - issue_count * 0.05)

    def _run_type_check(self) -> bool:
        """Run pyright and return pass/fail."""
        result = subprocess.run(
            ["pyright", "--outputjson"],
            capture_output=True, text=True,
            cwd=str(self.repo_path),
        )
        return result.returncode == 0

    def _run_security_scan(self) -> bool:
        """Run bandit and return pass/fail."""
        result = subprocess.run(
            ["bandit", "-r", "src/", "-f", "json", "-q"],
            capture_output=True, text=True,
            cwd=str(self.repo_path),
        )
        return result.returncode == 0

    def _measure_diff(self) -> dict:
        """Measure the size of the agent's changes."""
        result = subprocess.run(
            ["git", "diff", "--stat", "--numstat", "HEAD~1"],
            capture_output=True, text=True,
            cwd=str(self.repo_path),
        )
        files, additions, deletions = 0, 0, 0
        for line in result.stdout.strip().split("\n"):
            parts = line.split("\t")
            if len(parts) == 3:
                try:
                    additions += int(parts[0])
                    deletions += int(parts[1])
                    files += 1
                except ValueError:
                    pass
        return {
            "files": files,
            "additions": additions,
            "deletions": deletions,
        }

    def _compute_confidence(
        self, tests, lint, type_ok, security_ok, diff
    ) -> float:
        """
        Weighted confidence from verification signals.

        Tests are the strongest signal (weight 0.4).
        Smaller diffs get a bonus (less risk).
        """
        score = 0.0
        total_tests = tests["passed"] + tests["failed"] + tests["error"]
        if total_tests > 0:
            score += 0.4 * (tests["passed"] / total_tests)
        score += 0.2 * lint
        score += 0.15 * (1.0 if type_ok else 0.0)
        score += 0.15 * (1.0 if security_ok else 0.0)
        # Diff size penalty: larger changes are riskier
        total_lines = diff["additions"] + diff["deletions"]
        size_bonus = max(0.0, 0.1 * (1.0 - total_lines / 500))
        score += size_bonus
        return min(1.0, score)

2. Reward Signals for Self-Improvement

With all five stages in place, the pipeline can process issues end to end, but it still behaves identically on its hundredth run as on its first unless it can learn from past outcomes.

The factory becomes self-improving when the verify stage feeds signals back to the triage stage. Over time, the triage classifier learns which types of issues the agent resolves successfully and which types it consistently fails on. This feedback loop is the key difference between a static L3 system and a progressing L4/L5 system (where L3, L4, and L5 refer to levels on the autonomy ladder from Section 24.1, ranging from supervised agent execution to fully autonomous operation).

We define four categories of reward signals, ordered by availability (how quickly you get the signal) and reliability (how accurately it predicts real-world success):

Signal Categories by Latency

Immediate signals (seconds). Test pass rate, lint score, type check result, diff size. Available immediately after execution. Reliable for catching obvious failures but blind to subtle correctness issues.

CI signals (minutes). Full CI pipeline results, integration test outcomes, build artifacts. Available after CI completes. More comprehensive than immediate signals because CI typically runs a broader test matrix (multiple Python versions, operating systems, dependency configurations).

Review signals (hours to days). Human reviewer comments, requested changes, approval/rejection. Available only for SUPERVISED tasks where a human reviews the PR. The richest signal because it captures correctness judgments that automated checks miss, but the slowest and most expensive to collect.

Production signals (days to weeks). User-reported bugs against the change, rollback events, performance metrics after deployment. The ground truth signal: did the change actually work in production? The most reliable but also the most delayed, making it suitable only for long-term calibration, not real-time routing decisions.

from dataclasses import dataclass, field
import datetime

@dataclass
class RewardSignal:
    """A single reward signal from the verification/monitoring pipeline."""
    source: str          # "tests", "ci", "review", "production"
    value: float         # normalized to [0, 1]
    confidence: float    # how reliable this signal is
    timestamp: datetime.datetime = field(
        default_factory=datetime.datetime.now
    )
    metadata: dict = field(default_factory=dict)

@dataclass
class OutcomeRecord:
    """
    Complete outcome record for a resolved issue.

    Stored in the feedback database and used to train
    the triage classifier over time.
    """
    issue_id: str
    triage_decision: TriageDecision
    plan_complexity: float
    execution_cost_usd: float
    execution_time_seconds: float
    signals: list[RewardSignal] = field(default_factory=list)

    @property
    def composite_reward(self) -> float:
        """
        Weighted average of all reward signals.

        Later signals (review, production) are weighted higher
        because they are more reliable, but earlier signals
        dominate until later ones arrive.
        """
        if not self.signals:
            return 0.0
        weights = {
            "tests": 0.2,
            "ci": 0.25,
            "review": 0.3,
            "production": 0.25,
        }
        weighted_sum = sum(
            s.value * weights.get(s.source, 0.1) * s.confidence
            for s in self.signals
        )
        weight_total = sum(
            weights.get(s.source, 0.1) * s.confidence
            for s in self.signals
        )
        return weighted_sum / weight_total if weight_total > 0 else 0.0

    @property
    def was_successful(self) -> bool:
        """Binary success: composite reward above threshold."""
        return self.composite_reward >= 0.7

3. The Feedback Loop as Online Learning

The triage classifier improves over time by learning from outcome records. This is an online learning problem, where online learning means the model updates its parameters incrementally after each new observation rather than retraining on the full dataset: the classifier makes a routing decision (autonomous vs. supervised vs. human), observes the outcome, and updates its parameters. Two challenges complicate this loop. First, feedback arrives delayed: production signals take days to materialize. Second, feedback is partial: the system routes some tasks to humans and therefore never generates autonomous outcome data for them, so it never learns whether the agent could have handled those tasks.

We formalize this as a contextual bandit problem, where a contextual bandit is a sequential decision framework in which the learner observes context features, selects one of several actions, and receives a reward only for the chosen action (never learning the counterfactual reward of unchosen actions). At each time step \(t\), the triage classifier observes a context \(x_t\) (issue features: title, body, labels, repository statistics), selects an action $a_t \in \{\text{autonomous}, \text{supervised}, \text{human}\}$, and receives a reward $r_t$ (the composite reward from the outcome record). The goal is to learn a policy \(\pi(a | x)\) that maximizes cumulative reward while maintaining a safety constraint:

The safety constraint \(\epsilon\) bounds the probability of an autonomous merge causing a production incident. In practice, \(\epsilon\) is set conservatively (e.g., 0.01) and relaxed as the system accumulates a track record.

from dataclasses import dataclass, field
import math

@dataclass
class TriageLearner:
    """
    Online learner for triage routing decisions.

    Uses Thompson sampling to balance exploration (trying
    autonomous on new task types) with exploitation (using
    the best known route). Thompson sampling draws a random
    sample from each action's posterior distribution (here a
    Beta distribution) and picks the action whose sample is
    highest, naturally balancing exploration and exploitation.
    The code below implements the conservative gating check
    (should_allow_autonomous); a full deployment would pair
    this gate with Thompson sampling for active exploration.
    """
    # Per-category success/failure counts (Beta distribution params)
    category_stats: dict[str, dict[str, list[float]]] = field(
        default_factory=lambda: {}
    )
    # Minimum observations before allowing autonomous routing
    min_observations: int = 10
    # Maximum autonomous failure rate
    max_failure_rate: float = 0.05

    def update(self, category: str, route: str, success: bool) -> None:
        """Update beliefs after observing an outcome."""
        if category not in self.category_stats:
            self.category_stats[category] = {}
        if route not in self.category_stats[category]:
            # Beta(1, 1) = uniform prior
            self.category_stats[category][route] = [1.0, 1.0]

        alpha_beta = self.category_stats[category][route]
        if success:
            alpha_beta[0] += 1  # increment successes
        else:
            alpha_beta[1] += 1  # increment failures

    def should_allow_autonomous(self, category: str) -> bool:
        """
        Decide if a task category has enough evidence
        for autonomous routing.

        Requires: (1) enough observations, (2) failure rate
        below threshold, (3) lower bound of Beta confidence
        interval above the safety threshold.
        """
        stats = self.category_stats.get(category, {})
        auto_stats = stats.get("autonomous", [1.0, 1.0])
        alpha, beta = auto_stats

        total_observations = alpha + beta - 2  # subtract prior
        if total_observations < self.min_observations:
            return False

        # Point estimate of success rate
        success_rate = alpha / (alpha + beta)
        failure_rate = 1.0 - success_rate

        if failure_rate > self.max_failure_rate:
            return False

        # 95% lower bound of Beta distribution
        # (A Beta distribution models the probability of
        # success given observed successes and failures;
        # its two parameters alpha and beta count successes
        # and failures plus prior pseudo-counts.)
        # Approximation using normal approximation to Beta
        std = math.sqrt(
            alpha * beta / ((alpha + beta) ** 2 * (alpha + beta + 1))
        )
        lower_bound = success_rate - 1.96 * std

        return lower_bound >= (1.0 - self.max_failure_rate)

4. Guardrails and Safety Architecture

A self-improving triage learner increases throughput, but higher throughput also increases the damage an unchecked failure can cause, which is why the factory pairs its learning loop with layered safety mechanisms.

Autonomous operation requires guardrails at every stage. The guardrail architecture implements a defense in depth strategy (layered security in which multiple independent safeguards each catch failures the others miss): each stage has its own safety checks, and a failure at any stage halts progression and escalates to a human.

Repository isolation. Agents never operate on the production repository directly. Every task gets a fresh clone (or a git worktree, where a worktree is a secondary working directory linked to the same repository, allowing parallel checkouts on separate branches without duplicating the full history) on a feature branch. The agent cannot access the main branch, the production database, or any external service beyond the repository and its test infrastructure.

File-level permissions. The autonomy configuration from Section 24.1 restricts which files the agent may edit. Configuration files, secrets, CI pipelines, database migrations, and infrastructure definitions are off-limits. These restrictions are enforced at the tool level (the edit tool checks the path against the allowlist before writing) rather than relying on the agent to self-police.

Diff size limits. An upper bound on the number of files changed and lines modified prevents runaway agents that rewrite large portions of the codebase. If the agent exceeds the limit, the execution stage halts and the task is escalated.

Cost caps. Each task has a compute budget (in USD) that limits how many LLM tokens the agent can consume. This prevents infinite retry loops where the agent keeps trying different approaches, burning through budget without making progress. The cost cap connects to the AgentOps infrastructure from Chapter 22.

Incident response. When an autonomously merged change causes a production issue (detected through monitoring), the system automatically reverts the PR, creates a post-mortem issue, and downgrades the task category from "autonomous" to "supervised" in the triage learner. This negative feedback ensures the system contracts its autonomous scope in response to failures.

5. Integration with the Discovery Workbench

The autonomous factory extends the Discovery Workbench (introduced in Chapter 6) with a continuous development module. This module connects the factory's pipeline stages to the Workbench's existing components:

• The sense stage integrates with the Workbench's data ingestion layer, treating repository events as a data source alongside experimental results and literature updates.
• The triage stage uses the Workbench's knowledge graph (from Chapter 38) to understand dependencies between modules and assess blast radius.
• The verify stage publishes results to the Workbench's experiment registry (from Chapter 47), creating provenance records that trace every autonomous change back to its trigger, plan, and verification results.
• The feedback loop contributes to the Workbench's model of research team productivity, enabling the autonomous discovery systems in Chapter 53 to treat code changes as part of the scientific workflow.