強化 P4 source deployment runtime truth 回報
Some checks failed
CD Pipeline / deploy (push) Has been cancelled

This commit is contained in:
ogt
2026-07-02 15:04:41 +08:00
parent 3e83973db3
commit edc3f1fdc3
4 changed files with 585 additions and 0 deletions

View File

@@ -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、不寫 DB0 drift 時必須輸出 no-op evidencedrift 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 天業績、價差、下滑幅度與可信度。搜尋/排序後的行動摘要與明細列表必須使用同一批結果。

View File

@@ -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 |
## 後續回報格式

View File

@@ -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())

View File

@@ -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