From 4ad94a0fe342639602457ab0830330447278283c Mon Sep 17 00:00:00 2001 From: ogt Date: Thu, 2 Jul 2026 12:50:33 +0800 Subject: [PATCH] =?UTF-8?q?=E8=A3=9C=E9=BD=8A=20PChome=20drift=20verifier?= =?UTF-8?q?=20artifact?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- routes/ai_routes.py | 2 + services/pchome_mapping_backlog_service.py | 137 ++++++++++++++++---- tests/test_pchome_mapping_backlog_report.py | 13 ++ 3 files changed, 129 insertions(+), 23 deletions(-) diff --git a/routes/ai_routes.py b/routes/ai_routes.py index 97bfc69..ba4bd4b 100644 --- a/routes/ai_routes.py +++ b/routes/ai_routes.py @@ -2526,11 +2526,13 @@ def api_pchome_growth_direct_mapping_retry_candidate_exception_controlled_apply_ ) run_id = str(request.args.get('run_id') or '').strip() or None + materialize_artifacts = str(request.args.get('materialize_artifacts') or '').strip().lower() in {'1', 'true', 'yes'} engine = _create_icaim_dashboard_engine(DATABASE_PATH) try: package = build_pchome_direct_mapping_retry_candidate_exception_controlled_apply_drift_verifier_package( run_id=run_id, + materialize_artifacts=materialize_artifacts, engine=engine, ) finally: diff --git a/services/pchome_mapping_backlog_service.py b/services/pchome_mapping_backlog_service.py index a858260..a3564d0 100644 --- a/services/pchome_mapping_backlog_service.py +++ b/services/pchome_mapping_backlog_service.py @@ -4516,10 +4516,12 @@ def build_pchome_direct_mapping_retry_candidate_exception_controlled_apply_drift run_id: str | None = None, engine: Any = None, source_receipt_replay: dict[str, Any] | None = None, + materialize_artifacts: bool = False, ) -> dict[str, Any]: """Verify that applied PChome product matches still agree with the replay receipt.""" + root = Path(artifact_root) if artifact_root is not None else Path.cwd() / "data" replay = source_receipt_replay or build_pchome_direct_mapping_retry_candidate_exception_controlled_apply_receipt_replay_package( - artifact_root=artifact_root, + artifact_root=root, run_id=run_id, materialize_artifacts=False, engine=engine, @@ -4553,30 +4555,125 @@ def build_pchome_direct_mapping_retry_candidate_exception_controlled_apply_drift {"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": { - "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), - "writes_database_count": 0, - }, + "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), @@ -4584,18 +4681,7 @@ def build_pchome_direct_mapping_retry_candidate_exception_controlled_apply_drift "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": { - "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, - }, + "safety": safety, } @@ -5258,6 +5344,8 @@ def build_pchome_growth_ai_automation_readiness( controlled_apply_drift_count = int(drift_summary.get("drift_count") or 0) controlled_apply_drift_verified_count = int(drift_summary.get("drift_verified_count") or 0) controlled_apply_drift_selector_count = int(drift_summary.get("target_selector_count") or 0) + controlled_apply_drift_artifact_count = int(drift_summary.get("drift_verifier_artifact_materialized_count") or 0) + controlled_apply_drift_artifact_hash_match_count = int(drift_summary.get("drift_verifier_artifact_hash_match_count") or 0) exception_count = _summary_exception_count(receipt_summary) + int( decision_summary.get("machine_review_decision_count") or 0 ) @@ -5378,6 +5466,7 @@ def build_pchome_growth_ai_automation_readiness( controlled_apply_drift_count, ( f"verified {controlled_apply_drift_selector_count}/{controlled_apply_drift_selector_count}" + f" ยท artifact {controlled_apply_drift_artifact_count}" if controlled_apply_drift_verified_count else f"drift {controlled_apply_drift_count}" ), @@ -5428,6 +5517,8 @@ def build_pchome_growth_ai_automation_readiness( "controlled_apply_drift_count": controlled_apply_drift_count, "controlled_apply_drift_verified_count": controlled_apply_drift_verified_count, "controlled_apply_drift_selector_count": controlled_apply_drift_selector_count, + "controlled_apply_drift_artifact_count": controlled_apply_drift_artifact_count, + "controlled_apply_drift_artifact_hash_match_count": controlled_apply_drift_artifact_hash_match_count, "exception_count": exception_count, "ai_exception_count": exception_count, AI_EXCEPTION_REQUIRED_COUNT_KEY: exception_count, diff --git a/tests/test_pchome_mapping_backlog_report.py b/tests/test_pchome_mapping_backlog_report.py index 70c1889..935879d 100644 --- a/tests/test_pchome_mapping_backlog_report.py +++ b/tests/test_pchome_mapping_backlog_report.py @@ -1417,12 +1417,21 @@ def test_direct_mapping_retry_candidate_exception_controlled_apply_receipt_repla run_id=run_id, engine=engine, source_receipt_replay=read_only_package, + materialize_artifacts=True, ) assert verifier["result"] == "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_DRIFT_VERIFIED" assert verifier["summary"]["target_selector_count"] == 2 assert verifier["summary"]["drift_count"] == 0 assert verifier["summary"]["drift_verified_count"] == 1 + assert verifier["summary"]["drift_verifier_artifact_materialized_count"] == 1 + assert verifier["summary"]["drift_verifier_artifact_hash_match_count"] == 1 + assert verifier["post_drift_verifier_artifact_verifier"]["hash_match"] is True + drift_receipt = verifier["materialized_drift_artifacts"][0] + drift_receipt_path = Path(drift_receipt["absolute_path"]) + assert drift_receipt_path.exists() + assert hashlib.sha256(drift_receipt_path.read_bytes()).hexdigest() == drift_receipt["payload_sha256"] assert verifier["safety"]["writes_database"] is False + assert verifier["safety"]["writes_artifact_count"] == 1 with engine.begin() as conn: conn.execute(text(""" UPDATE pchome_product_matches @@ -1495,6 +1504,8 @@ def test_ai_automation_readiness_surfaces_controlled_apply_receipt_replay_closeo "drift_count": 0, "drift_verified_count": 1, "receipt_hash_match_count": 1, + "drift_verifier_artifact_materialized_count": 1, + "drift_verifier_artifact_hash_match_count": 1, }, }, ) @@ -1507,6 +1518,8 @@ def test_ai_automation_readiness_surfaces_controlled_apply_receipt_replay_closeo assert readiness["summary"]["controlled_apply_closeout_verified_count"] == 1 assert readiness["summary"]["controlled_apply_drift_count"] == 0 assert readiness["summary"]["controlled_apply_drift_verified_count"] == 1 + assert readiness["summary"]["controlled_apply_drift_artifact_count"] == 1 + assert readiness["summary"]["controlled_apply_drift_artifact_hash_match_count"] == 1 assert readiness["automation_policy"]["controlled_apply_closeout"] == "receipt_replay_machine_verified" assert readiness["automation_policy"]["controlled_apply_drift"] == "drift_verified" assert lanes["controlled_apply"]["status"] == "completed"