wooo/ewoooc
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

refactor(pchome): extract controlled drift verifier

Browse Source
This commit is contained in:
ogt
2026-07-11 12:15:01 +08:00
parent 08198c290d
commit b9e4948f99
6 changed files with 245 additions and 159 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
config.py
View File

@@ -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 # 用於模板顯示

2
docs/AI_INTELLIGENCE_MODULE_SOT.md
View File

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

2
docs/memory/code_modularization_inventory_20260430.md
View File

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

190
services/pchome_mapping_backlog/controlled_apply_verifier.py Normal file
View File

@@ -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",
)

164
services/pchome_mapping_backlog_service.py
View File

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

44
tests/test_pchome_mapping_backlog_modularization.py
View File

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