From b9e4948f99f15cd17bae77685564b3c8b5def1b9 Mon Sep 17 00:00:00 2001 From: ogt Date: Sat, 11 Jul 2026 12:15:01 +0800 Subject: [PATCH] refactor(pchome): extract controlled drift verifier --- config.py | 2 +- docs/AI_INTELLIGENCE_MODULE_SOT.md | 2 +- .../code_modularization_inventory_20260430.md | 2 +- .../controlled_apply_verifier.py | 190 ++++++++++++++++++ services/pchome_mapping_backlog_service.py | 164 +-------------- ...t_pchome_mapping_backlog_modularization.py | 44 +++- 6 files changed, 245 insertions(+), 159 deletions(-) create mode 100644 services/pchome_mapping_backlog/controlled_apply_verifier.py diff --git a/config.py b/config.py index 377a424..ec69a6f 100644 --- a/config.py +++ b/config.py @@ -402,7 +402,7 @@ YOUTUBE_API_KEY = os.getenv('YOUTUBE_API_KEY', '') # ========================================== # 系統版本與路徑 # ========================================== -SYSTEM_VERSION = "V10.781" +SYSTEM_VERSION = "V10.782" LOG_FILE_PATH = os.path.join(BASE_DIR, 'logs/system.log') public_url = PUBLIC_URL # 用於模板顯示 diff --git a/docs/AI_INTELLIGENCE_MODULE_SOT.md b/docs/AI_INTELLIGENCE_MODULE_SOT.md index b139c97..cfd23b4 100644 --- a/docs/AI_INTELLIGENCE_MODULE_SOT.md +++ b/docs/AI_INTELLIGENCE_MODULE_SOT.md @@ -2,7 +2,7 @@ > **最後更新**: 2026-07-11 (台北時間) > **狀態**: 🟠 Partial。四 Agent、PixelRAG、MCP/RAG registry、PChome controlled-preview families 與多項 smoke/readback 已建立；但 access control、database identity/RBAC、full asset reconciliation、software supply chain、same-run controlled apply、restore drill、internal RAG canary 與 MCP/RAG runtime closure 尚未全部完成。任何局部 receipt 不得代表整體閉環完成。 -> **適用版本**: V10.781 +> **適用版本**: V10.782 --- diff --git a/docs/memory/code_modularization_inventory_20260430.md b/docs/memory/code_modularization_inventory_20260430.md index a6162bd..bcad7e5 100644 --- a/docs/memory/code_modularization_inventory_20260430.md +++ b/docs/memory/code_modularization_inventory_20260430.md @@ -129,7 +129,7 @@ | 行數 | 檔案 | 分類 | 拆分方向 | |---:|---|---|---| -| 43946 | `services/pchome_mapping_backlog_service.py` | P0 gate 巨型 service、拆分進行中 | policy、AI exception contract、evidence parser、artifact path guard 與 bounded controlled executor 已移到 `services/pchome_mapping_backlog/`；下一步依序拆 verifier、receipt、reporter | +| 43802 | `services/pchome_mapping_backlog_service.py` | P0 gate 巨型 service、拆分進行中 | policy、AI exception contract、evidence parser、artifact guard、bounded executor 與 no-write drift verifier 已移到 `services/pchome_mapping_backlog/`；下一步依序拆 receipt、reporter | | 14289 | `services/ai_automation_smoke_service.py` | P0 smoke 巨型 service | 拆 family registry、collectors、health projection、metrics adapter | | 9383 | `routes/openclaw_bot_routes.py` | P0 巨型 Blueprint、拆分進行中 | scheduler hook 已移到 `services/openclaw_bot/scheduled_jobs.py`；下一步拆 report、command | | 7641 | `routes/ai_routes.py` | P0 巨型 Blueprint | PChome growth、AI automation、recommendation route extension 分離 | diff --git a/services/pchome_mapping_backlog/controlled_apply_verifier.py b/services/pchome_mapping_backlog/controlled_apply_verifier.py new file mode 100644 index 0000000..b3a890e --- /dev/null +++ b/services/pchome_mapping_backlog/controlled_apply_verifier.py @@ -0,0 +1,190 @@ +"""No-write drift verifier for controlled PChome mapping receipts.""" + +from __future__ import annotations + +import hashlib +import json +from pathlib import Path +from typing import Any + +from services.pchome_mapping_backlog.artifacts import ( + _canonical_retry_exception_artifact_bytes, + _resolve_retry_exception_artifact_path, +) +from services.pchome_mapping_backlog.policies import ( + DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CONTROLLED_APPLY_DRIFT_VERIFIER_POLICY, +) + + +def build_controlled_apply_drift_verifier_from_replay( + replay: dict[str, Any], + *, + root: Path, + run_id: str | None = None, + engine: Any = None, + source_receipt_replay: dict[str, Any] | None = None, + materialize_artifacts: bool = False, +) -> dict[str, Any]: + """Verify replay readbacks and emit a bounded drift receipt.""" + replay_summary = replay.get("summary") or {} + readbacks = list(replay.get("post_apply_readbacks") or []) + drift_items = [item for item in readbacks if item.get("passed") is not True] + selector_count = int(replay_summary.get("target_selector_count") or 0) + pass_count = int(replay_summary.get("post_apply_readback_pass_count") or 0) + receipt_hash_match_count = int(replay_summary.get("executor_receipt_hash_match_count") or 0) + source_ready = ( + replay.get("result") == "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_RECEIPT_REPLAYED" + and selector_count > 0 + and receipt_hash_match_count > 0 + ) + drift_verified = source_ready and not drift_items and pass_count == selector_count + + if drift_items: + result = "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_DRIFT_DETECTED" + elif drift_verified: + result = "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_DRIFT_VERIFIED" + elif replay.get("missing_artifacts"): + result = "WAITING_FOR_RETRY_EXCEPTION_CONTROLLED_APPLY_DRIFT_ARTIFACTS" + else: + result = "WAITING_FOR_RETRY_EXCEPTION_CONTROLLED_APPLY_DRIFT_BASELINE" + + checks = [ + {"check": "source_replay_loaded", "passed": bool(replay)}, + {"check": "source_receipt_hash_matches", "passed": receipt_hash_match_count > 0}, + {"check": "target_selectors_present", "passed": selector_count > 0}, + {"check": "all_current_readbacks_match_receipt", "passed": not drift_items and pass_count == selector_count}, + {"check": "drift_verifier_does_not_write_database", "passed": True}, + ] + verifier_id_payload = { + "run_id": (replay.get("receipt_replay") or {}).get("run_id") or run_id or "", + "result": result, + "post_apply_readbacks": readbacks, + } + verifier_id = ( + "pchome-retry-exception-controlled-apply-drift-verifier-" + + hashlib.sha256( + json.dumps(verifier_id_payload, ensure_ascii=False, sort_keys=True, default=str).encode("utf-8") + ).hexdigest()[:16] + ) + summary = { + "target_selector_count": selector_count, + "post_apply_readback_count": int(replay_summary.get("post_apply_readback_count") or 0), + "post_apply_readback_pass_count": pass_count, + "drift_count": len(drift_items), + "drift_verified_count": 1 if drift_verified else 0, + "receipt_hash_match_count": receipt_hash_match_count, + "missing_artifact_count": int(replay_summary.get("missing_artifact_count") or 0), + "drift_verifier_artifact_materialized_count": 0, + "drift_verifier_artifact_hash_match_count": 0, + "writes_database_count": 0, + } + safety = { + "ai_controlled_apply": True, + "reads_artifact_files": True, + "reads_database": engine is not None or bool(source_receipt_replay), + "writes_database": False, + "writes_database_count": 0, + "writes_artifact_count": 0, + "syncs_external_offers": False, + "dispatches_telegram": False, + "gemini_allowed": False, + "requires_production_version_truth": True, + } + artifact_payload = { + "artifact_key": "retry_exception_controlled_apply_drift_verifier_receipt", + "verifier_id": verifier_id, + "run_id": verifier_id_payload["run_id"], + "source_policy": DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CONTROLLED_APPLY_DRIFT_VERIFIER_POLICY, + "source_receipt_replay_result": replay.get("result"), + "result": result, + "summary": summary, + "drift_items": drift_items, + "post_apply_readbacks": readbacks, + "checks": checks, + "safety": safety, + } + artifact_bytes = _canonical_retry_exception_artifact_bytes(artifact_payload) + artifact_relative_path = ( + f"artifacts/pchome_growth/retry_exception_closeout/" + f"controlled_apply_drift_verifier/{verifier_id}.json" + ) + drift_verifier_artifact = { + "key": "retry_exception_controlled_apply_drift_verifier_receipt", + "artifact_type": "controlled_apply_drift_verifier_receipt", + "relative_path": artifact_relative_path, + "payload_sha256": hashlib.sha256(artifact_bytes).hexdigest(), + "byte_count": len(artifact_bytes), + "payload": artifact_payload, + "materialized": False, + "writes_database": False, + } + materialized_drift_artifacts: list[dict[str, Any]] = [] + if materialize_artifacts and selector_count: + target_path = _resolve_retry_exception_artifact_path(root, artifact_relative_path) + target_path.parent.mkdir(parents=True, exist_ok=True) + target_path.write_bytes(artifact_bytes) + materialized_drift_artifacts.append({ + "key": drift_verifier_artifact["key"], + "relative_path": artifact_relative_path, + "absolute_path": str(target_path), + "payload_sha256": drift_verifier_artifact["payload_sha256"], + "written_byte_count": target_path.stat().st_size, + "writes_database": False, + }) + drift_verifier_artifact["materialized"] = True + drift_verifier_artifact["absolute_path"] = str(target_path) + artifact_path = _resolve_retry_exception_artifact_path(root, artifact_relative_path) + artifact_sha = hashlib.sha256(artifact_path.read_bytes()).hexdigest() if artifact_path.exists() else "" + artifact_hash_match = bool(artifact_sha) and artifact_sha == drift_verifier_artifact["payload_sha256"] + summary["drift_verifier_artifact_materialized_count"] = len(materialized_drift_artifacts) or (1 if artifact_hash_match else 0) + summary["drift_verifier_artifact_hash_match_count"] = 1 if artifact_hash_match else 0 + safety["writes_artifact_count"] = len(materialized_drift_artifacts) + checks.extend([ + { + "check": "drift_artifact_materialized_when_requested", + "passed": (not materialize_artifacts) or (selector_count > 0 and artifact_path.exists()), + }, + { + "check": "drift_artifact_hash_matches_expected", + "passed": (not materialize_artifacts) or artifact_hash_match, + }, + ]) + return { + "policy": DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CONTROLLED_APPLY_DRIFT_VERIFIER_POLICY, + "result": result, + "success": result == "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_DRIFT_VERIFIED", + "summary": summary, + "drift_verifier": { + "verifier_id": verifier_id, + "stage": "P2_retry_exception_controlled_apply_drift_verifier", + "status": result, + "source_receipt_replay_result": replay.get("result"), + "ready": drift_verified, + "materialize_artifacts": bool(materialize_artifacts), + "requires_production_version_truth": True, + }, + "drift_items": drift_items, + "post_apply_readbacks": readbacks, + "source_receipt_replay_summary": replay_summary, + "drift_verifier_artifact": drift_verifier_artifact, + "materialized_drift_artifacts": materialized_drift_artifacts, + "post_drift_verifier_artifact_verifier": { + "expected_sha256": drift_verifier_artifact["payload_sha256"], + "actual_sha256": artifact_sha, + "hash_match": artifact_hash_match, + "writes_database": False, + }, + "checks": checks, + "check_count": len(checks), + "all_checks_passed": all(check.get("passed") is True for check in checks), + "next_actions": [ + "Keep this verifier on the readiness surface so DB drift is visible without manual table review.", + "If drift is detected, use the receipt replay readbacks as rollback or re-apply evidence.", + ], + "safety": safety, + } + + +__all__ = ( + "build_controlled_apply_drift_verifier_from_replay", +) diff --git a/services/pchome_mapping_backlog_service.py b/services/pchome_mapping_backlog_service.py index c8e6ef9..7ae267a 100644 --- a/services/pchome_mapping_backlog_service.py +++ b/services/pchome_mapping_backlog_service.py @@ -142,6 +142,9 @@ from services.pchome_mapping_backlog.controlled_apply_executor import ( _upsert_pchome_product_match, build_controlled_apply_executor_from_preflight, ) +from services.pchome_mapping_backlog.controlled_apply_verifier import ( + build_controlled_apply_drift_verifier_from_replay, +) from services.pchome_mapping_backlog.evidence import ( _action_code, _action_label, @@ -3879,163 +3882,16 @@ def build_pchome_direct_mapping_retry_candidate_exception_controlled_apply_drift materialize_artifacts=False, engine=engine, ) - replay_summary = replay.get("summary") or {} - readbacks = list(replay.get("post_apply_readbacks") or []) - drift_items = [item for item in readbacks if item.get("passed") is not True] - selector_count = int(replay_summary.get("target_selector_count") or 0) - pass_count = int(replay_summary.get("post_apply_readback_pass_count") or 0) - receipt_hash_match_count = int(replay_summary.get("executor_receipt_hash_match_count") or 0) - source_ready = ( - replay.get("result") == "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_RECEIPT_REPLAYED" - and selector_count > 0 - and receipt_hash_match_count > 0 + return build_controlled_apply_drift_verifier_from_replay( + replay, + root=root, + run_id=run_id, + engine=engine, + source_receipt_replay=source_receipt_replay, + materialize_artifacts=materialize_artifacts, ) - drift_verified = source_ready and not drift_items and pass_count == selector_count - if drift_items: - result = "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_DRIFT_DETECTED" - elif drift_verified: - result = "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_DRIFT_VERIFIED" - elif replay.get("missing_artifacts"): - result = "WAITING_FOR_RETRY_EXCEPTION_CONTROLLED_APPLY_DRIFT_ARTIFACTS" - else: - result = "WAITING_FOR_RETRY_EXCEPTION_CONTROLLED_APPLY_DRIFT_BASELINE" - checks = [ - {"check": "source_replay_loaded", "passed": bool(replay)}, - {"check": "source_receipt_hash_matches", "passed": receipt_hash_match_count > 0}, - {"check": "target_selectors_present", "passed": selector_count > 0}, - {"check": "all_current_readbacks_match_receipt", "passed": not drift_items and pass_count == selector_count}, - {"check": "drift_verifier_does_not_write_database", "passed": True}, - ] - verifier_id_payload = { - "run_id": (replay.get("receipt_replay") or {}).get("run_id") or run_id or "", - "result": result, - "post_apply_readbacks": readbacks, - } - verifier_id = ( - "pchome-retry-exception-controlled-apply-drift-verifier-" - + hashlib.sha256( - json.dumps(verifier_id_payload, ensure_ascii=False, sort_keys=True, default=str).encode("utf-8") - ).hexdigest()[:16] - ) - summary = { - "target_selector_count": selector_count, - "post_apply_readback_count": int(replay_summary.get("post_apply_readback_count") or 0), - "post_apply_readback_pass_count": pass_count, - "drift_count": len(drift_items), - "drift_verified_count": 1 if drift_verified else 0, - "receipt_hash_match_count": receipt_hash_match_count, - "missing_artifact_count": int(replay_summary.get("missing_artifact_count") or 0), - "drift_verifier_artifact_materialized_count": 0, - "drift_verifier_artifact_hash_match_count": 0, - "writes_database_count": 0, - } - safety = { - "ai_controlled_apply": True, - "reads_artifact_files": True, - "reads_database": engine is not None or bool(source_receipt_replay), - "writes_database": False, - "writes_database_count": 0, - "writes_artifact_count": 0, - "syncs_external_offers": False, - "dispatches_telegram": False, - "gemini_allowed": False, - "requires_production_version_truth": True, - } - artifact_payload = { - "artifact_key": "retry_exception_controlled_apply_drift_verifier_receipt", - "verifier_id": verifier_id, - "run_id": verifier_id_payload["run_id"], - "source_policy": DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CONTROLLED_APPLY_DRIFT_VERIFIER_POLICY, - "source_receipt_replay_result": replay.get("result"), - "result": result, - "summary": summary, - "drift_items": drift_items, - "post_apply_readbacks": readbacks, - "checks": checks, - "safety": safety, - } - artifact_bytes = _canonical_retry_exception_artifact_bytes(artifact_payload) - artifact_relative_path = ( - f"artifacts/pchome_growth/retry_exception_closeout/" - f"controlled_apply_drift_verifier/{verifier_id}.json" - ) - drift_verifier_artifact = { - "key": "retry_exception_controlled_apply_drift_verifier_receipt", - "artifact_type": "controlled_apply_drift_verifier_receipt", - "relative_path": artifact_relative_path, - "payload_sha256": hashlib.sha256(artifact_bytes).hexdigest(), - "byte_count": len(artifact_bytes), - "payload": artifact_payload, - "materialized": False, - "writes_database": False, - } - materialized_drift_artifacts: list[dict[str, Any]] = [] - if materialize_artifacts and selector_count: - target_path = _resolve_retry_exception_artifact_path(root, artifact_relative_path) - target_path.parent.mkdir(parents=True, exist_ok=True) - target_path.write_bytes(artifact_bytes) - materialized_drift_artifacts.append({ - "key": drift_verifier_artifact["key"], - "relative_path": artifact_relative_path, - "absolute_path": str(target_path), - "payload_sha256": drift_verifier_artifact["payload_sha256"], - "written_byte_count": target_path.stat().st_size, - "writes_database": False, - }) - drift_verifier_artifact["materialized"] = True - drift_verifier_artifact["absolute_path"] = str(target_path) - artifact_path = _resolve_retry_exception_artifact_path(root, artifact_relative_path) - artifact_sha = hashlib.sha256(artifact_path.read_bytes()).hexdigest() if artifact_path.exists() else "" - artifact_hash_match = bool(artifact_sha) and artifact_sha == drift_verifier_artifact["payload_sha256"] - summary["drift_verifier_artifact_materialized_count"] = len(materialized_drift_artifacts) or (1 if artifact_hash_match else 0) - summary["drift_verifier_artifact_hash_match_count"] = 1 if artifact_hash_match else 0 - safety["writes_artifact_count"] = len(materialized_drift_artifacts) - checks.extend([ - { - "check": "drift_artifact_materialized_when_requested", - "passed": (not materialize_artifacts) or (selector_count > 0 and artifact_path.exists()), - }, - { - "check": "drift_artifact_hash_matches_expected", - "passed": (not materialize_artifacts) or artifact_hash_match, - }, - ]) - return { - "policy": DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CONTROLLED_APPLY_DRIFT_VERIFIER_POLICY, - "result": result, - "success": result == "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_DRIFT_VERIFIED", - "summary": summary, - "drift_verifier": { - "verifier_id": verifier_id, - "stage": "P2_retry_exception_controlled_apply_drift_verifier", - "status": result, - "source_receipt_replay_result": replay.get("result"), - "ready": drift_verified, - "materialize_artifacts": bool(materialize_artifacts), - "requires_production_version_truth": True, - }, - "drift_items": drift_items, - "post_apply_readbacks": readbacks, - "source_receipt_replay_summary": replay_summary, - "drift_verifier_artifact": drift_verifier_artifact, - "materialized_drift_artifacts": materialized_drift_artifacts, - "post_drift_verifier_artifact_verifier": { - "expected_sha256": drift_verifier_artifact["payload_sha256"], - "actual_sha256": artifact_sha, - "hash_match": artifact_hash_match, - "writes_database": False, - }, - "checks": checks, - "check_count": len(checks), - "all_checks_passed": all(check.get("passed") is True for check in checks), - "next_actions": [ - "Keep this verifier on the readiness surface so DB drift is visible without manual table review.", - "If drift is detected, use the receipt replay readbacks as rollback or re-apply evidence.", - ], - "safety": safety, - } def _retry_exception_controlled_apply_drift_recovery_id( diff --git a/tests/test_pchome_mapping_backlog_modularization.py b/tests/test_pchome_mapping_backlog_modularization.py index dbb5fa1..2a033ce 100644 --- a/tests/test_pchome_mapping_backlog_modularization.py +++ b/tests/test_pchome_mapping_backlog_modularization.py @@ -5,6 +5,7 @@ from services.pchome_mapping_backlog import ( artifacts, contracts, controlled_apply_executor, + controlled_apply_verifier, evidence, policies, ) @@ -95,7 +96,46 @@ def test_controlled_apply_executor_core_is_no_write_without_apply_request(tmp_pa assert result["safety"]["writes_artifact_count"] == 0 -def test_legacy_facade_is_smaller_after_second_p0_extraction(): +def test_controlled_apply_drift_verifier_core_is_no_write(tmp_path): + replay = { + "result": "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_RECEIPT_REPLAYED", + "summary": { + "target_selector_count": 1, + "post_apply_readback_count": 1, + "post_apply_readback_pass_count": 1, + "executor_receipt_hash_match_count": 1, + "missing_artifact_count": 0, + }, + "post_apply_readbacks": [ + { + "selector_id": "selector-1", + "momo_icode": "MOMO-1", + "expected_pchome_id": "PCH-1", + "actual_pchome_id": "PCH-1", + "passed": True, + "writes_database": False, + } + ], + "receipt_replay": {"run_id": "run-1"}, + "missing_artifacts": [], + } + + result = controlled_apply_verifier.build_controlled_apply_drift_verifier_from_replay( + replay, + root=tmp_path, + source_receipt_replay=replay, + ) + + assert result["result"] == "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_DRIFT_VERIFIED" + assert result["success"] is True + assert result["summary"]["drift_count"] == 0 + assert result["summary"]["drift_verified_count"] == 1 + assert result["all_checks_passed"] is True + assert result["safety"]["writes_database_count"] == 0 + assert result["safety"]["writes_artifact_count"] == 0 + + +def test_legacy_facade_is_smaller_after_third_p0_extraction(): facade = ROOT / "services/pchome_mapping_backlog_service.py" - assert sum(1 for _ in facade.open(encoding="utf-8")) < 44_000 + assert sum(1 for _ in facade.open(encoding="utf-8")) < 43_850