AI Code Generation on Autopilot: Building a Production‑Ready Loop Engineering Loop

Problem: single‑pass AI code generation can cause production outages

A retail payment service added an idempotent Redis lock via an AI‑generated patch. The patch merged quickly and CI passed, but a network glitch caused massive callback retries. Root causes:

Lock release logic ran only on the success path; exception branches never cleared the lock.

Lock key was not bound to the request token, causing accidental deletion of other locks.

Unit tests covered only the happy path and missed concurrency, timeout retries, and out‑of‑order callbacks.

Engineers assumed the AI had already written correct idempotent logic, and reviewers treated the change as trivial.

The incident shows that the obstacle is not model capability but the "generate‑once, human‑fallback" pattern applied to an uncertain system.

Loop Engineering

Instead of a linear workflow Input → Model → Human review → Apply , Loop Engineering inserts the AI agent into a closed feedback loop:

Observe → Plan → Act → Verify → Reflect → Observe

Each step has explicit inputs, outputs, state changes, and termination conditions:

Observe : collect requirement, relevant code, dependency graph, historical failures, test baselines, runtime limits.

Plan : decide modification scope, select tools, generate subtasks, assess risk and budget.

Act : generate a patch in an isolated sandbox, apply it, run build.

Verify : static analysis, unit/integration/contract tests, security scans, performance thresholds.

Reflect : feed structured failure logs, diff results, and policy constraints back to the agent for the next iteration.

Failure modes in micro‑service environments

Correct code does not guarantee system correctness – local SDK calls may break transaction boundaries or cause downstream traffic spikes.

Passing unit tests does not ensure that external services (Kafka, Redis, MySQL, etc.) are available.

Traditional CI reports a failure but does not turn the reason into the next input, forcing engineers to manually adjust prompts.

Unrestricted agent access to repos, CI credentials, production config, and the internet amplifies risk.

Without persisted state, a loop is merely repeated tool calls. A true loop must answer:

What is the current task state and is it persisted?

What changed between the last failure and this attempt?

When do cost, risk, or retry limits trigger human takeover?

Four‑plane architecture

Control plane (decision‑making, no direct execution)

Task entry from GitHub Issue, Jira, alert platform, or code‑review comment.

Orchestration of stage, dependencies, timeout policy, and human‑in‑the‑loop rules.

Risk grading (e.g., "add test", "fix low‑risk bug", "SQL change").

Budget control: token budget, max attempts, wall‑clock limit, max concurrency.

Execution plane (isolated sandbox)

Code Agent – reads requirement, context, failure records; generates patch or refactor suggestion.

Build Agent – restores dependencies, compiles, formats, runs static checks.

Test Agent – runs unit, integration, contract, migration checks.

Review Agent – performs SAST, license scan, secret scan, and change‑summary generation.

The execution plane runs in short‑lived containers or K8s Jobs with the weakest permissions.

State plane (traceability and replay)

Task state machine:

NEW → PLANNED → CODING → TESTING → REVIEWING → WAITING_HUMAN → DONE/FAILED

Attempt numbering, input snapshot (requirement, files, tool list, budget), output snapshot (patch hash, test results, security alerts, summary comment).

Upgrade reason: why the loop stopped and who took over.

Governance plane (constraints, audit, rollback)

Permission isolation: read‑only repo, read‑only dependency images, temporary credentials, outbound‑network limits.

Policy validation: forbid high‑risk commands, disallow privileged directories, block sensitive file changes.

Cost governance: track token usage and failure cost per task, team, repository.

Result audit: record trigger, model version, prompt template, patch summary, final decision.

Human fallback: automatically suspend on policy hits or multiple failures and notify owners.

Typical business flow (ordered list)

Engineer tags an Issue with ai-ready to allow the agent.

Control plane creates a task and decides whether automatic modification is permitted based on file type, service domain, and risk level.

Execution plane clones the repo in a temporary sandbox, retrieves relevant context, generates a patch, and pushes it to a work branch.

Test Agent runs build, unit, integration, dependency, and migration checks, producing a structured failure report.

State plane records the round result and feeds the failure summary back to the Code Agent.

If budget and risk thresholds allow, the loop proceeds to the next round; otherwise it escalates to human review.

Governance plane attaches an audit summary to the final PR, including change scope, failure rounds, triggered policies, and recommended attention points.

Core principle: convergent state machine, not endless retries

Loop termination requires all of the following to hold:

Failure type is within the automatically fixable range.

Budget (token usage, wall‑clock time) is still under limits.

Failure information is structured enough to become the next input.

Risk level still permits automatic execution.

Example task state (YAML‑style) shows attempts, budget usage, and last failure type:

