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

新增 PChome controlled apply artifact retention
Some checks failed
CD Pipeline / deploy (push) Has been cancelled
Details

Browse Source
This commit is contained in:
ogt
2026-07-02 13:39:57 +08:00
parent 99003c171f
commit 5a7ff9879a
5 changed files with 373 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

293
services/pchome_mapping_backlog_service.py
View File

@@ -83,6 +83,9 @@ DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CONTROLLED_APPLY_DRIFT_RECOVERY_POLICY
DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CONTROLLED_APPLY_COMPACT_READBACK_POLICY = (
"read_only_pchome_growth_direct_mapping_retry_candidate_exception_controlled_apply_compact_readback"
)
DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CONTROLLED_APPLY_ARTIFACT_RETENTION_POLICY = (
"read_only_pchome_growth_direct_mapping_retry_candidate_exception_controlled_apply_artifact_retention"
)
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"
@@ -5284,6 +5287,296 @@ def build_pchome_direct_mapping_retry_candidate_exception_controlled_apply_compa
}
_CONTROLLED_APPLY_RETENTION_FAMILIES = [
{
"family": "verifier_inputs",
"artifact_key": "retry_exception_closeout_verifier_input_artifact",
"subdir": "verifier_inputs",
},
{
"family": "identity_readback",
"artifact_key": "retry_exception_closeout_identity_readback_artifact",
"subdir": "identity_readback",
},
{
"family": "controlled_apply_preflight",
"artifact_key": "retry_exception_controlled_apply_preflight_artifact",
"subdir": "controlled_apply_preflight",
},
{
"family": "controlled_apply_executor",
"artifact_key": "retry_exception_controlled_apply_executor_receipt",
"subdir": "controlled_apply_executor",
},
{
"family": "controlled_apply_executor_replay",
"artifact_key": "retry_exception_controlled_apply_executor_replay_receipt",
"subdir": "controlled_apply_executor_replay",
},
{
"family": "controlled_apply_drift_verifier",
"artifact_key": "retry_exception_controlled_apply_drift_verifier_receipt",
"subdir": "controlled_apply_drift_verifier",
},
{
"family": "controlled_apply_drift_recovery",
"artifact_key": "retry_exception_controlled_apply_drift_recovery_receipt",
"subdir": "controlled_apply_drift_recovery",
},
{
"family": "controlled_apply_compact_readback",
"artifact_key": "retry_exception_controlled_apply_compact_readback_receipt",
"subdir": "controlled_apply_compact_readback",
},
]
def _retry_exception_artifact_retention_id(summary: dict[str, Any], protected_paths: list[str]) -> str:
payload = {"summary": summary, "protected_paths": protected_paths}
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-artifact-retention-{digest}"
def _scan_retry_exception_retention_family(
root: Path,
family: dict[str, str],
*,
keep_latest_per_family: int,
protected_relative_paths: set[str],
) -> dict[str, Any]:
artifact_dir = root / "artifacts" / "pchome_growth" / "retry_exception_closeout" / family["subdir"]
paths = sorted(
artifact_dir.glob("*.json") if artifact_dir.exists() else [],
key=lambda path: path.stat().st_mtime,
reverse=True,
)
artifacts: list[dict[str, Any]] = []
keep_count = 0
prune_candidate_count = 0
total_bytes = 0
prune_candidate_bytes = 0
for index, path in enumerate(paths, start=1):
relative_path = str(path.relative_to(root))
byte_count = path.stat().st_size
total_bytes += byte_count
protected = index <= keep_latest_per_family or relative_path in protected_relative_paths
sha = hashlib.sha256(path.read_bytes()).hexdigest()
decision = "keep" if protected else "candidate_for_retention_prune"
if protected:
keep_count += 1
else:
prune_candidate_count += 1
prune_candidate_bytes += byte_count
artifacts.append({
"family": family["family"],
"artifact_key": family["artifact_key"],
"relative_path": relative_path,
"payload_sha256": sha,
"byte_count": byte_count,
"latest_rank": index,
"protected_by_latest_window": index <= keep_latest_per_family,
"protected_by_active_chain": relative_path in protected_relative_paths,
"retention_decision": decision,
"delete_in_package": False,
"writes_database": False,
})
return {
"family": family["family"],
"artifact_key": family["artifact_key"],
"subdir": family["subdir"],
"artifact_count": len(artifacts),
"keep_count": keep_count,
"prune_candidate_count": prune_candidate_count,
"total_byte_count": total_bytes,
"prune_candidate_byte_count": prune_candidate_bytes,
"artifacts": artifacts,
}
def _compact_readback_protected_relative_paths(compact_readback: dict[str, Any]) -> set[str]:
protected_paths: set[str] = set()
for receipt in (compact_readback.get("receipts") or {}).values():
relative_path = receipt.get("relative_path")
if relative_path:
protected_paths.add(str(relative_path))
compact_artifact = compact_readback.get("compact_artifact") or {}
if compact_artifact.get("relative_path"):
protected_paths.add(str(compact_artifact["relative_path"]))
return protected_paths
def build_pchome_direct_mapping_retry_candidate_exception_controlled_apply_artifact_retention_package(
*,
artifact_root: str | Path | None = None,
run_id: str | None = None,
engine: Any = None,
source_compact_readback: dict[str, Any] | None = None,
keep_latest_per_family: int = 3,
materialize_artifacts: bool = False,
) -> dict[str, Any]:
"""Build a no-delete retention policy package for controlled-apply artifacts."""
root = Path(artifact_root) if artifact_root is not None else Path.cwd() / "data"
keep_latest = max(1, int(keep_latest_per_family or 3))
compact_readback = 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,
materialize_artifacts=False,
)
protected_paths = _compact_readback_protected_relative_paths(compact_readback)
family_reports = [
_scan_retry_exception_retention_family(
root,
family,
keep_latest_per_family=keep_latest,
protected_relative_paths=protected_paths,
)
for family in _CONTROLLED_APPLY_RETENTION_FAMILIES
]
artifact_count = sum(int(report.get("artifact_count") or 0) for report in family_reports)
keep_count = sum(int(report.get("keep_count") or 0) for report in family_reports)
prune_candidate_count = sum(int(report.get("prune_candidate_count") or 0) for report in family_reports)
total_bytes = sum(int(report.get("total_byte_count") or 0) for report in family_reports)
prune_candidate_bytes = sum(int(report.get("prune_candidate_byte_count") or 0) for report in family_reports)
protected_path_list = sorted(protected_paths)
summary = {
"retention_family_count": len(family_reports),
"artifact_count": artifact_count,
"retained_artifact_count": keep_count,
"prune_candidate_count": prune_candidate_count,
"total_byte_count": total_bytes,
"prune_candidate_byte_count": prune_candidate_bytes,
"keep_latest_per_family": keep_latest,
"protected_active_chain_count": len(protected_path_list),
"retention_prune_executes_count": 0,
"retention_artifact_materialized_count": 0,
"retention_artifact_hash_match_count": 0,
"writes_database_count": 0,
}
result = (
"DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_ARTIFACT_RETENTION_POLICY_READY"
if artifact_count
else "WAITING_FOR_RETRY_EXCEPTION_CONTROLLED_APPLY_ARTIFACT_RETENTION_INPUTS"
)
retention_id = _retry_exception_artifact_retention_id(summary, protected_path_list)
safety = {
"ai_controlled_apply": True,
"artifact_retention": True,
"reads_artifact_files": True,
"reads_database": engine is not None,
"deletes_artifacts": False,
"retention_prune_executes": 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": "compact_readback_loaded", "passed": bool(compact_readback)},
{"check": "retention_families_scanned", "passed": len(family_reports) == len(_CONTROLLED_APPLY_RETENTION_FAMILIES)},
{"check": "active_chain_paths_protected", "passed": bool(protected_path_list)},
{"check": "retention_policy_does_not_delete_artifacts", "passed": True},
{"check": "retention_policy_does_not_write_database", "passed": True},
]
artifact_payload = {
"artifact_key": "retry_exception_controlled_apply_artifact_retention_policy_receipt",
"retention_id": retention_id,
"source_policy": DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CONTROLLED_APPLY_ARTIFACT_RETENTION_POLICY,
"source_compact_readback_result": compact_readback.get("result"),
"result": result,
"summary": summary,
"protected_active_chain_paths": protected_path_list,
"family_reports": family_reports,
"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_artifact_retention/{retention_id}.json"
)
retention_artifact = {
"key": "retry_exception_controlled_apply_artifact_retention_policy_receipt",
"artifact_type": "controlled_apply_artifact_retention_policy_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_retention_artifacts: list[dict[str, Any]] = []
if materialize_artifacts and artifact_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_retention_artifacts.append({
"key": retention_artifact["key"],
"relative_path": artifact_relative_path,
"absolute_path": str(target_path),
"payload_sha256": retention_artifact["payload_sha256"],
"written_byte_count": target_path.stat().st_size,
"writes_database": False,
})
retention_artifact["materialized"] = True
retention_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 == retention_artifact["payload_sha256"]
summary["retention_artifact_materialized_count"] = len(materialized_retention_artifacts) or (1 if artifact_hash_match else 0)
summary["retention_artifact_hash_match_count"] = 1 if artifact_hash_match else 0
safety["writes_artifact_count"] = len(materialized_retention_artifacts)
checks.extend([
{
"check": "retention_artifact_materialized_when_requested",
"passed": (not materialize_artifacts) or (artifact_count > 0 and artifact_path.exists()),
},
{
"check": "retention_artifact_hash_matches_expected",
"passed": (not materialize_artifacts) or artifact_hash_match,
},
])
return {
"policy": DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CONTROLLED_APPLY_ARTIFACT_RETENTION_POLICY,
"result": result,
"success": artifact_count > 0,
"summary": summary,
"artifact_retention": {
"retention_id": retention_id,
"stage": "P2_retry_exception_controlled_apply_artifact_retention",
"status": "ready" if artifact_count else "waiting",
"keep_latest_per_family": keep_latest,
"protected_active_chain_count": len(protected_path_list),
"materialize_artifacts": bool(materialize_artifacts),
"requires_production_version_truth": True,
},
"protected_active_chain_paths": protected_path_list,
"family_reports": family_reports,
"retention_artifact": retention_artifact,
"materialized_retention_artifacts": materialized_retention_artifacts,
"post_retention_artifact_verifier": {
"expected_sha256": retention_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": [
"Use prune candidates only after a separate controlled delete executor is added.",
"Keep the latest active compact readback chain protected before any artifact pruning.",
"Expose retained/prune candidate counts on the product dashboard before enabling prune execution.",
],
"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)