diff --git a/config.py b/config.py index 1710fab..377a424 100644 --- a/config.py +++ b/config.py @@ -402,7 +402,7 @@ YOUTUBE_API_KEY = os.getenv('YOUTUBE_API_KEY', '') # ========================================== # 系統版本與路徑 # ========================================== -SYSTEM_VERSION = "V10.780" +SYSTEM_VERSION = "V10.781" LOG_FILE_PATH = os.path.join(BASE_DIR, 'logs/system.log') public_url = PUBLIC_URL # 用於模板顯示 diff --git a/docs/AI_INTELLIGENCE_MODULE_SOT.md b/docs/AI_INTELLIGENCE_MODULE_SOT.md index c578255..b139c97 100644 --- a/docs/AI_INTELLIGENCE_MODULE_SOT.md +++ b/docs/AI_INTELLIGENCE_MODULE_SOT.md @@ -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.780 +> **適用版本**: V10.781 --- diff --git a/docs/memory/code_modularization_inventory_20260430.md b/docs/memory/code_modularization_inventory_20260430.md index c8cb98a..abb43fa 100644 --- a/docs/memory/code_modularization_inventory_20260430.md +++ b/docs/memory/code_modularization_inventory_20260430.md @@ -119,7 +119,7 @@ | 行數 | 檔案 | 分類 | 拆分方向 | |---:|---|---|---| -| 44261 | `services/pchome_mapping_backlog_service.py` | P0 gate 巨型 service、拆分進行中 | 96 個 policy、AI exception contract、商品頁與單位 evidence parser 已移到 `services/pchome_mapping_backlog/`；下一步依序拆 executor、verifier、receipt、reporter | +| 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 | | 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 分離 | diff --git a/services/pchome_mapping_backlog/artifacts.py b/services/pchome_mapping_backlog/artifacts.py new file mode 100644 index 0000000..0cecd5a --- /dev/null +++ b/services/pchome_mapping_backlog/artifacts.py @@ -0,0 +1,26 @@ +"""Shared artifact serialization and path guards for PChome mapping.""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any + + +def _canonical_retry_exception_artifact_bytes(payload: dict[str, Any]) -> bytes: + return ( + json.dumps(payload, ensure_ascii=False, sort_keys=True, indent=2, default=str) + "\n" + ).encode("utf-8") + + +def _resolve_retry_exception_artifact_path(root: Path, relative_path: str) -> Path: + relative = Path(relative_path) + if relative.is_absolute() or ".." in relative.parts: + raise ValueError(f"unsafe artifact path: {relative_path}") + return root / relative + + +__all__ = ( + "_canonical_retry_exception_artifact_bytes", + "_resolve_retry_exception_artifact_path", +) diff --git a/services/pchome_mapping_backlog/controlled_apply_executor.py b/services/pchome_mapping_backlog/controlled_apply_executor.py new file mode 100644 index 0000000..dc9f4bd --- /dev/null +++ b/services/pchome_mapping_backlog/controlled_apply_executor.py @@ -0,0 +1,356 @@ +"""Bounded database executor for verified PChome mapping selectors.""" + +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_EXECUTOR_POLICY, +) + + +def _retry_exception_controlled_apply_executor_id(preflight_package: dict[str, Any]) -> str: + preflight = preflight_package.get("controlled_apply_preflight") or {} + payload = { + "preflight_id": preflight.get("preflight_id") or "", + "run_id": preflight.get("run_id") or "", + "target_selector_count": (preflight_package.get("summary") or {}).get("target_selector_count") or 0, + "mutation_plan_count": (preflight_package.get("summary") or {}).get("mutation_plan_count") or 0, + } + digest = hashlib.sha256( + json.dumps(payload, ensure_ascii=False, sort_keys=True, default=str).encode("utf-8") + ).hexdigest() + return f"pchome-retry-exception-controlled-apply-executor-{digest[:16]}" + + +def _selector_to_pchome_match_params(selector: dict[str, Any]) -> dict[str, Any]: + target_pchome_product_id = str(selector.get("target_pchome_product_id") or "").strip() + return { + "momo_icode": str(selector.get("momo_product_id") or "").strip(), + "momo_name": str(selector.get("momo_product_name") or selector.get("momo_product_id") or "").strip(), + "momo_price": selector.get("momo_price"), + "pchome_id": target_pchome_product_id, + "pchome_name": str(selector.get("target_pchome_product_name") or target_pchome_product_id).strip(), + "pchome_url": f"https://24h.pchome.com.tw/prod/{target_pchome_product_id}" if target_pchome_product_id else None, + "similarity": selector.get("target_match_score"), + "advantage": "ai_match", + } + + +def _fetch_pchome_product_match_by_momo_icode(conn: Any, momo_icode: str) -> dict[str, Any] | None: + from sqlalchemy import text + + row = conn.execute(text(""" + SELECT id, momo_name, momo_icode, momo_price, pchome_id, pchome_name, + pchome_url, pchome_price, pchome_original, pchome_in_stock, + similarity, price_diff, price_diff_pct, advantage, last_checked + FROM pchome_product_matches + WHERE momo_icode = :momo_icode + """), {"momo_icode": momo_icode}).mappings().first() + return dict(row) if row else None + + +def _upsert_pchome_product_match(conn: Any, selector: dict[str, Any]) -> None: + from sqlalchemy import text + + params = _selector_to_pchome_match_params(selector) + conn.execute(text(""" + INSERT INTO pchome_product_matches ( + momo_name, momo_icode, momo_price, pchome_id, pchome_name, + pchome_url, similarity, advantage, last_checked + ) + VALUES ( + :momo_name, :momo_icode, :momo_price, :pchome_id, :pchome_name, + :pchome_url, :similarity, :advantage, CURRENT_TIMESTAMP + ) + ON CONFLICT (momo_icode) DO UPDATE SET + momo_name = EXCLUDED.momo_name, + momo_price = EXCLUDED.momo_price, + pchome_id = EXCLUDED.pchome_id, + pchome_name = EXCLUDED.pchome_name, + pchome_url = EXCLUDED.pchome_url, + similarity = EXCLUDED.similarity, + advantage = EXCLUDED.advantage, + last_checked = CURRENT_TIMESTAMP + """), params) + + +def build_controlled_apply_executor_from_preflight( + preflight_package: dict[str, Any], + *, + execute_apply: bool = False, + engine: Any = None, + materialize_artifacts: bool = False, + artifact_root: str | Path | None = None, +) -> dict[str, Any]: + """Apply verified selectors and emit independently verifiable receipts.""" + preflight = preflight_package.get("controlled_apply_preflight") or {} + selectors = list(preflight_package.get("target_selectors") or []) + executor_ready = bool(preflight.get("ready_for_controlled_apply_executor")) and bool(selectors) + executor_id = _retry_exception_controlled_apply_executor_id(preflight_package) + write_attempted = bool(execute_apply and executor_ready and engine is not None) + write_blockers: list[str] = [] + if execute_apply and engine is None: + write_blockers.append("engine_required_for_execute_apply") + if execute_apply and not executor_ready: + write_blockers.append("controlled_apply_preflight_not_ready") + + prewrite_snapshots: list[dict[str, Any]] = [] + applied_records: list[dict[str, Any]] = [] + post_apply_readbacks: list[dict[str, Any]] = [] + missing_tables: list[str] = [] + if write_attempted: + from sqlalchemy import inspect + + with engine.begin() as conn: + inspector = inspect(conn) + if not inspector.has_table("pchome_product_matches"): + missing_tables.append("pchome_product_matches") + else: + for selector in selectors: + momo_icode = str(selector.get("momo_product_id") or "").strip() + before = _fetch_pchome_product_match_by_momo_icode(conn, momo_icode) + prewrite_snapshots.append({ + "selector_id": selector.get("selector_id"), + "momo_icode": momo_icode, + "before": before, + "row_existed_before": before is not None, + }) + if not missing_tables: + for selector in selectors: + _upsert_pchome_product_match(conn, selector) + momo_icode = str(selector.get("momo_product_id") or "").strip() + after = _fetch_pchome_product_match_by_momo_icode(conn, momo_icode) + applied_records.append({ + "selector_id": selector.get("selector_id"), + "momo_icode": momo_icode, + "target_pchome_product_id": selector.get("target_pchome_product_id"), + "applied": bool(after), + }) + post_apply_readbacks.append({ + "selector_id": selector.get("selector_id"), + "momo_icode": momo_icode, + "expected_pchome_id": selector.get("target_pchome_product_id"), + "actual_pchome_id": (after or {}).get("pchome_id"), + "expected_momo_name": selector.get("momo_product_name"), + "actual_momo_name": (after or {}).get("momo_name"), + "passed": bool(after) + and str((after or {}).get("pchome_id") or "") == str(selector.get("target_pchome_product_id") or "") + and str((after or {}).get("momo_icode") or "") == momo_icode, + "writes_database": False, + }) + + readback_pass_count = sum(1 for item in post_apply_readbacks if item.get("passed")) + applied_count = len(applied_records) + if execute_apply and missing_tables: + result = "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_EXECUTOR_BLOCKED_MISSING_TABLE" + elif execute_apply and not selectors: + result = "WAITING_FOR_RETRY_EXCEPTION_CONTROLLED_APPLY_PREFLIGHT" + elif execute_apply and selectors and applied_count == len(selectors) and readback_pass_count == len(selectors): + result = "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_EXECUTED" + elif execute_apply: + result = "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_EXECUTION_INCOMPLETE" + elif executor_ready: + result = "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_EXECUTOR_READY" + else: + result = "WAITING_FOR_RETRY_EXCEPTION_CONTROLLED_APPLY_PREFLIGHT" + + rollback_steps = [ + { + "selector_id": snapshot.get("selector_id"), + "momo_icode": snapshot.get("momo_icode"), + "action": "restore_previous_pchome_product_match" if snapshot.get("row_existed_before") else "delete_inserted_pchome_product_match", + "prewrite_snapshot_available": True, + "executes_in_executor": False, + "writes_database": False, + } + for snapshot in prewrite_snapshots + ] or [ + { + "selector_id": selector.get("selector_id"), + "momo_icode": selector.get("momo_product_id"), + "action": "planned_restore_or_delete_pchome_product_match", + "prewrite_snapshot_available": False, + "executes_in_executor": False, + "writes_database": False, + } + for selector in selectors + ] + summary = { + "controlled_apply_executor_ready_count": 1 if executor_ready else 0, + "execute_apply_requested_count": 1 if execute_apply else 0, + "target_selector_count": len(selectors), + "prewrite_snapshot_count": len(prewrite_snapshots), + "applied_record_count": applied_count, + "post_apply_readback_count": len(post_apply_readbacks), + "post_apply_readback_pass_count": readback_pass_count, + "post_apply_readback_fail_count": len(post_apply_readbacks) - readback_pass_count, + "rollback_step_count": len(rollback_steps), + "missing_table_count": len(missing_tables), + "writes_database_count": applied_count, + "persists_candidate_count": applied_count, + } + executor_metadata = { + "executor_id": executor_id, + "source_preflight_id": preflight.get("preflight_id"), + "run_id": preflight.get("run_id"), + "stage": "P2_retry_exception_controlled_apply_executor", + "status": result, + "execute_apply": bool(execute_apply), + "target_table": "pchome_product_matches", + "ready_for_apply": executor_ready, + "write_attempted": write_attempted, + "missing_tables": missing_tables, + "requires_fresh_production_truth": True, + } + rollback_plan = { + "rollback_step_count": len(rollback_steps), + "rollback_steps": rollback_steps, + "executes_in_executor": False, + "writes_database": False, + } + safety = { + "ai_controlled_apply": True, + "execute_apply": bool(execute_apply), + "target_table": "pchome_product_matches", + "writes_database": bool(applied_count), + "writes_database_count": applied_count, + "persists_candidate": bool(applied_count), + "persists_candidate_count": applied_count, + "syncs_external_offers": False, + "dispatches_telegram": False, + "llm_calls_in_executor": False, + "gemini_allowed": False, + "requires_production_version_truth": True, + } + executor_receipt_ready = ( + result == "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_EXECUTED" + and bool(applied_records) + and readback_pass_count == len(selectors) + ) + root = Path(artifact_root) if artifact_root is not None else Path.cwd() / "data" + receipt_relative_path = ( + f"artifacts/pchome_growth/retry_exception_closeout/" + f"controlled_apply_executor/{executor_id}.json" + ) + executor_receipt_payload = { + "artifact_key": "retry_exception_controlled_apply_executor_receipt", + "executor_id": executor_id, + "source_preflight_id": preflight.get("preflight_id"), + "run_id": preflight.get("run_id"), + "source_policy": preflight_package.get("policy"), + "result": result, + "created_at": preflight_package.get("generated_at"), + "summary": summary, + "controlled_apply_executor": executor_metadata, + "target_selectors": selectors, + "prewrite_snapshots": prewrite_snapshots, + "applied_records": applied_records, + "post_apply_readbacks": post_apply_readbacks, + "rollback_plan": rollback_plan, + "safety": safety, + } + executor_receipt_bytes = _canonical_retry_exception_artifact_bytes(executor_receipt_payload) + executor_receipt_artifact = { + "key": "retry_exception_controlled_apply_executor_receipt", + "artifact_type": "controlled_apply_executor_receipt", + "relative_path": receipt_relative_path, + "payload_sha256": hashlib.sha256(executor_receipt_bytes).hexdigest(), + "byte_count": len(executor_receipt_bytes), + "payload": executor_receipt_payload, + "materialized": False, + "writes_database": False, + } + materialized_executor_artifacts: list[dict[str, Any]] = [] + if materialize_artifacts and executor_receipt_ready: + target_path = _resolve_retry_exception_artifact_path(root, receipt_relative_path) + target_path.parent.mkdir(parents=True, exist_ok=True) + target_path.write_bytes(executor_receipt_bytes) + materialized_executor_artifacts.append({ + "key": executor_receipt_artifact["key"], + "relative_path": receipt_relative_path, + "absolute_path": str(target_path), + "payload_sha256": executor_receipt_artifact["payload_sha256"], + "written_byte_count": target_path.stat().st_size, + "writes_database": False, + }) + executor_receipt_artifact["materialized"] = True + executor_receipt_artifact["absolute_path"] = str(target_path) + + receipt_path = _resolve_retry_exception_artifact_path(root, receipt_relative_path) + actual_receipt_sha = "" + receipt_file_exists = receipt_path.exists() + if receipt_file_exists: + actual_receipt_sha = hashlib.sha256(receipt_path.read_bytes()).hexdigest() + receipt_checks = [ + {"check": "executor_receipt_ready_after_apply", "passed": executor_receipt_ready}, + {"check": "all_post_apply_readbacks_passed", "passed": readback_pass_count == len(selectors)}, + {"check": "applied_record_count_matches_selectors", "passed": applied_count == len(selectors)}, + {"check": "receipt_payload_hash_is_sha256", "passed": len(executor_receipt_artifact["payload_sha256"]) == 64}, + { + "check": "materialized_receipt_exists_when_requested", + "passed": (not materialize_artifacts) or (executor_receipt_ready and receipt_file_exists), + }, + { + "check": "materialized_receipt_hash_matches_expected", + "passed": (not materialize_artifacts) + or (bool(actual_receipt_sha) and actual_receipt_sha == executor_receipt_artifact["payload_sha256"]), + }, + {"check": "receipt_safety_blocks_side_effects", "passed": safety["syncs_external_offers"] is False and safety["dispatches_telegram"] is False}, + ] + post_executor_receipt_verifier = { + "ready": executor_receipt_ready, + "checks": receipt_checks, + "check_count": len(receipt_checks), + "passed": all(check.get("passed") is True for check in receipt_checks), + "expected_sha256": executor_receipt_artifact["payload_sha256"], + "actual_sha256": actual_receipt_sha, + "hash_match": bool(actual_receipt_sha) and actual_receipt_sha == executor_receipt_artifact["payload_sha256"], + "reads_artifact_files": bool(materialize_artifacts), + "writes_database": False, + } + summary["executor_receipt_ready_count"] = 1 if executor_receipt_ready else 0 + summary["executor_receipt_payload_count"] = 1 + summary["executor_receipt_materialized_count"] = len(materialized_executor_artifacts) + summary["executor_receipt_hash_match_count"] = 1 if post_executor_receipt_verifier["hash_match"] else 0 + summary["post_executor_receipt_verifier_check_count"] = len(receipt_checks) + safety["writes_artifact_count"] = len(materialized_executor_artifacts) + return { + "policy": DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CONTROLLED_APPLY_EXECUTOR_POLICY, + "result": result, + "success": bool(preflight_package.get("success")), + "generated_at": preflight_package.get("generated_at"), + "source_policy": preflight_package.get("policy"), + "stats": preflight_package.get("stats") or {}, + "backlog": preflight_package.get("backlog") or {}, + "summary": summary, + "controlled_apply_executor": executor_metadata, + "target_selectors": selectors, + "prewrite_snapshots": prewrite_snapshots, + "applied_records": applied_records, + "post_apply_readbacks": post_apply_readbacks, + "rollback_plan": rollback_plan, + "executor_receipt_artifact": executor_receipt_artifact, + "materialized_executor_artifacts": materialized_executor_artifacts, + "post_executor_receipt_verifier": post_executor_receipt_verifier, + "write_blockers": write_blockers, + "source_preflight_summary": preflight_package.get("summary") or {}, + "next_actions": [ + "Use the executor receipt artifact as the machine-verifiable closeout source for every applied selector.", + "Run rollback steps only if post-apply readback fails or a future verifier detects drift.", + "Keep future writes bounded to selector IDs from this executor package.", + ], + "safety": safety, + } + + +__all__ = ( + "build_controlled_apply_executor_from_preflight", +) diff --git a/services/pchome_mapping_backlog_service.py b/services/pchome_mapping_backlog_service.py index 89879c8..c8e6ef9 100644 --- a/services/pchome_mapping_backlog_service.py +++ b/services/pchome_mapping_backlog_service.py @@ -125,12 +125,23 @@ from services.pchome_mapping_backlog.policies import ( PCHOME_FETCH_MAX_HTML_BYTES, EXTERNAL_BENCHMARK_REFERENCES, ) +from services.pchome_mapping_backlog.artifacts import ( + _canonical_retry_exception_artifact_bytes, + _resolve_retry_exception_artifact_path, +) from services.pchome_mapping_backlog.contracts import ( _ai_exception_compatibility_fields, _evidence_requires_ai_exception, _legacy_review_compatibility_fields, _summary_exception_count, ) +from services.pchome_mapping_backlog.controlled_apply_executor import ( + _fetch_pchome_product_match_by_momo_icode, + _retry_exception_controlled_apply_executor_id, + _selector_to_pchome_match_params, + _upsert_pchome_product_match, + build_controlled_apply_executor_from_preflight, +) from services.pchome_mapping_backlog.evidence import ( _action_code, _action_label, @@ -2824,17 +2835,6 @@ def _retry_exception_closeout_verifier_artifact_run_id(preview_package: dict[str return f"pchome-retry-closeout-verifier-run-{digest[:16]}" -def _canonical_retry_exception_artifact_bytes(payload: dict[str, Any]) -> bytes: - return ( - json.dumps(payload, ensure_ascii=False, sort_keys=True, indent=2, default=str) + "\n" - ).encode("utf-8") - - -def _resolve_retry_exception_artifact_path(root: Path, relative_path: str) -> Path: - relative = Path(relative_path) - if relative.is_absolute() or ".." in relative.parts: - raise ValueError(f"unsafe artifact path: {relative_path}") - return root / relative def _build_retry_exception_verifier_artifact_payloads( @@ -3557,72 +3557,6 @@ def build_pchome_direct_mapping_retry_candidate_exception_controlled_apply_prefl } -def _retry_exception_controlled_apply_executor_id(preflight_package: dict[str, Any]) -> str: - preflight = preflight_package.get("controlled_apply_preflight") or {} - payload = { - "preflight_id": preflight.get("preflight_id") or "", - "run_id": preflight.get("run_id") or "", - "target_selector_count": (preflight_package.get("summary") or {}).get("target_selector_count") or 0, - "mutation_plan_count": (preflight_package.get("summary") or {}).get("mutation_plan_count") or 0, - } - digest = hashlib.sha256( - json.dumps(payload, ensure_ascii=False, sort_keys=True, default=str).encode("utf-8") - ).hexdigest() - return f"pchome-retry-exception-controlled-apply-executor-{digest[:16]}" - - -def _selector_to_pchome_match_params(selector: dict[str, Any]) -> dict[str, Any]: - target_pchome_product_id = str(selector.get("target_pchome_product_id") or "").strip() - return { - "momo_icode": str(selector.get("momo_product_id") or "").strip(), - "momo_name": str(selector.get("momo_product_name") or selector.get("momo_product_id") or "").strip(), - "momo_price": selector.get("momo_price"), - "pchome_id": target_pchome_product_id, - "pchome_name": str(selector.get("target_pchome_product_name") or target_pchome_product_id).strip(), - "pchome_url": f"https://24h.pchome.com.tw/prod/{target_pchome_product_id}" if target_pchome_product_id else None, - "similarity": selector.get("target_match_score"), - "advantage": "ai_match", - } - - -def _fetch_pchome_product_match_by_momo_icode(conn: Any, momo_icode: str) -> dict[str, Any] | None: - from sqlalchemy import text - - row = conn.execute(text(""" - SELECT id, momo_name, momo_icode, momo_price, pchome_id, pchome_name, - pchome_url, pchome_price, pchome_original, pchome_in_stock, - similarity, price_diff, price_diff_pct, advantage, last_checked - FROM pchome_product_matches - WHERE momo_icode = :momo_icode - """), {"momo_icode": momo_icode}).mappings().first() - return dict(row) if row else None - - -def _upsert_pchome_product_match(conn: Any, selector: dict[str, Any]) -> None: - from sqlalchemy import text - - params = _selector_to_pchome_match_params(selector) - conn.execute(text(""" - INSERT INTO pchome_product_matches ( - momo_name, momo_icode, momo_price, pchome_id, pchome_name, - pchome_url, similarity, advantage, last_checked - ) - VALUES ( - :momo_name, :momo_icode, :momo_price, :pchome_id, :pchome_name, - :pchome_url, :similarity, :advantage, CURRENT_TIMESTAMP - ) - ON CONFLICT (momo_icode) DO UPDATE SET - momo_name = EXCLUDED.momo_name, - momo_price = EXCLUDED.momo_price, - pchome_id = EXCLUDED.pchome_id, - pchome_name = EXCLUDED.pchome_name, - pchome_url = EXCLUDED.pchome_url, - similarity = EXCLUDED.similarity, - advantage = EXCLUDED.advantage, - last_checked = CURRENT_TIMESTAMP - """), params) - - def build_pchome_direct_mapping_retry_candidate_exception_controlled_apply_executor_package( payload: dict[str, Any], batch_size: int = 5, @@ -3651,264 +3585,15 @@ def build_pchome_direct_mapping_retry_candidate_exception_controlled_apply_execu materialize_artifacts=materialize_artifacts, artifact_root=artifact_root, ) - preflight = preflight_package.get("controlled_apply_preflight") or {} - selectors = list(preflight_package.get("target_selectors") or []) - executor_ready = bool(preflight.get("ready_for_controlled_apply_executor")) and bool(selectors) - executor_id = _retry_exception_controlled_apply_executor_id(preflight_package) - write_attempted = bool(execute_apply and executor_ready and engine is not None) - write_blockers: list[str] = [] - if execute_apply and engine is None: - write_blockers.append("engine_required_for_execute_apply") - if execute_apply and not executor_ready: - write_blockers.append("controlled_apply_preflight_not_ready") - - prewrite_snapshots: list[dict[str, Any]] = [] - applied_records: list[dict[str, Any]] = [] - post_apply_readbacks: list[dict[str, Any]] = [] - missing_tables: list[str] = [] - if write_attempted: - from sqlalchemy import inspect - - with engine.begin() as conn: - inspector = inspect(conn) - if not inspector.has_table("pchome_product_matches"): - missing_tables.append("pchome_product_matches") - else: - for selector in selectors: - momo_icode = str(selector.get("momo_product_id") or "").strip() - before = _fetch_pchome_product_match_by_momo_icode(conn, momo_icode) - prewrite_snapshots.append({ - "selector_id": selector.get("selector_id"), - "momo_icode": momo_icode, - "before": before, - "row_existed_before": before is not None, - }) - if not missing_tables: - for selector in selectors: - _upsert_pchome_product_match(conn, selector) - momo_icode = str(selector.get("momo_product_id") or "").strip() - after = _fetch_pchome_product_match_by_momo_icode(conn, momo_icode) - applied_records.append({ - "selector_id": selector.get("selector_id"), - "momo_icode": momo_icode, - "target_pchome_product_id": selector.get("target_pchome_product_id"), - "applied": bool(after), - }) - post_apply_readbacks.append({ - "selector_id": selector.get("selector_id"), - "momo_icode": momo_icode, - "expected_pchome_id": selector.get("target_pchome_product_id"), - "actual_pchome_id": (after or {}).get("pchome_id"), - "expected_momo_name": selector.get("momo_product_name"), - "actual_momo_name": (after or {}).get("momo_name"), - "passed": bool(after) - and str((after or {}).get("pchome_id") or "") == str(selector.get("target_pchome_product_id") or "") - and str((after or {}).get("momo_icode") or "") == momo_icode, - "writes_database": False, - }) - - readback_pass_count = sum(1 for item in post_apply_readbacks if item.get("passed")) - applied_count = len(applied_records) - if execute_apply and missing_tables: - result = "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_EXECUTOR_BLOCKED_MISSING_TABLE" - elif execute_apply and not selectors: - result = "WAITING_FOR_RETRY_EXCEPTION_CONTROLLED_APPLY_PREFLIGHT" - elif execute_apply and selectors and applied_count == len(selectors) and readback_pass_count == len(selectors): - result = "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_EXECUTED" - elif execute_apply: - result = "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_EXECUTION_INCOMPLETE" - elif executor_ready: - result = "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_EXECUTOR_READY" - else: - result = "WAITING_FOR_RETRY_EXCEPTION_CONTROLLED_APPLY_PREFLIGHT" - - rollback_steps = [ - { - "selector_id": snapshot.get("selector_id"), - "momo_icode": snapshot.get("momo_icode"), - "action": "restore_previous_pchome_product_match" if snapshot.get("row_existed_before") else "delete_inserted_pchome_product_match", - "prewrite_snapshot_available": True, - "executes_in_executor": False, - "writes_database": False, - } - for snapshot in prewrite_snapshots - ] or [ - { - "selector_id": selector.get("selector_id"), - "momo_icode": selector.get("momo_product_id"), - "action": "planned_restore_or_delete_pchome_product_match", - "prewrite_snapshot_available": False, - "executes_in_executor": False, - "writes_database": False, - } - for selector in selectors - ] - summary = { - "controlled_apply_executor_ready_count": 1 if executor_ready else 0, - "execute_apply_requested_count": 1 if execute_apply else 0, - "target_selector_count": len(selectors), - "prewrite_snapshot_count": len(prewrite_snapshots), - "applied_record_count": applied_count, - "post_apply_readback_count": len(post_apply_readbacks), - "post_apply_readback_pass_count": readback_pass_count, - "post_apply_readback_fail_count": len(post_apply_readbacks) - readback_pass_count, - "rollback_step_count": len(rollback_steps), - "missing_table_count": len(missing_tables), - "writes_database_count": applied_count, - "persists_candidate_count": applied_count, - } - executor_metadata = { - "executor_id": executor_id, - "source_preflight_id": preflight.get("preflight_id"), - "run_id": preflight.get("run_id"), - "stage": "P2_retry_exception_controlled_apply_executor", - "status": result, - "execute_apply": bool(execute_apply), - "target_table": "pchome_product_matches", - "ready_for_apply": executor_ready, - "write_attempted": write_attempted, - "missing_tables": missing_tables, - "requires_fresh_production_truth": True, - } - rollback_plan = { - "rollback_step_count": len(rollback_steps), - "rollback_steps": rollback_steps, - "executes_in_executor": False, - "writes_database": False, - } - safety = { - "ai_controlled_apply": True, - "execute_apply": bool(execute_apply), - "target_table": "pchome_product_matches", - "writes_database": bool(applied_count), - "writes_database_count": applied_count, - "persists_candidate": bool(applied_count), - "persists_candidate_count": applied_count, - "syncs_external_offers": False, - "dispatches_telegram": False, - "llm_calls_in_executor": False, - "gemini_allowed": False, - "requires_production_version_truth": True, - } - executor_receipt_ready = ( - result == "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_EXECUTED" - and bool(applied_records) - and readback_pass_count == len(selectors) + return build_controlled_apply_executor_from_preflight( + preflight_package, + execute_apply=execute_apply, + engine=engine, + materialize_artifacts=materialize_artifacts, + artifact_root=artifact_root, ) - root = Path(artifact_root) if artifact_root is not None else Path.cwd() / "data" - receipt_relative_path = ( - f"artifacts/pchome_growth/retry_exception_closeout/" - f"controlled_apply_executor/{executor_id}.json" - ) - executor_receipt_payload = { - "artifact_key": "retry_exception_controlled_apply_executor_receipt", - "executor_id": executor_id, - "source_preflight_id": preflight.get("preflight_id"), - "run_id": preflight.get("run_id"), - "source_policy": preflight_package.get("policy"), - "result": result, - "created_at": preflight_package.get("generated_at"), - "summary": summary, - "controlled_apply_executor": executor_metadata, - "target_selectors": selectors, - "prewrite_snapshots": prewrite_snapshots, - "applied_records": applied_records, - "post_apply_readbacks": post_apply_readbacks, - "rollback_plan": rollback_plan, - "safety": safety, - } - executor_receipt_bytes = _canonical_retry_exception_artifact_bytes(executor_receipt_payload) - executor_receipt_artifact = { - "key": "retry_exception_controlled_apply_executor_receipt", - "artifact_type": "controlled_apply_executor_receipt", - "relative_path": receipt_relative_path, - "payload_sha256": hashlib.sha256(executor_receipt_bytes).hexdigest(), - "byte_count": len(executor_receipt_bytes), - "payload": executor_receipt_payload, - "materialized": False, - "writes_database": False, - } - materialized_executor_artifacts: list[dict[str, Any]] = [] - if materialize_artifacts and executor_receipt_ready: - target_path = _resolve_retry_exception_artifact_path(root, receipt_relative_path) - target_path.parent.mkdir(parents=True, exist_ok=True) - target_path.write_bytes(executor_receipt_bytes) - materialized_executor_artifacts.append({ - "key": executor_receipt_artifact["key"], - "relative_path": receipt_relative_path, - "absolute_path": str(target_path), - "payload_sha256": executor_receipt_artifact["payload_sha256"], - "written_byte_count": target_path.stat().st_size, - "writes_database": False, - }) - executor_receipt_artifact["materialized"] = True - executor_receipt_artifact["absolute_path"] = str(target_path) - receipt_path = _resolve_retry_exception_artifact_path(root, receipt_relative_path) - actual_receipt_sha = "" - receipt_file_exists = receipt_path.exists() - if receipt_file_exists: - actual_receipt_sha = hashlib.sha256(receipt_path.read_bytes()).hexdigest() - receipt_checks = [ - {"check": "executor_receipt_ready_after_apply", "passed": executor_receipt_ready}, - {"check": "all_post_apply_readbacks_passed", "passed": readback_pass_count == len(selectors)}, - {"check": "applied_record_count_matches_selectors", "passed": applied_count == len(selectors)}, - {"check": "receipt_payload_hash_is_sha256", "passed": len(executor_receipt_artifact["payload_sha256"]) == 64}, - { - "check": "materialized_receipt_exists_when_requested", - "passed": (not materialize_artifacts) or (executor_receipt_ready and receipt_file_exists), - }, - { - "check": "materialized_receipt_hash_matches_expected", - "passed": (not materialize_artifacts) - or (bool(actual_receipt_sha) and actual_receipt_sha == executor_receipt_artifact["payload_sha256"]), - }, - {"check": "receipt_safety_blocks_side_effects", "passed": safety["syncs_external_offers"] is False and safety["dispatches_telegram"] is False}, - ] - post_executor_receipt_verifier = { - "ready": executor_receipt_ready, - "checks": receipt_checks, - "check_count": len(receipt_checks), - "passed": all(check.get("passed") is True for check in receipt_checks), - "expected_sha256": executor_receipt_artifact["payload_sha256"], - "actual_sha256": actual_receipt_sha, - "hash_match": bool(actual_receipt_sha) and actual_receipt_sha == executor_receipt_artifact["payload_sha256"], - "reads_artifact_files": bool(materialize_artifacts), - "writes_database": False, - } - summary["executor_receipt_ready_count"] = 1 if executor_receipt_ready else 0 - summary["executor_receipt_payload_count"] = 1 - summary["executor_receipt_materialized_count"] = len(materialized_executor_artifacts) - summary["executor_receipt_hash_match_count"] = 1 if post_executor_receipt_verifier["hash_match"] else 0 - summary["post_executor_receipt_verifier_check_count"] = len(receipt_checks) - safety["writes_artifact_count"] = len(materialized_executor_artifacts) - return { - "policy": DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CONTROLLED_APPLY_EXECUTOR_POLICY, - "result": result, - "success": bool(preflight_package.get("success")), - "generated_at": preflight_package.get("generated_at"), - "source_policy": preflight_package.get("policy"), - "stats": preflight_package.get("stats") or {}, - "backlog": preflight_package.get("backlog") or {}, - "summary": summary, - "controlled_apply_executor": executor_metadata, - "target_selectors": selectors, - "prewrite_snapshots": prewrite_snapshots, - "applied_records": applied_records, - "post_apply_readbacks": post_apply_readbacks, - "rollback_plan": rollback_plan, - "executor_receipt_artifact": executor_receipt_artifact, - "materialized_executor_artifacts": materialized_executor_artifacts, - "post_executor_receipt_verifier": post_executor_receipt_verifier, - "write_blockers": write_blockers, - "source_preflight_summary": preflight_package.get("summary") or {}, - "next_actions": [ - "Use the executor receipt artifact as the machine-verifiable closeout source for every applied selector.", - "Run rollback steps only if post-apply readback fails or a future verifier detects drift.", - "Keep future writes bounded to selector IDs from this executor package.", - ], - "safety": safety, - } + def _find_retry_exception_artifact_file(root: Path, subdir: str, run_id: str | None = None) -> Path | None: diff --git a/tests/test_pchome_mapping_backlog_modularization.py b/tests/test_pchome_mapping_backlog_modularization.py index 0577795..dbb5fa1 100644 --- a/tests/test_pchome_mapping_backlog_modularization.py +++ b/tests/test_pchome_mapping_backlog_modularization.py @@ -1,7 +1,13 @@ from pathlib import Path from services import pchome_mapping_backlog_service as compatibility_facade -from services.pchome_mapping_backlog import contracts, evidence, policies +from services.pchome_mapping_backlog import ( + artifacts, + contracts, + controlled_apply_executor, + evidence, + policies, +) ROOT = Path(__file__).resolve().parents[1] @@ -39,7 +45,57 @@ def test_extracted_unit_parser_preserves_ai_exception_and_no_write_contract(): assert parsed["fetches_external_sites"] is False -def test_legacy_facade_is_smaller_after_first_p0_extraction(): +def test_controlled_apply_executor_helpers_keep_legacy_import_identity(): + assert ( + compatibility_facade._canonical_retry_exception_artifact_bytes + is artifacts._canonical_retry_exception_artifact_bytes + ) + assert ( + compatibility_facade._selector_to_pchome_match_params + is controlled_apply_executor._selector_to_pchome_match_params + ) + + +def test_controlled_apply_executor_core_is_no_write_without_apply_request(tmp_path): + preflight = { + "policy": "controlled_apply_preflight_fixture", + "success": True, + "generated_at": "2026-07-11T12:00:00+08:00", + "stats": {}, + "backlog": {}, + "summary": {"mutation_plan_count": 1}, + "controlled_apply_preflight": { + "preflight_id": "preflight-1", + "run_id": "run-1", + "ready_for_controlled_apply_executor": True, + }, + "target_selectors": [ + { + "selector_id": "selector-1", + "momo_product_id": "MOMO-1", + "momo_product_name": "Example 40ml", + "target_pchome_product_id": "PCH-1", + "target_pchome_product_name": "Example 40ml", + "target_match_score": 0.91, + } + ], + } + + result = controlled_apply_executor.build_controlled_apply_executor_from_preflight( + preflight, + artifact_root=tmp_path, + ) + + assert result["result"] == "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_EXECUTOR_READY" + assert result["summary"]["target_selector_count"] == 1 + assert result["summary"]["applied_record_count"] == 0 + assert result["summary"]["rollback_step_count"] == 1 + assert result["safety"]["execute_apply"] is False + assert result["safety"]["writes_database_count"] == 0 + assert result["safety"]["writes_artifact_count"] == 0 + + +def test_legacy_facade_is_smaller_after_second_p0_extraction(): facade = ROOT / "services/pchome_mapping_backlog_service.py" - assert sum(1 for _ in facade.open(encoding="utf-8")) < 44_300 + assert sum(1 for _ in facade.open(encoding="utf-8")) < 44_000