task_state:
  id: task-20260715-1024
  stage: TESTING
  risk_level: medium
  attempts: 2
  max_attempts: 5
  token_budget_used: 186000
  token_budget_limit: 300000
  wall_clock_seconds: 934
  last_failure_type: integration_test_failed
  stop_reason: null

Structured reflection replaces raw logs, enabling the model to focus on concise signals:

{
  "task_id": "task-20260715-1024",
  "attempt": 2,
  "failure_type": "contract_test_failed",
  "summary": "payment callback response schema changed",
  "changed_files": ["payment/callback_handler.py", "contracts/payment_callback.json"],
  "evidence": ["expected field `retryable` missing", "response status changed from 202 to 200"],
  "next_constraints": ["do not modify public response schema without migration note", "preserve existing status code semantics"]
}

Convergence also relies on shrinking the modification radius with each retry:

Round 1 – free generation of a solution.

Round 2 – limit changes to files directly related to the failure.

Round 3+ – treat as local patches, not full redesigns.

If failures cross multiple service boundaries, terminate the automatic loop.

Technical choices and practical deployment

Event bus (Kafka or Redpanda)

Using a persistent event stream avoids synchronous blocking, provides auditability, enables asynchronous queuing, and allows new agents (e.g., security, cost) to be added without changing the core pipeline.

Orchestration engine

Start with an explicit state‑machine persisted in a relational DB and event consumers. Adopt Temporal only when long‑running workflows, compensation steps, or visual retries become necessary.

Execution sandbox

Prefer one‑time Kubernetes Jobs or Firecracker VMs over long‑lived containers. Benefits:

Clean workspace per attempt.

No credential or cache leakage across tasks.

Easy audit evidence collection.

State storage

Store structured metadata (task state, attempts, patch hashes, budgets, policy hits) in a relational database; store large logs, build artifacts, and long‑form context in object storage.

Model integration

Wrap LLM calls behind a unified gateway that handles authentication, rate‑limiting, observability, and graceful degradation. Record per‑attempt model version, prompt template, and tool permissions in the state plane.

Runnable code example: a production‑close code‑generation loop

The following Python program demonstrates the key principles. It externalizes task state via Kafka events, enforces per‑attempt budgets, uses structured reflection, and separates generation from execution.

import json, os, subprocess, tempfile
from dataclasses import dataclass
from enum import Enum
from typing import Any
from kafka import KafkaConsumer, KafkaProducer
from openai import OpenAI

class FailureType(str, Enum):
    PATCH_APPLY_FAILED = "patch_apply_failed"
    BUILD_FAILED = "build_failed"
    UNIT_TEST_FAILED = "unit_test_failed"
    INTEGRATION_TEST_FAILED = "integration_test_failed"
    POLICY_BLOCKED = "policy_blocked"

@dataclass
class Task:
    task_id: str
    repo_url: str
    base_branch: str
    issue: str
    risk_level: str
    max_attempts: int
    token_budget_limit: int

client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])
producer = KafkaProducer(
    bootstrap_servers=os.environ["KAFKA_BROKER"],
    value_serializer=lambda v: json.dumps(v).encode("utf-8"),
)
consumer = KafkaConsumer(
    "devloop.tasks.codegen",
    bootstrap_servers=os.environ["KAFKA_BROKER"],
    group_id="codegen-agent",
    value_deserializer=lambda m: json.loads(m.decode("utf-8")),
)

def publish_event(event_type: str, payload: dict[str, Any]) -> None:
    producer.send("devloop.events", {"type": event_type, **payload})
    producer.flush()

def run(cmd: list[str], cwd: str) -> subprocess.CompletedProcess[str]:
    return subprocess.run(cmd, cwd=cwd, text=True, capture_output=True, check=False)

def clone_repo(repo_url: str, base_branch: str) -> str:
    workdir = tempfile.mkdtemp(prefix="devloop-")
    run(["git", "clone", "--depth", "20", repo_url, workdir], cwd="/tmp")
    run(["git", "checkout", base_branch], cwd=workdir)
    return workdir

