diff --git a/docs/AI_INTELLIGENCE_MODULE_SOT.md b/docs/AI_INTELLIGENCE_MODULE_SOT.md index f32ae49..ee24f55 100644 --- a/docs/AI_INTELLIGENCE_MODULE_SOT.md +++ b/docs/AI_INTELLIGENCE_MODULE_SOT.md @@ -95,6 +95,7 @@ - 2026-07-02 起 AI automation scheduled health summary 必須提供 machine-readable endpoint；`/api/ai-automation/scheduled-health-summary` 會只讀 smoke history，並可選擇 `include_current_smoke=1` 執行不寫 history 的 current smoke，收斂 AI smoke、PChome drift monitor、history freshness、daily summary delivery readiness 四個 family，輸出 `primary_human_gate_count=0`、`writes_database_count=0`、`next_machine_actions` 與 scheduled output endpoints。此 endpoint 不寄 Telegram、不寫 DB、不改排程，只提供排程/監控可消費的健康摘要。 - 2026-07-02 起 PChome controlled apply rollback evidence 必須提供聚合 endpoint；`/api/ai/pchome-growth/mapping-backlog/direct-mapping-retry-candidate-exception-controlled-apply-rollback-evidence-package` 會聚合 receipt replay、drift verifier、drift recovery、compact readback、artifact retention 五類 evidence，輸出 rollback required / ready actions / protected chain / next machine action。此 endpoint 不執行 rollback、不執行 re-apply、不執行 SQL、不寫 DB；0 drift 時必須輸出 no-op evidence，drift detected 時才輸出 check-mode reapply action。 - 2026-07-02 起 `/metrics` 必須匯出 AI automation scheduled health summary gauges：`momo_ai_automation_scheduled_health_summary_total`、`momo_ai_automation_scheduled_health_family_status`、`momo_ai_automation_scheduled_health_primary_human_gate_count`、`momo_ai_automation_scheduled_health_writes_database_count`。Prometheus scrape 不得寄 Telegram、不寫 DB、不執行 current smoke，只讀 scheduled health summary history。 +- 2026-07-02 起 P4 source / deployment governance 必須提供 machine-readable report：`scripts/ops/report_source_deploy_runtime_truth.py` 會分層輸出 Gitea / origin / local HEAD source truth、部署檔案 SHA256 readback、正式 `/health` runtime truth、optional container readback 與 GitHub freeze / `momo-db` protected / no DB write / no secret read 安全紅線。此 report 是推 Gitea 與正式部署後的 P4 收斂證據，不得把 source-control success 直接等同 deployment success 或 production runtime success。 - V10.644 起 `/ai_intelligence` 的商品明細列不得只用句子描述比價；每列必須顯示 PChome 價格、MOMO 參考價、差距、可信度四格價格證據，並保留下一步按鈕。單位價候選需顯示單位價與單位，候選待確認或缺資料則以「待補 / 候選待確認」呈現，不得捏造價格。 - V10.645 起 `/ai_intelligence` 的商品明細分流切換後，必須顯示「這類商品怎麼處理」的行動摘要，包含件數、近 7 天業績、平均可信度、最大價差、代表商品與主按鈕；使用者不得只能看到商品列表而不知道下一步。 - V10.646 起 `/ai_intelligence` 的商品明細必須提供搜尋與排序；搜尋至少涵蓋商品、分類、商品編號與 MOMO 候選資訊，排序至少支援優先級、近 7 天業績、價差、下滑幅度與可信度。搜尋/排序後的行動摘要與明細列表必須使用同一批結果。 diff --git a/docs/guides/pchome_ai_automation_priority_backlog.md b/docs/guides/pchome_ai_automation_priority_backlog.md index 42e3981..2536afc 100644 --- a/docs/guides/pchome_ai_automation_priority_backlog.md +++ b/docs/guides/pchome_ai_automation_priority_backlog.md @@ -198,6 +198,16 @@ - 不意外 bump version。 - 不 recreate / destroy / prune `momo-db`。 - source-control success、deployment success、production runtime readback 必須分開回報。 +- `scripts/ops/report_source_deploy_runtime_truth.py` 必須可輸出 machine-readable P4 report，明確拆開 local / origin / Gitea refs、部署檔案 hash、正式 `/health`、容器狀態與安全紅線。 + +已完成: + +- Source / deploy / runtime truth report 已建立: + - policy: `p4_source_deployment_runtime_truth_v1` + - source truth: local HEAD、origin `main` / `dev`、Gitea SSH `main` / `dev` + - deployment truth: tracked file SHA256 readback + - runtime truth: production `/health` version/status 與 optional container readback + - safety truth: GitHub freeze、`momo-db` protected、no `--remove-orphans`、no secret read、no DB write 完成標準: @@ -224,6 +234,7 @@ | P3.2 | Scheduled automation health summaries | 已完成 | `/api/ai-automation/scheduled-health-summary` + smoke service focused tests | P3.3 rollback evidence packages | | P3.3 | Rollback evidence packages | 已完成 | controlled apply rollback evidence route + focused tests | P3.4 observability metrics integration | | P3.4 | Observability metrics integration | 已完成 | `/metrics` exports scheduled health summary gauges + focused tests | P4 source / deployment governance ongoing | +| P4.1 | Source / deployment / runtime truth package | 已完成 | `scripts/ops/report_source_deploy_runtime_truth.py` + focused tests | 每次 Gitea push / production deploy 後執行 P4 report | ## 後續回報格式 diff --git a/scripts/ops/report_source_deploy_runtime_truth.py b/scripts/ops/report_source_deploy_runtime_truth.py new file mode 100644 index 0000000..4fdaf55 --- /dev/null +++ b/scripts/ops/report_source_deploy_runtime_truth.py @@ -0,0 +1,402 @@ +#!/usr/bin/env python3 +"""Machine-readable P4 source, deployment, and runtime truth report. + +This report deliberately keeps source-control success, deployment file +readback, and production runtime health as separate evidence layers. It is +read-only: no database writes, no container lifecycle actions, no secret reads. +""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import subprocess +import urllib.error +import urllib.request +from collections.abc import Callable, Iterable +from pathlib import Path +from typing import Any + +from scripts.ops.check_production_version_truth import parse_config_version + + +ROOT = Path(__file__).resolve().parents[2] +DEFAULT_HEALTH_URL = "https://mo.wooo.work/health" +DEFAULT_GITEA_REMOTE = "ssh://git@192.168.0.110:2222/wooo/ewoooc.git" +DEFAULT_TRACKED_FILES = ( + "config.py", + "scripts/ops/check_production_version_truth.py", + "scripts/ops/report_source_deploy_runtime_truth.py", + "docs/guides/pchome_ai_automation_priority_backlog.md", + "docs/AI_INTELLIGENCE_MODULE_SOT.md", +) + +CommandRunner = Callable[[list[str], Path], str] +HealthFetcher = Callable[[str, float], dict[str, Any]] + + +def run_command(args: list[str], cwd: Path) -> str: + result = subprocess.run( + args, + cwd=cwd, + check=True, + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + ) + return result.stdout.strip() + + +def fetch_json(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("runtime health payload must be a JSON object") + return data + + +def _run_git(args: list[str], root: Path, runner: CommandRunner) -> str: + return runner(["git", *args], root) + + +def parse_ls_remote(output: str) -> dict[str, str]: + refs: dict[str, str] = {} + for line in output.splitlines(): + parts = line.split() + if len(parts) >= 2: + refs[parts[1]] = parts[0] + return refs + + +def read_working_tree_config_version(root: Path) -> str: + return parse_config_version((root / "config.py").read_text(encoding="utf-8")) + + +def read_head_config_version(root: Path, runner: CommandRunner) -> str: + return parse_config_version(_run_git(["show", "HEAD:config.py"], root, runner)) + + +def collect_source_control( + root: Path, + gitea_remote: str, + tracked_files: Iterable[str], + runner: CommandRunner = run_command, +) -> dict[str, Any]: + origin_refs = parse_ls_remote( + _run_git(["ls-remote", "origin", "refs/heads/main", "refs/heads/dev"], root, runner) + ) + gitea_refs = parse_ls_remote( + _run_git(["ls-remote", gitea_remote, "refs/heads/main", "refs/heads/dev"], root, runner) + ) + local_head = _run_git(["rev-parse", "HEAD"], root, runner) + + return { + "truth_source": "Gitea", + "local": { + "branch": _run_git(["rev-parse", "--abbrev-ref", "HEAD"], root, runner), + "head": local_head, + "working_tree_config_version": read_working_tree_config_version(root), + "head_config_version": read_head_config_version(root, runner), + "tracked_file_status": _run_git( + ["status", "--porcelain", "--", *tracked_files], + root, + runner, + ).splitlines(), + }, + "origin": { + "remote": _run_git(["remote", "get-url", "origin"], root, runner), + "main": origin_refs.get("refs/heads/main"), + "dev": origin_refs.get("refs/heads/dev"), + }, + "gitea_ssh": { + "remote": gitea_remote, + "main": gitea_refs.get("refs/heads/main"), + "dev": gitea_refs.get("refs/heads/dev"), + }, + } + + +def sha256_file(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as handle: + for chunk in iter(lambda: handle.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + +def collect_deployment_files(root: Path, tracked_files: Iterable[str]) -> dict[str, Any]: + files: list[dict[str, Any]] = [] + for relpath in tracked_files: + path = root / relpath + exists = path.is_file() + files.append( + { + "path": relpath, + "exists": exists, + "sha256": sha256_file(path) if exists else None, + "size_bytes": path.stat().st_size if exists else None, + } + ) + + return { + "truth_source": "deployed_file_hash_readback", + "source_root": str(root), + "tracked_file_count": len(files), + "files": files, + } + + +def collect_container_state( + container_name: str | None, + root: Path, + runner: CommandRunner = run_command, +) -> dict[str, Any]: + if not container_name: + return {"requested": False, "name": None, "status": "skipped"} + + raw_state = runner(["docker", "inspect", "--format", "{{json .State}}", container_name], root) + state = json.loads(raw_state) + health = state.get("Health") or {} + return { + "requested": True, + "name": container_name, + "status": state.get("Status"), + "running": bool(state.get("Running")), + "health_status": health.get("Status"), + } + + +def collect_runtime( + health_url: str, + timeout: float, + root: Path, + container_name: str | None = None, + runner: CommandRunner = run_command, + health_fetcher: HealthFetcher = fetch_json, +) -> dict[str, Any]: + health = health_fetcher(health_url, timeout) + return { + "truth_source": "production_runtime_readback", + "health_url": health_url, + "health": { + "status": health.get("status"), + "database": health.get("database"), + "version": health.get("version"), + }, + "container": collect_container_state(container_name, root, runner), + } + + +def safety_gates() -> dict[str, Any]: + return { + "github_freeze_enforced": True, + "github_allowed_actions": 0, + "momo_db_protected": True, + "remove_orphans_forbidden": True, + "version_bump_forbidden_in_this_lane": True, + "secret_read_performed": False, + "database_write_performed": False, + "destructive_container_action_performed": False, + } + + +def summarize(report: dict[str, Any]) -> dict[str, Any]: + source = report["source_control"] + local_head = source["local"]["head"] + source_refs = [ + source["origin"]["main"], + source["origin"]["dev"], + source["gitea_ssh"]["main"], + source["gitea_ssh"]["dev"], + ] + source_control_ok = all(ref == local_head for ref in source_refs) + tracked_files_committed = not source["local"]["tracked_file_status"] + + runtime = report["runtime"] + production_version = runtime["health"]["version"] + head_version = source["local"]["head_config_version"] + working_tree_version = source["local"]["working_tree_config_version"] + production_health_ok = runtime["health"]["status"] == "healthy" + production_version_matches_head = production_version == head_version + version_bump_detected = working_tree_version != head_version or head_version != production_version + + deployment = report["deployment"] + deployment_hash_readback_ok = all(file["exists"] and file["sha256"] for file in deployment["files"]) + + container = runtime["container"] + container_readback_ok = not container["requested"] or ( + container.get("running") is True + and container.get("status") == "running" + and container.get("health_status") in {"healthy", None} + ) + + gates = report["safety_gates"] + safety_ok = ( + gates["github_freeze_enforced"] + and gates["github_allowed_actions"] == 0 + and gates["momo_db_protected"] + and gates["remove_orphans_forbidden"] + and not gates["secret_read_performed"] + and not gates["database_write_performed"] + and not gates["destructive_container_action_performed"] + ) + + return { + "source_control_ok": source_control_ok, + "tracked_files_committed": tracked_files_committed, + "deployment_hash_readback_ok": deployment_hash_readback_ok, + "production_health_ok": production_health_ok, + "production_version_matches_head": production_version_matches_head, + "version_bump_detected": version_bump_detected, + "container_readback_ok": container_readback_ok, + "github_freeze_enforced": gates["github_freeze_enforced"], + "momo_db_protected": gates["momo_db_protected"], + "truth_layers_separated": True, + "success": all( + [ + source_control_ok, + tracked_files_committed, + deployment_hash_readback_ok, + production_health_ok, + production_version_matches_head, + not version_bump_detected, + container_readback_ok, + safety_ok, + ] + ), + } + + +def evaluate(report: dict[str, Any]) -> tuple[bool, list[str]]: + summary = report["summary"] + errors: list[str] = [] + if not summary["source_control_ok"]: + errors.append("local HEAD, origin main/dev, and Gitea SSH main/dev are not aligned") + if not summary["tracked_files_committed"]: + errors.append("tracked deployment files have uncommitted source-control changes") + if not summary["deployment_hash_readback_ok"]: + errors.append("one or more tracked deployment files are missing or lack hash readback") + if not summary["production_health_ok"]: + errors.append("production runtime health is not healthy") + if not summary["production_version_matches_head"]: + errors.append("production /health version does not match HEAD config.py") + if summary["version_bump_detected"]: + errors.append("unexpected version drift or bump detected") + if not summary["container_readback_ok"]: + errors.append("container readback was requested but did not return running/healthy") + if not summary["github_freeze_enforced"]: + errors.append("GitHub freeze is not enforced") + if not summary["momo_db_protected"]: + errors.append("momo-db protection gate is not set") + return not errors, errors + + +def build_report( + *, + root: Path = ROOT, + health_url: str = DEFAULT_HEALTH_URL, + timeout: float = 10.0, + gitea_remote: str = DEFAULT_GITEA_REMOTE, + tracked_files: Iterable[str] = DEFAULT_TRACKED_FILES, + container_name: str | None = None, + runner: CommandRunner = run_command, + health_fetcher: HealthFetcher = fetch_json, +) -> dict[str, Any]: + tracked_file_tuple = tuple(tracked_files) + report = { + "policy": "p4_source_deployment_runtime_truth_v1", + "source_control": collect_source_control(root, gitea_remote, tracked_file_tuple, runner), + "deployment": collect_deployment_files(root, tracked_file_tuple), + "runtime": collect_runtime(health_url, timeout, root, container_name, runner, health_fetcher), + "safety_gates": safety_gates(), + } + report["summary"] = summarize(report) + ok, errors = evaluate(report) + report["result"] = "PASS" if ok else "BLOCKED" + report["errors"] = errors + return report + + +def _short_sha(value: str | None) -> str: + return value[:12] if value else "missing" + + +def format_text(report: dict[str, Any]) -> str: + source = report["source_control"] + runtime = report["runtime"] + summary = report["summary"] + container = runtime["container"] + + lines = [ + "source_deploy_runtime_truth:", + f"- policy: {report['policy']}", + f"- result: {report['result']}", + f"- local_branch: {source['local']['branch']}", + f"- local_head: {_short_sha(source['local']['head'])}", + f"- origin_main: {_short_sha(source['origin']['main'])}", + f"- origin_dev: {_short_sha(source['origin']['dev'])}", + f"- gitea_main: {_short_sha(source['gitea_ssh']['main'])}", + f"- gitea_dev: {_short_sha(source['gitea_ssh']['dev'])}", + f"- tracked_files_committed: {str(summary['tracked_files_committed']).lower()}", + f"- production_health: {runtime['health']['status']} {runtime['health']['database']} {runtime['health']['version']}", + f"- version_bump_detected: {str(summary['version_bump_detected']).lower()}", + f"- deployment_files_hashed: {report['deployment']['tracked_file_count']}", + f"- container: {container.get('name') or 'not_requested'} {container.get('status')}", + f"- truth_layers_separated: {str(summary['truth_layers_separated']).lower()}", + ] + for error in report["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("--source-root", type=Path, default=ROOT) + parser.add_argument("--health-url", default=DEFAULT_HEALTH_URL) + parser.add_argument("--timeout", type=float, default=10.0) + parser.add_argument("--gitea-remote", default=DEFAULT_GITEA_REMOTE) + parser.add_argument("--container-name") + parser.add_argument("--tracked-file", action="append", dest="tracked_files") + parser.add_argument("--json", action="store_true", help="Print machine-readable report") + args = parser.parse_args(argv) + + tracked_files = tuple(args.tracked_files) if args.tracked_files else DEFAULT_TRACKED_FILES + + try: + report = build_report( + root=args.source_root.resolve(), + health_url=args.health_url, + timeout=args.timeout, + gitea_remote=args.gitea_remote, + tracked_files=tracked_files, + container_name=args.container_name, + ) + except ( + OSError, + ValueError, + json.JSONDecodeError, + subprocess.CalledProcessError, + urllib.error.URLError, + ) as exc: + error_report = { + "policy": "p4_source_deployment_runtime_truth_v1", + "result": "BLOCKED", + "errors": [str(exc)], + } + if args.json: + print(json.dumps(error_report, ensure_ascii=False, indent=2)) + else: + print("source_deploy_runtime_truth:\n- result: BLOCKED\n- blocker: " + str(exc)) + return 2 + + if args.json: + print(json.dumps(report, ensure_ascii=False, indent=2)) + else: + print(format_text(report)) + return 0 if report["result"] == "PASS" else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_source_deploy_runtime_truth_report.py b/tests/test_source_deploy_runtime_truth_report.py new file mode 100644 index 0000000..97fc4ef --- /dev/null +++ b/tests/test_source_deploy_runtime_truth_report.py @@ -0,0 +1,171 @@ +import json +from pathlib import Path + +from scripts.ops import report_source_deploy_runtime_truth as report + + +LOCAL_HEAD = "a" * 40 +OTHER_HEAD = "b" * 40 + + +def _write_source_root(tmp_path: Path) -> Path: + (tmp_path / "config.py").write_text('SYSTEM_VERSION = "V10.725"\n', encoding="utf-8") + (tmp_path / "proof.txt").write_text("deployed proof\n", encoding="utf-8") + return tmp_path + + +def _runner( + *, + origin_main: str = LOCAL_HEAD, + origin_dev: str = LOCAL_HEAD, + gitea_main: str = LOCAL_HEAD, + gitea_dev: str = LOCAL_HEAD, + head_config_version: str = "V10.725", + tracked_file_status: str = "", + container_state: dict | None = None, +): + def run(args: list[str], cwd: Path) -> str: + if args == ["git", "rev-parse", "HEAD"]: + return LOCAL_HEAD + if args == ["git", "rev-parse", "--abbrev-ref", "HEAD"]: + return "codex/prod-version-truth-guard" + if args == ["git", "show", "HEAD:config.py"]: + return f'SYSTEM_VERSION = "{head_config_version}"\n' + if args == ["git", "remote", "get-url", "origin"]: + return "https://gitea.wooo.work/wooo/ewoooc.git" + if args[:3] == ["git", "ls-remote", "origin"]: + return "\n".join( + [ + f"{origin_main}\trefs/heads/main", + f"{origin_dev}\trefs/heads/dev", + ] + ) + if args[:3] == ["git", "ls-remote", report.DEFAULT_GITEA_REMOTE]: + return "\n".join( + [ + f"{gitea_main}\trefs/heads/main", + f"{gitea_dev}\trefs/heads/dev", + ] + ) + if args[:3] == ["git", "status", "--porcelain"]: + return tracked_file_status + if args[:4] == ["docker", "inspect", "--format", "{{json .State}}"]: + state = container_state or { + "Status": "running", + "Running": True, + "Health": {"Status": "healthy"}, + } + return json.dumps(state) + raise AssertionError(f"unexpected command: {args}") + + return run + + +def _health(version: str = "V10.725", status: str = "healthy"): + def fetch(url: str, timeout: float) -> dict: + return {"status": status, "database": "postgresql", "version": version} + + return fetch + + +def test_report_passes_when_source_deploy_runtime_truth_aligns(tmp_path): + source_root = _write_source_root(tmp_path) + + payload = report.build_report( + root=source_root, + tracked_files=("config.py", "proof.txt"), + container_name="momo-pro-system", + runner=_runner(), + health_fetcher=_health(), + ) + + assert payload["result"] == "PASS" + assert payload["summary"]["source_control_ok"] is True + assert payload["summary"]["tracked_files_committed"] is True + assert payload["summary"]["deployment_hash_readback_ok"] is True + assert payload["summary"]["production_health_ok"] is True + assert payload["summary"]["truth_layers_separated"] is True + assert payload["runtime"]["container"]["health_status"] == "healthy" + assert payload["safety_gates"]["github_allowed_actions"] == 0 + assert payload["safety_gates"]["database_write_performed"] is False + + +def test_report_blocks_when_gitea_main_differs_from_local_head(tmp_path): + source_root = _write_source_root(tmp_path) + + payload = report.build_report( + root=source_root, + tracked_files=("config.py", "proof.txt"), + runner=_runner(gitea_main=OTHER_HEAD), + health_fetcher=_health(), + ) + + assert payload["result"] == "BLOCKED" + assert payload["summary"]["source_control_ok"] is False + assert any("Gitea SSH main/dev are not aligned" in error for error in payload["errors"]) + + +def test_report_blocks_when_tracked_deployment_file_is_not_committed(tmp_path): + source_root = _write_source_root(tmp_path) + + payload = report.build_report( + root=source_root, + tracked_files=("config.py", "proof.txt"), + runner=_runner(tracked_file_status=" M proof.txt"), + health_fetcher=_health(), + ) + + assert payload["result"] == "BLOCKED" + assert payload["summary"]["tracked_files_committed"] is False + assert any("uncommitted source-control changes" in error for error in payload["errors"]) + + +def test_report_blocks_when_production_version_differs_from_head(tmp_path): + source_root = _write_source_root(tmp_path) + + payload = report.build_report( + root=source_root, + tracked_files=("config.py", "proof.txt"), + runner=_runner(), + health_fetcher=_health(version="V10.724"), + ) + + assert payload["result"] == "BLOCKED" + assert payload["summary"]["production_version_matches_head"] is False + assert payload["summary"]["version_bump_detected"] is True + assert any("production /health version does not match HEAD config.py" in error for error in payload["errors"]) + + +def test_missing_deployment_file_blocks_only_the_deployment_layer(tmp_path): + source_root = _write_source_root(tmp_path) + + payload = report.build_report( + root=source_root, + tracked_files=("config.py", "missing.txt"), + runner=_runner(), + health_fetcher=_health(), + ) + + assert payload["result"] == "BLOCKED" + assert payload["summary"]["source_control_ok"] is True + assert payload["summary"]["deployment_hash_readback_ok"] is False + assert any("tracked deployment files" in error for error in payload["errors"]) + + +def test_text_output_exposes_source_deployment_and_runtime_layers(tmp_path): + source_root = _write_source_root(tmp_path) + payload = report.build_report( + root=source_root, + tracked_files=("config.py", "proof.txt"), + runner=_runner(), + health_fetcher=_health(), + ) + + text = report.format_text(payload) + + assert "origin_main:" in text + assert "gitea_main:" in text + assert "tracked_files_committed: true" in text + assert "production_health: healthy postgresql V10.725" in text + assert "deployment_files_hashed: 2" in text + assert "truth_layers_separated: true" in text