Lab 16 · Production readiness review

"""Lab 16: a static production-readiness review of an agent project.

A static reviewer only sees structure: it proves a checkpointer is *configured*, never that it
is *correct*. Score the rest of the checklist by hand.

Run:  python readiness_review.py [project_dir] [--out gap-report.md]
"""

import argparse
import re
import sys
from pathlib import Path
from typing import Callable, NamedTuple

SKIP_DIRS = {".git", ".venv", "venv", "__pycache__", "node_modules", ".pytest_cache", ".ruff_cache"}
READABLE = {".py", ".json", ".toml", ".yml", ".yaml", ".txt", ".md", ".cfg", ".example"}
EXTRA_FILES = {"Dockerfile", ".env", ".gitignore"}

# Every marker below is searched as a regular expression, so brackets are escaped.
LEGACY_APIS = ("AgentExecutor", "initialize_agent", "MemorySaver", "LLMChain", "RetrievalQA",
               "ConversationBufferMemory", "MultiServerMCPClient")
DURABLE_SAVERS = ("SqliteSaver", "PostgresSaver", "AsyncPostgresSaver", "AsyncSqliteSaver")
DESTRUCTIVE_VERBS = ("refund", "delete", "cancel", "charge", "transfer", "revoke", "purge",
                     "terminate", "issue", "pay")
HITL_MARKERS = ("HumanInTheLoopMiddleware", "interrupt_on", r"interrupt\(")
EVAL_MARKERS = (r"client\.evaluate\(", r"aevaluate\(", "create_llm_as_judge", "create_dataset")
TRACING_MARKERS = ("LANGSMITH_TRACING", "LANGSMITH_API_KEY", "enable_tracing")
SECRET_PATTERNS = (
    (r"sk-[A-Za-z0-9]{20,}", "OpenAI-style key"),
    (r"lsv2_(?:pt|sk)_[A-Za-z0-9]{16,}", "LangSmith key"),
    (r"AKIA[0-9A-Z]{16}", "AWS access key id"),
    (r"(?i)(api_key|password|secret|token)\s*=\s*[\"'][^\"'\s$]{12,}[\"']", "hard-coded literal"),
)
TOOL_DEF = re.compile(r"@tool[^\n]*\n(?:\s*@[^\n]*\n)*\s*(?:async\s+)?def\s+(\w+)")
POINTS = {"pass": 1.0, "warn": 0.5, "fail": 0.0}

def load_files(root: Path) -> dict[str, str]:
    """Every text file in the project, keyed by its path relative to the root."""
    files: dict[str, str] = {}
    for path in sorted(root.rglob("*")):
        relative = path.relative_to(root)
        if not path.is_file() or SKIP_DIRS & set(relative.parts):
            continue
        if path.suffix in READABLE or path.name in EXTRA_FILES:
            files[relative.as_posix()] = path.read_text(encoding="utf-8", errors="ignore")
    return files

def line_of(text: str, match: re.Match[str]) -> int:
    """1-based line number of a regex match."""
    return text[: match.start()].count("\n") + 1


def find(files: dict[str, str], pattern: str, *, app_only: bool = False) -> str:
    """'path:line' of the first regex match, or ''. app_only skips tests and non-Python files."""
    for path, text in files.items():
        if app_only and (not path.endswith(".py") or "test" in path.lower()):
            continue
        if match := re.search(pattern, text):
            return f"{path}:{line_of(text, match)}"
    return ""


def exists(files: dict[str, str], *patterns: str) -> str:
    """The first file whose path matches one of the glob patterns, or ''."""
    return next((p for pattern in patterns for p in files if Path(p).match(pattern)), "")


def tool_names(files: dict[str, str]) -> set[str]:
    """Every @tool-decorated function name in the project, lowercased."""
    return {m.group(1).lower() for text in files.values() for m in TOOL_DEF.finditer(text)}