def load_relevant_context(repo_path: str) -> str:
    files = run(["rg", "-l", "payment|callback|idempotent|redis", "."], cwd=repo_path)
    interesting = files.stdout.strip().splitlines()[:8]
    chunks = []
    for path in interesting:
        content = run(["sed", "-n", "1,220p", path], cwd=repo_path).stdout
        chunks.append(f"FILE: {path}
{content}")
    return "

".join(chunks)

def build_reflection(last_failure: dict[str, Any] | None) -> str:
    if not last_failure:
        return "No previous attempt."
    return json.dumps({
        "failure_type": last_failure["failure_type"],
        "summary": last_failure["summary"],
        "constraints": last_failure["constraints"],
    }, ensure_ascii=False)

def request_diff(issue: str, context: str, reflection: str) -> str:
    prompt = f"""You are a senior backend engineer working in a guarded code generation loop.
Return only a unified diff.

Issue:
{issue}

Relevant context:
{context}

Previous failure reflection:
{reflection}

Rules:
- keep patch radius small
- preserve public API behavior unless issue explicitly requires change
- avoid adding new dependencies unless necessary
- include error handling for failure paths
""".strip()
    response = client.responses.create(
        model=os.getenv("LLM_MODEL", "gpt-4.1"),
        input=prompt,
    )
    return response.output_text

def apply_diff(repo_path: str, diff_text: str) -> tuple[bool, str]:
    patch_path = os.path.join(repo_path, "devloop.patch")
    with open(patch_path, "w", encoding="utf-8") as f:
        f.write(diff_text)
    result = run(["git", "apply", "--whitespace=fix", "devloop.patch"], cwd=repo_path)
    if result.returncode != 0:
        return False, result.stderr
    return True, ""

def run_checks(repo_path: str) -> tuple[bool, dict[str, Any]]:
    build = run(["pytest", "-q", "tests/unit"], cwd=repo_path)
    if build.returncode != 0:
        return False, {
            "failure_type": FailureType.UNIT_TEST_FAILED.value,
            "summary": "unit tests failed",
            "constraints": ["do not change unrelated tests"],
            "stderr": build.stderr[-4000:],
        }
    integration = run(["pytest", "-q", "tests/integration"], cwd=repo_path)
    if integration.returncode != 0:
        return False, {
            "failure_type": FailureType.INTEGRATION_TEST_FAILED.value,
            "summary": "integration tests failed",
            "constraints": ["preserve downstream contract", "keep idempotent semantics"],
            "stderr": integration.stderr[-4000:],
        }
    return True, {}

def handle_task(task_payload: dict[str, Any]) -> None:
    task = Task(**task_payload)
    repo_path = clone_repo(task.repo_url, task.base_branch)
    last_failure: dict[str, Any] | None = None
    publish_event("task.started", {"task_id": task.task_id, "risk_level": task.risk_level})
    for attempt in range(1, task.max_attempts + 1):
        context = load_relevant_context(repo_path)
        reflection = build_reflection(last_failure)
        diff_text = request_diff(task.issue, context, reflection)
        applied, error = apply_diff(repo_path, diff_text)
        if not applied:
            last_failure = {
                "failure_type": FailureType.PATCH_APPLY_FAILED.value,
                "summary": "patch could not be applied cleanly",
                "constraints": ["do not rename files unless required"],
                "stderr": error[-4000:],
            }
            publish_event("task.attempt_failed", {"task_id": task.task_id, "attempt": attempt, **last_failure})
            continue
        success, failure = run_checks(repo_path)
        if success:
            publish_event("task.codegen_succeeded", {"task_id": task.task_id, "attempt": attempt})
            return
        last_failure = failure
        run(["git", "reset", "--hard", "HEAD"], cwd=repo_path)
        publish_event("task.attempt_failed", {"task_id": task.task_id, "attempt": attempt, **failure})
    publish_event("task.escalated", {"task_id": task.task_id, "reason": "max_attempts_exhausted", "last_failure": last_failure})

def main() -> None:
    for message in consumer:
        handle_task(message.value)

if __name__ == "__main__":
    main()

Key observations:

Task state lives outside the process via events, enabling independent consumers.

Each round narrows input scope and uses a structured failure summary.

Final failure is treated as a normal output, allowing the loop to terminate gracefully.

Unified task event model (JSON)

{
  "type": "task.attempt_failed",
  "task_id": "task-20260715-1024",
  "attempt": 2,
  "stage": "TESTING",
  "failure_type": "integration_test_failed",
  "summary": "payment callback contract changed",
  "artifacts": {
    "patch_ref": "s3://devloop-artifacts/task-20260715-1024/attempt-2.patch",
    "log_ref": "s3://devloop-artifacts/task-20260715-1024/attempt-2.log"
  },
  "budget": {
    "token_used": 186000,
    "token_limit": 300000,
    "wall_clock_seconds": 934
  }
}

Dockerfile (minimal permissions)

FROM python:3.11-slim
RUN apt-get update \
    && apt-get install -y --no-install-recommends git ripgrep ca-certificates \
    && rm -rf /var/lib/apt/lists/*
RUN useradd -u 10001 -m agent
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY codegen_agent.py .
USER 10001
ENV PYTHONUNBUFFERED=1
CMD ["python", "codegen_agent.py"]

Kubernetes Deployment (budget & isolation as platform capabilities)

apiVersion: apps/v1
kind: Deployment
metadata:
  name: codegen-agent
spec:
  replicas: 2
  selector:
    matchLabels:
      app: codegen-agent
  template:
    metadata:
      labels:
        app: codegen-agent
    spec:
      serviceAccountName: devloop-agent
      automountServiceAccountToken: false
      securityContext:
        runAsNonRoot: true
        seccompProfile:
          type: RuntimeDefault
      containers:
        - name: agent
          image: registry.example.com/devloop/codegen-agent:1.0.0
          envFrom:
            - configMapRef:
                name: devloop-config
            - secretRef:
                name: devloop-secrets
          resources:
            requests:
              cpu: "500m"
              memory: "1Gi"
            limits:
              cpu: "2"
              memory: "4Gi"
          volumeMounts:
            - name: workspace
              mountPath: /tmp
          securityContext:
            allowPrivilegeEscalation: false
            readOnlyRootFilesystem: true
      volumes:
        - name: workspace
          emptyDir:
            sizeLimit: 4Gi

Test Agent Job template (one attempt, one environment)

apiVersion: batch/v1
kind: Job
metadata:
  name: test-agent-{{task_id}}-{{attempt}}
spec:
  backoffLimit: 0
  ttlSecondsAfterFinished: 1800
  template:
    spec:
      restartPolicy: Never
      serviceAccountName: devloop-runner
      automountServiceAccountToken: false
      containers:
        - name: runner
          image: registry.example.com/devloop/test-runner:1.0.0
          command: ["/bin/sh", "-c"]
          args:
            - |
              git clone --depth 20 -b {{branch}} {{repo_url}} /workspace && \
              cd /workspace && \
              ./scripts/bootstrap-test-env.sh && \
              pytest -q tests/unit && \
              pytest -q tests/integration --junitxml=/artifacts/results.xml
          resources:
            requests:
              cpu: "1"
              memory: "2Gi"
            limits:
              cpu: "4"
              memory: "8Gi"
          volumeMounts:
            - name: artifacts
              mountPath: /artifacts
      volumes:
        - name: artifacts
          emptyDir: {}

Real‑world pitfalls and mitigations

Infinite loops & budget explosion

Unbounded retries lead to token cost blow‑up and larger patches.

Mitigation: enforce max attempts (3‑5), token and wall‑clock caps, shrink modification radius after the third round, and suspend when failures cross service boundaries.

Security & compliance risks

Debug code, high‑risk dependencies, secret leakage, or unauthorized auth changes can slip in.

Governance plane adds three defense layers: pre‑patch policy checks, post‑patch SAST/secret/license scans, and an audit comment summarising security impact.

Tests that look smart but are business‑disconnected

Missing async‑message duplication, migration ordering, contract changes, config gaps, or edge‑case status codes.

Solution: tiered testing – unit tests added by the agent, platform‑provided integration tests, contract baselines, and smoke validation via deployment metrics.

Context window size

Switching to a larger‑window model adds irrelevant noise and increases the chance of plausible‑looking mistakes.

Better approach: code indexing or static dependency analysis to narrow candidate files, compress repository‑level info into summaries, and feed only the failure assertions to the model.

Multi‑agent coordination

Many agents without a unified state machine cause waiting, duplicate work, or conflicting decisions.

Rule: keep a single state entry point, allow concurrent actions, but let the control plane decide the final state; convert all agent outputs to structured events.

Optimization & evolution path

Phase 1 – low‑risk tasks

Test additions, lint fixes, documentation updates, small‑impact bugs.

Validate state machine clarity, structured failure feedback, and complete audit trails.

Phase 2 – repository knowledge & pattern library

Build code indexes and module summaries.

Record successful patches and failure reasons.

Create pattern libraries for common fixes (idempotency, null‑pointer, contract regression).

Phase 3 – canary & automatic rollback

Staging automatic deployments.

Gray‑scale decisions based on error rate, latency, and business‑critical metrics.

Automatic rollback on failure, feeding runtime evidence back into task state.

Phase 4 – organization‑wide governance

Define which repositories allow automatic changes and which only suggest.

Set policy customization boundaries per team.

Collect ROI, failure, and cost metrics per task, team, and repository.

Identify tasks that should never be handed to an intelligent agent.

Conclusion

Loop Engineering does not aim to make the model write perfect code on the first try. It embeds an uncertain generation system into a governed feedback loop that observes, plans, acts, verifies, reflects, and decides when to stop. By bounding retries, shrinking modification radius, persisting structured state, and enforcing sandboxed execution and policy checks, AI‑generated code can converge to a deployable, auditable, and rollback‑able state suitable for production micro‑service environments.