manual_slop/scripts/audit_optional_in_3_files.py

"""Audit: enforce the Optional[T] ban in the 4 baseline files.

The 3 refactored files (mcp_client.py, ai_client.py, rag_engine.py)
plus code_path_audit.py are held to the data-oriented error_handling
convention: Optional[T] return types are forbidden (use Result[T] +
NIL_T sentinel instead).

This script AST-scans the baseline files, reports any
Optional[T] return type as a violation, and exits 1 in --strict
mode on any violation.

Usage:
 uv run python scripts/audit_optional_in_3_files.py
 uv run python scripts/audit_optional_in_3_files.py --strict
 uv run python scripts/audit_optional_in_3_files.py --json
"""
from __future__ import annotations
import argparse
import ast
import json
import sys
from pathlib import Path

BASELINE_FILES: tuple[str, ...] = (
 "src/mcp_client.py",
 "src/ai_client.py",
 "src/rag_engine.py",
 "src/code_path_audit.py",
)

def _return_annotation_is_optional(node: ast.FunctionDef | ast.AsyncFunctionDef) -> bool:
 """Check if a function's return annotation is Optional[X]."""
 if node.returns is None:
  return False
 ann = node.returns
 if isinstance(ann, ast.Subscript):
  value = ann.value
  if isinstance(value, ast.Name) and value.id == "Optional":
   return True
  if isinstance(value, ast.Attribute) and value.attr == "Optional":
   return True
 return False

def _annotation_is_optional_arg(ann: ast.expr | None) -> bool:
 """Check if an annotation is Optional[X] (used for parameters)."""
 if ann is None:
  return False
 if isinstance(ann, ast.Subscript):
  value = ann.value
  if isinstance(value, ast.Name) and value.id == "Optional":
   return True
  if isinstance(value, ast.Attribute) and value.attr == "Optional":
   return True
 return False

def audit_file(filepath: Path) -> list[dict]:
 """Audit one file: scan function signatures for Optional[X] usage.

 Reports:
 - function return type Optional[X]
 - function parameter Optional[X] (warning, not strict violation)
 - variable annotation Optional[X] (info only)
 """
 if not filepath.exists():
  return [{"file": str(filepath), "line": 0, "function": "", "kind": "MISSING_FILE", "note": "file not found"}]
 try:
  source = filepath.read_text(encoding="utf-8")
 except (OSError, UnicodeDecodeError) as e:
  return [{"file": str(filepath), "line": 0, "function": "", "kind": "READ_ERROR", "note": str(e)}]
 try:
  tree = ast.parse(source)
 except SyntaxError as e:
  return [{"file": str(filepath), "line": e.lineno or 0, "function": "", "kind": "SYNTAX_ERROR", "note": str(e)}]
 findings: list[dict] = []
 for node in ast.walk(tree):
  if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
   if _return_annotation_is_optional(node):
    findings.append({
     "file": str(filepath),
     "line": node.lineno,
     "function": node.name,
     "kind": "RETURN_OPTIONAL",
     "note": "Optional[X] return type forbidden; use Result[X] + NIL_X sentinel",
    })
   for arg in node.args.args + node.args.kwonlyargs + node.args.posonlyargs:
    if _annotation_is_optional_arg(arg.annotation):
     findings.append({
      "file": str(filepath),
      "line": arg.lineno or node.lineno,
      "function": node.name,
      "kind": "PARAM_OPTIONAL",
      "note": f"parameter '{arg.arg}' typed Optional[X]",
     })
 return findings

def main() -> int:
 parser = argparse.ArgumentParser(description="Audit the 4 baseline files for the Optional[T] ban.")
 parser.add_argument("--strict", action="store_true", help="Exit 1 on any Optional[X] return type")
 parser.add_argument("--json", action="store_true", help="Output JSON")
 args = parser.parse_args()
 all_findings: list[dict] = []
 for rel in BASELINE_FILES:
  filepath = Path(rel)
  findings = audit_file(filepath)
  all_findings.extend(findings)
 if args.json:
  out = {
   "files_scanned": len(BASELINE_FILES),
   "files_with_findings": len({f["file"] for f in all_findings}),
   "total_findings": len(all_findings),
   "by_kind": {
    "RETURN_OPTIONAL": sum(1 for f in all_findings if f["kind"] == "RETURN_OPTIONAL"),
    "PARAM_OPTIONAL": sum(1 for f in all_findings if f["kind"] == "PARAM_OPTIONAL"),
   },
   "findings": all_findings,
  }
  print(json.dumps(out, indent=2))
  return 0
 return_violations = [f for f in all_findings if f["kind"] == "RETURN_OPTIONAL"]
 param_violations = [f for f in all_findings if f["kind"] == "PARAM_OPTIONAL"]
 print(f"Optional[T] audit: {len(all_findings)} total findings")
 print(f" - {len(return_violations)} return-type Optional[T] (strict violation)")
 print(f" - {len(param_violations)} parameter Optional[T] (warning)")
 for f in return_violations:
  print(f" VIOLATION: {f['file']}:{f['line']} {f['function']}() -> Optional[...]")
 if args.strict and return_violations:
  print(f"STRICT: {len(return_violations)} return-type Optional[T] violations")
  return 1
 return 0

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