AI Coding Agents Autonomous Production 2025: Benchmarked

# .github/workflows/ai-agent-deploy.yml
# Triggers on GitHub issues labeled 'ai-agent-task'
# Runs security scanning and performance profiling before any merge

name: AI Agent Autonomous Deployment

on:
  issues:
    types: [opened, labeled]
  workflow_dispatch:
    inputs:
      task_description:
        description: 'Task for AI agent'
        required: true
      auto_merge:
        description: 'Enable auto-merge if all gates pass'
        required: false
        default: 'false'

jobs:
  classify_task:
    runs-on: ubuntu-latest
    outputs:
      task_type: ${{ steps.classify.outputs.task_type }}
      severity: ${{ steps.classify.outputs.severity }}
      auto_merge_allowed: ${{ steps.classify.outputs.auto_merge_allowed }}
    steps:
      - name: Classify Task Severity
        id: classify
        env:
          ISSUE_BODY: ${{ github.event.issue.body }}
          ISSUE_LABELS: ${{ toJson(github.event.issue.labels) }}
        run: |
          python scripts/classify_task.py \
            --issue-body "$ISSUE_BODY" \
            --labels "$ISSUE_LABELS"

  agent_task:
    runs-on: ubuntu-latest
    needs: classify_task
    if: contains(github.event.issue.labels.*.name, 'ai-agent-task')

    steps:
      - uses: actions/checkout@v4
        with:
          fetch-depth: 0
          token: ${{ secrets.GITHUB_TOKEN }}

      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.11'

      - name: Install dependencies
        run: |
          pip install anthropic pydantic gitpython semgrep pip-audit pytest pytest-cov

      - name: Invoke AI Agent
        id: agent
        env:
          ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}
          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
          TASK_SEVERITY: ${{ needs.classify_task.outputs.severity }}
        run: |
          python scripts/invoke_agent.py \
            --issue-id ${{ github.event.issue.number }} \
            --repo ${{ github.repository }} \
            --task-type ${{ needs.classify_task.outputs.task_type }} \
            --auto-merge ${{ needs.classify_task.outputs.auto_merge_allowed }} \
            --security-scan true \
            --performance-profile true \
            --max-iterations 7

      - name: Run Security Scan (Semgrep)
        uses: returntocorp/semgrep-action@v1
        with:
          config: >-
            p/security-audit
            p/owasp-top-ten
            p/cwe-top-25
            p/secrets

      - name: Dependency Vulnerability Check
        run: pip-audit --desc --output json > audit-results.json

      - name: Performance Baseline Check
        env:
          PR_BRANCH: ${{ steps.agent.outputs.pr_branch }}
        run: |
          python scripts/benchmark_pr.py \
            --baseline main \
            --branch $PR_BRANCH \
            --threshold-latency 0.10 \
            --threshold-memory 0.05 \
            --output benchmark-results.json

      - name: Evaluate All Gates
        id: gates
        run: |
          python scripts/evaluate_gates.py \
            --task-type ${{ needs.classify_task.outputs.task_type }} \
            --severity ${{ needs.classify_task.outputs.severity }} \
            --audit-results audit-results.json \
            --benchmark-results benchmark-results.json

      - name: Conditional Auto-Merge
        if: |
          steps.gates.outputs.all_pass == 'true' &&
          needs.classify_task.outputs.auto_merge_allowed == 'true' &&
          needs.classify_task.outputs.severity == 'low'
        env:
          GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
        run: |
          gh pr merge ${{ steps.agent.outputs.pr_number }} \
            --auto \
            --squash \
            --delete-branch \
            --subject "Auto-merge: AI agent task #${{ github.event.issue.number }}"

      - name: Post Review Request (if not auto-merged)
        if: steps.gates.outputs.all_pass != 'true' || needs.classify_task.outputs.severity != 'low'
        env:
          GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
        run: |
          gh pr review ${{ steps.agent.outputs.pr_number }} \
            --request-reviewer ${{ github.event.issue.user.login }} \
            --body "AI agent completed task. Gates status: ${{ steps.gates.outputs.summary }}"

# scripts/agent_config.py
# Defines merge gates, rollback triggers, and review requirements
# per task category. Modify this file to tune your governance policy.

from dataclasses import dataclass, field
from typing import List
from enum import Enum


class TaskSeverity(Enum):
    LOW = "low"       # Auto-merge enabled for passing gates
    MEDIUM = "medium" # Review required before merge
    HIGH = "high"     # Manual approval + security review mandatory


