ss-tools/scripts/semantic_parse_gate.py

#!/usr/bin/env python3
# #region Tooling.SemanticParseGate.Module [C:2] [TYPE Module] [SEMANTICS semantic,gate,parser,warnings,ci]
# @BRIEF Fail the pre-merge gate when AXIOM's audit reports any parse/schema warning for the workspace.
# @PRE axiom-mcp binary is available via --axiom-bin, AXIOM_MCP_BIN, ../axiom-mcp/target/release/axiom-mcp or PATH.
# @POST Exit 0 when the parse/schema warning count is 0; exit 1 when it is greater than 0; exit 2 on a tool or environment failure.
# @INVARIANT This parser slice does not close whole-source INV_1–INV_7 acceptance.
# @RATIONALE The G-SEMANTIC-HEALTH gate is measured last, on the GO commit. Running this cheap check before every merge keeps the parse/schema class at zero instead of turning the final measurement into a multi-file cleanup.
# @REJECTED Grepping sources for missing #endregion markers — the AXIOM parser is the single source of truth for schema diagnostics.
# @REJECTED Marking the gate CLOSED between lanes — later lanes add markup and every intermediate measurement goes stale (owner decision 2026-10-20).
# @RELATION BINDS_TO -> [Tests.Tooling.SemanticParseGate]
import argparse

from semantic_gate_validation import (decode_tool, validate_audit, validate_rebuild, verify_config)
import json
import os
import pathlib
import shutil
import subprocess
import sys
import time

# #region Tooling.SemanticParseGate.Codes [C:2] [TYPE Block] [SEMANTICS semantic,gate,parser]
# @BRIEF Explicit current Axiom audit.rs/tags.rs/parser.rs vocabulary; unknown extensions are unavailable.
# @RATIONALE Missing optional tags are not violations; only present deprecated tags/formats are advisory here.
# @REJECTED Ignoring an unfamiliar code can convert a newly introduced schema violation into PASS.
PARSE_SCHEMA_CODES = (
    "schema_unclosed_anchor",
    "schema_unknown_tag",
    "schema_invalid_enum_value",
    "schema_invalid_relation_predicate",
    "flat_hierarchical_id",
    "tombstone_missing_deprecated",
    "tombstone_missing_replaced_by",
    "invalid_tier",
)
SIZE_CODES = ("contract_too_long", "module_too_long")
GRAPH_CODES = ("unresolved_relation", "test_verifies_missing_target")
ADVISORY_CODES = ("schema_deprecated_tag", "schema_def_format_deprecated")
# #endregion Tooling.SemanticParseGate.Codes


# #region Tooling.SemanticParseGate.Classify [C:1] [TYPE Function] [SEMANTICS semantic,gate,classify]
# @BRIEF Split severity_breakdown into parse/schema, size-advisory and graph buckets.
# @PRE severity is a mapping of warning code to count from AXIOM's audit_contracts response.
# @POST Returns a dict with parse_schema_total, size_total, graph_total and the per-code parse map.
def classify(severity: dict | None) -> dict:
    severity = severity or {}
    known = set(PARSE_SCHEMA_CODES + SIZE_CODES + GRAPH_CODES + ADVISORY_CODES)
    unknown = set(severity) - known
    if unknown:
        raise RuntimeError("[NEED_CONTEXT: unrecognized AXIOM diagnostic codes: " + ", ".join(sorted(unknown)) + "]")
    parse = {code: int(severity.get(code, 0) or 0) for code in PARSE_SCHEMA_CODES}
    size = {code: int(severity.get(code, 0) or 0) for code in SIZE_CODES}
    graph = {code: int(severity.get(code, 0) or 0) for code in GRAPH_CODES}
    advisory = {code: int(severity.get(code, 0) or 0) for code in ADVISORY_CODES}
    return {
        "parse_schema": parse,
        "parse_schema_total": sum(parse.values()),
        "size": size,
        "size_total": sum(size.values()),
        "graph": graph,
        "graph_total": sum(graph.values()),
        "advisory": advisory,
        "advisory_total": sum(advisory.values()),
    }
