from __future__ import annotations import argparse import fnmatch import json import pathlib import re import sys from collections.abc import Iterable, Mapping from typing import Any import yaml REPORT_BASENAME = "lifecycle-status-guard-scan" DEFAULT_EVIDENCE_WINDOW_CHARS = 240 def as_list(value: Any) -> list[str]: if value is None: return [] if isinstance(value, str): return [value] if isinstance(value, Iterable): return [str(item) for item in value if item is not None] return [str(value)] def lower_set(values: Iterable[str]) -> set[str]: return {value.casefold() for value in values} def normalize_config(raw_config: Mapping[str, Any]) -> dict[str, Any]: return { "watched_paths": as_list(raw_config.get("watched_paths")) or ["**/*"], "status_fields": as_list(raw_config.get("status_fields")), "forbidden_status_values": lower_set(as_list(raw_config.get("forbidden_status_values"))), "warning_status_values": lower_set(as_list(raw_config.get("warning_status_values"))), "required_evidence_markers": as_list(raw_config.get("required_evidence_markers")), "approved_phrases": as_list(raw_config.get("approved_phrases")), "forbidden_phrases": as_list(raw_config.get("forbidden_phrases")), "allowed_contexts": as_list(raw_config.get("allowed_contexts")), "evidence_window_chars": int( raw_config.get("evidence_window_chars", DEFAULT_EVIDENCE_WINDOW_CHARS) ), } def load_config(path: pathlib.Path) -> dict[str, Any]: text = path.read_text(encoding="utf-8") if path.suffix.lower() == ".json": parsed = json.loads(text) else: parsed = yaml.safe_load(text) if not isinstance(parsed, Mapping): raise ValueError("Config must be a JSON or YAML object.") return normalize_config(parsed) def rel_path(path: pathlib.Path, root: pathlib.Path) -> str: return path.relative_to(root).as_posix() def discover_files(scan_root: pathlib.Path, watched_paths: list[str]) -> list[pathlib.Path]: files: dict[pathlib.Path, None] = {} for pattern in watched_paths: for candidate in scan_root.glob(pattern): if candidate.is_file(): files[candidate.resolve()] = None return sorted(files) def parse_markdown_front_matter(text: str) -> Any | None: if not text.startswith("---"): return None lines = text.splitlines() if not lines or lines[0].strip() != "---": return None for index in range(1, len(lines)): if lines[index].strip() == "---": front_matter = "\n".join(lines[1:index]) parsed = yaml.safe_load(front_matter) if front_matter.strip() else {} return parsed if isinstance(parsed, Mapping) else None return None def parse_structured_content(path: pathlib.Path, text: str) -> Any | None: suffix = path.suffix.lower() try: if suffix == ".json": return json.loads(text) if suffix in {".yaml", ".yml"}: return yaml.safe_load(text) if suffix in {".md", ".markdown"}: return parse_markdown_front_matter(text) except Exception: return None return None def scalar_values(value: Any) -> list[str]: if isinstance(value, (str, int, float, bool)): return [str(value)] if isinstance(value, list): values: list[str] = [] for item in value: values.extend(scalar_values(item)) return values return [] def object_contains_marker(value: Any, markers: list[str]) -> bool: if not markers: return False marker_needles = [marker.casefold() for marker in markers] def walk(current: Any) -> bool: if isinstance(current, Mapping): for key, nested in current.items(): key_text = str(key).casefold() if any(marker in key_text for marker in marker_needles): return True if walk(nested): return True return False if isinstance(current, list): return any(walk(item) for item in current) text = str(current).casefold() return any(marker in text for marker in marker_needles) return walk(value) def line_col(text: str, index: int) -> tuple[int, int]: line = text.count("\n", 0, index) + 1 last_newline = text.rfind("\n", 0, index) column = index + 1 if last_newline == -1 else index - last_newline return line, column def make_context(text: str, index: int, length: int, window: int) -> str: start = max(0, index - window) end = min(len(text), index + length + window) return re.sub(r"\s+", " ", text[start:end]).strip() def context_has_any(context: str, values: list[str]) -> bool: folded = context.casefold() return any(value.casefold() in folded for value in values) def find_phrase_matches(text: str, phrase: str) -> Iterable[int]: if not phrase: return [] folded_text = text.casefold() folded_phrase = phrase.casefold() matches: list[int] = [] start = 0 while True: index = folded_text.find(folded_phrase, start) if index == -1: break matches.append(index) start = index + max(len(folded_phrase), 1) return matches def path_matches(field_path: str, key: str, status_fields: list[str]) -> bool: if not status_fields: return False candidates = {field_path, key} return any( field in candidates or fnmatch.fnmatch(field_path, field) or fnmatch.fnmatch(key, field) for field in status_fields ) def child_path(parent_path: str, key: str) -> str: return f"{parent_path}.{key}" if parent_path else key def list_path(parent_path: str, index: int) -> str: return f"{parent_path}[{index}]" if parent_path else f"[{index}]" def scan_structured( data: Any, file_rel: str, config: dict[str, Any], report: dict[str, Any], ) -> None: status_fields = config["status_fields"] forbidden_values = config["forbidden_status_values"] warning_values = config["warning_status_values"] evidence_markers = config["required_evidence_markers"] def walk(current: Any, path: str, parent: Any) -> None: if isinstance(current, Mapping): for key_obj, value in current.items(): key = str(key_obj) current_path = child_path(path, key) if path_matches(current_path, key, status_fields): for scalar in scalar_values(value): value_key = scalar.casefold() if value_key in forbidden_values: evidence_present = object_contains_marker(parent, evidence_markers) record = { "file": file_rel, "source": "structured", "path": current_path, "field": key, "value": scalar, "severity": "blocking" if not evidence_present else "observation", "evidence_present": evidence_present, "message": ( "Forbidden lifecycle/status value lacks local evidence marker." if not evidence_present else "Lifecycle/status value has a local evidence marker; approval validity is not decided by this scan." ), } if evidence_present: observation = dict(record) observation["kind"] = "field_evidence_present" report["observations"].append(observation) else: report["field_level_findings"].append(record) report["blocking_errors"].append(record["message"]) report["missing_evidence_markers"].append( { "file": file_rel, "path": current_path, "required_evidence_markers": evidence_markers, } ) elif value_key in warning_values: report["warnings"].append( { "file": file_rel, "source": "structured", "path": current_path, "field": key, "value": scalar, "severity": "warning", "message": "Warning lifecycle/status value matched configured policy.", } ) walk(value, current_path, value if isinstance(value, Mapping) else current) elif isinstance(current, list): for index, item in enumerate(current): walk(item, list_path(path, index), item if isinstance(item, Mapping) else parent) walk(data, "", data) def scan_phrases( text: str, file_rel: str, config: dict[str, Any], report: dict[str, Any], ) -> None: evidence_markers = config["required_evidence_markers"] allowed_contexts = config["allowed_contexts"] window = config["evidence_window_chars"] for phrase in config["approved_phrases"]: for index in find_phrase_matches(text, phrase): context = make_context(text, index, len(phrase), window) if context_has_any(context, allowed_contexts): continue evidence_present = context_has_any(context, evidence_markers) line, column = line_col(text, index) record = { "file": file_rel, "source": "text", "phrase": phrase, "line": line, "column": column, "severity": "blocking" if not evidence_present else "observation", "evidence_present": evidence_present, "context": context, "message": ( "Approval claim phrase lacks nearby evidence marker." if not evidence_present else "Approval claim phrase has nearby evidence marker; approval validity is not decided by this scan." ), } if evidence_present: observation = dict(record) observation["kind"] = "evidence_present_claim" report["observations"].append(observation) else: report["phrase_level_findings"].append(record) report["blocking_errors"].append(record["message"]) report["missing_evidence_markers"].append( { "file": file_rel, "line": line, "phrase": phrase, "required_evidence_markers": evidence_markers, } ) for phrase in config["forbidden_phrases"]: for index in find_phrase_matches(text, phrase): context = make_context(text, index, len(phrase), window) if context_has_any(context, allowed_contexts): continue line, column = line_col(text, index) record = { "file": file_rel, "source": "text", "phrase": phrase, "line": line, "column": column, "severity": "blocking", "evidence_present": context_has_any(context, evidence_markers), "context": context, "message": "Forbidden phrase matched configured policy.", } report["phrase_level_findings"].append(record) report["blocking_errors"].append(record["message"]) def empty_report(scan_root: pathlib.Path, config_path: pathlib.Path) -> dict[str, Any]: return { "scan_root": str(scan_root), "config_path": str(config_path), "files_scanned": [], "possible_overclaims": [], "field_level_findings": [], "phrase_level_findings": [], "missing_evidence_markers": [], "warnings": [], "blocking_errors": [], "observations": [], "machine_readable_summary": {}, } def finalize_report(report: dict[str, Any]) -> None: report["possible_overclaims"] = [ *report["field_level_findings"], *report["phrase_level_findings"], ] report["machine_readable_summary"] = { "status": "FAIL" if report["blocking_errors"] else "PASS", "files_scanned_count": len(report["files_scanned"]), "field_finding_count": len(report["field_level_findings"]), "phrase_finding_count": len(report["phrase_level_findings"]), "missing_evidence_count": len(report["missing_evidence_markers"]), "warning_count": len(report["warnings"]), "blocking_count": len(report["blocking_errors"]), "observation_count": len(report["observations"]), } def write_markdown_report(report: dict[str, Any], output_path: pathlib.Path) -> None: summary = report["machine_readable_summary"] lines = [ "# Lifecycle Status Guard Scan", "", "## Summary", "", f"- Status: `{summary['status']}`", f"- Files scanned: {summary['files_scanned_count']}", f"- Blocking findings: {summary['blocking_count']}", f"- Warnings: {summary['warning_count']}", f"- Evidence-present observations: {summary['observation_count']}", "", "## Blocking Findings", "", ] if report["possible_overclaims"]: for finding in report["possible_overclaims"]: location = finding.get("path") or f"line {finding.get('line')}" claim = finding.get("value") or finding.get("phrase") lines.append(f"- `{finding['file']}` {location}: `{claim}` - {finding['message']}") else: lines.append("- None") lines.extend(["", "## Warnings", ""]) if report["warnings"]: for warning in report["warnings"]: lines.append( f"- `{warning['file']}` {warning.get('path', '')}: `{warning.get('value', '')}` - {warning['message']}" ) else: lines.append("- None") lines.extend(["", "## Observations", ""]) if report["observations"]: for observation in report["observations"]: location = observation.get("path") or f"line {observation.get('line')}" claim = observation.get("value") or observation.get("phrase") lines.append( f"- `{observation['file']}` {location}: `{claim}` - evidence marker present; approval validity not decided." ) else: lines.append("- None") lines.extend(["", "## Files Scanned", ""]) for file_name in report["files_scanned"]: lines.append(f"- `{file_name}`") output_path.write_text("\n".join(lines) + "\n", encoding="utf-8") def run(scan_root: pathlib.Path, config_path: pathlib.Path, output_dir: pathlib.Path) -> int: scan_root = scan_root.resolve() config_path = config_path.resolve() if not scan_root.is_dir(): raise ValueError(f"Scan root is not a directory: {scan_root}") if not config_path.is_file(): raise ValueError(f"Config file does not exist: {config_path}") config = load_config(config_path) report = empty_report(scan_root, config_path) for path in discover_files(scan_root, config["watched_paths"]): file_rel = rel_path(path, scan_root) text = path.read_text(encoding="utf-8", errors="replace") report["files_scanned"].append(file_rel) structured = parse_structured_content(path, text) if isinstance(structured, (Mapping, list)): scan_structured(structured, file_rel, config, report) scan_phrases(text, file_rel, config, report) finalize_report(report) output_dir.mkdir(parents=True, exist_ok=True) json_path = output_dir / f"{REPORT_BASENAME}.json" md_path = output_dir / f"{REPORT_BASENAME}.md" json_path.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8") write_markdown_report(report, md_path) return 1 if report["blocking_errors"] else 0 def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description="Scan configured lifecycle/status claims.") parser.add_argument("--scan-root", required=True, type=pathlib.Path) parser.add_argument("--config", required=True, type=pathlib.Path) parser.add_argument("--output-dir", required=True, type=pathlib.Path) args = parser.parse_args(argv) try: return run(args.scan_root, args.config, args.output_dir) except Exception as exc: print(f"ERROR: {exc}", file=sys.stderr) return 2 if __name__ == "__main__": raise SystemExit(main())