@dataclass
class AgentTaskConfig:
    task_type: str
    severity: TaskSeverity
    auto_merge: bool
    required_gates: List[str]
    rollback_triggers: List[str]
    review_timeout_hours: int
    max_iterations: int = 7
    notify_channels: List[str] = field(default_factory=list)


TASK_CONFIGS: dict[str, AgentTaskConfig] = {
    "test_generation": AgentTaskConfig(
        task_type="test_generation",
        severity=TaskSeverity.LOW,
        auto_merge=True,
        required_gates=[
            "tests_pass",
            "coverage_maintained",
            "no_security_issues",
            "no_dependency_vulnerabilities",
        ],
        rollback_triggers=[
            "test_failure_in_staging",
            "coverage_drop_gt_5pct",
        ],
        review_timeout_hours=24,
        notify_channels=["#eng-bots"],
    ),

    "bug_fix_isolated": AgentTaskConfig(
        task_type="bug_fix_isolated",
        severity=TaskSeverity.LOW,
        auto_merge=True,
        required_gates=[
            "tests_pass",
            "no_regression",
            "no_security_issues",
            "performance_stable",
            "coverage_maintained",
        ],
        rollback_triggers=[
            "error_rate_spike_gt_5pct",
            "latency_p99_increase_gt_15pct",
            "security_alert_triggered",
        ],
        review_timeout_hours=12,
        notify_channels=["#eng-bots", "#on-call"],
    ),

    "cross_service_refactor": AgentTaskConfig(
        task_type="cross_service_refactor",
        severity=TaskSeverity.HIGH,
        auto_merge=False,
        required_gates=[
            "tests_pass",
            "integration_tests_pass",
            "no_security_issues",
            "performance_profile_reviewed",
            "manual_review_approved",
            "architecture_review_approved",
        ],
        rollback_triggers=[
            "any_error_in_prod",
            "latency_p99_increase_gt_5pct",
            "dependency_service_errors",
        ],
        review_timeout_hours=4,
        max_iterations=10,
        notify_channels=["#eng-leads", "#on-call"],
    ),

    "security_critical": AgentTaskConfig(
        task_type="security_critical",
        severity=TaskSeverity.HIGH,
        auto_merge=False,
        required_gates=[
            "tests_pass",
            "sast_scan_clean",
            "dast_scan_clean",
            "dependency_audit_clean",
            "security_team_approved",
            "manual_review_approved",
        ],
        rollback_triggers=[
            "any_security_alert",
            "any_error_in_prod",
            "anomalous_auth_pattern",
        ],
        review_timeout_hours=2,
        max_iterations=5,
        notify_channels=["#security", "#eng-leads", "#on-call"],
    ),

    "documentation": AgentTaskConfig(
        task_type="documentation",
        severity=TaskSeverity.LOW,
        auto_merge=True,
        required_gates=[
            "link_check_pass",
            "spelling_check_pass",
        ],
        rollback_triggers=[],
        review_timeout_hours=48,
        notify_channels=["#docs"],
    ),
}


def get_config(task_type: str) -> AgentTaskConfig:
    config = TASK_CONFIGS.get(task_type)
    if config is None:
        # Default to HIGH severity for unknown task types
        # Never auto-merge what you can't classify
        return AgentTaskConfig(
            task_type=task_type,
            severity=TaskSeverity.HIGH,
            auto_merge=False,
            required_gates=["manual_review_approved"],
            rollback_triggers=["any_error_in_prod"],
            review_timeout_hours=2,
        )
    return config

# scripts/pre_merge_validation.py
# Runs full validation suite before any merge decision.
# Designed to be called from CI and from the agent itself.

import subprocess
import json
import os
from typing import Tuple, Dict
from pathlib import Path