# #endregion Tooling.SemanticParseGate.Classify


# #region Tooling.SemanticParseGate.ResolveBinary [C:1] [TYPE Function] [SEMANTICS semantic,gate,binary]
# @BRIEF Locate the axiom-mcp executable: explicit flag, environment, sibling checkout, then PATH.
# @POST Returns an existing path as str; raises FileNotFoundError with the searched locations otherwise.
def resolve_binary(explicit: str | None) -> str:
    candidates = []
    if explicit:
        candidates.append(explicit)
    if os.environ.get("AXIOM_MCP_BIN"):
        candidates.append(os.environ["AXIOM_MCP_BIN"])
    repo_root = pathlib.Path(__file__).resolve().parents[1]
    # The main checkout keeps axiom-mcp as a sibling; a git worktree lives one level deeper
    # (<repo>/.worktrees/<name>), so both relative locations are tried.
    for ancestor in (repo_root, repo_root.parent, repo_root.parent.parent):
        candidates.append(str(ancestor.parent / "axiom-mcp" / "target" / "release" / "axiom-mcp"))
    for candidate in candidates:
        if candidate and pathlib.Path(candidate).is_file():
            return candidate
    found = shutil.which("axiom-mcp")
    if found:
        return found
    raise FileNotFoundError("axiom-mcp not found; tried: " + ", ".join(candidates) + ", PATH")
# #endregion Tooling.SemanticParseGate.ResolveBinary