def check_legacy_apis(files: dict[str, str]) -> tuple[str, str]:
    """ARCH-01: names from LangChain 0.x that must not survive the 1.x migration."""
    hits = [f"{api} ({where})" for api in LEGACY_APIS
            if (where := find(files, rf"(?<![A-Za-z0-9_]){api}"))]
    return ("fail", "; ".join(hits)) if hits else ("pass", "no legacy 0.x APIs found")


def check_checkpointer(files: dict[str, str]) -> tuple[str, str]:
    """ARCH-04: the most common gap is in-memory state in something that gets deployed."""
    if durable := next((s for s in DURABLE_SAVERS if find(files, s, app_only=True)), ""):
        return "pass", f"{durable} at {find(files, durable, app_only=True)}"
    in_memory = find(files, "InMemorySaver", app_only=True)
    if in_memory and exists(files, "langgraph.json", "**/langgraph.json", "Dockerfile"):
        return "fail", f"InMemorySaver in deployable code ({in_memory}); state is lost on restart"
    if in_memory:
        return "warn", f"only InMemorySaver ({in_memory}); fine locally, not in a deployment"
    return "fail", "no checkpointer at all: no threads, no resume, no HITL after a restart"

def check_secrets(files: dict[str, str]) -> tuple[str, str]:
    """SAFE-01: literal credentials in the repository, or a committed .env."""
    hits = []
    for path, text in files.items():
        if path.endswith((".example", ".md")):
            continue
        for pattern, label in SECRET_PATTERNS:
            if match := re.search(pattern, text):
                hits.append(f"{label} at {path}:{line_of(text, match)}")
    if ".env" in files and ".env" not in files.get(".gitignore", ""):
        hits.append("a committed .env that .gitignore does not exclude")
    return ("fail", "; ".join(hits)) if hits else ("pass", "no literal secrets found")


def check_hitl(files: dict[str, str]) -> tuple[str, str]:
    """SAFE-03: a destructive-sounding tool that can fire without a human."""
    risky = sorted(t for t in tool_names(files) if any(v in t for v in DESTRUCTIVE_VERBS))
    if not risky:
        return "n/a", "no destructive-sounding tools found"
    if where := next((w for m in HITL_MARKERS if (w := find(files, m))), ""):
        return "pass", f"approval gate at {where}, but check it covers {', '.join(risky)}"
    return "fail", f"destructive tools with no approval gate: {', '.join(risky)}"

def check_tests(files: dict[str, str]) -> tuple[str, str]:
    """QUAL-01: without tests, nothing stops a regression."""
    tests = [path for path in files if Path(path).name.startswith("test_")]
    if tests:
        return "pass", f"{len(tests)} test file(s), e.g. {tests[0]}"
    return "fail", "no test_*.py anywhere"


def check_evals(files: dict[str, str]) -> tuple[str, str]:
    """QUAL-03: unit tests catch API breaks; only evals catch behaviour changes."""
    where = next((w for m in EVAL_MARKERS if (w := find(files, m))), "")
    if where:
        return "pass", f"evaluation code at {where}"
    return "fail", "no evaluation suite: quality changes will be invisible"


def check_tracing(files: dict[str, str]) -> tuple[str, str]:
    """OBS-01: an incident with no trace is an incident with no evidence."""
    where = next((w for m in TRACING_MARKERS if (w := find(files, m))), "")
    return ("pass", f"tracing configured at {where}") if where else ("fail", "no tracing configuration")


def check_deployable(files: dict[str, str]) -> tuple[str, str]:
    """DEP-01: something a deployment target can actually build."""
    if path := exists(files, "langgraph.json", "**/langgraph.json"):
        return "pass", path
    if path := exists(files, "Dockerfile", "**/Dockerfile"):
        return "warn", f"{path} but no langgraph.json"
    return "fail", "no langgraph.json and no Dockerfile: nothing to deploy"

def check_pinned_deps(files: dict[str, str]) -> tuple[str, str]:
    """DEP-02 (your own check): an unpinned dependency makes every build a different build."""
    req_files = [p for p in files if Path(p).name.startswith("requirements") and p.endswith(".txt")]
    if not req_files:
        return "n/a", "no requirements*.txt (pins in pyproject.toml are not checked)"
    loose = []
    for path in req_files:
        for number, line in enumerate(files[path].splitlines(), 1):
            line = line.split("#")[0].strip()
            if line and not line.startswith("-") and "==" not in line:
                loose.append(f"{line} ({path}:{number})")
    if loose:
        return "warn", f"not pinned to an exact version: {', '.join(loose)}"
    return "pass", f"every dependency pinned in {', '.join(req_files)}"

