khadhroony-bot3/scripts/audit_pre_0_4_6_static_contracts.py

#!/usr/bin/env python3
# file: scripts/audit_pre_0_4_6_static_contracts.py
# version: 4

"""Audit static contracts that must remain true before the 0.4.6 alignment."""

from __future__ import annotations

import argparse
import pathlib
import re
import sys
import tomllib

def fail(code: str, message: str) -> None:
    """Print one violation."""

    print(f"{code}: {message}")

def frontend_invokes(root: pathlib.Path) -> set[str]:
    """Return literal Tauri commands invoked by the frontend."""

    commands: set[str] = set()
    pattern = re.compile(r"invoke(?:<[^>]+>)?\(\s*[\"']([A-Za-z0-9_]+)[\"']")
    for suffix in ("*.ts", "*.html"):
        for path in (root / "kb-app-demo-desktop" / "frontend").rglob(suffix):
            commands.update(pattern.findall(path.read_text(encoding="utf-8")))
    return commands

def registered_commands(root: pathlib.Path) -> set[str]:
    """Return commands registered in generate_handler."""

    path = root / "kb-app-demo-desktop" / "src" / "tauri.rs"
    text = path.read_text(encoding="utf-8")
    match = re.search(r"generate_handler!\[(.*?)\]\);", text, re.DOTALL)
    if match is None:
        return set()
    return set(re.findall(r"([A-Za-z_][A-Za-z0-9_]*)[ \t]*,", match.group(1)))

def main() -> int:
    """Run the pre-0.4.6 static audit."""

    parser = argparse.ArgumentParser()
    parser.add_argument("--root", default=".")
    parser.add_argument("--report-only", action="store_true")
    arguments = parser.parse_args()
    root = pathlib.Path(arguments.root).resolve()
    violations = 0

    dotenv_crates: list[str] = []
    for cargo in sorted(root.glob("*/Cargo.toml")):
        data = tomllib.loads(cargo.read_text(encoding="utf-8"))
        dependencies = data.get("dependencies", {})
        if "dotenvy" in dependencies:
            dotenv_crates.append(cargo.parent.name)
    if dotenv_crates != ["kb-config"]:
        fail("ALIGN001", f"dotenvy must be owned only by kb-config, found {dotenv_crates}")
        violations += 1

    missing = sorted(frontend_invokes(root) - registered_commands(root))
    if missing:
        fail("ALIGN002", f"frontend commands missing from Tauri registration: {missing}")
        violations += 1

    tauri_path = root / "kb-app-demo-desktop" / "src" / "tauri.rs"
    tauri_text = tauri_path.read_text(encoding="utf-8")
    forbidden_tauri_dependencies = (
        "kb_pipeline::",
        "kb_pipeline_demo_scenarios::",
        "kb_store::",
    )
    direct_dependencies = [
        dependency for dependency in forbidden_tauri_dependencies if dependency in tauri_text
    ]
    if direct_dependencies:
        fail(
            "ALIGN009",
            f"tauri.rs must remain a thin IPC/runtime layer, direct domain dependencies: {direct_dependencies}",
        )
        violations += 1
    if 'window.label() != "main"' not in tauri_text or "disconnect_demo_ws_app_state" not in tauri_text:
        fail("ALIGN003", "global WebSocket shutdown must be tied to main-window destruction")
        violations += 1
    if re.search(r'window\.label\(\)\s*==\s*"demo_ws".*disconnect_demo_ws', tauri_text, re.DOTALL):
        fail("ALIGN004", "demo_ws window destruction must not disconnect the application session")
        violations += 1

    app_state = (root / "kb-app-demo-desktop" / "src" / "app_state.rs").read_text(encoding="utf-8")
    if "demo_ws_session" not in app_state:
        fail("ALIGN005", "AppState must own the persistent demo WebSocket session")
        violations += 1

    forbidden_crates = (
        "kb_decoder_",
        "kb_executor_",
        "kb_materializer_",
        "kb_store_core",
        "kb_store_pg",
    )
    for path in sorted(root.rglob("*.rs")):
        if any(part in {"olddocs", "target"} for part in path.parts):
            continue
        text = path.read_text(encoding="utf-8")
        for name in forbidden_crates:
            if name in text:
                fail("ALIGN006", f"legacy crate identity {name!r} found in {path.relative_to(root)}")
                violations += 1

    required_docs = ("README.md", "TODO.md", "USAGE.md", "CHANGELOG.md")
    for cargo in sorted(root.glob("*/Cargo.toml")):
        for document in required_docs:
            if not (cargo.parent / document).is_file():
                fail("ALIGN007", f"missing {document} in {cargo.parent.name}")
                violations += 1

    matrix = root / "test-fixtures" / "contract-matrices" / "OPERATION_NAMING_MATRIX.json"
    if not matrix.is_file():
        fail("ALIGN008", "operation naming matrix is missing")
        violations += 1

    tauri_source = (root / "kb-app-demo-desktop/src/tauri.rs").read_text(encoding="utf-8")
    command_files = []
    for command_path in (root / "kb-app-demo-desktop/src").glob("*.rs"):
        command_source = command_path.read_text(encoding="utf-8")
        if "#[tauri::command]" in command_source:
            command_files.append(command_path.name)
    if command_files != ["tauri.rs"]:
        fail("ALIGN009", f"#[tauri::command] must exist only in tauri.rs, found: {command_files}")
        violations += 1
    forbidden_files = [
        "tauri_backfill.rs",
        "tauri_core_extraction.rs",
        "tauri_decode_replay.rs",
        "tauri_execution.rs",
        "tauri_general_commands.rs",
        "tauri_http.rs",
        "tauri_sql.rs",
        "tauri_support.rs",
        "tauri_ws.rs",
    ]
    for forbidden in forbidden_files:
        if (root / "kb-app-demo-desktop/src" / forbidden).exists():
            fail("ALIGN010", f"obsolete Tauri adapter module still exists: {forbidden}")
            violations += 1
    if "return crate::demo_decode_replay_options(state);" not in tauri_source:
        fail("ALIGN011", "tauri.rs must delegate demo_decode_replay_options through the crate façade")
        violations += 1
    if violations == 0:
        print("Pre-0.4.6 static contract audit: clean")
        return 0

    print(f"Pre-0.4.6 static contract audit: {violations} violation(s)")
    return 0 if arguments.report_only else 1

if __name__ == "__main__":
    raise SystemExit(main())