# #region Tooling.SemanticParseGate.McpClient [C:2] [TYPE Class] [SEMANTICS semantic,gate,mcp,stdio]
# @BRIEF Minimal MCP stdio client for the two server calls the gate needs.
# @INVARIANT The child process is always terminated, even when a call fails or times out.
class McpClient:
    # #region Tooling.SemanticParseGate.McpClient.Init [C:2] [TYPE Function]
    # @BRIEF Start the server and complete the MCP initialize handshake.
    # @POST Subsequent calls may use tool(name, arguments); raises RuntimeError when the handshake fails.
    def __init__(self, binary: str, ld_path: str | None, timeout: float = 120.0):
        env = dict(os.environ)
        if ld_path:
            env["LD_LIBRARY_PATH"] = ld_path + ":" + env.get("LD_LIBRARY_PATH", "")
        self.timeout = timeout
        self._id = 0
        self.process = subprocess.Popen(
            [binary],
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,
            bufsize=1,
            env=env,
        )
        try:
            self._initialize()
        except Exception:
            self.close()
            raise

    # #endregion Tooling.SemanticParseGate.McpClient.Init

    # #region Tooling.SemanticParseGate.McpClient.Initialize [C:2] [TYPE Function]
    # @BRIEF Complete initialization after the process exists so handshake errors can clean up.
    def _initialize(self):
        self._request("initialize", {
            "protocolVersion": "2024-11-05",
            "capabilities": {},
            "clientInfo": {"name": "semantic-parse-gate", "version": "1"},
        })
        self._notify("notifications/initialized", {})

    # #endregion Tooling.SemanticParseGate.McpClient.Initialize

    # #region Tooling.SemanticParseGate.McpClient.Request [C:2] [TYPE Function]
    def _request(self, method: str, params: dict) -> dict:
        self._id += 1
        request_id = self._id
        self._send({"jsonrpc": "2.0", "id": request_id, "method": method, "params": params})
        deadline = time.monotonic() + self.timeout
        while time.monotonic() < deadline:
            message = self._read(deadline - time.monotonic())
            if message is None:
                continue
            if message.get("id") != request_id:
                continue
            if "error" in message:
                raise RuntimeError(f"{method} failed: {message['error']}")
            return message.get("result", {})
        raise RuntimeError(f"{method} timed out after {self.timeout}s")
    # #endregion Tooling.SemanticParseGate.McpClient.Request

    # #region Tooling.SemanticParseGate.McpClient.Notify [C:1] [TYPE Function]
    def _notify(self, method: str, params: dict) -> None:
        self._send({"jsonrpc": "2.0", "method": method, "params": params})
    # #endregion Tooling.SemanticParseGate.McpClient.Notify

    # #region Tooling.SemanticParseGate.McpClient.Send [C:1] [TYPE Function]
    def _send(self, payload: dict) -> None:
        assert self.process.stdin is not None
        self.process.stdin.write(json.dumps(payload) + chr(10))
        self.process.stdin.flush()
    # #endregion Tooling.SemanticParseGate.McpClient.Send

    # #region Tooling.SemanticParseGate.McpClient.Read [C:1] [TYPE Function]
    def _read(self, timeout: float) -> dict | None:
        import select

        assert self.process.stdout is not None
        readable, _, _ = select.select([self.process.stdout], [], [], max(0.0, timeout))
        if not readable:
            return None
        line = self.process.stdout.readline()
        if not line.strip():
            return None
        return json.loads(line)
    # #endregion Tooling.SemanticParseGate.McpClient.Read

    # #region Tooling.SemanticParseGate.McpClient.Tool [C:2] [TYPE Function]
    # @BRIEF Call one MCP tool and return its structured content.
    def tool(self, name: str, arguments: dict) -> dict:
        result = self._request("tools/call", {"name": name, "arguments": arguments})
        return decode_tool(result)
    # #endregion Tooling.SemanticParseGate.McpClient.Tool

    # #region Tooling.SemanticParseGate.McpClient.NeedsRebuild [C:1] [TYPE Function]
    # @BRIEF Report whether the current index is stale for the workspace (no rebuild is started).
    def needs_rebuild(self, workspace: str) -> bool:
        status = self.tool("search", {"operation": "status", "workspace_path": workspace})
        return bool(status.get("is_stale") or not status.get("index_status", {}).get("is_ready"))

    # #endregion Tooling.SemanticParseGate.McpClient.NeedsRebuild

    # #region Tooling.SemanticParseGate.McpClient.Rebuild [C:2] [TYPE Function]
    # @BRIEF Reparse every source in this process; cold persisted diagnostics are incomplete.
    # @POST Returns the rebuild result mapping; raises RuntimeError when the server reports a failure.
    def rebuild(self, workspace: str) -> dict:
        result = self.tool("search", {"operation": "rebuild", "workspace_path": workspace, "rebuild_mode": "full"})
        validate_rebuild(result, workspace)
        return result
    # #endregion Tooling.SemanticParseGate.McpClient.Rebuild

    # #region Tooling.SemanticParseGate.McpClient.Close [C:1] [TYPE Function]
    def close(self) -> None:
        try:
            if self.process.poll() is None:
                self.process.terminate()
                try:
                    self.process.wait(timeout=5)
                except subprocess.TimeoutExpired:
                    self.process.kill()
        finally:
            for stream in (self.process.stdin, self.process.stdout, self.process.stderr):
                try:
                    if stream:
                        stream.close()
                except Exception:
                    pass
    # #endregion Tooling.SemanticParseGate.McpClient.Close
# #endregion Tooling.SemanticParseGate.McpClient


# #region Tooling.SemanticParseGate.Audit [C:2] [TYPE Function] [SEMANTICS semantic,gate,audit]
# @BRIEF Return a complete audit only after verifying rebuild generation and configuration.
def audit_severity(client: "McpClient", workspace: str, rebuild: dict, expected: str) -> dict:
    audit = client.tool("audit", {
        "operation": "audit_contracts",
        "workspace_path": workspace,
        "detail_level": "summary",
        "limit": 1,
    })
    validate_audit(audit, rebuild, expected)
    return audit
# #endregion Tooling.SemanticParseGate.Audit


