The Pipeline Deputy — an AI agent as a governed step in your CI pipeline

mkdir deputy-lab && cd deputy-lab
git init -b main
python3 -m venv .venv && source .venv/bin/activate
pip install anthropic pydantic pytest
export ANTHROPIC_API_KEY=sk-ant-...   # your key

ollama pull qwen2.5-coder     # strong at code/patch tasks; llama3.1 also works

def apply_discount(price_cents: int, pct: int) -> int:
    """Return the price after applying a percentage discount."""
    # BUG: returns the discount amount, not the discounted price
    return price_cents * pct // 100

from src.money import apply_discount

def test_ten_percent_off():
    assert apply_discount(10000, 10) == 9000

def test_zero_discount():
    assert apply_discount(5000, 0) == 5000

mkdir -p .pipeline && touch src/__init__.py tests/__init__.py
git add -A && git commit -m "app with settlement discount helper"

#!/usr/bin/env bash
# Local stand-in for your CI pipeline: lint -> test -> deputy on failure.
set -uo pipefail
mkdir -p .pipeline

echo "== Stage 1: lint =="
python -m py_compile src/money.py tests/test_money.py || { echo "LINT FAILED"; exit 1; }

echo "== Stage 2: tests =="
python -m pytest tests/ -x --tb=long > .pipeline/test.log 2>&1
STATUS=$?
tail -n 20 .pipeline/test.log

if [ $STATUS -ne 0 ]; then
  echo "== Stage 3: deputy step (invoked on failure only) =="
  python deputy.py --log .pipeline/test.log
fi
exit $STATUS

#!/usr/bin/env python3
"""Pipeline Deputy: a governed AI step invoked when the build goes red.

Governance: own identity (DEPUTY_IDENTITY), hard token budget
(DEPUTY_TOKEN_BUDGET), read-only inputs, append-only audit ledger,
scope-guarded writes. It NEVER writes to the failing branch.
"""
import argparse, json, os, pathlib, subprocess, sys, time
from typing import Literal, Optional
from anthropic import Anthropic
from pydantic import BaseModel

ALLOWED_PATHS = ("src/", "tests/")          # the deputy's entire write scope
STATE = pathlib.Path(".pipeline")
BUDGET_FILE, AUDIT_FILE = STATE / "budget.json", STATE / "audit.jsonl"

class Verdict(BaseModel):
    classification: Literal["code", "env", "flake"]
    explanation: str                         # cited evidence, 2-4 sentences
    fix_file: Optional[str]                  # single file, or null
    fix_content: Optional[str]               # full corrected file content
    retry_recommended: bool

def audit(identity, action, tokens=0, remaining=None, detail=""):
    entry = {"ts": time.strftime("%Y-%m-%dT%H:%M:%S"), "identity": identity,
             "action": action, "tokens_spent": tokens,
             "budget_remaining": remaining, "detail": detail}
    with AUDIT_FILE.open("a") as f:
        f.write(json.dumps(entry) + "\n")

def spent():
    return json.loads(BUDGET_FILE.read_text())["spent"] if BUDGET_FILE.exists() else 0

