M21C: Headless Agents

# triage_agent.py  —  run:  python triage_agent.py --query "disk full on web-03"
# WHAT: A headless agent that classifies an ops alert into a JSON verdict
# WHY:  No input() calls, no interactive prompts — safe to run from cron
# GOTCHA: We point the OpenAI client at Ollama; logs go to stderr only

import sys, json, argparse, time
from openai import OpenAI

# Ollama exposes an OpenAI-compatible API on localhost:11434
client = OpenAI(base_url="http://localhost:11434/v1", api_key="ollama")
MODEL = "mistral"

def log(msg: str) -> None:
    # Human-readable progress -> stderr, so stdout stays pure JSON
    print(f"[triage] {msg}", file=sys.stderr)

def read_task(args) -> str:
    # Priority: --query flag, else stdin (so you can pipe text in)
    if args.query:
        return args.query.strip()
    if not sys.stdin.isatty():            # data was piped in
        return sys.stdin.read().strip()
    raise SystemExit("no task: pass --query or pipe text on stdin")

def triage(task: str) -> dict:
    log(f"classifying: {task!r}")
    resp = client.chat.completions.create(
        model=MODEL,
        # Force JSON-shaped output; we still validate it below
        messages=[
            {"role": "system", "content": (
                "You are an ops triage agent. Reply with ONLY a JSON object: "
                '{"severity": "low|medium|high|critical", '
                '"category": str, "summary": str}. No prose.'
            )},
            {"role": "user", "content": task},
        ],
        temperature=0,
    )
    raw = resp.choices[0].message.content
    usage = resp.usage
    # Mistral sometimes wraps JSON in ```; strip a fenced block defensively
    raw = raw.strip().removeprefix("```json").removeprefix("```").removesuffix("```").strip()
    verdict = json.loads(raw)                       # raises if not valid JSON
    verdict["tokens"] = {
        "prompt": usage.prompt_tokens,
        "completion": usage.completion_tokens,
    }
    return verdict

def main() -> int:
    ap = argparse.ArgumentParser(description="Headless ops triage agent")
    ap.add_argument("--query", help="alert text; if omitted, read stdin")
    args = ap.parse_args()
    started = time.time()
    try:
        task = read_task(args)
        verdict = triage(task)
        verdict["latency_ms"] = int((time.time() - started) * 1000)
        # THE result -> stdout, as a single JSON line
        print(json.dumps(verdict))
        log(f"done in {verdict['latency_ms']}ms, severity={verdict.get('severity')}")
        return 0
    except json.JSONDecodeError as e:
        log(f"model did not return valid JSON: {e}")
        print(json.dumps({"error": "invalid_model_output"}))
        return 2                                    # distinct code: bad output
    except Exception as e:
        log(f"fatal: {type(e).__name__}: {e}")
        print(json.dumps({"error": str(e)}))
        return 1                                    # generic failure

if __name__ == "__main__":
    sys.exit(main())                                # exit code = the verdict flag

# contract.py — a stable envelope + exit-code mapping for any headless agent
import sys, json, time

# Custom exceptions let the agent declare *which* failure happened
class BadOutput(Exception):  code = 2     # model output failed schema
class NeedsReview(Exception): code = 3    # ran fine, but low confidence
# anything else -> code 1 (transient/operational), 0 -> success

def run_headless(agent_fn, payload: dict) -> int:
    """Run agent_fn(payload), print a fixed JSON envelope, return an exit code."""
    started = time.time()
    envelope = {"ok": False, "data": None, "error": None, "meta": {}}
    try:
        envelope["data"] = agent_fn(payload)        # your agent logic
        envelope["ok"] = True
        code = 0
    except NeedsReview as e:
        envelope["error"] = {"type": "needs_review", "message": str(e)}
        code = NeedsReview.code
    except BadOutput as e:
        envelope["error"] = {"type": "bad_output", "message": str(e)}
        code = BadOutput.code
    except Exception as e:                            # transient / unexpected
        envelope["error"] = {"type": type(e).__name__, "message": str(e)}
        code = 1
    envelope["meta"] = {
        "exit_code": code,
        "latency_ms": int((time.time() - started) * 1000),
    }
    print(json.dumps(envelope))                       # stdout: the one true line
    return code

# --- usage ---
def my_agent(payload):
    verdict = {"severity": "high"}                    # ... real work here ...
    if verdict["severity"] == "critical":
        raise NeedsReview("critical alerts are never auto-actioned")
    return verdict

if __name__ == "__main__":
    raw = sys.stdin.read() or "{}"
    sys.exit(run_headless(my_agent, json.loads(raw)))

# /etc/cron.d/triage  — run the triage agent every day at 06:00
# m h dom mon dow   user      command
0 6 * * *   deploy   /opt/agents/run_triage.sh >> /var/log/triage.log 2>&1