class PreMergeValidator:
    """
    Runs security scanning, dependency auditing, test execution,
    and performance profiling. Returns a pass/fail verdict with
    detailed results for each gate.
    """

    def __init__(self, base_branch: str = "main", coverage_threshold: int = 85):
        self.base_branch = base_branch
        self.coverage_threshold = coverage_threshold
        self.results: Dict[str, bool] = {}
        self.details: Dict[str, str] = {}

    def validate_pr(self, pr_branch: str) -> Tuple[bool, Dict]:
        """
        Run full validation suite. Returns (all_pass, detailed_results).
        Call this before any merge decision—auto or manual.
        """
        print(f"Running pre-merge validation: {pr_branch} → {self.base_branch}")

        self.results["tests_pass"] = self._run_tests()
        self.results["sast_clean"] = self._run_sast()
        self.results["deps_safe"] = self._check_dependencies()
        self.results["performance_stable"] = self._profile_performance(pr_branch)
        self.results["coverage_maintained"] = self._check_coverage(pr_branch)

        all_pass = all(self.results.values())

        return all_pass, {
            "gates": self.results,
            "details": self.details,
            "verdict": "PASS" if all_pass else "FAIL",
            "failed_gates": [k for k, v in self.results.items() if not v],
        }

    def _run_tests(self) -> bool:
        """Run pytest with verbose output. Fail fast on first error."""
        try:
            result = subprocess.run(
                ["pytest", "tests/", "-v", "--tb=short", "-x"],
                capture_output=True,
                text=True,
                timeout=300,  # 5 minute timeout
            )
            self.details["tests"] = result.stdout[-2000:]  # Last 2K chars
            return result.returncode == 0
        except subprocess.TimeoutExpired:
            self.details["tests"] = "TIMEOUT: Tests exceeded 5 minute limit"
            return False
        except FileNotFoundError:
            self.details["tests"] = "ERROR: pytest not found"
            return False

    def _run_sast(self) -> bool:
        """
        Run Semgrep with OWASP Top 10, CWE Top 25, and secrets detection.
        Zero tolerance for high-severity findings.
        """
        try:
            result = subprocess.run(
                [
                    "semgrep",
                    "--config=p/security-audit",
                    "--config=p/owasp-top-ten",
                    "--config=p/cwe-top-25",
                    "--config=p/secrets",
                    "--json",
                    "--severity=ERROR",  # Only flag high severity
                    ".",
                ],
                capture_output=True,
                text=True,
                timeout=120,
            )
            output = json.loads(result.stdout) if result.stdout else {"results": []}
            high_severity = [
                r for r in output.get("results", [])
                if r.get("extra", {}).get("severity") in ("ERROR", "WARNING")
            ]
            self.details["sast"] = f"{len(high_severity)} high-severity findings"
            return len(high_severity) == 0
        except (json.JSONDecodeError, subprocess.TimeoutExpired) as e:
            self.details["sast"] = f"SAST scan error: {str(e)}"
            return False

    def _check_dependencies(self) -> bool:
        """
        Audit Python dependencies for known CVEs using pip-audit.
        For Node projects, swap for 'npm audit --audit-level=high'.
        """
        try:
            result = subprocess.run(
                ["pip-audit", "--desc", "--output=json"],
                capture_output=True,
                text=True,
                timeout=60,
            )
            if result.returncode == 0:
                self.details["deps"] = "No vulnerable dependencies found"
                return True
            else:
                output = json.loads(result.stdout) if result.stdout else {}
                vulns = output.get("vulnerabilities", [])
                self.details["deps"] = f"{len(vulns)} vulnerable dependencies"
                return False
        except (json.JSONDecodeError, subprocess.TimeoutExpired, FileNotFoundError) as e:
            self.details["deps"] = f"Dependency audit error: {str(e)}"
            return False

    def _profile_performance(self, pr_branch: str) -> bool:
        """
        Compare benchmark results between base and PR branch.
        Rejects if latency increases >10% or memory increases >5%.
        """
        try:
            base_metrics = self._run_benchmarks(self.base_branch)
            pr_metrics = self._run_benchmarks(pr_branch)

            if not base_metrics or not pr_metrics:
                self.details["performance"] = "Benchmark data unavailable—skipping"
                return True  # Don't block on missing benchmarks

            latency_delta = (
                (pr_metrics["latency_ms"] - base_metrics["latency_ms"])
                / base_metrics["latency_ms"]
            )
            memory_delta = (
                (pr_metrics["memory_mb"] - base_metrics["memory_mb"])
                / base_metrics["memory_mb"]
            )

            self.details["performance"] = (
                f"Latency: {latency_delta:+.1%}, Memory: {memory_delta:+.1%}"
            )

            return latency_delta < 0.10 and memory_delta < 0.05

        except Exception as e:
            self.details["performance"] = f"Profiling error: {str(e)}"
            return False

    def _run_benchmarks(self, branch: str) -> Dict:
        """Run pytest-benchmark and parse results."""
        try:
            subprocess.run(
                ["git", "checkout", branch],
                check=True, capture_output=True
            )
            result = subprocess.run(
                [
                    "pytest",
                    "benchmarks/",
                    "--benchmark-json=benchmark_output.json",
                    "-q",
                ],
                capture_output=True,
                text=True,
                timeout=180,
            )
            if Path("benchmark_output.json").exists():
                with open("benchmark_output.json") as f:
                    data = json.load(f)
                benchmarks = data.get("benchmarks", [])
                if benchmarks:
                    avg_latency = sum(
                        b["stats"]["mean"] * 1000 for b in benchmarks
                    ) / len(benchmarks)
                    return {"latency_ms": avg_latency, "memory_mb": 256.0}
        except Exception:
            pass
        return {}

    def _check_coverage(self, pr_branch: str) -> bool:
        """Ensure coverage doesn't drop below threshold."""
        try:
            result = subprocess.run(
                [
                    "pytest",
                    "--cov=src",
                    "--cov-report=json:coverage.json",
                    "-q",
                ],
                capture_output=True,
                text=True,
                timeout=300,
            )
            if Path("coverage.json").exists():
                with open("coverage.json") as f:
                    data = json.load(f)
                coverage_pct = data.get("totals", {}).get("percent_covered", 0)
                self.details["coverage"] = f"{coverage_pct:.1f}%"
                return coverage_pct >= self.coverage_threshold
        except Exception as e:
            self.details["coverage"] = f"Coverage check error: {str(e)}"
        return False