class Finding(NamedTuple):
    """One check's verdict, with evidence the person who has to fix it can act on."""

    id: str
    title: str
    critical: bool
    status: str  # pass | warn | fail | n/a
    evidence: str


# id, title, critical (one failed critical check blocks the release), detector
CHECKS: list[tuple[str, str, bool, Callable[[dict[str, str]], tuple[str, str]]]] = [
    ("ARCH-01", "No legacy 0.x LangChain APIs", False, check_legacy_apis),
    ("ARCH-04", "Durable checkpointer in deployable code", True, check_checkpointer),
    ("SAFE-01", "No secrets in the repository", True, check_secrets),
    ("SAFE-03", "Destructive tools are gated by a human", True, check_hitl),
    ("QUAL-01", "Offline test suite exists", False, check_tests),
    ("QUAL-03", "Evaluation suite exists", False, check_evals),
    ("OBS-01", "Tracing is configured", True, check_tracing),
    ("DEP-01", "The project is deployable", False, check_deployable),
    ("DEP-02", "Dependencies are pinned", False, check_pinned_deps),
]

def review(project_dir: Path) -> list[Finding]:
    """Run every static check against a project directory."""
    files = load_files(Path(project_dir).resolve())
    return [Finding(i, title, critical, *detect(files)) for i, title, critical, detect in CHECKS]


def score(findings: list[Finding]) -> tuple[float, list[Finding], str]:
    """Score, blocking findings and verdict. One failed critical check blocks the release."""
    scored = [f for f in findings if f.status != "n/a"]
    total = round(sum(POINTS[f.status] for f in scored) / len(scored), 4) if scored else 0.0
    blocking = [f for f in findings if f.critical and f.status == "fail"]
    if blocking or total < 0.75:
        return total, blocking, "NOT READY"
    return total, blocking, "READY" if total >= 0.90 else "READY WITH CAVEATS"

def render_markdown(project: str, findings: list[Finding]) -> str:
    """The deliverable: a gap report someone can paste into a review ticket."""
    total, blocking, verdict = score(findings)
    lines = [f"# Production readiness gap report: `{project}`", "",
             f"Score **{total:.2f}** - **{verdict}**", "",
             "| ID | Check | Status | Evidence |", "|---|---|---|---|"]
    for f in findings:
        critical = " **(critical)**" if f.critical else ""
        lines.append(f"| {f.id} | {f.title}{critical} | {f.status.upper()} | {f.evidence} |")
    if blocking:
        lines += ["", "## Blocking gaps (fix before release)", ""]
        lines += [f"1. **{f.id}** {f.title} - {f.evidence}" for f in blocking]
    lines += ["", "Static checks cannot see least privilege, the threat model, the SLOs or the",
              "runbook. Score those by hand with `checklists/production-readiness.md`."]
    return "\n".join(lines) + "\n"

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Static production-readiness review.")
    parser.add_argument("project", nargs="?", default="sample_project", help="folder to review")
    parser.add_argument("--out", help="also write the report to this Markdown file")
    args = parser.parse_args()

    project = Path(args.project).resolve()
    if not project.is_dir():
        sys.exit(f"Not a folder: {project}. Unzip sample_project.zip here or pass a path.")
    report = render_markdown(project.name, review(project))
    print(report)
    if args.out:
        Path(args.out).write_text(report, encoding="utf-8")
        print(f"Report written to {args.out}")

"""Lab 16: score a filled-in production-readiness checklist.

Run:  python checklist_score.py [checklist.md]
"""

import re
import sys
from pathlib import Path

DEFAULT_CHECKLIST = Path(__file__).resolve().parent / "checklists" / "production-readiness.md"