# ----- /opt/agents/run_triage.sh -----
#!/usr/bin/env bash
set -euo pipefail
export PATH=/usr/local/bin:/usr/bin:/bin        # cron's PATH is minimal — be explicit

cd /opt/agents
# Feed the day's alerts in on stdin; capture stdout (JSON) and the exit code
OUT="$(cat alerts.txt | python3 triage_agent.py)" && CODE=0 || CODE=$?

case "$CODE" in
  0) echo "$OUT" | jq -c . >> results.jsonl ;;            # success: append result
  2) echo "BAD OUTPUT, escalating" | mail -s triage oncall@co ;;  # never retry
  3) echo "$OUT" >> review_queue.jsonl ;;                 # needs a human
  *) echo "transient failure (code $CODE), will retry next run" ;;  # 1 etc.
esac
exit 0    # the wrapper itself always succeeds; cron should not "fail"

# .github/workflows/agent-review.yml
name: Headless agent review
on: [push]

jobs:
  review:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Install Ollama
        run: curl -fsSL https://ollama.ai/install.sh | sh

      - name: Start Ollama and pull the model
        run: |
          ollama serve &                 # background the server
          sleep 5
          ollama pull mistral            # warm the model before the agent runs

      - name: Install deps
        run: pip install openai

      - name: Run headless agent (exit code gates the build)
        run: |
          # The agent reads the changelog and emits a JSON verdict.
          # If it exits non-zero, this step fails and the build goes red.
          git log -1 --pretty=%B | python triage_agent.py | tee verdict.json
          jq -e '.severity != "critical"' verdict.json   # fail build on critical

      - name: Upload verdict artifact
        if: always()                       # keep the JSON even when the gate fails
        uses: actions/upload-artifact@v4
        with:
          name: agent-verdict
          path: verdict.json

# guarded_loop.py — three hard limits replace the human circuit-breaker
import signal, sys, json
from openai import OpenAI

client = OpenAI(base_url="http://localhost:11434/v1", api_key="ollama")

class GuardTripped(Exception): pass

def _timeout_handler(signum, frame):
    raise GuardTripped("wall-clock timeout exceeded")

def run_guarded(task: str, *, max_seconds=60, max_iters=8, max_tokens=4000) -> dict:
    # GUARD 1: wall-clock timeout (Unix). SIGALRM fires after max_seconds.
    signal.signal(signal.SIGALRM, _timeout_handler)
    signal.alarm(max_seconds)

    messages = [{"role": "user", "content": task}]
    tokens_used = 0
    try:
        for i in range(max_iters):                  # GUARD 2: hard iteration cap
            resp = client.chat.completions.create(model="mistral", messages=messages)
            tokens_used += resp.usage.total_tokens
            print(f"[guard] iter={i} tokens={tokens_used}", file=sys.stderr)

            if tokens_used > max_tokens:             # GUARD 3: token budget
                raise GuardTripped(f"token budget {max_tokens} exceeded")

            msg = resp.choices[0].message
            messages.append({"role": "assistant", "content": msg.content})
            if not msg.tool_calls:                   # loop ends when no tool needed
                return {"result": msg.content, "iters": i + 1, "tokens": tokens_used}
            # ... (execute tool_calls, append role:"tool" replies) ...
        raise GuardTripped(f"max_iters {max_iters} reached without finishing")
    finally:
        signal.alarm(0)                              # always disarm the timer

if __name__ == "__main__":
    try:
        out = run_guarded(sys.stdin.read().strip())
        print(json.dumps({"ok": True, **out}))
        sys.exit(0)
    except GuardTripped as e:
        print(json.dumps({"ok": False, "error": str(e)}), file=sys.stdout)
        print(f"[guard] TRIPPED: {e}", file=sys.stderr)
        sys.exit(1)                                  # deterministic non-zero exit

# 1) Classify one alert and pull a single field with jq
echo "db replica lag 45s on orders-db" | python triage_agent.py 2>/dev/null \
  | jq -r '.severity'
# -> high

# 2) Triage a whole file, keep only the critical ones, count them
cat overnight_alerts.txt \
  | while read -r line; do echo "$line" | python triage_agent.py 2>/dev/null; done \
  | jq -c 'select(.severity=="critical")' \
  | tee critical.jsonl | wc -l
# -> 3

# 3) Chain TWO agents: triage, then for high-sev ones, draft a response
echo "$ALERT" | python triage_agent.py 2>/dev/null \
  | jq -r 'select(.severity=="high") | .summary' \
  | python draft_response_agent.py 2>/dev/null \
  | jq -r '.draft'

# 4) Fan an event from a webhook straight into the agent and post to Slack
curl -s "$EVENT_URL" \
  | python triage_agent.py 2>/dev/null \
  | jq '{text: ("Triage: " + .severity + " — " + .summary)}' \
  | curl -s -X POST -H 'Content-type: application/json' -d @- "$SLACK_WEBHOOK"