# scripts/rollback_monitor.py
# Monitors production metrics for 15 minutes post-merge.
# Triggers automatic rollback if thresholds are breached.
# Integrates with Datadog; swap _fetch_metrics() for Prometheus/New Relic.

import time
import subprocess
from datetime import datetime, timedelta
from typing import Dict, Optional
import requests


class RollbackMonitor:
    """
    Post-merge production monitor. Watches error rate, latency,
    and resource usage. Auto-reverts PR and creates incident if
    thresholds are breached within the observation window.
    """

    THRESHOLDS = {
        "error_rate_pct": 5.0,       # 5% error rate spike
        "latency_p99_multiplier": 1.5, # 50% increase in p99 latency
        "memory_multiplier": 1.3,      # 30% increase in memory usage
        "cpu_multiplier": 1.4,         # 40% increase in CPU usage
    }

    def __init__(
        self,
        github_token: str,
        datadog_api_key: str,
        datadog_app_key: str,
        slack_webhook: Optional[str] = None,
    ):
        self.github_token = github_token
        self.datadog_api_key = datadog_api_key
        self.datadog_app_key = datadog_app_key
        self.slack_webhook = slack_webhook

    def monitor(
        self,
        repo: str,
        pr_number: int,
        service_name: str,
        observation_minutes: int = 15,
        check_interval_seconds: int = 30,
    ) -> bool:
        """
        Monitor service health post-merge.
        Returns True if service stayed healthy, False if rollback triggered.
        """
        end_time = datetime.now() + timedelta(minutes=observation_minutes)
        baseline = self._fetch_baseline_metrics(service_name)

        print(
            f"[RollbackMonitor] Watching {service_name} for {observation_minutes}m "
            f"after PR #{pr_number} merge"
        )

        while datetime.now() < end_time:
            current = self._fetch_current_metrics(service_name)
            breach = self._check_thresholds(baseline, current)

            if breach:
                print(f"[RollbackMonitor] THRESHOLD BREACH: {breach}")
                self._trigger_rollback(repo, pr_number, breach)
                self._notify_team(repo, pr_number, breach, current)
                return False

            remaining = (end_time - datetime.now()).seconds // 60
            print(f"[RollbackMonitor] OK — {remaining}m remaining in observation window")
            time.sleep(check_interval_seconds)

        print(f"[RollbackMonitor] Observation window complete. No issues detected.")
        return True

    def _fetch_baseline_metrics(self, service: str) -> Dict:
        """Fetch 1-hour baseline metrics before the deploy."""
        # Query Datadog for pre-deploy baseline (last 1 hour, ending at deploy time)
        # Real implementation: use datadog-api-client-python
        return {
            "error_rate_pct": 0.8,
            "latency_p99_ms": 120.0,
            "memory_mb": 512.0,
            "cpu_pct": 35.0,
        }

    def _fetch_current_metrics(self, service: str) -> Dict:
        """Fetch current metrics from Datadog."""
        # Real implementation queries Datadog Metrics API
        # endpoint: https://api.datadoghq.com/api/v1/query
        return {
            "error_rate_pct": 0.9,
            "latency_p99_ms": 125.0,
            "memory_mb": 520.0,
            "cpu_pct": 37.0,
        }

    def _check_thresholds(self, baseline: Dict, current: Dict) -> Optional[str]:
        """Return breach description if any threshold is exceeded, else None."""
        if current["error_rate_pct"] > self.THRESHOLDS["error_rate_pct"]:
            return (
                f"Error rate: {current['error_rate_pct']:.1f}% "
                f"(threshold: {self.THRESHOLDS['error_rate_pct']}%)"
            )

        if baseline["latency_p99_ms"] > 0:
            latency_mult = current["latency_p99_ms"] / baseline["latency_p99_ms"]
            if latency_mult > self.THRESHOLDS["latency_p99_multiplier"]:
                return (
                    f"p99 latency: {current['latency_p99_ms']:.0f}ms "
                    f"({latency_mult:.1f}x baseline)"
                )

        if baseline["memory_mb"] > 0:
            mem_mult = current["memory_mb"] / baseline["memory_mb"]
            if mem_mult > self.THRESHOLDS["memory_multiplier"]:
                return (
                    f"Memory: {current['memory_mb']:.0f}MB "
                    f"({mem_mult:.1f}x baseline)"
                )

        return None

    def _trigger_rollback(self, repo: str, pr_number: int, reason: str):
        """Revert the PR commit and create a tracking issue."""
        print(f"[RollbackMonitor] TRIGGERING ROLLBACK: {repo}#{pr_number}")

        # Git revert via GitHub API
        headers = {
            "Authorization": f"Bearer {self.github_token}",
            "Accept": "application/vnd.github+json",
        }
        # Create revert PR via GitHub API
        revert_response = requests.post(
            f"https://api.github.com/repos/{repo}/pulls",
            headers=headers,
            json={
                "title": f"[AUTO-REVERT] Revert AI agent PR #{pr_number}",
                "body": (
                    f"Automatic rollback triggered.\n\n"
                    f"**Reason:** {reason}\n\n"
                    f"**Original PR:** #{pr_number}\n"
                    f"**Triggered by:** RollbackMonitor at {datetime.now().isoformat()}"
                ),
                "head": f"revert-{pr_number}",
                "base": "main",
            },
        )
        print(f"[RollbackMonitor] Revert PR status: {revert_response.status_code}")

    def _notify_team(
        self, repo: str, pr_number: int, breach: str, metrics: Dict
    ):
        """Send Slack alert with rollback details."""
        if not self.slack_webhook:
            return

        payload = {
            "text": (
                f":rotating_light: *AUTO-ROLLBACK TRIGGERED*\n"
                f"Repo: `{repo}` | PR: #{pr_number}\n"
                f"Breach: {breach}\n"
                f"Error rate: {metrics.get('error_rate_pct')}% | "
                f"p99: {metrics.get('latency_p99_ms')}ms"
            )
        }
        requests.post(self.slack_webhook, json=payload, timeout=10)

