M23: Agentic Coding Workflows & Quality Gates

---
name: security-review
description: Review a code diff for security and safety problems
model: mistral
output: json
---

You are a careful security reviewer. Review the DIFF below and report problems.

Look specifically for:
- Hardcoded secrets (passwords, API keys, tokens)
- Use of eval/exec or shelling out with untrusted input
- Swallowed exceptions (except: pass) that hide failures
- SQL or command strings built by concatenation

Respond with ONLY a JSON object, no prose:
{"issues": [{"severity": "low|medium|high", "line": int, "problem": str, "fix": str}],
 "approved": bool}

DIFF:
{{diff}}

# run_workflow.py — load a workflow file, fill params, run it on local Mistral
# WHAT: A generic executor for reusable workflow files
# WHY:  The procedure is data (a .md file); this code just drives it
# GOTCHA: Frontmatter is between the first two '---' lines; parse defensively
import sys, json, re
from openai import OpenAI

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

def load_workflow(path: str) -> tuple[dict, str]:
    text = open(path, encoding="utf-8").read()
    m = re.match(r"^---\n(.*?)\n---\n(.*)$", text, re.DOTALL)
    if not m:
        raise ValueError(f"{path}: missing --- frontmatter --- block")
    meta_block, body = m.group(1), m.group(2)
    # Tiny YAML-ish parser: "key: value" per line (good enough for flat metadata)
    meta = {}
    for line in meta_block.splitlines():
        if ":" in line:
            k, v = line.split(":", 1)
            meta[k.strip()] = v.strip()
    return meta, body

def run_workflow(path: str, params: dict) -> dict:
    meta, body = load_workflow(path)
    # Fill {{name}} slots; fail loudly if a required slot was left unfilled
    prompt = re.sub(r"\{\{(\w+)\}\}", lambda mm: str(params[mm.group(1)]), body)
    resp = client.chat.completions.create(
        model=meta.get("model", "mistral"),
        messages=[{"role": "user", "content": prompt}],
        temperature=0,
    )
    raw = resp.choices[0].message.content.strip()
    raw = raw.removeprefix("```json").removeprefix("```").removesuffix("```").strip()
    if meta.get("output") == "json":
        return json.loads(raw)          # raises if the model didn't honor the contract
    return {"text": raw}

if __name__ == "__main__":
    diff = sys.stdin.read()             # pipe a diff in:  git diff | python run_workflow.py
    try:
        result = run_workflow("workflows/review.md", {"diff": diff})
        print(json.dumps(result, indent=2))
        sys.exit(0 if result.get("approved") else 1)   # exit code = the verdict
    except json.JSONDecodeError as e:
        print(f"workflow output was not valid JSON: {e}", file=sys.stderr)
        sys.exit(2)
    except KeyError as e:
        print(f"workflow needs a parameter that wasn't supplied: {e}", file=sys.stderr)
        sys.exit(2)

# build_from_spec.py — a tiny spec-driven coding framework on local Mistral
# WHAT: Generate code to satisfy a spec, run its tests, feed failures back, repeat
# WHY:  The test suite is the objective grader; the loop iterates until it's green
# GOTCHA: A hard max_iters cap is mandatory — the model may never satisfy the spec
import subprocess, sys, re
from openai import OpenAI

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

SPEC = """Write a Python function `slugify(text: str) -> str` that lowercases text,
replaces any run of non-alphanumeric characters with a single hyphen, and strips
leading/trailing hyphens. slugify('Hello,  World!') must return 'hello-world'."""

TESTS = '''import pytest
from solution import slugify
def test_basic(): assert slugify("Hello,  World!") == "hello-world"
def test_trim():  assert slugify("--Already--Slugged--") == "already-slugged"
def test_empty(): assert slugify("") == ""
'''

def extract_code(reply: str) -> str:
    # Pull the first fenced code block; fall back to the whole reply
    m = re.search(r"```(?:python)?\n(.*?)```", reply, re.DOTALL)
    return (m.group(1) if m else reply).strip()

def run_tests() -> tuple[bool, str]:
    proc = subprocess.run([sys.executable, "-m", "pytest", "-q", "test_solution.py"],
                          capture_output=True, text=True, timeout=60)
    return proc.returncode == 0, proc.stdout + proc.stderr

def build(max_iters: int = 4) -> bool:
    open("test_solution.py", "w").write(TESTS)
    messages = [{"role": "system", "content": "You are a precise Python engineer. Return only code."},
                {"role": "user", "content": SPEC}]
    for i in range(max_iters):                       # the mandatory iteration cap
        reply = client.chat.completions.create(model="mistral", messages=messages,
                                               temperature=0).choices[0].message.content
        open("solution.py", "w").write(extract_code(reply))
        passed, output = run_tests()                 # the OBJECTIVE grader
        print(f"[iter {i}] tests {'PASSED' if passed else 'FAILED'}", file=sys.stderr)
        if passed:
            return True
        # Feed the real failure back so the next attempt is informed, not random
        messages += [{"role": "assistant", "content": reply},
                     {"role": "user", "content": f"The tests failed:\n{output}\nFix the code."}]
    return False