# #region Tooling.SemanticParseGate.Arguments [C:2] [TYPE Function]
# @BRIEF Parse the CLI; a reviewed semantic configuration pin is mandatory.
def arguments(argv):
    parser = argparse.ArgumentParser(description="Verified AXIOM parse/schema slice; global semantic acceptance remains required.")
    parser.add_argument("--workspace", default=str(pathlib.Path(__file__).resolve().parents[1]))
    parser.add_argument("--axiom-bin")
    parser.add_argument("--expected-config-hash", default=os.environ.get("AXIOM_EXPECTED_CONFIG_HASH"))
    parser.add_argument("--no-rebuild", action="store_true", help="Unsupported: returns unavailable because cold diagnostics are incomplete.")
    parser.add_argument("--timeout", type=float, default=180.0)
    parser.add_argument("--json", action="store_true")
    return parser.parse_args(argv)
# #endregion Tooling.SemanticParseGate.Arguments


# #region Tooling.SemanticParseGate.Run [C:3] [TYPE Function]
# @BRIEF Full rebuild and audit share exactly one process and one verified generation.
# @RATIONALE Cold DuckDB loads lose schema warnings; incremental rebuild reuses those incomplete nodes.
# @REJECTED Retrying across processes until zero warnings converts a real failure into a false PASS.
def run(args, workspace):
    if args.no_rebuild:
        raise RuntimeError("full rebuild required; --no-rebuild is unavailable")
    verify_config(workspace, args.expected_config_hash)
    binary = resolve_binary(args.axiom_bin)
    ld_path = str(pathlib.Path(binary).parent) if pathlib.Path(binary).parent.name == "release" else None
    client = McpClient(binary, ld_path, timeout=args.timeout)
    try:
        rebuild = client.rebuild(workspace)
        validate_rebuild(rebuild, workspace)
        audit = audit_severity(client, workspace, rebuild, args.expected_config_hash)
        verify_config(workspace, args.expected_config_hash)
    finally:
        client.close()
    result = classify(audit["severity_breakdown"])
    result.update(workspace=workspace, rebuild=rebuild, index_provenance=audit["index_provenance"],
                  audit_attempts=1, config_hash=str(audit["index_provenance"]["config_hash"]),
                  severity_breakdown=audit["severity_breakdown"])
    result["index_provenance"]["config_hash"] = result["config_hash"]
    result["verdict"] = "FAIL" if result["parse_schema_total"] else "PASS"
    return result
# #endregion Tooling.SemanticParseGate.Run


# #region Tooling.SemanticParseGate.Main [C:3] [TYPE Function] [SEMANTICS semantic,gate,cli]
# @BRIEF Return RC0 for verified zero, RC1 for parse/schema violations, RC2 for unavailable evidence.
def main(argv: list[str] | None = None) -> int:
    args = arguments(argv)
    workspace = str(pathlib.Path(args.workspace).resolve())
    try:
        result = run(args, workspace)
    except Exception as error:  # noqa: BLE001 — CLI environment boundary
        print(f"semantic_parse_gate: UNAVAILABLE: {type(error).__name__}: {error}", file=sys.stderr)
        return 2
    if args.json:
        print(json.dumps(result, indent=1, sort_keys=True))
    else:
        print(f"workspace: {workspace}")
        print(f"parse/schema warnings: {result['parse_schema_total']}")
        print(f"size warnings (global INV_7 acceptance required): {result['size_total']}")
        print(f"advisory warnings: {result['advisory_total']}")
        print(f"graph warnings: {result['graph_total']}")
        print(f"generation: {result['index_provenance']['index_generation_id']}")
        print(f"config hash: {result['config_hash']}")
        print(f"parser slice verdict: {result['verdict']}")
    return 1 if result["parse_schema_total"] else 0
# #endregion Tooling.SemanticParseGate.Main


if __name__ == "__main__":
    raise SystemExit(main())
# #endregion Tooling.SemanticParseGate.Module