Scenario A: Early-stage startup (10 devs, 5 repos, ~200 PRs/month)
├─ AI Agent API cost:     $500/month
├─ Review overhead:       20 hrs × $100/hr = $2,000/month
├─ Total AI cost:         $2,500/month
├─ Developer time saved:  ~40 hrs/month = $4,000
├─ Net benefit:           $1,500/month
├─ Verdict: Marginal. Only worthwhile if review overhead stays below 15 hrs.

Scenario B: Growth-stage (50 devs, 30 repos, ~2,000 PRs/month)
├─ AI Agent API cost:     $1,500/month
├─ Review overhead:       80 hrs × $100/hr = $8,000/month
├─ Total AI cost:         $9,500/month
├─ Developer time saved:  ~1,000 hrs/month = $100,000
├─ Net benefit:           $90,500/month
├─ Verdict: Strong ROI. Prioritize governance infrastructure investment.

Scenario C: Enterprise (200 devs, 100 repos, ~10,000 PRs/month)
├─ AI Agent fleet cost:   $5,000/month
├─ Review overhead:       300 hrs × $100/hr = $30,000/month
├─ Total AI cost:         $35,000/month
├─ Developer time saved:  ~5,000 hrs/month = $500,000
├─ Net benefit:           $465,000/month
├─ Verdict: Essential infrastructure, not optional tooling.

GitHub Issue: "Fix the bug in the payment processor"

