Add PChome AI controlled dry-run closeout chain

scripts/ops/check_production_version_truth.py

@@ -0,0 +1,176 @@
+#!/usr/bin/env python3
+"""Read-only guard for EwoooC production version truth.
+
+Production /health is the authoritative latest runtime version. Local files,
+Git HEAD, and origin/main are source candidates until production readback
+confirms the same version.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import re
+import subprocess
+import sys
+import urllib.error
+import urllib.request
+from pathlib import Path
+from typing import Any
+
+
+ROOT = Path(__file__).resolve().parents[2]
+DEFAULT_HEALTH_URL = "https://mo.wooo.work/health"
+VERSION_RE = re.compile(r'^SYSTEM_VERSION\s*=\s*["\']([^"\']+)["\']', re.MULTILINE)
+
+
+def _run_git(args: list[str], cwd: Path = ROOT) -> str:
+    result = subprocess.run(
+        ["git", *args],
+        cwd=cwd,
+        check=True,
+        text=True,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+    )
+    return result.stdout.strip()
+
+
+def parse_config_version(source: str) -> str:
+    match = VERSION_RE.search(source)
+    if not match:
+        raise ValueError("SYSTEM_VERSION not found")
+    return match.group(1)
+
+
+def read_local_config_version(root: Path = ROOT) -> str:
+    return parse_config_version((root / "config.py").read_text(encoding="utf-8"))
+
+
+def read_head_config_version() -> str:
+    return parse_config_version(_run_git(["show", "HEAD:config.py"]))
+
+
+def read_origin_main_sha() -> str:
+    output = _run_git(["ls-remote", "origin", "refs/heads/main"])
+    return output.split()[0]
+
+
+def fetch_health(url: str, timeout: float) -> dict[str, Any]:
+    with urllib.request.urlopen(url, timeout=timeout) as response:
+        payload = response.read().decode("utf-8")
+    data = json.loads(payload)
+    if not isinstance(data, dict):
+        raise ValueError("health payload must be a JSON object")
+    return data
+
+
+def build_report(health_url: str, timeout: float) -> dict[str, Any]:
+    health = fetch_health(health_url, timeout)
+    local_sha = _run_git(["rev-parse", "HEAD"])
+    local_branch = _run_git(["rev-parse", "--abbrev-ref", "HEAD"])
+    origin_sha = read_origin_main_sha()
+
+    return {
+        "policy": "production_health_is_latest_version_truth",
+        "health_url": health_url,
+        "production": {
+            "status": health.get("status"),
+            "database": health.get("database"),
+            "version": health.get("version"),
+        },
+        "local": {
+            "branch": local_branch,
+            "head": local_sha,
+            "config_version": read_local_config_version(),
+            "head_config_version": read_head_config_version(),
+        },
+        "origin_main": {
+            "head": origin_sha,
+            "matches_local_head": origin_sha == local_sha,
+        },
+    }
+
+
+def evaluate(report: dict[str, Any], allow_local_version_drift: bool) -> tuple[bool, list[str]]:
+    errors: list[str] = []
+    production = report["production"]
+    local = report["local"]
+
+    if production["status"] != "healthy":
+        errors.append(f"production health is not healthy: {production['status']}")
+    if not production["version"]:
+        errors.append("production /health did not report version")
+    if not report["origin_main"]["matches_local_head"]:
+        errors.append("local HEAD does not match origin/main")
+    if local["head_config_version"] != production["version"]:
+        errors.append(
+            "HEAD config.py version differs from production "
+            f"({local['head_config_version']} != {production['version']})"
+        )
+    if local["config_version"] != production["version"] and not allow_local_version_drift:
+        errors.append(
+            "working-tree config.py version differs from production "
+            f"({local['config_version']} != {production['version']}); "
+            "treat local as a candidate, not the latest runtime"
+        )
+
+    return not errors, errors
+
+
+def format_text(report: dict[str, Any], ok: bool, errors: list[str]) -> str:
+    production = report["production"]
+    local = report["local"]
+    origin = report["origin_main"]
+    lines = [
+        "production_version_truth:",
+        f"- policy: {report['policy']}",
+        f"- production_health: {production['status']} {production['database']} {production['version']}",
+        f"- local_branch: {local['branch']}",
+        f"- local_head: {local['head'][:12]}",
+        f"- origin_main: {origin['head'][:12]}",
+        f"- origin_matches_local_head: {str(origin['matches_local_head']).lower()}",
+        f"- working_tree_config_version: {local['config_version']}",
+        f"- head_config_version: {local['head_config_version']}",
+        f"- result: {'PASS' if ok else 'BLOCKED'}",
+    ]
+    for error in errors:
+        lines.append(f"- blocker: {error}")
+    return "\n".join(lines)
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--health-url", default=DEFAULT_HEALTH_URL)
+    parser.add_argument("--timeout", type=float, default=10.0)
+    parser.add_argument("--json", action="store_true", help="Print machine-readable report")
+    parser.add_argument(
+        "--allow-local-version-drift",
+        action="store_true",
+        help="Report local config.py drift without failing; use only for explicit release prep.",
+    )
+    args = parser.parse_args(argv)
+
+    try:
+        report = build_report(args.health_url, args.timeout)
+        ok, errors = evaluate(report, args.allow_local_version_drift)
+    except (OSError, ValueError, subprocess.CalledProcessError, urllib.error.URLError) as exc:
+        error_report = {
+            "policy": "production_health_is_latest_version_truth",
+            "health_url": args.health_url,
+            "result": "BLOCKED",
+            "errors": [str(exc)],
+        }
+        print(json.dumps(error_report, ensure_ascii=False, indent=2) if args.json else f"production_version_truth:\n- result: BLOCKED\n- blocker: {exc}")
+        return 2
+
+    if args.json:
+        output = {**report, "result": "PASS" if ok else "BLOCKED", "errors": errors}
+        print(json.dumps(output, ensure_ascii=False, indent=2))
+    else:
+        print(format_text(report, ok, errors))
+    return 0 if ok else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())