補齊 PChome drift verifier artifact

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:

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,

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"