{
  "task_type": "bug_fix_isolated",
  "component": "payment_processor",
  "module_path": "src/payments/processor.py",
  "bug_description": "PaymentProcessor.charge() returns success=true when Stripe returns HTTP 402",
  "input_contract": {
    "amount_cents": "integer, > 0, < 100000000",
    "currency": "string, ISO 4217 (USD, EUR, GBP)",
    "idempotency_key": "UUID v4 string"
  },
  "output_contract": {
    "success": "boolean",
    "transaction_id": "string or null",
    "error_code": "string or null (INSUFFICIENT_FUNDS, CARD_DECLINED, NETWORK_ERROR)",
    "error_message": "human-readable string or null"
  },
  "test_cases": [
    {
      "name": "stripe_402_returns_failure",
      "mock": {"stripe_status": 402, "stripe_error": "insufficient_funds"},
      "expected": {"success": false, "transaction_id": null, "error_code": "INSUFFICIENT_FUNDS"}
    },
    {
      "name": "stripe_200_returns_success",
      "mock": {"stripe_status": 200, "stripe_charge_id": "ch_test_123"},
      "expected": {"success": true, "transaction_id": "ch_test_123", "error_code": null}
    }
  ],
  "constraints": [
    "Do not modify the Stripe API client initialization",
    "Maintain backward compatibility with existing charge() callers",
    "All monetary amounts remain in cents (integer), never float"
  ],
  "rollback_condition": "error_rate in payments service exceeds 2% within 15 minutes of merge"
}

# scripts/index_codebase.py
# Run this on every commit via GitHub Actions hook.
# Stores embeddings in Pinecone for agent retrieval.

import os
import json
from pathlib import Path
from anthropic import Anthropic
import pinecone

client = Anthropic()

def chunk_file(file_path: Path, chunk_size: int = 100) -> list[dict]:
    """Split file into overlapping chunks for embedding."""
    with open(file_path, "r", encoding="utf-8", errors="ignore") as f:
        lines = f.readlines()

    chunks = []
    for i in range(0, len(lines), chunk_size // 2):  # 50% overlap
        chunk_lines = lines[i : i + chunk_size]
        chunks.append({
            "path": str(file_path),
            "start_line": i + 1,
            "content": "".join(chunk_lines),
        })
    return chunks


def index_codebase(repo_path: str, index_name: str = "codebase-index"):
    """Index all source files in the repository."""
    pc = pinecone.Pinecone(api_key=os.environ["PINECONE_API_KEY"])
    index = pc.Index(index_name)

    source_extensions = {".py", ".ts", ".js", ".go", ".rs", ".java"}
    all_chunks = []

    for path in Path(repo_path).rglob("*"):
        if path.suffix in source_extensions and ".git" not in str(path):
            all_chunks.extend(chunk_file(path))

    # Batch embed and upsert
    batch_size = 50
    for i in range(0, len(all_chunks), batch_size):
        batch = all_chunks[i : i + batch_size]
        texts = [c["content"] for c in batch]

        # Use Voyage AI or OpenAI for embeddings (Anthropic doesn't offer embeddings)
        # Placeholder: replace with your embedding provider
        embeddings = get_embeddings(texts)

        vectors = [
            {
                "id": f"{c['path']}:{c['start_line']}",
                "values": emb,
                "metadata": {"path": c["path"], "start_line": c["start_line"]},
            }
            for c, emb in zip(batch, embeddings)
        ]
        index.upsert(vectors=vectors)

    print(f"Indexed {len(all_chunks)} chunks from {repo_path}")

SYSTEM_PROMPT = """
You are a senior engineer at Acme Corp working in a Python/FastAPI codebase.

ARCHITECTURE CONVENTIONS:
- All database access goes through the Repository pattern (src/repositories/)
- Never import database models directly in route handlers
- All monetary amounts are stored and computed as integers (cents), never floats
- Authentication is handled by the AuthMiddleware—never implement auth logic in route handlers
- Use structured logging (structlog) not print() or logging.info()

PERFORMANCE BUDGETS:
- API endpoint p99 latency budget: 200ms
- Database queries per request: maximum 5 (use joins, not N+1 loops)
- Memory per request: maximum 50MB

FORBIDDEN PATTERNS:
- No global mutable state
- No synchronous HTTP calls in async route handlers (use httpx.AsyncClient)
- No raw SQL strings—use SQLAlchemy ORM or parameterized queries only

SECURITY REQUIREMENTS:
- All user input must be validated with Pydantic models before processing
- Never log PII (email, phone, payment data)
- All external API keys come from environment variables, never hardcoded

When generating code, verify it against each of these constraints before submitting.
If a constraint would be violated, explain why and propose an alternative approach.
"""