if __name__ == "__main__":
    sys.exit(0 if build() else 1)

# rules/no-swallowed-except.yml  —  run:  ast-grep scan -r rules/no-swallowed-except.yml .
id: no-swallowed-except
language: python
severity: error
message: "Swallowed exception: handle the error or re-raise, never 'except: pass'."
rule:
  kind: try_statement
  has:
    kind: except_clause
    has:
      # the body of the except is a single 'pass' statement
      kind: block
      has:
        kind: pass_statement
      not:
        has:
          kind: expression_statement   # ...and nothing else in the block

# .pre-commit-config.yaml — the gate every commit must pass
repos:
  # Layer 1: cheap hygiene — whitespace, file size, valid YAML/JSON
  - repo: https://github.com/pre-commit/pre-commit-hooks
    rev: v4.6.0
    hooks:
      - id: trailing-whitespace
      - id: end-of-file-fixer
      - id: check-added-large-files     # block a 200MB blob from an agent gone wrong
      - id: check-yaml

  # Layer 2: ruff — lint + format, auto-fixing what it safely can
  - repo: https://github.com/astral-sh/ruff-pre-commit
    rev: v0.6.9
    hooks:
      - id: ruff
        args: [--fix]
      - id: ruff-format

  # Layer 3: bandit — security scan, fails the commit on a finding
  - repo: https://github.com/PyCQA/bandit
    rev: 1.7.10
    hooks:
      - id: bandit
        args: ["-q"]

  # Layer 4: your structural policy via ast-grep (local hook)
  - repo: local
    hooks:
      - id: ast-grep-policy
        name: ast-grep structural rules
        entry: ast-grep scan -r rules/
        language: system
        types: [python]

# commit_with_fixups.py — let the gate teach the agent until the commit is clean
# WHAT: Try to commit; if the gate blocks it, send the error back to Mistral, fix, retry
# WHY:  Turns "write good code" into a concrete loop graded by deterministic hooks
# GOTCHA: Bound the retries; a gate the model can't satisfy must fail loudly, not spin
import subprocess, sys
from openai import OpenAI

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

def run_gate() -> tuple[bool, str]:
    # Run the full pre-commit gate on staged files; non-zero = blocked
    proc = subprocess.run(["pre-commit", "run", "--files", TARGET],
                          capture_output=True, text=True, timeout=120)
    return proc.returncode == 0, proc.stdout + proc.stderr

def ask_fix(code: str, gate_output: str) -> str:
    resp = client.chat.completions.create(
        model="mistral", temperature=0,
        messages=[
            {"role": "system", "content": "You are a Python engineer. Return only the corrected file."},
            {"role": "user", "content":
                f"This file was rejected by our commit gate.\n\nFILE ({TARGET}):\n{code}\n\n"
                f"GATE OUTPUT:\n{gate_output}\n\nReturn the full corrected file."},
        ])
    reply = resp.choices[0].message.content.strip()
    return reply.removeprefix("```python").removeprefix("```").removesuffix("```").strip()

def commit_clean(max_fixups: int = 3) -> int:
    for attempt in range(max_fixups + 1):
        subprocess.run(["git", "add", TARGET], check=True)
        passed, output = run_gate()
        if passed:
            subprocess.run(["git", "commit", "-m", "agent change (gate-clean)"], check=True)
            print(f"[gate] passed on attempt {attempt}; commit landed", file=sys.stderr)
            return 0
        print(f"[gate] blocked on attempt {attempt}:\n{output}", file=sys.stderr)
        if attempt == max_fixups:
            break
        fixed = ask_fix(open(TARGET, encoding="utf-8").read(), output)
        open(TARGET, "w", encoding="utf-8").write(fixed)   # the agent's correction
    print("[gate] still failing after retries — escalate to a human", file=sys.stderr)
    return 2                                               # bad output: do not auto-merge

if __name__ == "__main__":
    sys.exit(commit_clean())

# .github/workflows/gate.yml — the same checks, re-run where they can't be skipped
name: Quality gate
on: [push, pull_request]

# Agents push many small commits; cancel superseded runs to save CI minutes.
concurrency:
  group: gate-${{ github.ref }}
  cancel-in-progress: true

jobs:
  gate:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: "3.12" }

      - name: Install the gate
        run: pip install pre-commit

      # Re-run the IDENTICAL hooks from .pre-commit-config.yaml on a clean machine.
      # A non-zero exit fails the job and blocks the merge on a protected branch.
      - name: Run pre-commit on all files
        run: pre-commit run --all-files --show-diff-on-failure

mkdir gate-lab && cd gate-lab && git init
pip install openai ruff bandit pre-commit
# ast-grep:  brew install ast-grep   (or)   cargo install ast-grep   (or)   npm i -g @ast-grep/cli
ollama pull mistral            # the local model the agent uses
mkdir -p rules workflows