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

feat(ai): add PChome rollback evidence package
Some checks failed
CD Pipeline / deploy (push) Has been cancelled
Details

Browse Source
This commit is contained in:
ogt
2026-07-02 14:47:17 +08:00
parent 1e97aef3c4
commit 32e3c4baa5
5 changed files with 600 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

331
services/pchome_mapping_backlog_service.py
View File

@@ -89,6 +89,9 @@ DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CONTROLLED_APPLY_COMPACT_READBACK_POLIC
DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CONTROLLED_APPLY_ARTIFACT_RETENTION_POLICY = (
"read_only_pchome_growth_direct_mapping_retry_candidate_exception_controlled_apply_artifact_retention"
)
DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CONTROLLED_APPLY_ROLLBACK_EVIDENCE_POLICY = (
"read_only_pchome_growth_direct_mapping_retry_candidate_exception_controlled_apply_rollback_evidence"
)
AI_AUTOMATION_READINESS_POLICY = "read_only_pchome_growth_ai_automation_readiness"
EVIDENCE_ENRICHMENT_PREVIEW_POLICY = "read_only_pchome_growth_evidence_enrichment_preview"
EVIDENCE_SOURCE_PREVIEW_POLICY = "read_only_pchome_growth_evidence_source_preview"
@@ -5857,6 +5860,334 @@ def build_pchome_direct_mapping_retry_candidate_exception_controlled_apply_artif
}
def _retry_exception_controlled_apply_rollback_evidence_id(
summary: dict[str, Any],
rollback_actions: list[dict[str, Any]],
) -> str:
payload = {"summary": summary, "rollback_actions": rollback_actions}
digest = hashlib.sha256(
json.dumps(payload, ensure_ascii=False, sort_keys=True, default=str).encode("utf-8")
).hexdigest()[:16]
return f"pchome-retry-exception-controlled-apply-rollback-evidence-{digest}"
def build_pchome_direct_mapping_retry_candidate_exception_controlled_apply_rollback_evidence_package(
*,
artifact_root: str | Path | None = None,
run_id: str | None = None,
engine: Any = None,
source_receipt_replay: dict[str, Any] | None = None,
source_drift_verifier: dict[str, Any] | None = None,
source_drift_recovery: dict[str, Any] | None = None,
source_compact_readback: dict[str, Any] | None = None,
source_artifact_retention: dict[str, Any] | None = None,
materialize_artifacts: bool = False,
) -> dict[str, Any]:
"""Aggregate rollback evidence for the controlled-apply family without executing rollback."""
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=root,
run_id=run_id,
materialize_artifacts=False,
engine=engine,
)
drift = source_drift_verifier or build_pchome_direct_mapping_retry_candidate_exception_controlled_apply_drift_verifier_package(
artifact_root=root,
run_id=run_id,
engine=engine,
source_receipt_replay=replay,
materialize_artifacts=False,
)
recovery = source_drift_recovery or build_pchome_direct_mapping_retry_candidate_exception_controlled_apply_drift_recovery_package(
artifact_root=root,
run_id=run_id,
engine=engine,
source_receipt_replay=replay,
source_drift_verifier=drift,
materialize_artifacts=False,
)
compact = source_compact_readback or build_pchome_direct_mapping_retry_candidate_exception_controlled_apply_compact_readback_package(
artifact_root=root,
run_id=run_id,
engine=engine,
source_receipt_replay=replay,
source_drift_verifier=drift,
source_drift_recovery=recovery,
materialize_artifacts=False,
)
retention = source_artifact_retention or build_pchome_direct_mapping_retry_candidate_exception_controlled_apply_artifact_retention_package(
artifact_root=root,
run_id=run_id,
engine=engine,
source_compact_readback=compact,
materialize_artifacts=False,
)
replay_summary = replay.get("summary") or {}
drift_summary = drift.get("summary") or {}
recovery_summary = recovery.get("summary") or {}
compact_summary = compact.get("summary") or {}
retention_summary = retention.get("summary") or {}
drift_count = int(drift_summary.get("drift_count") or recovery_summary.get("drift_count") or compact_summary.get("drift_count") or 0)
recovery_actions = list(recovery.get("recovery_actions") or [])
ready_actions = [
item for item in recovery_actions
if item.get("status") == "ready_for_controlled_reapply"
]
rollback_actions = []
for action in recovery_actions:
rollback_actions.append({
"action_id": action.get("action_id"),
"selector_id": action.get("selector_id"),
"momo_icode": action.get("momo_icode"),
"expected_pchome_id": action.get("expected_pchome_id"),
"actual_pchome_id": action.get("actual_pchome_id"),
"status": action.get("status"),
"rollback_sql_shape": action.get("rollback_sql_shape"),
"controlled_reapply_sql_shape": action.get("controlled_reapply_sql_shape"),
"selector_bindings": action.get("selector_bindings") or {},
"acceptance_gates": list(action.get("acceptance_gates") or []),
"executes_in_package": False,
"writes_database": False,
})
rollback_required = drift_count > 0
rollback_ready = rollback_required and bool(ready_actions) and len(ready_actions) == drift_count
no_rollback_required = (
not rollback_required
and (
drift.get("result") == "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_DRIFT_VERIFIED"
or recovery.get("result") == "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_DRIFT_RECOVERY_NOT_REQUIRED"
or compact.get("result") == "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_COMPACT_READBACK_VERIFIED"
)
)
missing_artifacts = list(replay.get("missing_artifacts") or [])
retention_ready = bool(retention_summary.get("artifact_count") or retention.get("success"))
family_evidence = [
{
"family": "controlled_apply_receipt_replay",
"result": replay.get("result"),
"selector_count": int(replay_summary.get("target_selector_count") or 0),
"readback_pass_count": int(replay_summary.get("post_apply_readback_pass_count") or 0),
"hash_match_count": int(replay_summary.get("executor_receipt_hash_match_count") or 0),
"rollback_role": "baseline_expected_state",
"writes_database": False,
},
{
"family": "controlled_apply_drift_verifier",
"result": drift.get("result"),
"drift_count": drift_count,
"hash_match_count": int(drift_summary.get("drift_verifier_artifact_hash_match_count") or 0),
"rollback_role": "current_state_delta_detector",
"writes_database": False,
},
{
"family": "controlled_apply_drift_recovery",
"result": recovery.get("result"),
"action_count": len(recovery_actions),
"ready_action_count": len(ready_actions),
"hash_match_count": int(recovery_summary.get("recovery_artifact_hash_match_count") or 0),
"rollback_role": "rollback_and_reapply_action_source",
"writes_database": False,
},
{
"family": "controlled_apply_compact_readback",
"result": compact.get("result"),
"product_status": (compact.get("compact_readback") or {}).get("status"),
"next_machine_action": (compact.get("compact_readback") or {}).get("next_machine_action"),
"hash_match_count": int(compact_summary.get("compact_readback_artifact_hash_match_count") or 0),
"rollback_role": "product_facing_status_source",
"writes_database": False,
},
{
"family": "controlled_apply_artifact_retention",
"result": retention.get("result"),
"artifact_count": int(retention_summary.get("artifact_count") or 0),
"protected_active_chain_count": int(retention_summary.get("protected_active_chain_count") or 0),
"hash_match_count": int(retention_summary.get("retention_artifact_hash_match_count") or 0),
"rollback_role": "evidence_chain_protection_source",
"writes_database": False,
},
]
if rollback_ready:
result = "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_ROLLBACK_EVIDENCE_READY"
product_status = "rollback_ready"
next_machine_action = "run_controlled_reapply_check_mode"
elif rollback_required:
result = "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_ROLLBACK_EVIDENCE_BLOCKED"
product_status = "rollback_blocked"
next_machine_action = "rebuild_selector_identity_before_rollback"
elif no_rollback_required:
result = "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_ROLLBACK_EVIDENCE_NOT_REQUIRED"
product_status = "rollback_not_required"
next_machine_action = "keep_monitoring_drift"
elif missing_artifacts:
result = "WAITING_FOR_RETRY_EXCEPTION_CONTROLLED_APPLY_ROLLBACK_EVIDENCE_ARTIFACTS"
product_status = "waiting"
next_machine_action = "restore_or_materialize_source_receipts"
else:
result = "WAITING_FOR_RETRY_EXCEPTION_CONTROLLED_APPLY_ROLLBACK_EVIDENCE_BASELINE"
product_status = "waiting"
next_machine_action = "run_receipt_replay_and_drift_verifier"
summary = {
"controlled_apply_family_count": len(family_evidence),
"target_selector_count": int(replay_summary.get("target_selector_count") or compact_summary.get("target_selector_count") or 0),
"post_apply_readback_pass_count": int(replay_summary.get("post_apply_readback_pass_count") or compact_summary.get("post_apply_readback_pass_count") or 0),
"drift_count": drift_count,
"rollback_required_count": 1 if rollback_required else 0,
"rollback_action_count": len(rollback_actions),
"rollback_ready_action_count": len(ready_actions),
"rollback_blocked_action_count": len(rollback_actions) - len(ready_actions),
"retention_artifact_count": int(retention_summary.get("artifact_count") or 0),
"protected_active_chain_count": int(retention_summary.get("protected_active_chain_count") or 0),
"rollback_evidence_ready_count": 1 if (rollback_ready or no_rollback_required) else 0,
"rollback_evidence_artifact_materialized_count": 0,
"rollback_evidence_artifact_hash_match_count": 0,
"primary_human_gate_count": 0,
"writes_database_count": 0,
}
rollback_evidence_id = _retry_exception_controlled_apply_rollback_evidence_id(summary, rollback_actions)
safety = {
"ai_controlled_apply": True,
"rollback_evidence": True,
"reads_artifact_files": True,
"reads_database": engine is not None,
"executes_rollback": False,
"executes_reapply": False,
"executes_sql": False,
"deletes_artifacts": False,
"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,
}
checks = [
{"check": "receipt_replay_loaded", "passed": bool(replay)},
{"check": "drift_verifier_loaded", "passed": bool(drift)},
{"check": "drift_recovery_loaded", "passed": bool(recovery)},
{"check": "compact_readback_loaded", "passed": bool(compact)},
{"check": "artifact_retention_loaded", "passed": bool(retention)},
{"check": "rollback_actions_cover_detected_drift", "passed": (not rollback_required) or len(ready_actions) == drift_count},
{"check": "artifact_retention_protects_evidence_chain", "passed": retention_ready or not rollback_required},
{"check": "rollback_evidence_has_no_primary_human_gate", "passed": True},
{"check": "rollback_evidence_does_not_execute_sql", "passed": True},
{"check": "rollback_evidence_does_not_write_database", "passed": True},
]
artifact_payload = {
"artifact_key": "retry_exception_controlled_apply_rollback_evidence_receipt",
"rollback_evidence_id": rollback_evidence_id,
"source_policy": DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CONTROLLED_APPLY_ROLLBACK_EVIDENCE_POLICY,
"source_compact_readback_result": compact.get("result"),
"source_drift_recovery_result": recovery.get("result"),
"source_artifact_retention_result": retention.get("result"),
"result": result,
"product_status": product_status,
"next_machine_action": next_machine_action,
"summary": summary,
"family_evidence": family_evidence,
"rollback_actions": rollback_actions,
"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_rollback_evidence/{rollback_evidence_id}.json"
)
rollback_evidence_artifact = {
"key": "retry_exception_controlled_apply_rollback_evidence_receipt",
"artifact_type": "controlled_apply_rollback_evidence_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_rollback_evidence_artifacts: list[dict[str, Any]] = []
if materialize_artifacts and (rollback_ready or no_rollback_required):
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_rollback_evidence_artifacts.append({
"key": rollback_evidence_artifact["key"],
"relative_path": artifact_relative_path,
"absolute_path": str(target_path),
"payload_sha256": rollback_evidence_artifact["payload_sha256"],
"written_byte_count": target_path.stat().st_size,
"writes_database": False,
})
rollback_evidence_artifact["materialized"] = True
rollback_evidence_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 == rollback_evidence_artifact["payload_sha256"]
summary["rollback_evidence_artifact_materialized_count"] = (
len(materialized_rollback_evidence_artifacts) or (1 if artifact_hash_match else 0)
)
summary["rollback_evidence_artifact_hash_match_count"] = 1 if artifact_hash_match else 0
safety["writes_artifact_count"] = len(materialized_rollback_evidence_artifacts)
checks.extend([
{
"check": "rollback_evidence_artifact_materialized_when_requested",
"passed": (not materialize_artifacts) or ((rollback_ready or no_rollback_required) and artifact_path.exists()),
},
{
"check": "rollback_evidence_artifact_hash_matches_expected",
"passed": (not materialize_artifacts) or artifact_hash_match,
},
])
return {
"policy": DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CONTROLLED_APPLY_ROLLBACK_EVIDENCE_POLICY,
"result": result,
"success": result in {
"DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_ROLLBACK_EVIDENCE_READY",
"DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_ROLLBACK_EVIDENCE_NOT_REQUIRED",
},
"summary": summary,
"rollback_evidence": {
"rollback_evidence_id": rollback_evidence_id,
"stage": "P3_retry_exception_controlled_apply_rollback_evidence",
"status": product_status,
"rollback_required": rollback_required,
"next_machine_action": next_machine_action,
"materialize_artifacts": bool(materialize_artifacts),
"requires_production_version_truth": True,
},
"family_evidence": family_evidence,
"rollback_actions": rollback_actions,
"rollback_evidence_artifact": rollback_evidence_artifact,
"materialized_rollback_evidence_artifacts": materialized_rollback_evidence_artifacts,
"post_rollback_evidence_artifact_verifier": {
"expected_sha256": rollback_evidence_artifact["payload_sha256"],
"actual_sha256": artifact_sha,
"hash_match": artifact_hash_match,
"writes_database": False,
},
"source_results": {
"receipt_replay": replay.get("result"),
"drift_verifier": drift.get("result"),
"drift_recovery": recovery.get("result"),
"compact_readback": compact.get("result"),
"artifact_retention": retention.get("result"),
},
"checks": checks,
"check_count": len(checks),
"all_checks_passed": all(check.get("passed") is True for check in checks),
"next_actions": [
"If rollback_required is false, keep monitoring drift using the scheduled health summary.",
"If rollback evidence is ready, run the controlled re-apply path in check-mode before any write.",
"After any controlled re-apply, replay receipts and regenerate this rollback evidence package.",
],
"safety": safety,
}
def build_pchome_evidence_enrichment_preview(payload: dict[str, Any], batch_size: int = 5) -> dict[str, Any]:
"""Build a read-only evidence enrichment package for mapping targets."""
operator_preview = build_pchome_mapping_operator_preview(payload, batch_size=batch_size)