#!/usr/bin/env python3 """LLM audit for semantic string-routing debt and scenario-dependent prompts. This is an explicit, non-pytest helper script. It scans project Python files excluding tests plus prompt-pack files, sends each whole file to Gemini or OpenAI, and writes file:line findings for: 1. open-world semantic decisions made by regex / substring / token lists, and 2. scenario-dependent code or prompt pollution. Default output: backend/tests/_audit_outputs/semantic_string_debt// Example: python3 backend/tests/tools/semantic_string_debt_llm_audit.py \ --threads 8 \ --model gemini-3-flash-preview \ --candidate-only \ --prompt-mode latest-dir Dry-run file count only: python3 backend/tests/tools/semantic_string_debt_llm_audit.py --dry-run Resume a previous interrupted run: python3 backend/tests/tools/semantic_string_debt_llm_audit.py \ --output-dir backend/tests/_audit_outputs/semantic_string_debt/20260515_120000 \ --resume Chunk only unusually large files: python3 backend/tests/tools/semantic_string_debt_llm_audit.py \ --chunk-large-files --max-chars 1200000 """ from __future__ import annotations import argparse import ast import concurrent.futures import hashlib import json import os from pathlib import Path import re import socket import subprocess import sys import threading import time from typing import Any, Dict, Iterable, Iterator, List, Optional, Sequence, Tuple import urllib.error import urllib.request REPO_ROOT = Path(__file__).resolve().parents[3] BACKEND_DIR = REPO_ROOT / "backend" PROMPTS_BASE = REPO_ROOT / "prompts" / "_base" DEFAULT_OUTPUT_BASE = BACKEND_DIR / "tests" / "_audit_outputs" / "semantic_string_debt" OPENAI_RESPONSES_URL = "https://api.openai.com/v1/responses" GEMINI_API_URL_TEMPLATE = ( "https://generativelanguage.googleapis.com/v1beta/models/{model}:generateContent?key={api_key}" ) APP_DIR = BACKEND_DIR / "app" PROMPT_EXTENSIONS = {".md", ".json", ".txt", ".yaml", ".yml"} DEFAULT_CANDIDATE_IMPORT_DEPTH = 2 _GEMINI_KEYS: List[str] = [] _GEMINI_KEY_COUNTER = 0 _GEMINI_KEY_LOCK = threading.Lock() SYSTEM_PROMPT = """You are a high-precision code/prompt auditor for TheRoad-I1. Goal: Find only high-confidence, actionable places where open-world scenario/visual meaning is decided by brittle string patterns, or where active prompt text contains concrete scenario pollution. Core rule: - Report a code finding only when the file both (a) defines/uses a string pattern over natural-language scenario or generated prompt/checkpoint text and (b) the match result changes behavior: routing, validation pass/fail, reference attachment, visible-entity membership, image prompt content, prompt mutation, or fail-fast behavior. - Report a prompt finding only when the prompt asks an LLM to classify open-world meaning from a closed phrase/example list, or when it contains concrete scenario-specific names/IDs/places/props/style examples that can bias arbitrary future scenarios. Forbidden targets when they satisfy the core rule: - Regex/search/find, substring checks, lower().in(...), noun/token/keyword/ phrase lists, or blind replace used over scenario text, scene/shot descriptions, LLM checkpoint outputs, generated t2i_prompt text, prompt-card prose, validator/review text, or intermediate analysis outputs. - Prompt-side phrase lists that are treated as semantic classifiers rather than neutral examples. - Blind target/suggestion mutation of generated prompts or checkpoint prose. Allowed targets: - Closed-world syntax validation: machine identifiers with formally defined formats, JSON/schema shape, file paths/extensions, hashes, version strings, exact enum validation, status constants, database column names, and tests/fixtures if present. These are allowed because the pattern validates a known technical format; it is not inferring open-world story/visual meaning. - A schema enum or central SOT enum is not debt by itself. Report it only if another site uses strings/regex to infer that enum from open-world natural language, or if the prompt duplicates/drifts from the canonical enum. - Soft diagnostics are allowed if they do not mutate routing, attach refs, fail/pass semantic validation, or instruct production LLM behavior. - Technical metadata encoded in non-natural-language fields is allowed unless it parses generated natural-language prompt text or drives visual/story routing. Examples: migration text, API provider strings, model aliases, prompt version stamps, prompt_used technical tags such as composite:/state_variant:, DB status filters, and normal prompt section headers. - Do not flag Korean/English labels, section headers, or prompt assembly by themselves. Flag them only if they contain scenario-specific examples or closed-list semantic classifiers. - Do not flag abstract placeholders such as Character A/B, Figure A/B, 캐릭터A/캐릭터B, , , , , [outfit], or similar template variables. They are placeholders, not scenario pollution. - Do not flag generic cinematic/style preferences (photorealistic, cinematic, 35mm, warm/cold lighting) unless they are tied to a specific story, character, culture, location, prop, or function as a semantic classifier. - Do not flag low-level infrastructure, migrations, CRUD APIs, logging, or model provider routing unless the finding directly affects scenario analysis, visual prompt generation, validator/review behavior, reference attachment, or image generation semantics. Evidence standard: - For code, evidence must identify the string pattern and the behavioral use site. If only a constant/list is visible without a behavior-changing use in this file, omit it. - For prompts, evidence must quote the exact instruction/example that is scenario-specific or classifier-like. - Keep JSON validity more important than verbatim quoting. In evidence fields, prefer symbol names, short phrases, and line references. Do not paste long source-code fragments with nested quotes; paraphrase them if needed. - Prefer omission over noisy speculation. A clean file is expected. - If a file has many possible matches, return only the top 5 highest-confidence findings. - Deduplicate by root mechanism. If one mechanism has a constant, helper, and caller in the same file, report one finding that cites the mechanism instead of three separate findings. Severity and category calibration: - Use P0/P1 semantic_string_judgment only when natural-language scenario or generated semantic text is pattern-matched and the result changes behavior. - Use P1 blind_string_mutation for blind replacement of generated prompt or checkpoint prose. - Use P2 schema_or_enum_drift for exact LLM-output categories that are listed only in descriptions but consumed by exact string comparisons. This is schema enforcement drift, not regex debt. - Use P2 schema_or_enum_drift for centralized vocabularies that contain scenario-like words and must be manually synchronized with prompts, unless the same file also infers the value from natural language. Do not call those P1 semantic string judgment. - Use P2 scenario_dependent_prompt for concrete prompt pollution examples that may bias generation but do not directly define a behavior-changing classifier. Calibration examples from this repo: Report these patterns: - Code like `_FACE_CLOSE_UP_PATTERNS` plus `pat.search(prompt)` that decides whether ID/outlook enforcement is exempt or denied. This reads generated prompt prose and changes ID policy. - Code like `_ELEMENT_ID_CLOSE_REGEX` / `_DESCRIPTION_CLOSE_KEYWORDS` plus `return "close"` / `return "full"`. This classifies visual framing from element IDs/descriptions. - Code like Korean gaze/offscreen/body-part lexicons plus regex over shot descriptions that changes visible/offscreen entity membership. - Code like `from the reference` regex plus character/background/object token windows that classifies what a reference phrase points to. - Code like `if target in old_prompt: old_prompt.replace(target, suggestion)` when `target` and `suggestion` are LLM-produced fixes for generated T2I prompt text. This is blind semantic string mutation; report it even when no regex is used. - Prompt text that says an LLM should apply semantic rules from phrase patterns, for example "if this phrase appears, treat the actor as off-camera", "target must be an exact substring to replace", or specific close-up / body-part phrase lists that control ID/reference behavior. - Prompt text like a shot-director visibility rule that gives gaze/offscreen/ blocking/reaction-only phrase patterns and says visible entities must be included/excluded from those patterns. This is prompt-side semantic string routing. - Prompt/schema text like `target must be an exact substring` when production code later applies that target by substring replacement. This is a contract for blind semantic mutation. - Schema descriptions that list values for a field but do not enforce them as a JSON enum, when downstream code/prompt expects exact values. Report as schema_or_enum_drift. This is not because the words are examples; it is because the field is an unenforced string contract. - A field whose name says one meaning but whose values encode another meaning, when downstream code consumes those values. Example pattern: a "gaze target" field also carrying physical state values such as motionless/injured/dead states, and downstream code uses those values to select references or enforce policy. Report the overloaded semantic channel, not just the words. Do NOT report these patterns: - A central enum/SOT constant such as a state/location vocabulary when the code only validates that an LLM-emitted field is one of the allowed values and raises on invalid values. If it is duplicated with prompt text and worth tracking, report at most one P2 schema_or_enum_drift item, not a P1 semantic judgment. - A prompt/schema field that defines a canonical enum, such as a shot framing enum, if the enum itself is the structured SOT. Debt is the downstream natural-language regex that tries to infer the enum, not the enum definition. - A soft diagnostic such as checking whether a label/zone phrase appears in a prompt when it only returns warning data and does not change routing, validation pass/fail, reference attachment, or generated prompt text. - A small structured LLM-output category such as an internal edit type routed by exact value comparison, if the code does not infer that category from natural-language prose. It may be schema design debt, but it is not semantic string-pattern debt for this audit; report only as P2 schema_or_enum_drift when the schema fails to enforce the exact values that code consumes. - Structured payload kind checks such as character/prop/location/background, or schemas requiring arrays of names/IDs for downstream lookup compatibility, are not semantic string-pattern debt by themselves. Report only if code later searches natural-language prose with those names/kinds to infer open-world meaning. - Charset/language/canonical-output guards are not semantic string-pattern debt when they only enforce the format of a structured output field and fail fast. Example: rejecting non-ASCII values in a structured "owned objects" list before normalization is an output-format contract, not an attempt to infer object identity or visual meaning from arbitrary prose. If a later normalizer uses noun lists to decide story/visual meaning, report that later use instead of the charset guard. - A prompt/schema output-shape rule like "always return names/IDs" or "empty array means none" is not debt by itself. It is a contract shape, not a scenario-specific string classifier. - Technical model/provider/API/status/path/version strings, prompt version stamps, database metadata, or internal improvement types unless they parse scenario prose or generated semantic text to make a story/visual decision. - A generic example in a prompt/schema is not enough by itself. Report prompt pollution only when the example is concrete enough to bias arbitrary future scenarios (proper name, project-specific ID, culturally/era-specific trope, specific prop/place/style inserted into a general rule), or when examples are used as a classifier list. - Centralized policy or safety rewrite tables are not automatically in scope. Report them only if they alter scenario/visual meaning in this pipeline, not merely because they contain sensitive or style words. Report only actionable findings in this file chunk. Do not invent problems. If a pattern is allowed closed-world syntax, omit it unless it is clearly being used as open-world semantic judgment. If you are unsure whether a site affects story/visual semantics, omit it. Return line numbers from the provided numbered source. """ RESPONSE_SCHEMA: Dict[str, Any] = { "type": "object", "additionalProperties": False, "required": ["findings", "chunk_summary"], "properties": { "chunk_summary": { "type": "string", "description": "One sentence summary; say no actionable findings if clean.", }, "findings": { "type": "array", "items": { "type": "object", "additionalProperties": False, "required": [ "line_start", "line_end", "category", "severity", "evidence", "why_problematic", "recommended_fix", ], "properties": { "line_start": {"type": "integer", "minimum": 1}, "line_end": {"type": "integer", "minimum": 1}, "category": { "type": "string", "enum": [ "semantic_string_judgment", "scenario_dependent_code", "scenario_dependent_prompt", "llm_closed_list_instruction", "blind_string_mutation", "schema_or_enum_drift", ], }, "severity": { "type": "string", "enum": ["P0", "P1", "P2"], "description": "P0 routing/fail-fast risk, P1 important debt, P2 low-risk cleanup.", }, "evidence": { "type": "string", "description": "Short quote or exact symbol names from the file.", }, "why_problematic": { "type": "string", "description": "Why this is pattern-based semantic judgment or scenario pollution.", }, "recommended_fix": { "type": "string", "description": "Concrete replacement structure or audit action.", }, }, }, }, }, } def _load_env_file(path: Path) -> Dict[str, str]: env: Dict[str, str] = {} if not path.exists(): return env for raw_line in path.read_text(encoding="utf-8", errors="ignore").splitlines(): line = raw_line.strip() if not line or line.startswith("#") or "=" not in line: continue key, value = line.split("=", 1) key = key.strip() value = value.strip().strip('"').strip("'") if key: env[key] = value return env def _api_key() -> str: env = _load_env_file(BACKEND_DIR / ".env") key = os.environ.get("OPENAI_API_KEY") or env.get("OPENAI_API_KEY") or env.get("openai_api_key") if not key: raise SystemExit("OPENAI_API_KEY not found in environment or backend/.env") return key def _gemini_keys() -> List[str]: global _GEMINI_KEYS if _GEMINI_KEYS: return _GEMINI_KEYS env = _load_env_file(BACKEND_DIR / ".env") all_sources = {**env, **dict(os.environ)} keys: List[str] = [] base_key = all_sources.get("GEMINI_API_KEY") if base_key: keys.append(base_key) idx = 1 while True: key = all_sources.get(f"GEMINI_API_KEY{idx}") if not key: break if key not in keys: keys.append(key) idx += 1 if not keys: raise SystemExit("GEMINI_API_KEY not found in environment or backend/.env") _GEMINI_KEYS = keys return _GEMINI_KEYS def _next_gemini_key() -> str: global _GEMINI_KEY_COUNTER keys = _gemini_keys() if len(keys) == 1: return keys[0] with _GEMINI_KEY_LOCK: key = keys[_GEMINI_KEY_COUNTER % len(keys)] _GEMINI_KEY_COUNTER += 1 return key def _version_sort_key(name: str) -> Tuple[int, str]: parts = name.split(".", 1) try: return int(parts[0]), parts[1] if len(parts) > 1 else "" except (ValueError, IndexError): return 0, name def _rg_files(args: Sequence[str]) -> List[Path]: out = subprocess.check_output(["rg", "--files", *args], cwd=REPO_ROOT, text=True) return [REPO_ROOT / line for line in out.splitlines() if line.strip()] def discover_python_files(mode: str) -> List[Path]: if mode == "none": return [] base_args = [ "-g", "*.py", "-g", "!**/tests/**", "-g", "!**/test_*.py", "-g", "!**/*_test.py", "-g", "!**/.venv/**", "-g", "!**/venv/**", "-g", "!**/__pycache__/**", "-g", "!**/node_modules/**", ] if mode == "backend-app": return _rg_files(["backend/app", *base_args]) if mode == "all": return _rg_files(base_args) raise ValueError(f"unknown python mode: {mode}") def discover_prompt_files(mode: str) -> List[Path]: if mode == "none": return [] if not PROMPTS_BASE.exists(): return [] if mode == "all": return sorted( p for p in PROMPTS_BASE.rglob("*") if p.is_file() and p.suffix in PROMPT_EXTENSIONS ) modules = [p for p in PROMPTS_BASE.iterdir() if p.is_dir()] if mode == "latest-dir": files: List[Path] = [] for module_dir in modules: versions = sorted( [d for d in module_dir.iterdir() if d.is_dir()], key=lambda p: _version_sort_key(p.name), reverse=True, ) if not versions: continue files.extend( p for p in versions[0].rglob("*") if p.is_file() and p.suffix in PROMPT_EXTENSIONS ) return sorted(files) if mode == "effective": selected: List[Path] = [] for module_dir in modules: choices: Dict[Tuple[str, str, str], Tuple[Path, Path]] = {} for version_dir in [d for d in module_dir.iterdir() if d.is_dir()]: for file_path in version_dir.rglob("*"): if not file_path.is_file() or file_path.suffix not in PROMPT_EXTENSIONS: continue rel = file_path.relative_to(version_dir) key = ( module_dir.name, str(rel.with_suffix("")), file_path.suffix, ) old = choices.get(key) if old is None or _version_sort_key(version_dir.name) > _version_sort_key(old[0].name): choices[key] = (version_dir, file_path) selected.extend(file_path for _, file_path in choices.values()) return sorted(selected) raise ValueError(f"unknown prompt mode: {mode}") def module_to_path(module_name: str) -> Optional[Path]: if not module_name.startswith("app."): return None rel = module_name.split(".")[1:] file_path = APP_DIR / Path(*rel).with_suffix(".py") if file_path.exists(): return file_path init_path = APP_DIR / Path(*rel) / "__init__.py" if init_path.exists(): return init_path return None def path_to_module(path: Path) -> Optional[str]: try: rel = path.relative_to(BACKEND_DIR) except ValueError: return None if rel.suffix != ".py": return None parts = list(rel.with_suffix("").parts) if parts[-1] == "__init__": parts = parts[:-1] return ".".join(parts) def parse_python_ast(path: Path) -> ast.Module: return ast.parse(path.read_text(encoding="utf-8", errors="ignore"), filename=str(path)) def literal_step_manifest() -> Dict[str, Dict[str, Any]]: tree = parse_python_ast(APP_DIR / "core" / "step_manifest.py") for node in tree.body: if isinstance(node, ast.Assign): for target in node.targets: if isinstance(target, ast.Name) and target.id == "STEP_MANIFEST": return ast.literal_eval(node.value) if ( isinstance(node, ast.AnnAssign) and isinstance(node.target, ast.Name) and node.target.id == "STEP_MANIFEST" and node.value is not None ): return ast.literal_eval(node.value) raise RuntimeError("STEP_MANIFEST assignment not found") def step_class_file_map() -> Dict[str, Path]: init_path = APP_DIR / "core" / "steps" / "__init__.py" tree = parse_python_ast(init_path) imported: Dict[str, Path] = {} for node in tree.body: if not isinstance(node, ast.ImportFrom) or not node.module: continue source_path = module_to_path(node.module) if not source_path: continue for alias in node.names: imported[alias.asname or alias.name] = source_path result: Dict[str, Path] = {} for node in tree.body: if not isinstance(node, ast.Assign): continue if not any(isinstance(t, ast.Name) and t.id == "STEP_CLASSES" for t in node.targets): continue if not isinstance(node.value, ast.Dict): continue for key_node, value_node in zip(node.value.keys, node.value.values): if not isinstance(key_node, ast.Constant) or not isinstance(key_node.value, str): continue if isinstance(value_node, ast.Name) and value_node.id in imported: result[key_node.value] = imported[value_node.id] return result def active_main_step_ids() -> List[str]: manifest = literal_step_manifest() active: List[Tuple[float, str]] = [] for step_id, info in manifest.items(): if info.get("lifecycle") != "active": continue if info.get("applicability") in {"disabled", "on_demand"}: continue category = info.get("category") if category not in {"analysis", "image"}: continue active.append((float(info.get("order", 0)), step_id)) return [step_id for _, step_id in sorted(active)] def local_imports(path: Path) -> List[Path]: tree = parse_python_ast(path) imported: List[Path] = [] current_module = path_to_module(path) current_package = ".".join(current_module.split(".")[:-1]) if current_module else "" for node in ast.walk(tree): if isinstance(node, ast.Import): for alias in node.names: source = module_to_path(alias.name) if source: imported.append(source) elif isinstance(node, ast.ImportFrom): module_name = node.module or "" if node.level and current_package: base_parts = current_package.split(".") module_parts = base_parts[: max(0, len(base_parts) - node.level + 1)] if module_name: module_parts.extend(module_name.split(".")) module_name = ".".join(module_parts) source = module_to_path(module_name) if source: imported.append(source) for alias in node.names: child = module_to_path(f"{module_name}.{alias.name}") if module_name else None if child: imported.append(child) return sorted(set(imported)) def should_follow_import(path: Path) -> bool: try: rel = path.relative_to(APP_DIR) except ValueError: return False if rel.parts[:1] in {("models",), ("schemas",), ("api",)}: return False if rel.parts[:1] == ("core",) and rel.parts[1:2] in {("database",), ("config",)}: return False return True def collect_structural_python_candidates(import_depth: int) -> Dict[Path, List[str]]: reasons: Dict[Path, List[str]] = {} def add(path: Path, reason: str) -> None: if path.exists() and path.suffix == ".py": reasons.setdefault(path, []) if reason not in reasons[path]: reasons[path].append(reason) infrastructure = [ APP_DIR / "core" / "step_manifest.py", APP_DIR / "core" / "step_catalog.py", APP_DIR / "core" / "step_runner.py", APP_DIR / "core" / "steps" / "__init__.py", APP_DIR / "modules" / "prompt_loader.py", APP_DIR / "services" / "analysis_dispatch_service.py", ] for path in infrastructure: add(path, "pipeline infrastructure: dispatch/step registry/prompt loading") class_map = step_class_file_map() queue: List[Tuple[Path, int, str]] = [] for step_id in active_main_step_ids(): path = class_map.get(step_id) if not path: continue reason = f"active main pipeline step runner: {step_id}" add(path, reason) queue.append((path, 0, reason)) seen_edges: set[Tuple[Path, Path]] = set() while queue: source, depth, source_reason = queue.pop(0) if depth >= import_depth: continue for imported in local_imports(source): if not should_follow_import(imported): continue edge = (source, imported) if edge in seen_edges: continue seen_edges.add(edge) reason = f"imported by {relative_path(source)} ({source_reason})" add(imported, reason) queue.append((imported, depth + 1, reason)) return reasons def constant_string_assignments(tree: ast.Module) -> Dict[str, str]: values: Dict[str, str] = {} for node in tree.body: if not isinstance(node, ast.Assign): continue if not isinstance(node.value, ast.Constant) or not isinstance(node.value.value, str): continue for target in node.targets: if isinstance(target, ast.Name): values[target.id] = node.value.value return values def prompt_modules_from_python(path: Path) -> List[str]: tree = parse_python_ast(path) constants = constant_string_assignments(tree) modules: set[str] = set() known_prompt_modules = {p.name for p in PROMPTS_BASE.iterdir() if p.is_dir()} if PROMPTS_BASE.exists() else set() for node in ast.walk(tree): if isinstance(node, ast.Call): func_name = "" if isinstance(node.func, ast.Name): func_name = node.func.id elif isinstance(node.func, ast.Attribute): func_name = node.func.attr if func_name in {"load_prompt", "load_schema", "_load_pl"} and node.args: first = node.args[0] module: Optional[str] = None if isinstance(first, ast.Constant) and isinstance(first.value, str): module = first.value elif isinstance(first, ast.Name): module = constants.get(first.id) if module and module in known_prompt_modules: modules.add(module) if isinstance(node, ast.Assign): target_names = [t.id for t in node.targets if isinstance(t, ast.Name)] if not any(name.endswith("PROMPT_DIR") or name == "PROMPT_DIR" for name in target_names): continue for child in ast.walk(node.value): if isinstance(child, ast.Constant) and isinstance(child.value, str): if child.value in known_prompt_modules: modules.add(child.value) return sorted(modules) def latest_prompt_files_for_module(module: str) -> List[Path]: module_dir = PROMPTS_BASE / module if not module_dir.exists(): return [] versions = sorted( [d for d in module_dir.iterdir() if d.is_dir()], key=lambda p: _version_sort_key(p.name), reverse=True, ) if not versions: return [] latest = versions[0] return sorted(p for p in latest.rglob("*") if p.is_file() and p.suffix in PROMPT_EXTENSIONS) def collect_structural_candidates(import_depth: int) -> Dict[Path, Tuple[str, List[str]]]: python_reasons = collect_structural_python_candidates(import_depth) candidates: Dict[Path, Tuple[str, List[str]]] = { path: ("python", reasons) for path, reasons in python_reasons.items() } prompt_module_reasons: Dict[str, List[str]] = {} for path, reasons in python_reasons.items(): for module in prompt_modules_from_python(path): prompt_module_reasons.setdefault(module, []) reason = f"latest prompt pack loaded by {relative_path(path)}" if reason not in prompt_module_reasons[module]: prompt_module_reasons[module].append(reason) for module, reasons in prompt_module_reasons.items(): for path in latest_prompt_files_for_module(module): candidates[path] = ("prompt", reasons) return candidates def relative_path(path: Path) -> str: return str(path.relative_to(REPO_ROOT)) def file_sha256(path: Path) -> str: h = hashlib.sha256() with path.open("rb") as fh: for block in iter(lambda: fh.read(1024 * 1024), b""): h.update(block) return h.hexdigest() def line_chunks( text: str, *, max_chars: int, max_lines: int, ) -> Iterator[Tuple[int, int, str]]: lines = text.splitlines() start = 1 current: List[str] = [] current_chars = 0 for idx, line in enumerate(lines, start=1): numbered = f"{idx}: {line}" add_chars = len(numbered) + 1 if current and ( current_chars + add_chars > max_chars or len(current) >= max_lines ): end = start + len(current) - 1 yield start, end, "\n".join(current) start = idx current = [] current_chars = 0 current.append(numbered) current_chars += add_chars if current: end = start + len(current) - 1 yield start, end, "\n".join(current) def whole_file_chunk(text: str) -> Optional[Tuple[int, int, str]]: lines = text.splitlines() if not lines: return None numbered = [f"{idx}: {line}" for idx, line in enumerate(lines, start=1)] return 1, len(lines), "\n".join(numbered) def response_text_from_payload(payload: Dict[str, Any]) -> str: output_text = payload.get("output_text") if isinstance(output_text, str) and output_text.strip(): return output_text output = payload.get("output") if isinstance(output, list): for item in output: if not isinstance(item, dict): continue content = item.get("content") if not isinstance(content, list): continue for part in content: if isinstance(part, dict) and part.get("type") == "output_text": text = part.get("text") if isinstance(text, str) and text.strip(): return text status = payload.get("status") incomplete = payload.get("incomplete_details") output_types = [] output = payload.get("output") if isinstance(output, list): output_types = [str(item.get("type")) for item in output if isinstance(item, dict)] raise RuntimeError( "OpenAI response did not include output_text " f"(status={status!r}, incomplete_details={incomplete!r}, output_types={output_types!r})" ) def _strip_json_fence(text: str) -> str: stripped = text.strip() if not stripped.startswith("```"): return stripped lines = stripped.splitlines() if lines and lines[0].strip().startswith("```"): lines = lines[1:] if lines and lines[-1].strip() == "```": lines = lines[:-1] return "\n".join(lines).strip() def _extract_outer_json_object(text: str) -> str: stripped = _strip_json_fence(text) start = stripped.find("{") end = stripped.rfind("}") if start == -1 or end == -1 or end <= start: return stripped return stripped[start : end + 1] def _remove_json_comments(text: str) -> str: lines = [] for line in text.splitlines(): stripped = line.lstrip() if stripped.startswith("//") or stripped.startswith("#"): continue lines.append(line) return "\n".join(lines) def _repair_common_llm_json(text: str) -> str: repaired = _extract_outer_json_object(text) repaired = _remove_json_comments(repaired) repaired = re.sub(r",(\s*[}\]])", r"\1", repaired) # Gemini occasionally emits JS-style keys despite responseMimeType=json. # Keep this limited to line-start object keys so evidence strings are not # rewritten. repaired = re.sub( r'(?m)^(\s*)([A-Za-z_][A-Za-z0-9_]*)(\s*:)', r'\1"\2"\3', repaired, ) return repaired def loads_llm_json(text: str) -> Dict[str, Any]: try: loaded = json.loads(text) except json.JSONDecodeError: loaded = json.loads(_repair_common_llm_json(text)) if not isinstance(loaded, dict): raise ValueError("LLM JSON root must be an object") return loaded def call_openai_json( *, api_key: str, model: str, reasoning_effort: Optional[str], user_prompt: str, max_output_tokens: int, timeout: int, retries: int, ) -> Dict[str, Any]: body: Dict[str, Any] = { "model": model, "instructions": SYSTEM_PROMPT, "input": user_prompt, "store": False, "text": { "format": { "type": "json_schema", "name": "semantic_string_debt_audit", "strict": True, "schema": RESPONSE_SCHEMA, } }, "max_output_tokens": max_output_tokens, } if reasoning_effort: body["reasoning"] = {"effort": reasoning_effort} last_error: Optional[BaseException] = None for attempt in range(1, retries + 2): req = urllib.request.Request( OPENAI_RESPONSES_URL, data=json.dumps(body).encode("utf-8"), headers={ "Authorization": f"Bearer {api_key}", "Content-Type": "application/json", }, method="POST", ) try: with urllib.request.urlopen(req, timeout=timeout) as resp: payload = json.loads(resp.read().decode("utf-8")) text = response_text_from_payload(payload) return loads_llm_json(text) except urllib.error.HTTPError as exc: error_text = exc.read().decode("utf-8", errors="replace") last_error = RuntimeError(f"OpenAI API error {exc.code}: {error_text}") # Some deployments may not accept a reasoning field for a model. # Retry once without it only for schema/param 400s. if exc.code == 400 and reasoning_effort and "reasoning" in error_text.lower(): body.pop("reasoning", None) reasoning_effort = None continue if exc.code in {408, 409, 429, 500, 502, 503, 504} and attempt <= retries: time.sleep(min(30, 2 * attempt)) continue raise last_error from exc except (urllib.error.URLError, socket.timeout, TimeoutError) as exc: last_error = exc if attempt <= retries: time.sleep(min(30, 2 * attempt)) continue raise RuntimeError(f"OpenAI API failed after {retries} retries: {exc}") from exc except json.JSONDecodeError as exc: raw = locals().get("text", "") raise RuntimeError( f"OpenAI response was not valid JSON: {exc}; raw={raw[:2000]!r}" ) from exc raise RuntimeError(f"OpenAI API failed: {last_error}") def gemini_text_from_payload(payload: Dict[str, Any]) -> str: candidates = payload.get("candidates") if not isinstance(candidates, list) or not candidates: feedback = payload.get("promptFeedback", {}) raise RuntimeError(f"Gemini returned no candidates: {json.dumps(feedback, ensure_ascii=False)[:500]}") content = candidates[0].get("content", {}) parts = content.get("parts", []) texts = [part.get("text", "") for part in parts if isinstance(part, dict) and "text" in part] if not texts: raise RuntimeError(f"Gemini response has no text parts: {json.dumps(payload, ensure_ascii=False)[:500]}") return "".join(texts) def call_gemini_json( *, model: str, user_prompt: str, max_output_tokens: int, timeout: int, retries: int, ) -> Dict[str, Any]: body: Dict[str, Any] = { "systemInstruction": {"parts": [{"text": SYSTEM_PROMPT}]}, "contents": [{"role": "user", "parts": [{"text": user_prompt}]}], "generationConfig": { "temperature": 0, "maxOutputTokens": max_output_tokens, "responseMimeType": "application/json", "responseJsonSchema": RESPONSE_SCHEMA, }, } last_error: Optional[BaseException] = None for attempt in range(1, retries + 2): api_key = _next_gemini_key() url = GEMINI_API_URL_TEMPLATE.format(model=model, api_key=api_key) req = urllib.request.Request( url, data=json.dumps(body).encode("utf-8"), headers={"Content-Type": "application/json"}, method="POST", ) try: with urllib.request.urlopen(req, timeout=timeout) as resp: payload = json.loads(resp.read().decode("utf-8")) text = gemini_text_from_payload(payload) return loads_llm_json(text) except urllib.error.HTTPError as exc: error_text = exc.read().decode("utf-8", errors="replace") last_error = RuntimeError(f"Gemini API error {exc.code}: {error_text}") if exc.code in {408, 409, 429, 500, 502, 503, 504} and attempt <= retries: time.sleep(min(30, 2 * attempt)) continue raise last_error from exc except (urllib.error.URLError, socket.timeout, TimeoutError) as exc: last_error = exc if attempt <= retries: time.sleep(min(30, 2 * attempt)) continue raise RuntimeError(f"Gemini API failed after {retries} retries: {exc}") from exc except json.JSONDecodeError as exc: raw = locals().get("text", "") raise RuntimeError( f"Gemini response was not valid JSON: {exc}; raw={raw[:2000]!r}" ) from exc raise RuntimeError(f"Gemini API failed: {last_error}") def build_user_prompt( *, path: Path, scan_kind: str, candidate_reason: str, chunk_start: int, chunk_end: int, chunk_text: str, ) -> str: rel = relative_path(path) return f"""Audit this {scan_kind} file chunk. File: {rel} Line range: {chunk_start}-{chunk_end} Candidate reason: {candidate_reason} Important: - The source below is numbered. Return exact line_start/line_end from these numbers. - Flag only high-confidence, actionable findings. - Candidate reason explains why this file is part of the main pipeline audit scope; do not treat it as evidence of debt. - This repo allows pattern matching for closed technical formats: machine identifiers with formally defined shapes, schema/path/status/version/hash metadata, and canonical schema/SOT enum validation. Do not flag those unless they are used to infer open-world story or visual semantics from natural-language text. - For prompts, flag only concrete scenario pollution or classifier-like phrase lists, not ordinary examples, labels, or generic style preferences. - For code, report only if a string/regex/list match over scenario text, LLM output, checkpoint strings, prompt-card prose, generated t2i_prompt, or review/validation text changes behavior. - For code, include both the pattern and the behavior-changing use site in the evidence; omit constant-only findings. - Do not flag ordinary infrastructure, prompt assembly labels, model/provider names, DB status strings, migrations, or API plumbing unless they directly route/validate/mutate scenario or image-generation semantics. - If uncertain, return no findings for this file. High precision is more important than recall. Source: ```text {chunk_text} ``` """ def processed_keys(results_path: Path) -> set[Tuple[str, str, int, int]]: keys: set[Tuple[str, str, int, int]] = set() if not results_path.exists(): return keys for line in results_path.read_text(encoding="utf-8", errors="ignore").splitlines(): if not line.strip(): continue try: item = json.loads(line) except json.JSONDecodeError: continue keys.add(( item.get("path", ""), item.get("sha256", ""), int(item.get("chunk_start", 0)), int(item.get("chunk_end", 0)), )) return keys def audit_chunk( *, path: Path, scan_kind: str, candidate_reason: str, sha: str, chunk_start: int, chunk_end: int, chunk_text: str, args: argparse.Namespace, api_key: str, ) -> Dict[str, Any]: prompt = build_user_prompt( path=path, scan_kind=scan_kind, candidate_reason=candidate_reason, chunk_start=chunk_start, chunk_end=chunk_end, chunk_text=chunk_text, ) started = time.time() if args.provider == "gemini": payload = call_gemini_json( model=args.model, user_prompt=prompt, max_output_tokens=args.max_output_tokens, timeout=args.timeout, retries=args.retries, ) else: payload = call_openai_json( api_key=api_key, model=args.model, reasoning_effort=args.reasoning_effort, user_prompt=prompt, max_output_tokens=args.max_output_tokens, timeout=args.timeout, retries=args.retries, ) duration_ms = int((time.time() - started) * 1000) findings = payload.get("findings") or [] if not isinstance(findings, list): findings = [] return { "path": relative_path(path), "scan_kind": scan_kind, "candidate_reason": candidate_reason, "sha256": sha, "chunk_start": chunk_start, "chunk_end": chunk_end, "duration_ms": duration_ms, "chunk_summary": payload.get("chunk_summary", ""), "findings": findings, } def write_jsonl(path: Path, item: Dict[str, Any], lock: threading.Lock) -> None: with lock: with path.open("a", encoding="utf-8") as fh: fh.write(json.dumps(item, ensure_ascii=False, sort_keys=True) + "\n") def load_results(results_path: Path) -> List[Dict[str, Any]]: if not results_path.exists(): return [] rows: List[Dict[str, Any]] = [] for line in results_path.read_text(encoding="utf-8", errors="ignore").splitlines(): if not line.strip(): continue try: rows.append(json.loads(line)) except json.JSONDecodeError: continue return rows def write_markdown(results_path: Path, markdown_path: Path) -> None: rows = load_results(results_path) flattened: List[Dict[str, Any]] = [] for row in rows: for finding in row.get("findings") or []: item = dict(finding) item["path"] = row["path"] item["scan_kind"] = row["scan_kind"] flattened.append(item) flattened.sort(key=lambda x: (x["path"], int(x.get("line_start", 0)), x.get("category", ""))) lines = [ "# Semantic String Debt LLM Audit Findings", "", f"- result chunks: `{len(rows)}`", f"- findings: `{len(flattened)}`", "", ] current_path = None for finding in flattened: if finding["path"] != current_path: current_path = finding["path"] lines.append(f"## `{current_path}`") lines.append("") line_start = finding.get("line_start") line_end = finding.get("line_end") span = f"{line_start}" if line_start == line_end else f"{line_start}-{line_end}" lines.append( f"- `{span}` **{finding.get('severity')} / {finding.get('category')}**" ) lines.append(f" - evidence: {finding.get('evidence', '').strip()}") lines.append(f" - why: {finding.get('why_problematic', '').strip()}") lines.append(f" - fix: {finding.get('recommended_fix', '').strip()}") lines.append("") markdown_path.write_text("\n".join(lines), encoding="utf-8") def parse_args(argv: Optional[Sequence[str]] = None) -> argparse.Namespace: parser = argparse.ArgumentParser( description="LLM audit for semantic string-routing debt and scenario-dependent prompts.", ) parser.add_argument("--provider", choices=["gemini", "openai"], default="gemini") parser.add_argument("--model", default="gemini-3-flash-preview") parser.add_argument( "--reasoning-effort", default="xhigh", choices=["low", "medium", "high", "xhigh", ""], help="Responses API reasoning effort. Empty string disables the field.", ) parser.add_argument("--threads", type=int, default=8) parser.add_argument( "--python-mode", choices=["all", "backend-app", "none"], default="all", help="Python scan scope excluding tests.", ) parser.add_argument( "--prompt-mode", choices=["all", "effective", "latest-dir", "none"], default="latest-dir", help="Prompt scan scope. 'all' scans full prompt archive.", ) parser.add_argument( "--only-path", action="append", default=[], help="Restrict to one path or prefix. Can be repeated.", ) parser.add_argument("--limit", type=int, default=0, help="Limit file count after discovery.") parser.add_argument( "--candidate-only", action="store_true", help="Use structural main-pipeline candidates instead of all discovered files.", ) parser.add_argument( "--candidate-import-depth", type=int, default=DEFAULT_CANDIDATE_IMPORT_DEPTH, help="Local import graph depth from active step runner files.", ) parser.add_argument( "--list-candidates", action="store_true", help="Print candidate files with structural candidate_reason and exit.", ) parser.add_argument( "--chunk-large-files", action="store_true", help="Default is whole-file calls. Enable this only for files exceeding model input limits.", ) parser.add_argument( "--max-chars", type=int, default=1_200_000, help="Only used with --chunk-large-files.", ) parser.add_argument( "--max-lines", type=int, default=100_000, help="Only used with --chunk-large-files.", ) parser.add_argument("--max-output-tokens", type=int, default=65_536) parser.add_argument("--timeout", type=int, default=900) parser.add_argument("--retries", type=int, default=0) parser.add_argument("--dry-run", action="store_true") parser.add_argument("--resume", action="store_true") parser.add_argument("--output-dir", default="") return parser.parse_args(argv) def main(argv: Optional[Sequence[str]] = None) -> int: args = parse_args(argv) args.reasoning_effort = args.reasoning_effort or None file_reasons: Dict[Path, Tuple[str, List[str]]] = {} if args.candidate_only: file_reasons = collect_structural_candidates(args.candidate_import_depth) else: py_files = [(p, "python") for p in discover_python_files(args.python_mode)] prompt_files = [(p, "prompt") for p in discover_prompt_files(args.prompt_mode)] for p, kind in py_files + prompt_files: file_reasons[p] = (kind, [f"{kind} scope discovery"]) if args.only_path: prefixes = [str((REPO_ROOT / item).resolve()) if not Path(item).is_absolute() else str(Path(item).resolve()) for item in args.only_path] file_reasons = { p: item for p, item in file_reasons.items() if any(str(p.resolve()).startswith(prefix) for prefix in prefixes) } files = sorted(file_reasons.items(), key=lambda item: relative_path(item[0])) if args.limit: files = files[: args.limit] total_bytes = sum(p.stat().st_size for p, _ in files) total_lines = 0 for p, _ in files: total_lines += len(p.read_text(encoding="utf-8", errors="ignore").splitlines()) print(f"files={len(files)} bytes={total_bytes} lines={total_lines}", flush=True) print( f"provider={args.provider} model={args.model} python_mode={args.python_mode} " f"prompt_mode={args.prompt_mode} candidate_only={args.candidate_only} threads={args.threads}", flush=True, ) if args.dry_run or args.list_candidates: for p, (kind, reasons) in files[:200]: print(f"{kind}: {relative_path(p)} | candidate_reason={'; '.join(reasons)}", flush=True) if len(files) > 200: print(f"... {len(files) - 200} more", flush=True) if args.list_candidates: return 0 if args.dry_run: return 0 output_dir = Path(args.output_dir) if args.output_dir else DEFAULT_OUTPUT_BASE / time.strftime("%Y%m%d_%H%M%S") if not output_dir.is_absolute(): output_dir = REPO_ROOT / output_dir output_dir.mkdir(parents=True, exist_ok=True) results_path = output_dir / "findings.jsonl" failures_path = output_dir / "failures.jsonl" manifest_path = output_dir / "manifest.json" markdown_path = output_dir / "findings.md" already_done = processed_keys(results_path) if args.resume else set() api_key = _api_key() if args.provider == "openai" else "" write_lock = threading.Lock() work_items: List[Dict[str, Any]] = [] for path, (kind, reasons) in files: text = path.read_text(encoding="utf-8", errors="ignore") sha = file_sha256(path) candidate_reason = "; ".join(reasons) if args.chunk_large_files: chunks = list(line_chunks(text, max_chars=args.max_chars, max_lines=args.max_lines)) else: whole = whole_file_chunk(text) chunks = [whole] if whole else [] for start, end, chunk in chunks: key = (relative_path(path), sha, start, end) if key in already_done: continue work_items.append({ "path": path, "scan_kind": kind, "candidate_reason": candidate_reason, "sha": sha, "chunk_start": start, "chunk_end": end, "chunk_text": chunk, }) manifest = { "repo_root": str(REPO_ROOT), "created_at": time.strftime("%Y-%m-%dT%H:%M:%S%z"), "model": args.model, "provider": args.provider, "reasoning_effort": args.reasoning_effort, "threads": args.threads, "python_mode": args.python_mode, "prompt_mode": args.prompt_mode, "candidate_only": args.candidate_only, "candidate_import_depth": args.candidate_import_depth, "file_count": len(files), "chunk_count": len(work_items), "total_bytes": total_bytes, "total_lines": total_lines, "results_path": str(results_path.relative_to(REPO_ROOT)), "failures_path": str(failures_path.relative_to(REPO_ROOT)), "markdown_path": str(markdown_path.relative_to(REPO_ROOT)), } manifest_path.write_text(json.dumps(manifest, indent=2, ensure_ascii=False), encoding="utf-8") candidates_path = output_dir / "candidates.tsv" candidate_lines = ["path\tscan_kind\tcandidate_reason"] for path, (kind, reasons) in files: candidate_lines.append(f"{relative_path(path)}\t{kind}\t{'; '.join(reasons)}") candidates_path.write_text("\n".join(candidate_lines) + "\n", encoding="utf-8") print(f"output_dir={output_dir}", flush=True) print(f"candidates={candidates_path}", flush=True) print(f"chunks_to_process={len(work_items)}", flush=True) completed = 0 failed = 0 started = time.time() def _run(item: Dict[str, Any]) -> Dict[str, Any]: return audit_chunk(args=args, api_key=api_key, **item) with concurrent.futures.ThreadPoolExecutor(max_workers=args.threads) as executor: future_map = {executor.submit(_run, item): item for item in work_items} for future in concurrent.futures.as_completed(future_map): item = future_map[future] try: result = future.result() write_jsonl(results_path, result, write_lock) completed += 1 finding_count = len(result.get("findings") or []) print( f"[ok] {completed}/{len(work_items)} " f"{relative_path(item['path'])}:{item['chunk_start']}-{item['chunk_end']} " f"findings={finding_count}", flush=True, ) except Exception as exc: # keep audit running and record failed chunk failed += 1 failure = { "path": relative_path(item["path"]), "scan_kind": item["scan_kind"], "chunk_start": item["chunk_start"], "chunk_end": item["chunk_end"], "error": repr(exc), } write_jsonl(failures_path, failure, write_lock) print( f"[fail] {relative_path(item['path'])}:{item['chunk_start']}-{item['chunk_end']} {exc}", file=sys.stderr, flush=True, ) write_markdown(results_path, markdown_path) duration = int(time.time() - started) print(f"done completed={completed} failed={failed} seconds={duration}", flush=True) print(f"findings={markdown_path}", flush=True) print(f"raw={results_path}", flush=True) print(f"failures={failures_path}", flush=True) return 1 if failed else 0 if __name__ == "__main__": raise SystemExit(main())