Codex 智能体工程实践：上下文收集、沙箱执行与变更审查

from dataclasses import dataclass
from pathlib import Path

IGNORE_DIRS = {".git", "node_modules", "__pycache__", ".venv",
               "dist", "build", ".next"}
ALLOW_EXT = {".py", ".ts", ".tsx", ".go", ".java",
             ".md", ".toml", ".yaml", ".yml"}

@dataclass
class FileMeta:
    path: str
    tokens: int

def scan_repo(root: Path, max_files: int = 3000) -> list[FileMeta]:
    files = []
    for p in root.rglob("*"):
        if not p.is_file() or p.suffix not in ALLOW_EXT:
            continue
        if any(part in IGNORE_DIRS for part in p.parts):
            continue
        text = p.read_text(encoding="utf-8", errors="ignore")
        files.append(FileMeta(
            str(p.relative_to(root)), len(text) // 4))
        if len(files) >= max_files:
            break
    return files

def select_context(files: list[FileMeta], keywords: list[str],
                   budget: int = 60000) -> list[str]:
    scored = []
    for f in files:
        score = sum(1 for k in keywords if k.lower() in f.path.lower())
        scored.append((score, f))
    scored.sort(key=lambda x: (-x[0], x[1].tokens))

    chosen, used = [], 0
    for _, f in scored:
        if used + f.tokens > budget:
            continue
        chosen.append(f.path)
        used += f.tokens
    return chosen

import subprocess
from pathlib import Path

ALLOWED_CMDS = {"ruff", "mypy", "pytest", "npm", "go", "cargo"}
FORBIDDEN_ARGS = ("rm", "-rf", "sudo", "curl", "wget", "chmod")

def safe_run(cmd: list[str], cwd: Path,
             timeout: int = 120) -> dict:
    if not cmd or cmd[0] not in ALLOWED_CMDS:
        return {"code": -1, "err": f"命令不在白名单：{cmd[:1]}"}

    joined = " ".join(cmd).lower()
    if any(f in joined for f in FORBIDDEN_ARGS):
        return {"code": -1, "err": "命中禁止参数"}

    try:
        p = subprocess.run(
            cmd, cwd=cwd, capture_output=True, text=True,
            timeout=timeout, shell=False,
        )
        return {"code": p.returncode,
                "out": p.stdout[-4000:],
                "err": p.stderr[-2000:]}
    except subprocess.TimeoutExpired:
        return {"code": -1, "err": "timeout"}

import difflib
from pathlib import Path

def make_diff(old: str, new: str, path: str) -> str:
    return "".join(difflib.unified_diff(
        old.splitlines(keepends=True),
        new.splitlines(keepends=True),
        fromfile=f"a/{path}", tofile=f"b/{path}",
    ))

def summarize(diff: str) -> dict:
    added = sum(1 for l in diff.splitlines()
                if l.startswith("+") and not l.startswith("+++"))
    removed = sum(1 for l in diff.splitlines()
                  if l.startswith("-") and not l.startswith("---"))
    files = diff.count("--- ")
    return {"added": added, "removed": removed, "files": files}

RISKY_PATTERNS = [
    ("eval(", "疑似动态执行"),
    ("shell=True", "疑似 shell 注入"),
    ("verify=False", "疑似关闭证书校验"),
    ("# type: ignore", "疑似绕过类型检查"),
    ("except: pass", "疑似吞异常"),
]

def risk_scan(diff: str) -> list[str]:
    flags = []
    for line in diff.splitlines():
        if not line.startswith("+"):
            continue
        for pattern, desc in RISKY_PATTERNS:
            if pattern in line:
                flags.append(f"{desc}: {line.strip()[:60]}")
    return flags

name: ci
on: [push, pull_request]
jobs:
  quality:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: "3.12" }
      - run: pip install ruff mypy pytest bandit pip-audit
      - run: ruff check .
      - run: mypy .
      - run: pytest -q
      - run: bandit -r app
      - run: pip-audit

import time, logging

log = logging.getLogger("codex-agent")

def agent_loop(plan: list[dict], repo: Path,
               apply_patch, max_rounds: int = 5) -> list[dict]:
    for rnd in range(max_rounds):
        pending = [s for s in plan if s["status"] != "done"]
        if not pending:
            break

        for step in pending:
            start = time.time()
            patch = apply_patch(step)          # 模型生成补丁
            checks = {
                "lint": safe_run(["ruff", "check", "."], repo),
                "type": safe_run(["mypy", "."], repo),
                "test": safe_run(["pytest", "-q"], repo),
            }
            ok = all(c["code"] == 0 for c in checks.values())

            if ok:
                step["status"] = "done"
            else:
                step["status"] = "retry"
                failed = [k for k, v in checks.items() if v["code"] != 0]
                step["goal"] += f"\n上次失败：{failed}"

            log.info("round=%d step=%s ok=%s cost=%.2fs",
                     rnd, step.get("id"), ok, time.time() - start)

    return plan