# "- [x] **ARCH-01** · statement". A "-" or "/" mark means "not applicable here".
ITEM_RE = re.compile(r"^- \[([ xX~/\-])\] \*\*([A-Z]{3,4}-\d{1,2})\*\* ·? ?(.+)$", re.MULTILINE)
POINTS = {"x": 1.0, "X": 1.0, "~": 0.5, " ": 0.0}

def parse_checklist(text: str) -> list[dict[str, object]]:
    """One dict per applicable checklist line; prose, legends and tables are ignored."""
    return [
        {"id": item_id, "points": POINTS[mark], "critical": "(critical)" in statement}
        for mark, item_id, statement in ITEM_RE.findall(text)
        if mark in POINTS
    ]

def score_checklist(items: list[dict[str, object]]) -> tuple[float, list[str]]:
    """The overall score, plus the ids of critical items that are not fully done."""
    if not items:
        return 0.0, []
    total = sum(float(item["points"]) for item in items) / len(items)
    return round(total, 4), [str(i["id"]) for i in items if i["critical"] and float(i["points"]) < 1.0]


def verdict(score: float, blocking: list[str]) -> str:
    """One unchecked critical item blocks the release, whatever the total says."""
    if blocking or score < 0.50:
        return "NOT READY"
    if score >= 0.90:
        return "READY"
    return "READY WITH CAVEATS" if score >= 0.75 else "NOT READY"

if __name__ == "__main__":
    checklist = Path(sys.argv[1]) if len(sys.argv) > 1 else DEFAULT_CHECKLIST
    if not checklist.is_file():
        sys.exit(f"Checklist not found: {checklist}")
    items = parse_checklist(checklist.read_text(encoding="utf-8"))
    score, blocking = score_checklist(items)
    print(f"{checklist.name}: {len(items)} applicable items, score {score:.2f} - {verdict(score, blocking)}")
    print(f"blocking critical items ({len(blocking)}): {', '.join(blocking) or 'none'}")

# Production readiness gap report: `sample_project`

Score **0.50** - **NOT READY**

| ID | Check | Status | Evidence |
|---|---|---|---|
| ARCH-01 | No legacy 0.x LangChain APIs | PASS | no legacy 0.x APIs found |
| ARCH-04 | Durable checkpointer in deployable code **(critical)** | FAIL | InMemorySaver in deployable code (app/agent.py:5); state is lost on restart |
| SAFE-01 | No secrets in the repository **(critical)** | FAIL | OpenAI-style key at app/config.py:4; hard-coded literal at app/config.py:4 |
| SAFE-03 | Destructive tools are gated by a human **(critical)** | FAIL | destructive tools with no approval gate: issue_refund |
| QUAL-01 | Offline test suite exists | PASS | 1 test file(s), e.g. tests/test_tools.py |
| QUAL-03 | Evaluation suite exists | FAIL | no evaluation suite: quality changes will be invisible |
| OBS-01 | Tracing is configured **(critical)** | PASS | tracing configured at .env.example:3 |
| DEP-01 | The project is deployable | PASS | langgraph.json |
| DEP-02 | Dependencies are pinned | WARN | not pinned to an exact version: langchain>=1.0 (requirements.txt:1), langgraph (requirements.txt:3) |

## Blocking gaps (fix before release)

1. **ARCH-04** Durable checkpointer in deployable code - InMemorySaver in deployable code (app/agent.py:5); state is lost on restart
1. **SAFE-01** No secrets in the repository - OpenAI-style key at app/config.py:4; hard-coded literal at app/config.py:4
1. **SAFE-03** Destructive tools are gated by a human - destructive tools with no approval gate: issue_refund

Static checks cannot see least privilege, the threat model, the SLOs or the
runbook. Score those by hand with `checklists/production-readiness.md`.

production-readiness.md: 60 applicable items, score 0.00 - NOT READY
blocking critical items (14): ARCH-03, ARCH-04, ARCH-09, QUAL-07, SAFE-01, SAFE-03, SAFE-04, SAFE-07, SAFE-09, OBS-06, DEP-05, DEP-07, OPS-01, OPS-05