diff --git a/config.py b/config.py
index 87ed2c8..d0d72cb 100644
--- a/config.py
+++ b/config.py
@@ -414,7 +414,7 @@ YOUTUBE_API_KEY = os.getenv('YOUTUBE_API_KEY', '')
# ==========================================
# 系統版本與路徑
# ==========================================
-SYSTEM_VERSION = "V10.790"
+SYSTEM_VERSION = "V10.791"
LOG_FILE_PATH = os.path.join(BASE_DIR, 'logs/system.log')
public_url = PUBLIC_URL # 用於模板顯示
diff --git a/routes/ai_routes.py b/routes/ai_routes.py
index dcc4c81..5ae2f2e 100644
--- a/routes/ai_routes.py
+++ b/routes/ai_routes.py
@@ -6444,6 +6444,35 @@ def api_pchome_growth_backfill_momo_candidates():
engine.dispose()
+@ai_bp.route('/api/ai/pchome-growth/same-item-reconciliation')
+@login_required
+def api_pchome_growth_same_item_reconciliation():
+ """Return public-safe durable readback for the latest bounded runs."""
+ try:
+ limit = max(1, min(int(request.args.get('limit', 20)), 100))
+ except (TypeError, ValueError):
+ limit = 20
+
+ engine = None
+ try:
+ from config import DATABASE_PATH
+ from services.pchome_growth_same_item_reconciliation import (
+ read_latest_reconciliation_receipts,
+ )
+
+ engine = _create_icaim_dashboard_engine(DATABASE_PATH)
+ return jsonify(read_latest_reconciliation_receipts(engine, limit=limit))
+ except Exception as exc:
+ logger.error("[PChomeGrowth] 同商品 reconciliation readback 失敗: %s", exc, exc_info=True)
+ return jsonify({
+ "success": False,
+ "error": "同商品自動化回讀暫時無法取得。",
+ }), 500
+ finally:
+ if engine is not None:
+ engine.dispose()
+
+
@ai_bp.route('/api/ai/pchome-growth/source-contract')
@login_required
def api_pchome_growth_source_contract():
diff --git a/routes/dashboard_routes.py b/routes/dashboard_routes.py
index c4d1df8..664f597 100644
--- a/routes/dashboard_routes.py
+++ b/routes/dashboard_routes.py
@@ -1355,6 +1355,8 @@ def _load_pchome_growth_command_center(session):
'mapped_count': 0,
'mapping_rate': 0,
'mapping_rate_width': 0,
+ 'comparison_revenue_coverage_rate': 0,
+ 'comparison_ready_sales_7d': 0,
'needs_mapping_count': 0,
'mapping_backlog': {
'direct_mapping_count': 0,
@@ -1449,7 +1451,7 @@ def _load_pchome_growth_command_center(session):
)
engine = session.get_bind()
- payload = build_pchome_growth_opportunities(engine, limit=16)
+ payload = build_pchome_growth_opportunities(engine, limit=50)
stats = payload.get('stats') or {}
opportunities = payload.get('opportunities') or []
mapping_summary = summarize_pchome_mapping_backlog(payload)
@@ -1521,6 +1523,18 @@ def _load_pchome_growth_command_center(session):
sales_delta_value = _to_float(sales_delta_pct)
max_sales = max(sales_7d, sales_prev_7d, 1)
mapping_rate = _to_float(stats.get('mapping_rate')) or 0
+ comparison_contract = stats.get('comparison_metric_contract') or {}
+ revenue_coverage = comparison_contract.get('revenue_coverage') or {}
+ comparison_revenue_coverage_rate = (
+ _to_float(revenue_coverage.get('rate'))
+ or _to_float(stats.get('comparison_revenue_coverage_rate'))
+ or 0
+ )
+ comparison_ready_sales_7d = (
+ _to_float(revenue_coverage.get('numerator'))
+ or _to_float(stats.get('comparison_ready_sales_7d'))
+ or 0
+ )
action_code_counts = stats.get('action_code_counts') or {}
candidate_count = int(stats.get('candidate_count') or 0)
mapped_count = int(stats.get('mapped_count') or 0)
@@ -1776,6 +1790,11 @@ def _load_pchome_growth_command_center(session):
'mapped_count': mapped_count,
'mapping_rate': round(mapping_rate, 1),
'mapping_rate_width': round(max(0, min(100, mapping_rate)), 1),
+ 'comparison_revenue_coverage_rate': round(
+ comparison_revenue_coverage_rate,
+ 1,
+ ),
+ 'comparison_ready_sales_7d': comparison_ready_sales_7d,
'needs_mapping_count': needs_mapping,
'mapping_backlog': mapping_backlog,
'ai_automation_readiness': ai_automation_readiness,
diff --git a/run_scheduler.py b/run_scheduler.py
index f022adf..c006293 100644
--- a/run_scheduler.py
+++ b/run_scheduler.py
@@ -93,7 +93,7 @@ def _pchome_growth_backfill_due(*, now=None) -> tuple[bool, str]:
status = str(latest.get("status") or "").lower()
cooldown_hours = float(os.getenv("PCHOME_GROWTH_BACKFILL_COOLDOWN_HOURS", "4"))
- if status == "failed":
+ if status in {"failed", "degraded", "partial"}:
cooldown_hours = float(os.getenv("PCHOME_GROWTH_BACKFILL_RETRY_HOURS", "1"))
elapsed_hours = max(0.0, (current - last_run).total_seconds() / 3600)
if elapsed_hours >= cooldown_hours:
diff --git a/scheduler.py b/scheduler.py
index a42fdfa..9441f02 100644
--- a/scheduler.py
+++ b/scheduler.py
@@ -2517,8 +2517,26 @@ def run_pchome_growth_momo_backfill_task():
before_stats = data.get("before_stats") or {}
after_stats = data.get("after_stats") or {}
sync_result = data.get("external_offer_sync") or {}
+ post_verifier = data.get("post_verifier") or {}
+ receipt_persistence = data.get("receipt_persistence") or {}
+ identity = data.get("identity") or {}
+ terminal_status = str(data.get("terminal_status") or "")
stats = {
- "status": "Success" if result.get("success") else "Skipped",
+ "status": (
+ "Success"
+ if result.get("success") and terminal_status.startswith("verified")
+ else "Degraded"
+ if terminal_status.startswith(("degraded", "partial"))
+ else "Failed"
+ if not result.get("success")
+ else "Skipped"
+ ),
+ "contract_version": data.get("contract_version"),
+ "trace_id": identity.get("trace_id"),
+ "run_id": identity.get("run_id"),
+ "work_item_id": identity.get("work_item_id"),
+ "terminal_status": terminal_status,
+ "next_machine_action": data.get("next_machine_action"),
"scanned_products": data.get("scanned_products", 0),
"candidate_count": data.get("candidate_count", 0),
"auto_compare_count": data.get("auto_compare_count", 0),
@@ -2530,10 +2548,21 @@ def run_pchome_growth_momo_backfill_task():
"mapped_count_before": before_stats.get("mapped_count"),
"mapped_count_after": after_stats.get("mapped_count"),
"needs_mapping_count_after": after_stats.get("needs_mapping_count"),
+ "comparison_ready_delta": post_verifier.get("comparison_ready_delta", 0),
+ "count_coverage_delta": post_verifier.get("count_coverage_delta", 0),
+ "revenue_coverage_before": (post_verifier.get("before") or {}).get("revenue_coverage_rate"),
+ "revenue_coverage_after": (post_verifier.get("after") or {}).get("revenue_coverage_rate"),
+ "revenue_coverage_delta": post_verifier.get("revenue_coverage_delta", 0),
+ "post_verifier_status": post_verifier.get("status"),
+ "post_verifier_passed": post_verifier.get("all_checks_passed", False),
+ "receipt_written_count": receipt_persistence.get("written_count", 0),
+ "receipt_readback_count": receipt_persistence.get("readback_count", 0),
+ "receipt_status": receipt_persistence.get("status"),
"message": result.get("message"),
}
logging.info(
- "[Scheduler] [PChomeGrowthMomoBackfill] ✅ 完成 | scanned=%s candidates=%s auto=%s written=%s review=%s mapping=%s%%→%s%%",
+ "[Scheduler] [PChomeGrowthMomoBackfill] 完成 | terminal=%s scanned=%s candidates=%s auto=%s written=%s review=%s mapping=%s%%→%s%% revenue_coverage=%s%%→%s%% receipts=%s/%s",
+ stats["terminal_status"],
stats["scanned_products"],
stats["candidate_count"],
stats["auto_compare_count"],
@@ -2541,6 +2570,10 @@ def run_pchome_growth_momo_backfill_task():
stats["review_count"],
stats["mapping_rate_before"],
stats["mapping_rate_after"],
+ stats["revenue_coverage_before"],
+ stats["revenue_coverage_after"],
+ stats["receipt_readback_count"],
+ stats["receipt_written_count"],
)
_save_stats('pchome_growth_momo_backfill', stats)
diff --git a/services/external_market_offer_service.py b/services/external_market_offer_service.py
index 912e629..8e5ccdd 100644
--- a/services/external_market_offer_service.py
+++ b/services/external_market_offer_service.py
@@ -1107,6 +1107,10 @@ def _targeted_candidate_to_external_offer(
"target_gap_pct": candidate.get("target_gap_pct"),
"unit_price_comparison": unit_price_comparison,
"search_term": candidate.get("target_search_term"),
+ "same_item_reconciliation": candidate.get("same_item_reconciliation") or {},
+ "same_item_candidate_fingerprint": candidate.get(
+ "same_item_candidate_fingerprint"
+ ),
"tags": [
"identity_v2",
"source_targeted_momo_search",
@@ -1193,6 +1197,7 @@ def _targeted_review_candidate_to_external_offer(
"target_gap_pct": candidate.get("target_gap_pct"),
"unit_price_comparison": unit_price_comparison,
"search_term": candidate.get("target_search_term"),
+ "same_item_reconciliation": candidate.get("same_item_reconciliation") or {},
"tags": [
"identity_v2",
"source_targeted_momo_search",
diff --git a/services/momo_crawler.py b/services/momo_crawler.py
index ee6cd0b..a41e3fe 100644
--- a/services/momo_crawler.py
+++ b/services/momo_crawler.py
@@ -708,7 +708,8 @@ def search_momo_products_for_pchome_products(
product_id = str(row.get("product_id") or row.get("goodsCode") or row.get("id") or "").strip()
if not product_id:
product_id = f"momo_candidate_{len(candidates_by_id)}"
- existing = candidates_by_id.get(product_id)
+ candidate_key = f"{pchome_id}:{product_id}"
+ existing = candidates_by_id.get(candidate_key)
if existing and float(existing.get("target_match_score") or 0.0) >= score:
continue
@@ -732,7 +733,7 @@ def search_momo_products_for_pchome_products(
"target_review_status": review_status,
"source_strategy": "pchome_targeted_momo_search",
})
- candidates_by_id[product_id] = row
+ candidates_by_id[candidate_key] = row
candidates = sorted(
candidates_by_id.values(),
diff --git a/services/pchome_growth_momo_backfill_service.py b/services/pchome_growth_momo_backfill_service.py
index c1835aa..227a7e8 100644
--- a/services/pchome_growth_momo_backfill_service.py
+++ b/services/pchome_growth_momo_backfill_service.py
@@ -5,8 +5,19 @@
from __future__ import annotations
import os
+from datetime import datetime, timezone
from typing import Any, Callable
+from services.pchome_growth_same_item_reconciliation import (
+ CONTRACT_VERSION,
+ build_coverage_post_verifier,
+ build_revenue_weighted_targets,
+ ensure_evidence_receipt_table,
+ new_reconciliation_identity,
+ persist_reconciliation_receipts,
+ verify_same_item_candidates,
+)
+
def _int_env(name: str, default: int, *, minimum: int, maximum: int) -> int:
try:
@@ -34,28 +45,7 @@ def candidate_auto_compare_type(candidate: dict[str, Any]) -> str:
def build_momo_backfill_targets(payload: dict[str, Any], limit: int) -> list[dict[str, Any]]:
- opportunities = list((payload or {}).get("opportunities") or [])
- targets: list[dict[str, Any]] = []
- for item in opportunities:
- action = item.get("recommended_action") or {}
- if item.get("external_price"):
- continue
- if action.get("code") != "map_external_product":
- continue
- product_id = str(item.get("pchome_product_id") or "").strip()
- product_name = str(item.get("product_name") or "").strip()
- if not product_id or not product_name:
- continue
- targets.append({
- "product_id": product_id,
- "name": product_name,
- "price": item.get("pchome_price"),
- "sales_7d": item.get("sales_7d"),
- "priority_score": item.get("priority_score"),
- })
- if len(targets) >= limit:
- break
- return targets
+ return build_revenue_weighted_targets(payload, limit)
def _default_build_payload(engine, limit: int) -> dict[str, Any]:
@@ -112,64 +102,61 @@ def run_pchome_growth_momo_backfill(
sync_func: Callable[[Any, list[dict[str, Any]]], dict[str, Any]] | None = None,
sync_review_func: Callable[[Any, list[dict[str, Any]]], dict[str, Any]] | None = None,
) -> dict[str, Any]:
- """補高業績 PChome 商品的 MOMO 對應。
-
- 不呼叫 LLM，只搜尋 MOMO 候選，並只把可自動判斷的 total_price / unit_price
- 寫入 external_offers；需 AI 自動驗證確認的候選會以 needs_review 保存，不進價格判斷。
- """
+ """Run a bounded search, independent verifier, apply, and readback loop."""
limit = max(1, min(int(limit or 12), 20))
build_payload = build_payload_func or _default_build_payload
search_candidates = search_func or _default_search_candidates
sync_candidates = sync_func or _default_sync_candidates
sync_review_candidates = sync_review_func or _default_sync_review_candidates
+ identity = new_reconciliation_identity()
+ generated_at = datetime.now(timezone.utc).isoformat()
- before_payload = build_payload(engine, max(limit, 16))
+ # The command center contract is top-50. Ranking a smaller pre-sorted slice
+ # can hide the highest-revenue unresolved products.
+ before_payload = build_payload(engine, 50)
targets = build_momo_backfill_targets(before_payload, limit)
- if not targets:
+ schema_receipt = ensure_evidence_receipt_table(engine)
+ if not schema_receipt.get("success"):
return {
- "success": True,
- "message": "目前高業績清單沒有需要補 MOMO 對應的商品。",
+ "success": False,
+ "message": "同商品收據資料層尚未就緒，本輪已安全停止且沒有寫入價格。",
"data": {
+ "contract_version": CONTRACT_VERSION,
+ "identity": identity,
+ "terminal_status": "blocked_receipt_schema_unavailable",
+ "target_count": len(targets),
"scanned_products": 0,
- "target_count": 0,
"candidate_count": 0,
- "exact_compare_count": 0,
- "unit_compare_count": 0,
"auto_compare_count": 0,
"review_count": 0,
- "external_offer_sync": {
- "success": True,
- "status": "not_needed",
- "written_count": 0,
- "message": "沒有需要同步的自動候選。",
- },
- "review_candidate_sync": {
- "success": True,
- "status": "not_needed",
- "written_count": 0,
- "message": "沒有需要保存的待確認候選。",
- },
+ "external_offer_sync": {"status": "not_executed", "written_count": 0},
+ "review_candidate_sync": {"status": "not_executed", "written_count": 0},
+ "receipt_schema": schema_receipt,
"before_stats": before_payload.get("stats") or {},
"after_stats": before_payload.get("stats") or {},
- "targets": [],
+ "targets": targets,
"review_candidates": [],
},
}
- search_success, search_message, candidates = search_candidates(targets, limit)
+ if targets:
+ search_success, search_message, candidates = search_candidates(targets, limit)
+ else:
+ search_success, search_message, candidates = (
+ True,
+ "目前 TOP 50 沒有未解決的同商品對應。",
+ [],
+ )
candidates = list(candidates or [])
- exact_candidates = [
- item for item in candidates
- if candidate_auto_compare_type(item) == "total_price"
- ]
- unit_candidates = [
- item for item in candidates
- if candidate_auto_compare_type(item) == "unit_price"
- ]
- review_candidates = [
- item for item in candidates
- if candidate_auto_compare_type(item) not in {"total_price", "unit_price"}
- ]
+ candidate_verification = verify_same_item_candidates(
+ targets,
+ candidates,
+ identity=identity,
+ )
+ auto_candidates = list(candidate_verification.get("verified_candidates") or [])
+ exact_candidates = [item for item in auto_candidates if candidate_auto_compare_type(item) == "total_price"]
+ unit_candidates = [item for item in auto_candidates if candidate_auto_compare_type(item) == "unit_price"]
+ review_candidates = list(candidate_verification.get("review_candidates") or [])
auto_candidates = [*exact_candidates, *unit_candidates]
external_offer_sync = {
"success": True,
@@ -188,19 +175,118 @@ def run_pchome_growth_momo_backfill(
if review_candidates:
review_candidate_sync = sync_review_candidates(engine, review_candidates)
- after_payload = build_payload(engine, max(limit, 16))
+ after_payload = build_payload(engine, 50)
+ post_verifier = build_coverage_post_verifier(
+ before_payload=before_payload,
+ after_payload=after_payload,
+ verified_candidates=auto_candidates,
+ sync_result=external_offer_sync,
+ )
written_count = int(external_offer_sync.get("written_count") or 0)
+ after_snapshot = post_verifier.get("after") or {}
+ unresolved_after = int(after_snapshot.get("candidate_validation_count") or 0) + int(
+ after_snapshot.get("unmatched_count") or 0
+ )
+ if not post_verifier.get("all_checks_passed"):
+ terminal_status = "partial_post_verifier_failed"
+ next_machine_action = "retry_bounded_batch_after_drift_diagnosis"
+ elif written_count > 0:
+ terminal_status = "verified_with_coverage_gain"
+ next_machine_action = (
+ "continue_next_revenue_weighted_batch"
+ if unresolved_after
+ else "maintain_same_item_coverage"
+ )
+ elif unresolved_after:
+ terminal_status = "degraded_no_safe_candidate"
+ next_machine_action = "retry_unresolved_batch_with_fresh_source_evidence"
+ else:
+ terminal_status = "verified_no_work_remaining"
+ next_machine_action = "maintain_same_item_coverage"
+
+ run_receipt = {
+ "contract_version": CONTRACT_VERSION,
+ "generated_at": generated_at,
+ "identity": identity,
+ "source_receipt": {
+ "source": "momo_targeted_search",
+ "search_success": bool(search_success),
+ "search_message": search_message,
+ "target_count": len(targets),
+ "candidate_count": len(candidates),
+ },
+ "normalization": {
+ "target_count": len(targets),
+ "candidate_count": len(candidates),
+ "candidate_fingerprint_count": len({
+ str(item.get("same_item_candidate_fingerprint") or "")
+ for item in auto_candidates
+ if item.get("same_item_candidate_fingerprint")
+ }),
+ },
+ "candidate_verification": candidate_verification,
+ "coverage_diff": {
+ "comparison_ready_delta": post_verifier.get("comparison_ready_delta"),
+ "count_coverage_delta": post_verifier.get("count_coverage_delta"),
+ "revenue_coverage_delta": post_verifier.get("revenue_coverage_delta"),
+ },
+ "risk_policy": {
+ "risk": "medium",
+ "bounded_target_limit": limit,
+ "formal_price_write_requires_independent_verifier": True,
+ "pixelrag_direct_price_write_allowed": False,
+ },
+ "check_mode": {
+ "receipt_schema_ready": bool(schema_receipt.get("success")),
+ "verified_candidate_count": len(auto_candidates),
+ "blocked_candidate_count": int(candidate_verification.get("blocked_candidate_count") or 0),
+ },
+ "execution": {
+ "idempotent_bounded_execution": True,
+ "written_count": written_count,
+ "review_candidate_written_count": int(review_candidate_sync.get("written_count") or 0),
+ "sync_status": external_offer_sync.get("status"),
+ },
+ "post_verifier": post_verifier,
+ "terminal": {
+ "status": terminal_status,
+ "next_machine_action": next_machine_action,
+ "unresolved_after": unresolved_after,
+ },
+ }
+ receipt_persistence = persist_reconciliation_receipts(
+ engine,
+ targets=targets,
+ run_receipt=run_receipt,
+ )
+ if not receipt_persistence.get("success"):
+ terminal_status = "partial_receipt_persistence_failed"
+ next_machine_action = "retry_receipt_persistence_before_next_apply"
+ run_receipt["terminal"] = {
+ "status": terminal_status,
+ "next_machine_action": next_machine_action,
+ "unresolved_after": unresolved_after,
+ }
+
+ success = bool(
+ post_verifier.get("all_checks_passed")
+ and receipt_persistence.get("success")
+ )
message = (
- f"已掃描 {len(targets)} 個高業績商品，找到 {len(candidates)} 筆 MOMO 候選，"
- f"自動寫入 {written_count} 筆。"
+ f"已依業績權重掃描 {len(targets)} 個商品，找到 {len(candidates)} 筆候選，"
+ f"獨立驗證 {len(auto_candidates)} 筆，正式寫入並回讀 {post_verifier.get('readback_pass_count', 0)} 筆。"
)
if not search_success and not candidates:
- message = search_message or "已搜尋 MOMO，但沒有找到可用候選。"
+ message = search_message or "已完成 bounded 搜尋，但本輪沒有安全候選。"
return {
- "success": True,
+ "success": success,
"message": message,
"data": {
+ "contract_version": CONTRACT_VERSION,
+ "identity": identity,
+ "terminal_status": terminal_status,
+ "next_machine_action": next_machine_action,
"search_success": bool(search_success),
"search_message": search_message,
"scanned_products": len(targets),
@@ -214,7 +300,11 @@ def run_pchome_growth_momo_backfill(
"review_candidate_sync": review_candidate_sync,
"before_stats": before_payload.get("stats") or {},
"after_stats": after_payload.get("stats") or {},
- "targets": targets[:8],
+ "post_verifier": post_verifier,
+ "receipt_schema": schema_receipt,
+ "receipt_persistence": receipt_persistence,
+ "reconciliation_receipt": run_receipt,
+ "targets": targets,
"review_candidates": review_candidates[:8],
},
}
diff --git a/services/pchome_growth_same_item_reconciliation.py b/services/pchome_growth_same_item_reconciliation.py
new file mode 100644
index 0000000..e5ea3c9
--- /dev/null
+++ b/services/pchome_growth_same_item_reconciliation.py
@@ -0,0 +1,718 @@
+"""Revenue-weighted same-item reconciliation for PChome growth automation."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import uuid
+from datetime import datetime, timezone
+from typing import Any, Callable
+
+from sqlalchemy import bindparam, inspect, text
+
+
+CONTRACT_VERSION = "pchome_same_item_reconciliation_v1"
+WORK_ITEM_ID = "GROWTH-P0-001-B"
+ALLOWED_TARGET_ACTIONS = {
+ "map_external_product",
+ "review_external_candidate",
+ "validate_external_source",
+}
+
+
+def _float(value: Any, default: float = 0.0) -> float:
+ try:
+ return float(value)
+ except (TypeError, ValueError):
+ return default
+
+
+def _canonical_hash(payload: Any) -> str:
+ encoded = json.dumps(
+ payload,
+ ensure_ascii=False,
+ sort_keys=True,
+ separators=(",", ":"),
+ default=str,
+ ).encode("utf-8")
+ return hashlib.sha256(encoded).hexdigest()
+
+
+def new_reconciliation_identity() -> dict[str, str]:
+ run_id = uuid.uuid4().hex
+ return {
+ "trace_id": uuid.uuid4().hex,
+ "run_id": run_id,
+ "work_item_id": WORK_ITEM_ID,
+ }
+
+
+def build_revenue_weighted_targets(
+ payload: dict[str, Any],
+ limit: int,
+) -> list[dict[str, Any]]:
+ """Rank unresolved products by revenue at risk before generic priority."""
+ unresolved: list[dict[str, Any]] = []
+ for item in list((payload or {}).get("opportunities") or []):
+ action = item.get("recommended_action") or {}
+ if item.get("external_price"):
+ continue
+ if str(action.get("code") or "") not in ALLOWED_TARGET_ACTIONS:
+ continue
+ product_id = str(item.get("pchome_product_id") or "").strip()
+ product_name = str(item.get("product_name") or "").strip()
+ if not product_id or not product_name:
+ continue
+ sales_7d = max(0.0, _float(item.get("sales_7d")))
+ sales_prev_7d = max(0.0, _float(item.get("sales_prev_7d")))
+ unresolved.append({
+ "product_id": product_id,
+ "name": product_name,
+ "price": item.get("pchome_price"),
+ "sales_7d": sales_7d,
+ "sales_prev_7d": sales_prev_7d,
+ "revenue_at_risk": max(sales_7d, sales_prev_7d),
+ "priority_score": _float(item.get("priority_score")),
+ "target_state": (
+ "candidate_validation"
+ if item.get("review_candidate") or item.get("market_offers")
+ else "unmatched"
+ ),
+ "existing_review_candidate": item.get("review_candidate") or None,
+ })
+
+ if not unresolved:
+ return []
+
+ total_revenue_at_risk = sum(item["revenue_at_risk"] for item in unresolved)
+ max_revenue_at_risk = max(item["revenue_at_risk"] for item in unresolved) or 1.0
+ for item in unresolved:
+ revenue_index = item["revenue_at_risk"] / max_revenue_at_risk * 100
+ validation_boost = 5.0 if item["target_state"] == "candidate_validation" else 0.0
+ item["revenue_share_pct"] = round(
+ item["revenue_at_risk"] / max(total_revenue_at_risk, 1.0) * 100,
+ 3,
+ )
+ item["execution_priority_score"] = round(
+ revenue_index * 0.75 + item["priority_score"] * 0.20 + validation_boost,
+ 3,
+ )
+
+ unresolved.sort(
+ key=lambda item: (
+ item["execution_priority_score"],
+ item["revenue_at_risk"],
+ item["priority_score"],
+ item["product_id"],
+ ),
+ reverse=True,
+ )
+ bounded_limit = max(1, min(int(limit or 12), 20))
+ selected = unresolved[:bounded_limit]
+ for rank, item in enumerate(selected, start=1):
+ item["execution_rank"] = rank
+ return selected
+
+
+def _candidate_source_id(candidate: dict[str, Any]) -> str:
+ return str(
+ candidate.get("product_id")
+ or candidate.get("goodsCode")
+ or candidate.get("id")
+ or ""
+ ).strip()
+
+
+def _candidate_name(candidate: dict[str, Any]) -> str:
+ return str(candidate.get("name") or candidate.get("title") or "").strip()
+
+
+def _verification_rank(candidate: dict[str, Any]) -> tuple[int, float, str]:
+ compare_type = str(candidate.get("auto_compare_type") or "")
+ type_rank = 2 if compare_type == "total_price" else 1
+ return (
+ type_rank,
+ _float(candidate.get("target_match_score")),
+ _candidate_source_id(candidate),
+ )
+
+
+def verify_same_item_candidates(
+ targets: list[dict[str, Any]],
+ candidates: list[dict[str, Any]],
+ *,
+ identity: dict[str, str],
+ score_func: Callable[..., Any] | None = None,
+ unit_price_func: Callable[..., dict[str, Any]] | None = None,
+ min_exact_score: float = 0.76,
+) -> dict[str, Any]:
+ """Recompute identity and unit evidence without trusting crawler labels."""
+ if score_func is None or unit_price_func is None:
+ from services.marketplace_product_matcher import (
+ build_unit_price_comparison,
+ score_marketplace_match,
+ )
+
+ score_func = score_func or score_marketplace_match
+ unit_price_func = unit_price_func or build_unit_price_comparison
+
+ target_map = {str(item.get("product_id") or ""): item for item in targets}
+ verified: list[dict[str, Any]] = []
+ blocked: list[dict[str, Any]] = []
+
+ for candidate in list(candidates or []):
+ target_id = str(candidate.get("target_pchome_product_id") or "").strip()
+ target = target_map.get(target_id)
+ source_id = _candidate_source_id(candidate)
+ source_name = _candidate_name(candidate)
+ if not target:
+ blocked.append({
+ "target_pchome_product_id": target_id,
+ "source_product_id": source_id,
+ "reason_code": "target_not_in_bounded_batch",
+ })
+ continue
+ if not source_id or not source_name:
+ blocked.append({
+ "target_pchome_product_id": target_id,
+ "source_product_id": source_id,
+ "reason_code": "source_identity_missing",
+ })
+ continue
+
+ source_price = _float(candidate.get("price"))
+ target_price = _float(target.get("price"))
+ if source_price <= 0 or target_price <= 0:
+ blocked.append({
+ "target_pchome_product_id": target_id,
+ "source_product_id": source_id,
+ "reason_code": "positive_price_evidence_missing",
+ })
+ continue
+
+ diagnostics = score_func(
+ source_name,
+ str(target.get("name") or ""),
+ momo_price=source_price,
+ competitor_price=target_price,
+ )
+ exact_ready = bool(
+ not getattr(diagnostics, "hard_veto", True)
+ and _float(getattr(diagnostics, "score", 0.0)) >= min_exact_score
+ and getattr(diagnostics, "comparison_mode", "") == "exact_identity"
+ and getattr(diagnostics, "match_type", "") == "exact"
+ and getattr(diagnostics, "price_basis", "") == "total_price"
+ and getattr(diagnostics, "alert_tier", "") == "price_alert_exact"
+ )
+ unit_comparison: dict[str, Any] = {}
+ unit_ready = False
+ if (
+ getattr(diagnostics, "comparison_mode", "") == "unit_comparable"
+ or getattr(diagnostics, "price_basis", "") == "unit_price"
+ ):
+ unit_comparison = unit_price_func(
+ source_name,
+ str(target.get("name") or ""),
+ momo_price=source_price,
+ competitor_price=target_price,
+ )
+ unit_ready = bool(
+ unit_comparison.get("comparable")
+ and unit_comparison.get("unit_label")
+ and _float(unit_comparison.get("momo_total_quantity")) > 0
+ and _float(unit_comparison.get("competitor_total_quantity")) > 0
+ )
+
+ if exact_ready:
+ compare_type = "total_price"
+ price_basis = "total_price"
+ review_status = "independent_verifier_exact"
+ elif unit_ready:
+ compare_type = "unit_price"
+ price_basis = "unit_price"
+ review_status = "independent_verifier_unit"
+ else:
+ blocked.append({
+ "target_pchome_product_id": target_id,
+ "source_product_id": source_id,
+ "reason_code": "same_item_verifier_rejected",
+ "match_score": _float(getattr(diagnostics, "score", 0.0)),
+ "hard_veto": bool(getattr(diagnostics, "hard_veto", False)),
+ "reasons": list(getattr(diagnostics, "reasons", ()) or ())[:8],
+ })
+ continue
+
+ claimed_type = str(candidate.get("auto_compare_type") or "manual_review")
+ fingerprint_payload = {
+ "target_id": target_id,
+ "source_id": source_id,
+ "target_name": target.get("name"),
+ "source_name": source_name,
+ "target_price": target_price,
+ "source_price": source_price,
+ "compare_type": compare_type,
+ "match_score": _float(getattr(diagnostics, "score", 0.0)),
+ }
+ enriched = dict(candidate)
+ enriched.update({
+ "target_pchome_product_id": target_id,
+ "target_pchome_name": target.get("name"),
+ "target_pchome_price": target_price,
+ "target_match_score": round(_float(getattr(diagnostics, "score", 0.0)), 3),
+ "target_match_reasons": list(getattr(diagnostics, "reasons", ()) or ()),
+ "target_comparison_mode": getattr(diagnostics, "comparison_mode", ""),
+ "target_match_type": getattr(diagnostics, "match_type", ""),
+ "target_alert_tier": getattr(diagnostics, "alert_tier", ""),
+ "target_hard_veto": bool(getattr(diagnostics, "hard_veto", False)),
+ "target_price_basis": price_basis,
+ "target_unit_price_comparison": unit_comparison,
+ "target_review_status": review_status,
+ "auto_compare_type": compare_type,
+ "can_auto_compare": True,
+ "same_item_candidate_fingerprint": _canonical_hash(fingerprint_payload)[:24],
+ "same_item_reconciliation": {
+ **identity,
+ "contract_version": CONTRACT_VERSION,
+ "execution_rank": target.get("execution_rank"),
+ "execution_priority_score": target.get("execution_priority_score"),
+ "revenue_share_pct": target.get("revenue_share_pct"),
+ "target_state_before": target.get("target_state"),
+ "candidate_claimed_type": claimed_type,
+ "candidate_claim_drift": claimed_type != compare_type,
+ "independent_verifier_passed": True,
+ "candidate_fingerprint": _canonical_hash(fingerprint_payload)[:24],
+ },
+ })
+ verified.append(enriched)
+
+ selected_by_target: dict[str, dict[str, Any]] = {}
+ for candidate in verified:
+ target_id = str(candidate.get("target_pchome_product_id") or "")
+ current = selected_by_target.get(target_id)
+ if current is None or _verification_rank(candidate) > _verification_rank(current):
+ selected_by_target[target_id] = candidate
+
+ selected = list(selected_by_target.values())
+ selected.sort(
+ key=lambda item: int(
+ (item.get("same_item_reconciliation") or {}).get("execution_rank") or 999
+ )
+ )
+
+ review_by_target: dict[str, dict[str, Any]] = {}
+ selected_ids = set(selected_by_target)
+ for candidate in list(candidates or []):
+ target_id = str(candidate.get("target_pchome_product_id") or "").strip()
+ if not target_id or target_id in selected_ids or target_id not in target_map:
+ continue
+ current = review_by_target.get(target_id)
+ if current is None or _float(candidate.get("target_match_score")) > _float(
+ current.get("target_match_score")
+ ):
+ review_candidate = dict(candidate)
+ review_candidate["same_item_reconciliation"] = {
+ **identity,
+ "contract_version": CONTRACT_VERSION,
+ "independent_verifier_passed": False,
+ "target_state_before": target_map[target_id].get("target_state"),
+ }
+ review_by_target[target_id] = review_candidate
+
+ return {
+ "contract_version": CONTRACT_VERSION,
+ "candidate_count": len(candidates or []),
+ "verified_candidate_count": len(verified),
+ "selected_candidate_count": len(selected),
+ "blocked_candidate_count": len(blocked),
+ "claim_drift_count": sum(
+ 1
+ for item in selected
+ if (item.get("same_item_reconciliation") or {}).get("candidate_claim_drift")
+ ),
+ "verified_candidates": selected,
+ "review_candidates": list(review_by_target.values()),
+ "blocked_candidates": blocked,
+ }
+
+
+def build_coverage_snapshot(payload: dict[str, Any]) -> dict[str, Any]:
+ opportunities = list((payload or {}).get("opportunities") or [])
+ total_revenue = sum(max(0.0, _float(item.get("sales_7d"))) for item in opportunities)
+ ready_revenue = sum(
+ max(0.0, _float(item.get("sales_7d")))
+ for item in opportunities
+ if item.get("external_price")
+ )
+ validation_revenue = sum(
+ max(0.0, _float(item.get("sales_7d")))
+ for item in opportunities
+ if not item.get("external_price")
+ and (item.get("review_candidate") or item.get("market_offers"))
+ )
+ unmatched_revenue = max(0.0, total_revenue - ready_revenue - validation_revenue)
+ stats = (payload or {}).get("stats") or {}
+ contract = stats.get("comparison_metric_contract") or {}
+ partition = contract.get("state_partition") or {}
+ return {
+ "scope": contract.get("scope") or f"top_{len(opportunities)}_revenue_opportunities",
+ "comparison_ready_count": int(
+ partition.get("comparison_ready") or stats.get("mapped_count") or 0
+ ),
+ "candidate_validation_count": int(
+ partition.get("candidate_validation") or stats.get("candidate_validation_count") or 0
+ ),
+ "unmatched_count": int(partition.get("unmatched") or stats.get("unmatched_count") or 0),
+ "total_count": int(partition.get("total") or len(opportunities)),
+ "count_coverage_rate": round(_float(contract.get("rate") or stats.get("mapping_rate")), 3),
+ "comparison_ready_sales_7d": round(ready_revenue, 2),
+ "candidate_validation_sales_7d": round(validation_revenue, 2),
+ "unmatched_sales_7d": round(unmatched_revenue, 2),
+ "total_sales_7d": round(total_revenue, 2),
+ "revenue_coverage_rate": round(ready_revenue / max(total_revenue, 1.0) * 100, 3),
+ }
+
+
+def build_coverage_post_verifier(
+ *,
+ before_payload: dict[str, Any],
+ after_payload: dict[str, Any],
+ verified_candidates: list[dict[str, Any]],
+ sync_result: dict[str, Any],
+) -> dict[str, Any]:
+ before = build_coverage_snapshot(before_payload)
+ after = build_coverage_snapshot(after_payload)
+ after_by_id = {
+ str(item.get("pchome_product_id") or ""): item
+ for item in list((after_payload or {}).get("opportunities") or [])
+ }
+ readback: list[dict[str, Any]] = []
+ for candidate in verified_candidates:
+ target_id = str(candidate.get("target_pchome_product_id") or "")
+ source_id = _candidate_source_id(candidate)
+ external_price = (after_by_id.get(target_id) or {}).get("external_price") or {}
+ observed_source_id = str(external_price.get("momo_sku") or "")
+ passed = bool(external_price and observed_source_id == source_id)
+ readback.append({
+ "target_pchome_product_id": target_id,
+ "source_product_id": source_id,
+ "observed_source_product_id": observed_source_id,
+ "passed": passed,
+ })
+
+ selected_count = len(verified_candidates)
+ written_count = int((sync_result or {}).get("written_count") or 0)
+ readback_pass_count = sum(1 for item in readback if item["passed"])
+ no_write_verified = selected_count == 0 and written_count == 0
+ apply_verified = bool(
+ selected_count > 0
+ and written_count == selected_count
+ and readback_pass_count == selected_count
+ )
+ no_regression = (
+ after["comparison_ready_count"] >= before["comparison_ready_count"]
+ and after["count_coverage_rate"] >= before["count_coverage_rate"]
+ and after["revenue_coverage_rate"] >= before["revenue_coverage_rate"]
+ )
+ all_checks_passed = bool((no_write_verified or apply_verified) and no_regression)
+ return {
+ "contract_version": CONTRACT_VERSION,
+ "status": (
+ "verified"
+ if apply_verified and no_regression
+ else "no_write_verified"
+ if no_write_verified and no_regression
+ else "failed"
+ ),
+ "selected_candidate_count": selected_count,
+ "written_count": written_count,
+ "readback_pass_count": readback_pass_count,
+ "count_coverage_delta": round(
+ after["count_coverage_rate"] - before["count_coverage_rate"], 3
+ ),
+ "comparison_ready_delta": (
+ after["comparison_ready_count"] - before["comparison_ready_count"]
+ ),
+ "revenue_coverage_delta": round(
+ after["revenue_coverage_rate"] - before["revenue_coverage_rate"], 3
+ ),
+ "no_regression": no_regression,
+ "all_checks_passed": all_checks_passed,
+ "before": before,
+ "after": after,
+ "readback": readback,
+ }
+
+
+def ensure_evidence_receipt_table(engine) -> dict[str, Any]:
+ """Idempotently apply the existing additive Migration 045 schema contract."""
+ table_name = "external_offer_evidence_receipts"
+ if inspect(engine).has_table(table_name):
+ return {
+ "success": True,
+ "status": "already_ready",
+ "table": table_name,
+ "created": False,
+ }
+
+ dialect = engine.dialect.name
+ if dialect == "postgresql":
+ create_sql = """
+ CREATE TABLE IF NOT EXISTS external_offer_evidence_receipts (
+ receipt_id TEXT PRIMARY KEY,
+ pchome_product_id TEXT NOT NULL,
+ automation_decision TEXT NOT NULL,
+ payload_hash TEXT NOT NULL,
+ evidence_delta_json JSONB NOT NULL DEFAULT '{}'::jsonb,
+ created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+ applied_at TIMESTAMPTZ,
+ apply_status TEXT NOT NULL DEFAULT 'previewed'
+ )
+ """
+ else:
+ create_sql = """
+ CREATE TABLE IF NOT EXISTS external_offer_evidence_receipts (
+ receipt_id TEXT PRIMARY KEY,
+ pchome_product_id TEXT NOT NULL,
+ automation_decision TEXT NOT NULL,
+ payload_hash TEXT NOT NULL,
+ evidence_delta_json TEXT NOT NULL DEFAULT '{}',
+ created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
+ applied_at TEXT,
+ apply_status TEXT NOT NULL DEFAULT 'previewed'
+ )
+ """
+ index_sql = (
+ "CREATE INDEX IF NOT EXISTS idx_external_offer_evidence_receipts_pchome_product_id "
+ "ON external_offer_evidence_receipts (pchome_product_id)",
+ "CREATE INDEX IF NOT EXISTS idx_external_offer_evidence_receipts_payload_hash "
+ "ON external_offer_evidence_receipts (payload_hash)",
+ "CREATE INDEX IF NOT EXISTS idx_external_offer_evidence_receipts_apply_status "
+ "ON external_offer_evidence_receipts (apply_status)",
+ )
+ try:
+ with engine.begin() as conn:
+ conn.execute(text(create_sql))
+ for statement in index_sql:
+ conn.execute(text(statement))
+ ready = inspect(engine).has_table(table_name)
+ return {
+ "success": ready,
+ "status": "created_and_verified" if ready else "verification_failed",
+ "table": table_name,
+ "created": ready,
+ }
+ except Exception as exc:
+ return {
+ "success": False,
+ "status": "apply_failed",
+ "table": table_name,
+ "created": False,
+ "error_type": type(exc).__name__,
+ }
+
+
+def _target_receipt_decision(
+ target: dict[str, Any],
+ run_receipt: dict[str, Any],
+) -> tuple[str, str]:
+ target_id = str(target.get("product_id") or "")
+ verification = run_receipt.get("candidate_verification") or {}
+ selected_ids = {
+ str(item.get("target_pchome_product_id") or "")
+ for item in verification.get("verified_candidates") or []
+ }
+ blocked_ids = {
+ str(item.get("target_pchome_product_id") or "")
+ for item in verification.get("blocked_candidates") or []
+ }
+ passed_ids = {
+ str(item.get("target_pchome_product_id") or "")
+ for item in (run_receipt.get("post_verifier") or {}).get("readback") or []
+ if item.get("passed")
+ }
+ if target_id in passed_ids:
+ return "promoted_verified", "verified"
+ if target_id in selected_ids:
+ return "apply_not_verified", "failed"
+ if target_id in blocked_ids:
+ return "candidate_rejected", "no_write"
+ return "no_candidate", "no_write"
+
+
+def persist_reconciliation_receipts(
+ engine,
+ *,
+ targets: list[dict[str, Any]],
+ run_receipt: dict[str, Any],
+) -> dict[str, Any]:
+ """Persist one bounded, replayable receipt for each targeted product."""
+ identity = run_receipt.get("identity") or {}
+ run_id = str(identity.get("run_id") or uuid.uuid4().hex)
+ generated_at = str(run_receipt.get("generated_at") or datetime.now(timezone.utc).isoformat())
+ receipt_targets = list(targets or []) or [{
+ "product_id": "__run__",
+ "target_state": "not_needed",
+ "execution_rank": 0,
+ "revenue_share_pct": 0,
+ }]
+ rows: list[dict[str, Any]] = []
+ for target in receipt_targets:
+ target_id = str(target.get("product_id") or "__run__")
+ decision, apply_status = _target_receipt_decision(target, run_receipt)
+ evidence = {
+ "contract_version": CONTRACT_VERSION,
+ "identity": identity,
+ "generated_at": generated_at,
+ "target": {
+ "pchome_product_id": target_id,
+ "target_state_before": target.get("target_state"),
+ "execution_rank": target.get("execution_rank"),
+ "execution_priority_score": target.get("execution_priority_score"),
+ "revenue_share_pct": target.get("revenue_share_pct"),
+ },
+ "source_receipt": run_receipt.get("source_receipt") or {},
+ "source_of_truth_diff": run_receipt.get("coverage_diff") or {},
+ "ai_decision": decision,
+ "risk_policy": run_receipt.get("risk_policy") or {},
+ "check_mode": run_receipt.get("check_mode") or {},
+ "execution": run_receipt.get("execution") or {},
+ "post_verifier": run_receipt.get("post_verifier") or {},
+ "terminal": run_receipt.get("terminal") or {},
+ }
+ payload_hash = _canonical_hash(evidence)
+ rows.append({
+ "receipt_id": f"sameitem-{_canonical_hash([run_id, target_id])[:24]}",
+ "pchome_product_id": target_id,
+ "automation_decision": decision,
+ "payload_hash": payload_hash,
+ "evidence_delta_json": json.dumps(evidence, ensure_ascii=False, default=str),
+ "applied_at": generated_at if apply_status == "verified" else None,
+ "apply_status": apply_status,
+ })
+
+ if engine.dialect.name == "postgresql":
+ insert_sql = """
+ INSERT INTO external_offer_evidence_receipts (
+ receipt_id, pchome_product_id, automation_decision, payload_hash,
+ evidence_delta_json, applied_at, apply_status
+ ) VALUES (
+ :receipt_id, :pchome_product_id, :automation_decision, :payload_hash,
+ CAST(:evidence_delta_json AS JSONB), CAST(:applied_at AS TIMESTAMPTZ), :apply_status
+ )
+ ON CONFLICT (receipt_id) DO UPDATE SET
+ automation_decision = EXCLUDED.automation_decision,
+ payload_hash = EXCLUDED.payload_hash,
+ evidence_delta_json = EXCLUDED.evidence_delta_json,
+ applied_at = EXCLUDED.applied_at,
+ apply_status = EXCLUDED.apply_status
+ """
+ else:
+ insert_sql = """
+ INSERT INTO external_offer_evidence_receipts (
+ receipt_id, pchome_product_id, automation_decision, payload_hash,
+ evidence_delta_json, applied_at, apply_status
+ ) VALUES (
+ :receipt_id, :pchome_product_id, :automation_decision, :payload_hash,
+ :evidence_delta_json, :applied_at, :apply_status
+ )
+ ON CONFLICT (receipt_id) DO UPDATE SET
+ automation_decision = excluded.automation_decision,
+ payload_hash = excluded.payload_hash,
+ evidence_delta_json = excluded.evidence_delta_json,
+ applied_at = excluded.applied_at,
+ apply_status = excluded.apply_status
+ """
+ try:
+ with engine.begin() as conn:
+ conn.execute(text(insert_sql), rows)
+ ids = [row["receipt_id"] for row in rows]
+ readback_stmt = text(
+ "SELECT receipt_id, apply_status FROM external_offer_evidence_receipts "
+ "WHERE receipt_id IN :receipt_ids"
+ ).bindparams(bindparam("receipt_ids", expanding=True))
+ readback = conn.execute(readback_stmt, {"receipt_ids": ids}).mappings().all()
+ readback_ids = {str(row.get("receipt_id") or "") for row in readback}
+ return {
+ "success": len(readback_ids) == len(rows),
+ "status": "persisted_and_verified" if len(readback_ids) == len(rows) else "readback_mismatch",
+ "written_count": len(rows),
+ "readback_count": len(readback_ids),
+ "receipt_ids": [row["receipt_id"] for row in rows],
+ }
+ except Exception as exc:
+ return {
+ "success": False,
+ "status": "persistence_failed",
+ "written_count": 0,
+ "readback_count": 0,
+ "receipt_ids": [],
+ "error_type": type(exc).__name__,
+ }
+
+
+def read_latest_reconciliation_receipts(engine, *, limit: int = 20) -> dict[str, Any]:
+ bounded_limit = max(1, min(int(limit or 20), 100))
+ if not inspect(engine).has_table("external_offer_evidence_receipts"):
+ return {
+ "contract_version": CONTRACT_VERSION,
+ "status": "schema_missing",
+ "receipt_count": 0,
+ "receipts": [],
+ }
+ with engine.connect() as conn:
+ rows = conn.execute(
+ text(
+ "SELECT receipt_id, pchome_product_id, automation_decision, payload_hash, "
+ "evidence_delta_json, created_at, applied_at, apply_status "
+ "FROM external_offer_evidence_receipts "
+ "WHERE receipt_id LIKE 'sameitem-%' "
+ "ORDER BY created_at DESC LIMIT :limit"
+ ),
+ {"limit": bounded_limit},
+ ).mappings().all()
+ receipts: list[dict[str, Any]] = []
+ for row in rows:
+ evidence = row.get("evidence_delta_json")
+ if isinstance(evidence, str):
+ try:
+ evidence = json.loads(evidence)
+ except json.JSONDecodeError:
+ evidence = {}
+ evidence = evidence if isinstance(evidence, dict) else {}
+ receipts.append({
+ "receipt_id": row.get("receipt_id"),
+ "pchome_product_id": row.get("pchome_product_id"),
+ "automation_decision": row.get("automation_decision"),
+ "apply_status": row.get("apply_status"),
+ "payload_hash": row.get("payload_hash"),
+ "created_at": str(row.get("created_at") or ""),
+ "applied_at": str(row.get("applied_at") or ""),
+ "trace_id": (evidence.get("identity") or {}).get("trace_id"),
+ "run_id": (evidence.get("identity") or {}).get("run_id"),
+ "work_item_id": (evidence.get("identity") or {}).get("work_item_id"),
+ "post_verifier_status": (evidence.get("post_verifier") or {}).get("status"),
+ })
+ return {
+ "contract_version": CONTRACT_VERSION,
+ "status": "ready",
+ "receipt_count": len(receipts),
+ "latest_run_id": receipts[0].get("run_id") if receipts else None,
+ "verified_count": sum(1 for item in receipts if item.get("apply_status") == "verified"),
+ "no_write_count": sum(1 for item in receipts if item.get("apply_status") == "no_write"),
+ "failed_count": sum(1 for item in receipts if item.get("apply_status") == "failed"),
+ "receipts": receipts,
+ }
+
+
+__all__ = (
+ "CONTRACT_VERSION",
+ "WORK_ITEM_ID",
+ "build_coverage_post_verifier",
+ "build_coverage_snapshot",
+ "build_revenue_weighted_targets",
+ "ensure_evidence_receipt_table",
+ "new_reconciliation_identity",
+ "persist_reconciliation_receipts",
+ "read_latest_reconciliation_receipts",
+ "verify_same_item_candidates",
+)
diff --git a/services/pchome_revenue_growth_service.py b/services/pchome_revenue_growth_service.py
index 1249fad..89e2426 100644
--- a/services/pchome_revenue_growth_service.py
+++ b/services/pchome_revenue_growth_service.py
@@ -1013,6 +1013,7 @@ def _score_opportunity(
"gap_pct": review_candidate.get("gap_pct"),
"match_reasons": review_candidate.get("match_reasons") or [],
"product_url": review_candidate.get("product_url"),
+ "observed_at": review_candidate.get("observed_at"),
}
data_quality_score = max(data_quality_score, 62)
action_code = "review_external_candidate"
@@ -1247,6 +1248,30 @@ def build_pchome_growth_opportunities(engine, limit: int = 20) -> dict[str, Any]
)
mapping_rate = round(mapped_count / max(len(opportunities), 1) * 100, 1)
observed_mapping_rate = round(observed_mapping_count / max(len(opportunities), 1) * 100, 1)
+ total_opportunity_sales_7d = sum(
+ _to_float(item.get("sales_7d")) for item in opportunities
+ )
+ comparison_ready_sales_7d = sum(
+ _to_float(item.get("sales_7d"))
+ for item in opportunities
+ if item.get("external_price")
+ )
+ candidate_validation_sales_7d = sum(
+ _to_float(item.get("sales_7d"))
+ for item in opportunities
+ if not item.get("external_price")
+ and (item.get("review_candidate") or item.get("market_offers"))
+ )
+ unmatched_sales_7d = max(
+ 0.0,
+ total_opportunity_sales_7d
+ - comparison_ready_sales_7d
+ - candidate_validation_sales_7d,
+ )
+ comparison_revenue_coverage_rate = round(
+ comparison_ready_sales_7d / max(total_opportunity_sales_7d, 1.0) * 100,
+ 1,
+ )
action_counts: dict[str, int] = {}
action_code_counts: dict[str, int] = {}
external_data_source_counts: dict[str, int] = {}
@@ -1312,6 +1337,18 @@ def build_pchome_growth_opportunities(engine, limit: int = 20) -> dict[str, Any]
"rate": _to_float(catalog_mapping_summary.get("active_catalog_mapping_rate")),
"scope": "products_with_positive_sales_in_latest_7_day_window",
},
+ "revenue_coverage": {
+ "numerator": round(comparison_ready_sales_7d, 2),
+ "denominator": round(total_opportunity_sales_7d, 2),
+ "rate": comparison_revenue_coverage_rate,
+ "meaning": "高優先商品近 7 天業績中，已有正式同商品外部價格證據的占比",
+ "state_partition": {
+ "comparison_ready": round(comparison_ready_sales_7d, 2),
+ "candidate_validation": round(candidate_validation_sales_7d, 2),
+ "unmatched": round(unmatched_sales_7d, 2),
+ "total": round(total_opportunity_sales_7d, 2),
+ },
+ },
"platform_runtime": {
"numerator": runtime_external_source_count,
"denominator": registered_external_source_count,
@@ -1342,8 +1379,12 @@ def build_pchome_growth_opportunities(engine, limit: int = 20) -> dict[str, Any]
"candidate_validation_count": candidate_validation_count,
"unmatched_count": unmatched_count,
"unresolved_count": max(0, len(opportunities) - mapped_count),
- "total_sales_7d": round(sum(_to_float(item.get("sales_7d")) for item in opportunities), 2),
- "opportunity_sales_7d": round(sum(_to_float(item.get("sales_7d")) for item in opportunities), 2),
+ "total_sales_7d": round(total_opportunity_sales_7d, 2),
+ "opportunity_sales_7d": round(total_opportunity_sales_7d, 2),
+ "comparison_ready_sales_7d": round(comparison_ready_sales_7d, 2),
+ "candidate_validation_sales_7d": round(candidate_validation_sales_7d, 2),
+ "unmatched_sales_7d": round(unmatched_sales_7d, 2),
+ "comparison_revenue_coverage_rate": comparison_revenue_coverage_rate,
"action_counts": action_counts,
"action_code_counts": action_code_counts,
"external_data_source_counts": external_data_source_counts,
diff --git a/templates/dashboard_v2.html b/templates/dashboard_v2.html
index f3b0589..2f2cbc2 100644
--- a/templates/dashboard_v2.html
+++ b/templates/dashboard_v2.html
@@ -47,10 +47,10 @@
{{ growth.active_product_count | default(0) | number_format }} 個有銷售商品
-
比價可用率
+
正式比價覆蓋
{{ growth.mapping_rate | default(0) }}%
-
{{ growth.mapped_count | default(0) }} / {{ growth.candidate_count | default(0) }} 個高業績商品已對應
+
{{ growth.mapped_count | default(0) }}/{{ growth.candidate_count | default(0) }} 商品 · 業績涵蓋 {{ growth.comparison_revenue_coverage_rate | default(0) }}%
最大業績分類
diff --git a/tests/test_external_market_offer_service.py b/tests/test_external_market_offer_service.py
index 141eb51..723a99a 100644
--- a/tests/test_external_market_offer_service.py
+++ b/tests/test_external_market_offer_service.py
@@ -334,6 +334,13 @@ def test_sync_targeted_momo_candidates_writes_unit_price_offer(monkeypatch):
"competitor_unit_price": 23.0,
"unit_gap_pct": -49.13,
},
+ "same_item_reconciliation": {
+ "trace_id": "trace-unit",
+ "run_id": "run-unit",
+ "work_item_id": "GROWTH-P0-001-B",
+ "independent_verifier_passed": True,
+ },
+ "same_item_candidate_fingerprint": "candidate-unit-1",
},
{
"product_id": "REVIEW-1",
@@ -370,6 +377,9 @@ def test_sync_targeted_momo_candidates_writes_unit_price_offer(monkeypatch):
assert raw_payload["price_basis"] == "unit_price"
assert raw_payload["pchome_public_price"] == 920
assert raw_payload["unit_price_comparison"]["unit_gap_pct"] == -49.13
+ assert raw_payload["same_item_reconciliation"]["run_id"] == "run-unit"
+ assert raw_payload["same_item_reconciliation"]["independent_verifier_passed"] is True
+ assert raw_payload["same_item_candidate_fingerprint"] == "candidate-unit-1"
assert stale_marks == [True]
diff --git a/tests/test_frontend_v2_assets.py b/tests/test_frontend_v2_assets.py
index cb7f9d7..ca572ea 100644
--- a/tests/test_frontend_v2_assets.py
+++ b/tests/test_frontend_v2_assets.py
@@ -340,7 +340,7 @@ def test_dashboard_v2_is_production_default_and_uses_real_dashboard_data():
assert "force_rebuild=False" in route_source
assert "def _load_competitor_decision_overview(session, latest_items=None)" in route_source
assert "def _load_pchome_growth_command_center(session)" in route_source
- assert "build_pchome_growth_opportunities(engine, limit=16)" in route_source
+ assert "build_pchome_growth_opportunities(engine, limit=50)" in route_source
assert "build_pchome_direct_mapping_candidate_decision_package(" in route_source
assert "build_pchome_direct_mapping_retry_candidate_exception_controlled_apply_receipt_replay_package(" in route_source
assert "build_pchome_direct_mapping_retry_candidate_exception_controlled_apply_compact_readback_package(" in route_source
diff --git a/tests/test_pchome_revenue_growth_service.py b/tests/test_pchome_revenue_growth_service.py
index b7dd45a..49428d7 100644
--- a/tests/test_pchome_revenue_growth_service.py
+++ b/tests/test_pchome_revenue_growth_service.py
@@ -287,6 +287,10 @@ def test_pchome_growth_opportunities_use_plain_language_and_pause_shopee_coupang
assert contract["operational_decision"]["ready"] is False
assert contract["operational_decision"]["ready_count"] == 0
assert "sales_data_stale" in contract["operational_decision"]["blocked_reason_codes"]
+ revenue_coverage = contract["revenue_coverage"]
+ assert 0 < revenue_coverage["rate"] < 100
+ assert revenue_coverage["numerator"] < revenue_coverage["denominator"]
+ assert revenue_coverage["state_partition"]["total"] == revenue_coverage["denominator"]
assert contract["platform_runtime"]["denominator"] == 15
actions = {item["pchome_product_id"]: item["recommended_action"]["label"] for item in payload["opportunities"]}
@@ -454,8 +458,7 @@ def test_pchome_growth_momo_backfill_service_targets_unmapped_high_sales_items()
captured = {}
- class FakeEngine:
- pass
+ engine = create_engine("sqlite:///:memory:")
before_payload = {
"success": True,
@@ -463,8 +466,8 @@ def test_pchome_growth_momo_backfill_service_targets_unmapped_high_sales_items()
"opportunities": [
{
"pchome_product_id": "PCH-NEEDS-1",
- "product_name": "需要補對應商品一",
- "pchome_price": 920,
+ "product_name": "理膚寶水 全面修復霜 B5 40ml",
+ "pchome_price": 679,
"sales_7d": 120000,
"priority_score": 91,
"external_price": None,
@@ -479,8 +482,8 @@ def test_pchome_growth_momo_backfill_service_targets_unmapped_high_sales_items()
},
{
"pchome_product_id": "PCH-NEEDS-2",
- "product_name": "需要補對應商品二",
- "pchome_price": 760,
+ "product_name": "理膚寶水 B5 全面修復霜 40ml",
+ "pchome_price": 679,
"sales_7d": 90000,
"priority_score": 82,
"external_price": None,
@@ -490,8 +493,19 @@ def test_pchome_growth_momo_backfill_service_targets_unmapped_high_sales_items()
}
after_payload = {
"success": True,
- "stats": {"mapping_rate": 66.7, "candidate_count": 3, "mapped_count": 2},
- "opportunities": [],
+ "stats": {"mapping_rate": 100, "candidate_count": 2, "mapped_count": 2},
+ "opportunities": [
+ {
+ "pchome_product_id": "PCH-NEEDS-1",
+ "sales_7d": 120000,
+ "external_price": {"momo_sku": "MOMO-AUTO"},
+ },
+ {
+ "pchome_product_id": "PCH-NEEDS-2",
+ "sales_7d": 90000,
+ "external_price": {"momo_sku": "MOMO-UNIT"},
+ },
+ ],
}
payload_calls = []
@@ -504,20 +518,29 @@ def test_pchome_growth_momo_backfill_service_targets_unmapped_high_sales_items()
captured["targets"] = targets
captured["search_limit"] = limit
return True, "找到候選", [
- {
- "product_id": "MOMO-AUTO",
- "auto_compare_type": "total_price",
- "can_auto_compare": True,
- },
- {
- "product_id": "MOMO-UNIT",
- "auto_compare_type": "unit_price",
- "can_auto_compare": True,
- },
- {
- "product_id": "MOMO-REVIEW",
- "auto_compare_type": "manual_review",
- "can_auto_compare": False,
+ {
+ "product_id": "MOMO-AUTO",
+ "name": "理膚寶水 B5 修復霜 40ml",
+ "price": 699,
+ "target_pchome_product_id": "PCH-NEEDS-1",
+ "auto_compare_type": "total_price",
+ "can_auto_compare": True,
+ },
+ {
+ "product_id": "MOMO-UNIT",
+ "name": "理膚寶水 B5 全面修復霜 40ml x2 超值組",
+ "price": 1199,
+ "target_pchome_product_id": "PCH-NEEDS-2",
+ "auto_compare_type": "unit_price",
+ "can_auto_compare": True,
+ },
+ {
+ "product_id": "MOMO-REVIEW",
+ "name": "蘭蔻 玫瑰霜 60ml",
+ "price": 5000,
+ "target_pchome_product_id": "PCH-NEEDS-1",
+ "auto_compare_type": "manual_review",
+ "can_auto_compare": False,
},
]
@@ -540,7 +563,7 @@ def test_pchome_growth_momo_backfill_service_targets_unmapped_high_sales_items()
}
payload = run_pchome_growth_momo_backfill(
- FakeEngine(),
+ engine,
limit=2,
build_payload_func=fake_build_payload,
search_func=fake_search,
@@ -552,14 +575,16 @@ def test_pchome_growth_momo_backfill_service_targets_unmapped_high_sales_items()
assert payload["data"]["scanned_products"] == 2
assert payload["data"]["candidate_count"] == 3
assert payload["data"]["auto_compare_count"] == 2
- assert payload["data"]["review_count"] == 1
+ assert payload["data"]["review_count"] == 0
assert payload["data"]["external_offer_sync"]["written_count"] == 2
- assert payload["data"]["review_candidate_sync"]["written_count"] == 1
- assert payload["data"]["after_stats"]["mapping_rate"] == 66.7
+ assert payload["data"]["review_candidate_sync"]["written_count"] == 0
+ assert payload["data"]["after_stats"]["mapping_rate"] == 100
+ assert payload["data"]["post_verifier"]["all_checks_passed"] is True
+ assert payload["data"]["receipt_persistence"]["readback_count"] == 2
assert [item["product_id"] for item in captured["targets"]] == ["PCH-NEEDS-1", "PCH-NEEDS-2"]
- assert [item["price"] for item in captured["targets"]] == [920, 760]
+ assert [item["price"] for item in captured["targets"]] == [679, 679]
assert [item["product_id"] for item in captured["sync_candidates"]] == ["MOMO-AUTO", "MOMO-UNIT"]
- assert [item["product_id"] for item in captured["review_sync_candidates"]] == ["MOMO-REVIEW"]
+ assert "review_sync_candidates" not in captured
assert captured["search_limit"] == 2
diff --git a/tests/test_pchome_same_item_reconciliation.py b/tests/test_pchome_same_item_reconciliation.py
new file mode 100644
index 0000000..9f807d0
--- /dev/null
+++ b/tests/test_pchome_same_item_reconciliation.py
@@ -0,0 +1,281 @@
+import json
+
+from sqlalchemy import create_engine, text
+
+
+def _target(product_id, name, price, rank):
+ return {
+ "product_id": product_id,
+ "name": name,
+ "price": price,
+ "execution_rank": rank,
+ "execution_priority_score": 90 - rank,
+ "revenue_share_pct": 20,
+ "target_state": "unmatched",
+ }
+
+
+def test_revenue_weighted_targets_include_candidate_validation_and_sort_by_revenue():
+ from services.pchome_growth_same_item_reconciliation import (
+ build_revenue_weighted_targets,
+ )
+
+ payload = {
+ "opportunities": [
+ {
+ "pchome_product_id": "LOW",
+ "product_name": "低業績未配對",
+ "sales_7d": 100,
+ "sales_prev_7d": 100,
+ "priority_score": 95,
+ "external_price": None,
+ "recommended_action": {"code": "map_external_product"},
+ },
+ {
+ "pchome_product_id": "HIGH",
+ "product_name": "高業績候選",
+ "sales_7d": 10000,
+ "sales_prev_7d": 9000,
+ "priority_score": 40,
+ "external_price": None,
+ "review_candidate": {"momo_sku": "M-1"},
+ "recommended_action": {"code": "review_external_candidate"},
+ },
+ {
+ "pchome_product_id": "READY",
+ "product_name": "已有正式比價",
+ "sales_7d": 50000,
+ "priority_score": 100,
+ "external_price": {"momo_sku": "M-2"},
+ "recommended_action": {"code": "monitor"},
+ },
+ ]
+ }
+
+ targets = build_revenue_weighted_targets(payload, 10)
+
+ assert [item["product_id"] for item in targets] == ["HIGH", "LOW"]
+ assert targets[0]["target_state"] == "candidate_validation"
+ assert targets[0]["execution_rank"] == 1
+ assert targets[0]["execution_priority_score"] > targets[1]["execution_priority_score"]
+
+
+def test_independent_verifier_promotes_exact_and_unit_candidates_but_rejects_conflict():
+ from services.pchome_growth_same_item_reconciliation import (
+ verify_same_item_candidates,
+ )
+
+ targets = [
+ _target("P-EXACT", "理膚寶水 全面修復霜 B5 40ml", 679, 1),
+ _target("P-UNIT", "理膚寶水 全面修復霜 B5 40ml", 679, 2),
+ _target("P-BLOCK", "巴黎萊雅 膠原輕盈乳霜 60ml", 1249, 3),
+ ]
+ candidates = [
+ {
+ "product_id": "M-EXACT",
+ "name": "理膚寶水 B5 修復霜 40ml",
+ "price": 699,
+ "target_pchome_product_id": "P-EXACT",
+ "auto_compare_type": "manual_review",
+ },
+ {
+ "product_id": "M-UNIT",
+ "name": "理膚寶水 B5 全面修復霜 40ml x2 超值組",
+ "price": 1199,
+ "target_pchome_product_id": "P-UNIT",
+ "auto_compare_type": "unit_price",
+ },
+ {
+ "product_id": "M-BLOCK",
+ "name": "蘭蔻 玫瑰霜 60ml",
+ "price": 5000,
+ "target_pchome_product_id": "P-BLOCK",
+ "auto_compare_type": "total_price",
+ },
+ ]
+ identity = {
+ "trace_id": "trace-1",
+ "run_id": "run-1",
+ "work_item_id": "GROWTH-P0-001-B",
+ }
+
+ result = verify_same_item_candidates(targets, candidates, identity=identity)
+ selected = {item["target_pchome_product_id"]: item for item in result["verified_candidates"]}
+
+ assert result["selected_candidate_count"] == 2
+ assert result["blocked_candidate_count"] == 1
+ assert result["claim_drift_count"] == 1
+ assert selected["P-EXACT"]["auto_compare_type"] == "total_price"
+ assert selected["P-UNIT"]["auto_compare_type"] == "unit_price"
+ assert selected["P-UNIT"]["target_unit_price_comparison"]["comparable"] is True
+ assert selected["P-EXACT"]["same_item_candidate_fingerprint"]
+ assert result["blocked_candidates"][0]["reason_code"] == "same_item_verifier_rejected"
+
+
+def test_coverage_post_verifier_requires_exact_source_readback_and_reports_revenue_delta():
+ from services.pchome_growth_same_item_reconciliation import (
+ build_coverage_post_verifier,
+ )
+
+ before = {
+ "stats": {"mapping_rate": 0, "mapped_count": 0},
+ "opportunities": [
+ {"pchome_product_id": "P-1", "sales_7d": 8000, "external_price": None},
+ {"pchome_product_id": "P-2", "sales_7d": 2000, "external_price": None},
+ ],
+ }
+ after = {
+ "stats": {"mapping_rate": 50, "mapped_count": 1},
+ "opportunities": [
+ {
+ "pchome_product_id": "P-1",
+ "sales_7d": 8000,
+ "external_price": {"momo_sku": "M-1"},
+ },
+ {"pchome_product_id": "P-2", "sales_7d": 2000, "external_price": None},
+ ],
+ }
+
+ result = build_coverage_post_verifier(
+ before_payload=before,
+ after_payload=after,
+ verified_candidates=[{
+ "target_pchome_product_id": "P-1",
+ "product_id": "M-1",
+ }],
+ sync_result={"written_count": 1},
+ )
+
+ assert result["status"] == "verified"
+ assert result["all_checks_passed"] is True
+ assert result["comparison_ready_delta"] == 1
+ assert result["count_coverage_delta"] == 50
+ assert result["revenue_coverage_delta"] == 80
+
+
+def test_receipt_schema_apply_persistence_and_public_safe_readback():
+ from services.pchome_growth_same_item_reconciliation import (
+ ensure_evidence_receipt_table,
+ persist_reconciliation_receipts,
+ read_latest_reconciliation_receipts,
+ )
+
+ engine = create_engine("sqlite:///:memory:")
+ schema = ensure_evidence_receipt_table(engine)
+ target = _target("P-1", "商品", 100, 1)
+ run_receipt = {
+ "generated_at": "2026-07-14T12:00:00+00:00",
+ "identity": {
+ "trace_id": "trace-1",
+ "run_id": "run-1",
+ "work_item_id": "GROWTH-P0-001-B",
+ },
+ "source_receipt": {"candidate_count": 1},
+ "candidate_verification": {
+ "verified_candidates": [{"target_pchome_product_id": "P-1"}],
+ "blocked_candidates": [],
+ },
+ "coverage_diff": {"comparison_ready_delta": 1},
+ "risk_policy": {"risk": "medium"},
+ "check_mode": {"verified_candidate_count": 1},
+ "execution": {"written_count": 1},
+ "post_verifier": {
+ "status": "verified",
+ "readback": [{"target_pchome_product_id": "P-1", "passed": True}],
+ },
+ "terminal": {"status": "verified_with_coverage_gain"},
+ }
+
+ persistence = persist_reconciliation_receipts(
+ engine,
+ targets=[target],
+ run_receipt=run_receipt,
+ )
+ readback = read_latest_reconciliation_receipts(engine)
+
+ assert schema["status"] == "created_and_verified"
+ assert persistence["status"] == "persisted_and_verified"
+ assert persistence["written_count"] == 1
+ assert readback["status"] == "ready"
+ assert readback["verified_count"] == 1
+ assert readback["latest_run_id"] == "run-1"
+ assert readback["receipts"][0]["work_item_id"] == "GROWTH-P0-001-B"
+
+
+def test_targeted_momo_search_keeps_same_source_product_per_pchome_target():
+ from services.momo_crawler import search_momo_products_for_pchome_products
+
+ class FakeCrawler:
+ def search_products(self, term, limit):
+ return True, "ok", [{
+ "product_id": "MOMO-SAME",
+ "name": "理膚寶水 B5 全面修復霜 40ml",
+ "price": 699,
+ }]
+
+ success, _, candidates = search_momo_products_for_pchome_products(
+ [
+ {"product_id": "P-1", "name": "理膚寶水 全面修復霜 B5 40ml", "price": 679},
+ {"product_id": "P-2", "name": "理膚寶水 全面修復霜 B5 40ml", "price": 679},
+ ],
+ limit_per_product=3,
+ max_products=2,
+ max_terms_per_product=2,
+ crawler=FakeCrawler(),
+ )
+
+ assert success is True
+ assert len(candidates) == 2
+ assert {item["target_pchome_product_id"] for item in candidates} == {"P-1", "P-2"}
+
+
+def test_dashboard_exposes_count_and_revenue_coverage_without_a_text_wall():
+ from pathlib import Path
+
+ template = Path("templates/dashboard_v2.html").read_text(encoding="utf-8")
+
+ assert "正式比價覆蓋" in template
+ assert "業績涵蓋" in template
+ assert "growth.comparison_revenue_coverage_rate" in template
+
+
+def test_same_item_reconciliation_route_returns_durable_readback(monkeypatch):
+ from flask import Flask
+ from routes import ai_routes as routes
+ from services.pchome_growth_same_item_reconciliation import (
+ ensure_evidence_receipt_table,
+ )
+
+ engine = create_engine("sqlite:///:memory:")
+ ensure_evidence_receipt_table(engine)
+ with engine.begin() as conn:
+ conn.execute(text("""
+ INSERT INTO external_offer_evidence_receipts (
+ receipt_id, pchome_product_id, automation_decision, payload_hash,
+ evidence_delta_json, apply_status
+ ) VALUES (
+ 'sameitem-route', 'P-1', 'candidate_rejected', 'hash-1',
+ :evidence, 'no_write'
+ )
+ """), {
+ "evidence": json.dumps({
+ "identity": {
+ "trace_id": "trace-route",
+ "run_id": "run-route",
+ "work_item_id": "GROWTH-P0-001-B",
+ },
+ "post_verifier": {"status": "no_write_verified"},
+ })
+ })
+
+ monkeypatch.setattr(routes, "_create_icaim_dashboard_engine", lambda _: engine)
+ app = Flask(__name__)
+ with app.test_request_context(
+ "/api/ai/pchome-growth/same-item-reconciliation?limit=5"
+ ):
+ response = routes.api_pchome_growth_same_item_reconciliation.__wrapped__()
+
+ payload = response.get_json()
+ assert payload["status"] == "ready"
+ assert payload["receipt_count"] == 1
+ assert payload["latest_run_id"] == "run-route"
diff --git a/tests/test_run_scheduler_embed_consistency.py b/tests/test_run_scheduler_embed_consistency.py
index 656a803..4d39f9e 100644
--- a/tests/test_run_scheduler_embed_consistency.py
+++ b/tests/test_run_scheduler_embed_consistency.py
@@ -261,6 +261,31 @@ def test_pchome_growth_backfill_catchup_uses_runtime_receipt_cooldown(monkeypatc
assert reason == "stale_receipt_4.5h"
+def test_pchome_growth_backfill_degraded_receipt_retries_after_one_hour(monkeypatch, tmp_path):
+ run_scheduler = _load_run_scheduler(monkeypatch)
+ stats_path = tmp_path / "scheduler_stats.json"
+ stats_path.write_text(json.dumps({
+ "pchome_growth_momo_backfill": [{
+ "status": "Degraded",
+ "last_run": "2026-07-10 09:00:00",
+ "terminal_status": "degraded_no_safe_candidate",
+ }],
+ }), encoding="utf-8")
+ monkeypatch.setenv("SCHEDULER_STATS_PATH", str(stats_path))
+
+ due, reason = run_scheduler._pchome_growth_backfill_due(
+ now=datetime(2026, 7, 10, 9, 30, 0)
+ )
+ assert due is False
+ assert reason == "cooldown_1.0h"
+
+ due, reason = run_scheduler._pchome_growth_backfill_due(
+ now=datetime(2026, 7, 10, 10, 0, 0)
+ )
+ assert due is True
+ assert reason == "stale_receipt_1.0h"
+
+
def test_pchome_growth_backfill_catchup_executes_missing_receipt(monkeypatch, tmp_path):
run_scheduler = _load_run_scheduler(monkeypatch)
monkeypatch.setenv("SCHEDULER_STATS_PATH", str(tmp_path / "missing.json"))