def main():
    ap = argparse.ArgumentParser(); ap.add_argument("--log", required=True)
    args = ap.parse_args()

    identity = os.environ.get("DEPUTY_IDENTITY")
    if not identity:
        sys.exit("deputy: refusing to run without DEPUTY_IDENTITY set")
    budget = int(os.environ.get("DEPUTY_TOKEN_BUDGET", "8000"))

    used = spent()
    if used >= budget:                       # graceful, audited refusal
        audit(identity, "budget_exhausted", remaining=0,
              detail=f"spent {used} of {budget}; declining to run")
        pathlib.Path("review-comment.md").write_text(
            f"## Deputy verdict ({identity})\n\n"
            f"Token budget exhausted ({used}/{budget}). No analysis performed. "
            f"A platform owner must raise the budget or wait for the reset.\n")
        print("deputy: budget exhausted — declined gracefully"); return

    log_tail = "\n".join(
        pathlib.Path(args.log).read_text().splitlines()[-120:])
    diff = subprocess.run(["git", "diff", "HEAD~1"],
                          capture_output=True, text=True).stdout or "(no diff)"

    client = Anthropic()
    response = client.messages.parse(
        model="claude-opus-4-8",
        max_tokens=4096,
        thinking={"type": "adaptive"},
        system=(
            "You are a CI pipeline deputy. Classify the failure as exactly one "
            "of: code (a defect in the changed code), env (toolchain/runner/"
            "dependency problem), or flake (nondeterministic test). Cite the "
            "log lines and diff hunks that justify the verdict. Propose "
            "fix_file/fix_content ONLY if a single-file change under src/ or "
            "tests/ fixes a code defect; otherwise set them to null. Set "
            "retry_recommended true only for flakes."),
        messages=[{"role": "user", "content":
            f"## Failure log (tail)\n{log_tail}\n\n## Diff (HEAD~1)\n{diff}"}],
        output_format=Verdict,
    )
    verdict = response.parsed_output
    used += response.usage.output_tokens
    BUDGET_FILE.write_text(json.dumps({"spent": used}))
    audit(identity, "verdict", tokens=response.usage.output_tokens,
          remaining=budget - used, detail=verdict.classification)

    comment = [f"## Deputy verdict ({identity})",
               f"**Classification:** {verdict.classification}",
               f"**Retry recommended:** {verdict.retry_recommended}",
               f"\n{verdict.explanation}",
               f"\n_Tokens: {used}/{budget} spent._"]

    if verdict.fix_file and verdict.fix_content:
        if not verdict.fix_file.startswith(ALLOWED_PATHS):   # SCOPE GUARD
            comment.append(f"\n**Autofix refused:** proposed change to "
                           f"`{verdict.fix_file}` is outside the deputy's "
                           f"scope ({', '.join(ALLOWED_PATHS)}). Escalating "
                           f"to a human.")
            audit(identity, "scope_refusal", detail=verdict.fix_file)
        else:
            base = subprocess.run(["git", "rev-parse", "--abbrev-ref", "HEAD"],
                                  capture_output=True, text=True).stdout.strip()
            branch = f"deputy/autofix-{int(time.time())}"
            subprocess.run(["git", "switch", "-c", branch], check=True)
            pathlib.Path(verdict.fix_file).write_text(verdict.fix_content)
            subprocess.run(["git", "add", verdict.fix_file], check=True)
            subprocess.run(["git", "-c", f"user.name={identity}",
                            "-c", f"user.email={identity}@agents.local",
                            "commit", "-m",
                            f"deputy autofix: {verdict.explanation[:72]}"],
                           check=True)
            subprocess.run(["git", "switch", base], check=True)
            comment.append(f"\n**Autofix branch:** `{branch}` "
                           f"(never pushed to the failing branch)")
            audit(identity, "autofix_branch", detail=branch)

    pathlib.Path("review-comment.md").write_text("\n".join(comment) + "\n")
    print(f"deputy: verdict={verdict.classification} -> review-comment.md")

if __name__ == "__main__":
    main()

export DEPUTY_IDENTITY=deputy-ci
export DEPUTY_TOKEN_BUDGET=8000
./pipeline.sh
cat review-comment.md
cat .pipeline/audit.jsonl
git branch --list 'deputy/*'

import configparser

def test_policy_allows_ten_percent():
    cfg = configparser.ConfigParser()
    cfg.read("settings.ini")
    assert int(cfg["policy"]["max_discount_pct"]) >= 10

export DEPUTY_TOKEN_BUDGET=1
rm -f .pipeline/budget.json && echo '{"spent": 5}' > .pipeline/budget.json
./pipeline.sh
cat review-comment.md && tail -1 .pipeline/audit.jsonl