feat(growth): automate sales source reconciliation

config.py

@@ -414,7 +414,7 @@ YOUTUBE_API_KEY = os.getenv('YOUTUBE_API_KEY', '')
 # ==========================================
 # 系統版本與路徑
 # ==========================================
-SYSTEM_VERSION = "V10.789"
+SYSTEM_VERSION = "V10.790"
 LOG_FILE_PATH = os.path.join(BASE_DIR, 'logs/system.log')
 public_url = PUBLIC_URL  # 用於模板顯示
 

docs/AI_INTELLIGENCE_MODULE_SOT.md

@@ -344,7 +344,7 @@ SQL漏斗(~300筆)
 > **重要**: 此表由 `import_service.py` 使用 `df.to_sql()` 動態建立。
 > 欄位名稱**完全繼承自 PChome 後台匯出的業績 Excel 原始欄位**，加上程式碼追加的 `snapshot_date`。
 > V10.605 起，匯入器會掃描所有 worksheet 與前 15 列表頭，優先選擇含「日期 / 商品名稱 / 總業績或銷售金額」的明細工作表；格式或日期真的不合格的檔案會移到 Google Drive `匯入失敗`，避免每 30 分鐘重複告警。
-> V10.776 起，`pchome_sales_acquisition_service.py` 依序探測 Google Drive、顯式授權 HTTPS、顯式授權 IMAP 與受控落地目錄；所有非 Drive 檔案先做副檔名/魔術碼/大小檢查與 SHA-256 冪等，再由 `import_service.py` 在同一 DB transaction 內 replace、寫入及驗證 `daily_sales_snapshot` 與 `realtime_sales_monthly`。任一步失敗必須整批 rollback，不得留下半同步狀態。每次 run 需把 `trace_id`、OpenTelemetry-compatible `span_id`、`run_id`、`work_item_id`、來源、政策、check、execution、post-verifier 與 terminal 寫入 `pchome_sales_acquisition_receipts`；舊版人工暫停旗標不得停止此 medium-risk controlled-apply lane。正式完成仍要求最新業績 <= 1 天，不能以 source/test/UI 綠燈取代 production freshness closure。
+> V10.776 起，`pchome_sales_acquisition_service.py` 依序探測 Google Drive、顯式授權 HTTPS、顯式授權 IMAP 與受控落地目錄；所有非 Drive 檔案先做副檔名/魔術碼/大小檢查與 SHA-256 冪等，再由 `import_service.py` 在同一 DB transaction 內 replace、寫入及驗證 `daily_sales_snapshot` 與 `realtime_sales_monthly`。任一步失敗必須整批 rollback，不得留下半同步狀態。每次 run 需把 `trace_id`、OpenTelemetry-compatible `span_id`、`run_id`、`work_item_id`、來源、政策、check、execution、post-verifier 與 terminal 寫入 `pchome_sales_acquisition_receipts`；舊版人工暫停旗標不得停止此 medium-risk controlled-apply lane。V10.790 起再以 `pchome_sales_source_reconciliation_v1` 跨 run 累積 provider 空跑/失敗次數，disabled 來源不得冒充 `no_candidate`，ready-empty 走固定重試、provider failure 走有界指數退避，所有已啟用 fallback 仍依序探測；相同狀態 fingerprint 在 cooldown 內抑制重複通知，決策與 `next_retry_at` 一併寫入原 receipt。正式完成仍要求最新業績 <= 1 天，不能以 source/test/UI 綠燈取代 production freshness closure。
 
 #### 已確認的關鍵欄位（PChome 後台報表欄位名稱）
 

docs/guides/pchome_sales_import_automation.md

@@ -73,3 +73,17 @@ PChome 後台業績匯出
 - `daily_sales_snapshot` 與 `realtime_sales_monthly` 最新日期落後不超過 1 天。
 - `/daily_sales` 與 `/growth_analysis` 圖表 smoke 通過。
 - 失敗檔只告警一次，並被移到 `匯入失敗`。
+
+## 來源 Reconciliation 合約
+
+`PChomeSalesAcquisitionService` 每輪必須把四個授權來源的設定狀態、探測結果、
+連續空跑/失敗次數、fallback chain、下一次重試時間與通知去重決策寫入同一張
+`pchome_sales_acquisition_receipts`。正式合約名稱為
+`pchome_sales_source_reconciliation_v1`。
+
+- ready 但無檔：維持 30 分鐘固定重試，並繼續探測所有已啟用 fallback。
+- provider 失敗：使用 30、60、120、240 分鐘有界指數退避。
+- disabled 來源：必須顯示為 `disabled`，不得誤報為已探測但沒有候選。
+- 相同 stale/error fingerprint：6 小時內抑制重複通知；狀態改變立即通知。
+- 低階 Drive importer 由 acquisition controller 呼叫時不直接發 stale 告警，避免雙重通知。
+- receipt 與 API 只能輸出 public-safe 狀態，不得回傳 URL、帳號、token 或郵件內容。

scheduler.py

@@ -2060,7 +2060,40 @@ def run_auto_import_task():
                 "decision_ready": result.get('decision_ready'),
                 "requires_upstream_acquisition": result.get('requires_upstream_acquisition', True),
                 "safe_next_action": result.get('safe_next_action'),
+                "next_retry_at": result.get('next_retry_at'),
+                "notification_policy": result.get('notification_policy'),
             }
+            notification_policy = result.get('notification_policy') or {}
+            if notification_policy.get('send'):
+                try:
+                    now_str = datetime.now(TAIPEI_TZ).strftime('%Y-%m-%d %H:%M')
+                    next_retry = result.get('next_retry_at') or '下一輪排程'
+                    message = (
+                        f"[EwoooC] PChome 業績來源需自動重試 ({now_str})\n"
+                        f"狀態：{run_status}\n"
+                        f"最新業績：{result.get('latest_sales_date') or '尚未建立'}\n"
+                        f"落後：{result.get('data_lag_days')} 天\n"
+                        f"下一輪：{next_retry}\n"
+                        f"Trace：{str(result.get('trace_id') or '')[:12]}"
+                    )
+                    notifier = NotificationManager()
+                    notifier._send_line_messages([message])
+                    notifier._send_telegram_messages([message])
+                    notification_sent = True
+                    logging.info(
+                        "[Scheduler] [AutoImport] source-state notification sent | fingerprint=%s",
+                        notification_policy.get('fingerprint'),
+                    )
+                except Exception as notify_error:
+                    logging.error(
+                        "[Scheduler] [AutoImport] source-state notification failed | Error: %s",
+                        notify_error,
+                    )
+            else:
+                logging.info(
+                    "[Scheduler] [AutoImport] duplicate source-state notification suppressed | reason=%s",
+                    notification_policy.get('reason') or 'policy_suppressed',
+                )
         elif result.get('success'):
             logging.info(f"[Scheduler] [AutoImport] ✅ 自動匯入完成 | Message: {result.get('message')}")
             stats = {
@@ -2164,27 +2197,37 @@ def run_auto_import_task():
                 "receipt_id": result.get('receipt_id'),
                 "receipt_persisted": result.get('receipt_persisted'),
                 "safe_next_action": result.get('safe_next_action'),
+                "next_retry_at": result.get('next_retry_at'),
+                "notification_policy": result.get('notification_policy'),
             }
 
             # V-New: 匯入失敗時也發送通知
-            try:
-                logging.info("[Scheduler] [AutoImport] 📢 準備發送自動匯入失敗通知...")
-                now_str = datetime.now(TAIPEI_TZ).strftime('%Y-%m-%d %H:%M')
-                message = (
-                    f"⚠️ 當日業績自動匯入失敗 ({now_str})\n"
-                    f"{'='*30}\n"
-                    f"❌ 匯入狀態：失敗\n"
-                    f"📌 錯誤訊息：{result.get('message', 'Unknown error')}\n"
-                    f"🔎 Trace：{str(result.get('trace_id') or '')[:12]}\n"
-                    f"{'='*30}\n"
-                    f"系統將依安全重試策略再次取得授權來源"
+            notification_policy = result.get('notification_policy') or {}
+            if notification_policy and not notification_policy.get('send'):
+                logging.info(
+                    "[Scheduler] [AutoImport] duplicate failure notification suppressed | reason=%s",
+                    notification_policy.get('reason') or 'policy_suppressed',
                 )
-                notifier = NotificationManager()
-                notifier._send_line_messages([message])
-                notifier._send_telegram_messages([message])
-                logging.info("[Scheduler] [AutoImport] ✅ 匯入失敗通知已發送")
-            except Exception as e:
-                logging.error(f"[Scheduler] [AutoImport] ❌ 發送失敗通知時發生錯誤 | Error: {e}")
+            else:
+                try:
+                    logging.info("[Scheduler] [AutoImport] 📢 準備發送自動匯入失敗通知...")
+                    now_str = datetime.now(TAIPEI_TZ).strftime('%Y-%m-%d %H:%M')
+                    message = (
+                        f"⚠️ 當日業績自動匯入失敗 ({now_str})\n"
+                        f"{'='*30}\n"
+                        f"❌ 匯入狀態：失敗\n"
+                        f"📌 錯誤訊息：{result.get('message', 'Unknown error')}\n"
+                        f"🔎 Trace：{str(result.get('trace_id') or '')[:12]}\n"
+                        f"{'='*30}\n"
+                        f"系統將依安全重試策略再次取得授權來源"
+                    )
+                    notifier = NotificationManager()
+                    notifier._send_line_messages([message])
+                    notifier._send_telegram_messages([message])
+                    notification_sent = True
+                    logging.info("[Scheduler] [AutoImport] ✅ 匯入失敗通知已發送")
+                except Exception as e:
+                    logging.error(f"[Scheduler] [AutoImport] ❌ 發送失敗通知時發生錯誤 | Error: {e}")
 
         _save_stats('auto_import_task', stats)
 

services/import_service.py

@@ -1175,7 +1175,7 @@ class ImportService:
             logger.error(traceback.format_exc())
             return False
 
-    def auto_import_from_drive(self) -> Dict[str, Any]:
+    def auto_import_from_drive(self, *, send_stale_alert: bool = True) -> Dict[str, Any]:
         """
         從 Google Drive 自動匯入檔案
 
@@ -1243,7 +1243,7 @@ class ImportService:
                         days_since = (datetime.now(TAIPEI_TZ).date() - normalized_last_date).days
                         data_lag_days = days_since
                         latest_sales_date = str(normalized_last_date)
-                        if days_since >= 3:
+                        if days_since >= 3 and send_stale_alert:
                             _send_data_stale_alert(
                                 report_type="upstream_drive",
                                 last_date=str(normalized_last_date),

services/pchome_sales_acquisition_service.py

@@ -23,6 +23,7 @@ from services.pchome_sales_acquisition_providers import (
     env_bool,
     env_int,
 )
+from services.pchome_sales_source_reconciliation import build_source_reconciliation
 
 
 logger = logging.getLogger(__name__)
@@ -80,7 +81,20 @@ class PChomeSalesAcquisitionService:
         finally:
             session.close()
 
-    def readiness(self) -> Dict[str, object]:
+    def _recent_receipts(self, limit: int = 48) -> List[Dict[str, object]]:
+        session = Session()
+        try:
+            receipts = session.query(SalesAcquisitionReceipt).order_by(
+                SalesAcquisitionReceipt.started_at.desc()
+            ).limit(max(1, min(limit, 336))).all()
+            return [receipt.to_public_dict() for receipt in receipts]
+        except Exception:
+            logger.warning("Unable to read sales acquisition history", exc_info=True)
+            return []
+        finally:
+            session.close()
+
+    def _source_inventory(self) -> List[Dict[str, object]]:
         try:
             drive = drive_service.check_auth_readiness(refresh_expired=False)
             drive_ready = bool(drive.get("ready"))
@@ -104,7 +118,7 @@ class PChomeSalesAcquisitionService:
         )
         local_path = os.getenv("PCHOME_SALES_LOCAL_DROP_DIR", "").strip()
         local_ready = bool(local_path and Path(local_path).is_dir())
-        sources = [
+        return [
             {"id": "google_drive", "enabled": True, "ready": drive_ready, "status": drive_status},
             {
                 "id": "authorized_https",
@@ -125,8 +139,13 @@ class PChomeSalesAcquisitionService:
                 "status": "ready" if local_ready else "disabled" if not local_path else "path_missing",
             },
         ]
+
+    def readiness(self) -> Dict[str, object]:
+        sources = self._source_inventory()
         ready_count = sum(1 for source in sources if source["ready"])
         freshness = self._freshness()
+        last_receipt = self._latest_receipt()
+        last_stages = (last_receipt or {}).get("stages") or {}
         return {
             "success": True,
             "automation_mode": "controlled_apply",
@@ -141,7 +160,8 @@ class PChomeSalesAcquisitionService:
             },
             "freshness": freshness,
             "runtime_closure": "closed" if freshness["decision_ready"] else "blocked_by_upstream_freshness",
-            "last_receipt": self._latest_receipt(),
+            "source_reconciliation": last_stages.get("source_reconciliation"),
+            "last_receipt": last_receipt,
         }
 
     def _persist_receipt(
@@ -206,6 +226,9 @@ class PChomeSalesAcquisitionService:
         span_id = uuid.uuid4().hex[:16]
         run_id = uuid.uuid4().hex
         before = self._freshness()
+        source_inventory = self._source_inventory()
+        source_by_id = {str(source["id"]): source for source in source_inventory}
+        prior_receipts = self._recent_receipts()
         stages: Dict[str, object] = {
             "sensor": {"status": "started", "trigger": trigger, "trace_id": trace_id, "span_id": span_id},
             "normalize": {"asset_id": "pchome.daily_sales.authorized_report"},
@@ -224,9 +247,11 @@ class PChomeSalesAcquisitionService:
         error_kind = None
 
         try:
-            drive_result = import_service.auto_import_from_drive()
+            drive_result = import_service.auto_import_from_drive(send_stale_alert=False)
             stages["providers"].append({
                 "source": "google_drive",
+                "enabled": True,
+                "ready": bool(source_by_id.get("google_drive", {}).get("ready")),
                 "status": drive_result.get("status") or ("completed" if drive_result.get("success") else "failed"),
                 "file_count": drive_result.get("file_count", 0),
                 "imported_count": drive_result.get("imported_count", 0),
@@ -258,17 +283,49 @@ class PChomeSalesAcquisitionService:
                     ("authorized_imap", self.providers.imap_candidates),
                     ("controlled_local_drop", self.providers.local_candidates),
                 ):
+                    source_state = source_by_id.get(source_name) or {}
+                    if not source_state.get("enabled"):
+                        stages["providers"].append({
+                            "source": source_name,
+                            "enabled": False,
+                            "ready": False,
+                            "status": "disabled",
+                            "candidate_count": 0,
+                        })
+                        continue
+                    if not source_state.get("ready"):
+                        failed_results.append({
+                            "source_type": source_name,
+                            "error_kind": "provider_not_ready",
+                            "message": "授權來源設定尚未達到自動執行條件。",
+                        })
+                        stages["providers"].append({
+                            "source": source_name,
+                            "enabled": True,
+                            "ready": False,
+                            "status": "not_ready",
+                            "error_kind": "provider_not_ready",
+                        })
+                        continue
                     try:
                         provider_candidates = provider()
                         candidates.extend(provider_candidates)
                         stages["providers"].append({
                             "source": source_name,
+                            "enabled": True,
+                            "ready": True,
                             "status": "candidate_found" if provider_candidates else "no_candidate",
                             "candidate_count": len(provider_candidates),
                         })
                     except ProviderError as exc:
                         failed_results.append({"source_type": source_name, "error_kind": exc.kind, "message": exc.public_message})
-                        stages["providers"].append({"source": source_name, "status": "failed", "error_kind": exc.kind})
+                        stages["providers"].append({
+                            "source": source_name,
+                            "enabled": True,
+                            "ready": True,
+                            "status": "failed",
+                            "error_kind": exc.kind,
+                        })
 
                 max_files = env_int("PCHOME_SALES_MAX_FILES_PER_RUN", 5, 1, 25)
                 for candidate in candidates[:max_files]:
@@ -342,6 +399,36 @@ class PChomeSalesAcquisitionService:
                 except Exception:
                     logger.warning("Unable to clean sales acquisition temp file", exc_info=True)
 
+        for provider_stage in stages["providers"]:
+            provider_source = provider_stage.get("source")
+            imported_for_source = [
+                item for item in successful_results
+                if item.get("source_type") == provider_source
+            ]
+            duplicates_for_source = [
+                item for item in duplicate_results
+                if item.get("source_type") == provider_source
+            ]
+            failures_for_source = [
+                item for item in failed_results
+                if item.get("source_type") == provider_source
+            ]
+            if imported_for_source:
+                provider_stage["status"] = "completed"
+                provider_stage["imported_count"] = len(imported_for_source)
+                provider_stage["rows_imported"] = sum(
+                    int(item.get("rows") or 0) for item in imported_for_source
+                )
+            elif duplicates_for_source:
+                provider_stage["status"] = "duplicate"
+                provider_stage["duplicate_count"] = len(duplicates_for_source)
+            elif failures_for_source and provider_stage.get("status") != "disabled":
+                provider_stage["status"] = "failed"
+                provider_stage["error_kind"] = (
+                    provider_stage.get("error_kind")
+                    or failures_for_source[0].get("error_kind")
+                )
+
         verifier = self._freshness()
         rows_imported = sum(int(item.get("rows") or 0) for item in successful_results)
         import_job_id = next((item.get("job_id") for item in reversed(successful_results) if item.get("job_id")), None)
@@ -409,6 +496,24 @@ class PChomeSalesAcquisitionService:
             decision = "no_authorized_candidate"
             message = "所有已啟用授權來源目前都沒有新業績檔，正式資料未變更。"
 
+        source_reconciliation = build_source_reconciliation(
+            sources=source_inventory,
+            provider_attempts=stages["providers"],
+            prior_receipts=prior_receipts,
+            freshness=verifier,
+            now=datetime.now(TAIPEI_TZ),
+            retry_interval_minutes=env_int(
+                "PCHOME_SALES_RETRY_INTERVAL_MINUTES", 30, 1, 1440
+            ),
+            max_failure_backoff_minutes=env_int(
+                "PCHOME_SALES_MAX_FAILURE_BACKOFF_MINUTES", 240, 30, 1440
+            ),
+            alert_cooldown_minutes=env_int(
+                "PCHOME_SALES_ALERT_COOLDOWN_MINUTES", 360, 30, 1440
+            ),
+        )
+        stages["source_reconciliation"] = source_reconciliation
+
         stages["execution"] = {
             "status": status,
             "successful_files": len(successful_results),
@@ -458,6 +563,13 @@ class PChomeSalesAcquisitionService:
             decision = "runtime_verified_receipt_write_failed"
             message = "業績資料已通過驗證，但治理 receipt 寫入失敗，閉環尚未完成。"
 
+        if error_kind == "source_finalize_failed":
+            safe_next_action = "retry_source_finalization"
+        elif status == "failed_rolled_back":
+            safe_next_action = "replay_transaction_after_rollback"
+        else:
+            safe_next_action = source_reconciliation.get("safe_next_action") or "none"
+
         return {
             "success": status in {"completed", "completed_no_write"},
             "status": status,
@@ -481,15 +593,10 @@ class PChomeSalesAcquisitionService:
             "decision_ready": verifier.get("decision_ready"),
             "requires_upstream_acquisition": not bool(verifier.get("decision_ready")),
             "manual_review_required": False,
-            "safe_next_action": (
-                "scheduler_retry_authorized_sources"
-                if status == "blocked_with_safe_next_action"
-                else "repair_provider_or_replay_transaction"
-                if status in {"failed_rolled_back", "degraded_no_write"}
-                else "retry_source_finalization"
-                if error_kind == "source_finalize_failed"
-                else "none"
-            ),
+            "safe_next_action": safe_next_action,
+            "next_retry_at": source_reconciliation.get("next_retry_at"),
+            "notification_policy": source_reconciliation.get("notification_policy"),
+            "source_reconciliation": source_reconciliation,
             "providers": stages["providers"],
             "verifier": verifier,
         }

services/pchome_sales_source_reconciliation.py

@@ -0,0 +1,326 @@
+"""Cross-run source health, retry, fallback, and alert decisions for PChome sales."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from datetime import datetime, timedelta
+from typing import Dict, Iterable, List, Mapping, Optional
+
+
+SOURCE_PRIORITY = (
+    "google_drive",
+    "authorized_https",
+    "authorized_imap",
+    "controlled_local_drop",
+)
+
+_EMPTY_STATUSES = {
+    "no_candidate",
+    "no_pending_file",
+    "upstream_missing",
+    "upstream_stale",
+}
+_FAILED_STATUSES = {"failed", "not_ready", "probe_failed"}
+
+
+def _provider_category(
+    source: Mapping[str, object],
+    attempt: Optional[Mapping[str, object]],
+) -> str:
+    if not bool(source.get("enabled")):
+        return "disabled"
+    if not attempt:
+        return "not_probed"
+
+    status = str(attempt.get("status") or "unknown")
+    if int(attempt.get("imported_count") or 0) > 0:
+        return "imported"
+    if int(attempt.get("duplicate_count") or 0) > 0 or status == "duplicate":
+        return "duplicate"
+    if int(attempt.get("candidate_count") or 0) > 0:
+        return "candidate_found"
+    if attempt.get("error_kind") or status in _FAILED_STATUSES:
+        return "failed"
+    if status in _EMPTY_STATUSES:
+        return "empty"
+    if not bool(source.get("ready")):
+        return "not_ready"
+    return "ready"
+
+
+def _history_attempts(
+    receipts: Iterable[Mapping[str, object]],
+    source_id: str,
+) -> List[Dict[str, object]]:
+    attempts: List[Dict[str, object]] = []
+    for receipt in receipts:
+        stages = receipt.get("stages") or {}
+        if not isinstance(stages, Mapping):
+            continue
+        providers = stages.get("providers") or []
+        if not isinstance(providers, list):
+            continue
+        for provider in providers:
+            if isinstance(provider, Mapping) and provider.get("source") == source_id:
+                attempts.append({
+                    **dict(provider),
+                    "receipt_status": receipt.get("status"),
+                    "attempted_at": receipt.get("completed_at") or receipt.get("started_at"),
+                })
+                break
+    return attempts
+
+
+def _consecutive_count(categories: Iterable[str], target: str) -> int:
+    count = 0
+    for category in categories:
+        if category != target:
+            break
+        count += 1
+    return count
+
+
+def _parse_datetime(value: object) -> Optional[datetime]:
+    if not value:
+        return None
+    try:
+        parsed = datetime.fromisoformat(str(value).replace("Z", "+00:00"))
+    except (TypeError, ValueError):
+        return None
+    return parsed
+
+
+def _lag_bucket(lag_days: object) -> str:
+    if lag_days is None:
+        return "missing"
+    try:
+        lag = int(lag_days)
+    except (TypeError, ValueError):
+        return "missing"
+    if lag <= 1:
+        return "fresh"
+    if lag <= 3:
+        return "warning"
+    if lag <= 7:
+        return "stale"
+    return "critical"
+
+
+def _previous_alert(receipts: Iterable[Mapping[str, object]]) -> Dict[str, object]:
+    for receipt in receipts:
+        stages = receipt.get("stages") or {}
+        if not isinstance(stages, Mapping):
+            continue
+        reconciliation = stages.get("source_reconciliation") or {}
+        if not isinstance(reconciliation, Mapping):
+            continue
+        alert = reconciliation.get("notification_policy") or {}
+        if isinstance(alert, Mapping) and alert.get("fingerprint"):
+            return {
+                **dict(alert),
+                "completed_at": receipt.get("completed_at") or receipt.get("started_at"),
+            }
+    return {}
+
+
+def build_source_reconciliation(
+    *,
+    sources: Iterable[Mapping[str, object]],
+    provider_attempts: Iterable[Mapping[str, object]],
+    prior_receipts: Iterable[Mapping[str, object]],
+    freshness: Mapping[str, object],
+    now: datetime,
+    retry_interval_minutes: int = 30,
+    max_failure_backoff_minutes: int = 240,
+    alert_cooldown_minutes: int = 360,
+) -> Dict[str, object]:
+    """Build a public-safe decision without exposing provider configuration values."""
+    source_list = [dict(source) for source in sources]
+    attempt_by_source = {
+        str(item.get("source")): dict(item)
+        for item in provider_attempts
+        if item.get("source")
+    }
+    receipt_list = list(prior_receipts)
+    source_states: List[Dict[str, object]] = []
+
+    for source in source_list:
+        source_id = str(source.get("id") or "unknown")
+        attempt = attempt_by_source.get(source_id)
+        category = _provider_category(source, attempt)
+        history = _history_attempts(receipt_list, source_id)
+        historical_categories = [
+            _provider_category(source, historical_attempt)
+            for historical_attempt in history
+        ]
+        current_and_history = [category, *historical_categories]
+        consecutive_empty = _consecutive_count(current_and_history, "empty")
+        consecutive_failures = _consecutive_count(current_and_history, "failed")
+
+        retry_minutes = None
+        next_retry_at = None
+        if bool(source.get("enabled")) and category in {"duplicate", "empty", "failed", "not_ready"}:
+            retry_minutes = max(1, retry_interval_minutes)
+            if category in {"failed", "not_ready"}:
+                failure_run = max(consecutive_failures, 1)
+                retry_minutes = min(
+                    retry_minutes * (2 ** min(failure_run - 1, 3)),
+                    max_failure_backoff_minutes,
+                )
+            next_retry_at = (now + timedelta(minutes=retry_minutes)).isoformat()
+
+        action = {
+            "disabled": "activate_when_authorized_configuration_is_ready",
+            "not_probed": "probe_on_next_scheduler_run",
+            "not_ready": "reconcile_provider_then_retry",
+            "failed": "retry_with_bounded_backoff_and_probe_fallback",
+            "duplicate": "retry_on_schedule_and_probe_fallback",
+            "empty": "retry_on_schedule_and_probe_fallback",
+            "candidate_found": "validate_and_import_candidate",
+            "imported": "verify_freshness_and_monitor",
+            "ready": "monitor_on_schedule",
+        }[category]
+        source_states.append({
+            "id": source_id,
+            "enabled": bool(source.get("enabled")),
+            "ready": bool(source.get("ready")),
+            "configuration_status": source.get("status") or "unknown",
+            "runtime_state": category,
+            "error_kind": attempt.get("error_kind") if attempt else None,
+            "candidate_count": int((attempt or {}).get("candidate_count") or 0),
+            "imported_count": int((attempt or {}).get("imported_count") or 0),
+            "duplicate_count": int((attempt or {}).get("duplicate_count") or 0),
+            "last_attempt_at": (
+                now.isoformat()
+                if attempt and category != "disabled"
+                else history[0].get("attempted_at")
+                if history
+                else None
+            ),
+            "last_success_at": (
+                now.isoformat()
+                if category == "imported"
+                else next(
+                    (
+                        historical_attempt.get("attempted_at")
+                        for historical_attempt in history
+                        if _provider_category(source, historical_attempt) == "imported"
+                    ),
+                    None,
+                )
+            ),
+            "consecutive_empty_runs": consecutive_empty,
+            "consecutive_failure_runs": consecutive_failures,
+            "retry_minutes": retry_minutes,
+            "next_retry_at": next_retry_at,
+            "action": action,
+        })
+
+    enabled_states = [item for item in source_states if item["enabled"]]
+    ready_states = [item for item in source_states if item["ready"]]
+    due_times = [
+        str(item["next_retry_at"])
+        for item in source_states
+        if item.get("next_retry_at")
+    ]
+    decision_ready = bool(freshness.get("decision_ready"))
+    any_candidate = any(
+        item["runtime_state"] in {"candidate_found", "imported"}
+        for item in source_states
+    )
+
+    if decision_ready:
+        status = "closed"
+        automation_decision = "monitor_scheduled_sources"
+        safe_next_action = "none"
+    elif any_candidate:
+        status = "partial"
+        automation_decision = "verify_candidate_import_and_freshness"
+        safe_next_action = "post_import_freshness_verification"
+    elif ready_states:
+        status = "degraded"
+        automation_decision = "retry_ready_sources_and_probe_enabled_fallbacks"
+        safe_next_action = "scheduled_retry_and_enabled_fallback_probe"
+    elif enabled_states:
+        status = "blocked"
+        automation_decision = "repair_enabled_sources_with_bounded_retry"
+        safe_next_action = "reconcile_enabled_source_health_then_retry"
+    else:
+        status = "blocked"
+        automation_decision = "reconcile_authorized_source_configuration"
+        safe_next_action = "reconcile_authorized_source_configuration"
+
+    lag_bucket = _lag_bucket(freshness.get("data_lag_days"))
+    severity = (
+        "critical"
+        if lag_bucket in {"critical", "missing"} or not ready_states
+        else "warning"
+        if lag_bucket in {"warning", "stale"}
+        else "info"
+    )
+    fingerprint_payload = {
+        "lag_bucket": lag_bucket,
+        "source_states": [
+            [item["id"], item["runtime_state"], item.get("error_kind")]
+            for item in source_states
+        ],
+    }
+    fingerprint = hashlib.sha256(
+        json.dumps(fingerprint_payload, sort_keys=True, ensure_ascii=True).encode("utf-8")
+    ).hexdigest()[:20]
+    previous_alert = _previous_alert(receipt_list)
+    previous_at = _parse_datetime(previous_alert.get("completed_at"))
+    elapsed_minutes = None
+    if previous_at:
+        comparable_now = now
+        if comparable_now.tzinfo and not previous_at.tzinfo:
+            previous_at = previous_at.replace(tzinfo=comparable_now.tzinfo)
+        elif previous_at.tzinfo and not comparable_now.tzinfo:
+            comparable_now = comparable_now.replace(tzinfo=previous_at.tzinfo)
+        elapsed_minutes = max(0, int((comparable_now - previous_at).total_seconds() // 60))
+    state_changed = previous_alert.get("fingerprint") != fingerprint
+    send_notification = severity in {"warning", "critical"} and (
+        state_changed
+        or elapsed_minutes is None
+        or elapsed_minutes >= max(1, alert_cooldown_minutes)
+    )
+
+    notification_policy = {
+        "event": "pchome_sales_source_freshness",
+        "severity": severity,
+        "fingerprint": fingerprint,
+        "send": send_notification,
+        "reason": (
+            "state_changed"
+            if state_changed
+            else "cooldown_elapsed"
+            if send_notification
+            else "duplicate_within_cooldown"
+        ),
+        "cooldown_minutes": max(1, alert_cooldown_minutes),
+    }
+
+    return {
+        "contract": "pchome_sales_source_reconciliation_v1",
+        "status": status,
+        "automation_decision": automation_decision,
+        "manual_review_required": False,
+        "freshness": dict(freshness),
+        "source_priority": [
+            source_id
+            for source_id in SOURCE_PRIORITY
+            if any(item["id"] == source_id for item in source_states)
+        ],
+        "fallback_chain": [item["id"] for item in ready_states],
+        "single_source_dependency": len(ready_states) == 1,
+        "coverage": {
+            "ready": len(ready_states),
+            "enabled": len(enabled_states),
+            "total": len(source_states),
+        },
+        "sources": source_states,
+        "next_retry_at": min(due_times) if due_times else None,
+        "safe_next_action": safe_next_action,
+        "notification_policy": notification_policy,
+    }

templates/auto_import_index.html

@@ -155,7 +155,19 @@
     element.hidden = false;
   }
 
-  function sourceStatusLabel(source) {
+  function sourceStatusLabel(source, runtimeState) {
+    const runtimeLabels = {
+      imported: '已匯入',
+      duplicate: '無需更新',
+      candidate_found: '處理中',
+      empty: '待新檔',
+      failed: '重試中',
+      not_ready: '修復中',
+      not_probed: '待探測',
+      ready: '可用',
+      disabled: '未啟用'
+    };
+    if (runtimeState && runtimeLabels[runtimeState]) return runtimeLabels[runtimeState];
     if (source.ready) return '可用';
     if (!source.enabled) return '未啟用';
     return source.status === 'incomplete' ? '設定未完整' : source.status === 'path_missing' ? '路徑不存在' : '不可用';
@@ -176,14 +188,23 @@
     const freshness = payload.freshness || {};
     const coverage = payload.asset_coverage || {};
     const receipt = payload.last_receipt || null;
+    const reconciliation = payload.source_reconciliation || {};
+    const runtimeSources = new Map((reconciliation.sources || []).map(source => [source.id, source]));
+    const nextRetry = reconciliation.next_retry_at;
     const lag = freshness.data_lag_days;
     document.getElementById('freshnessValue').textContent = freshness.latest_sales_date || '無資料';
     document.getElementById('freshnessMeta').textContent = lag === null || lag === undefined ? '尚未建立基準' : lag === 0 ? '今天已更新' : `落後 ${lag} 天`;
     document.getElementById('closureValue').textContent = freshness.decision_ready ? '可決策' : '受阻';
     document.getElementById('closureValue').dataset.state = freshness.decision_ready ? 'ready' : 'blocked';
-    document.getElementById('closureMeta').textContent = freshness.decision_ready ? '新鮮度守門通過' : '等待最新業績';
+    document.getElementById('closureMeta').textContent = freshness.decision_ready
+      ? '新鮮度守門通過'
+      : nextRetry
+        ? `已排程 ${formatTime(nextRetry)}`
+        : '等待最新業績';
     document.getElementById('sourceCoverageValue').textContent = `${coverage.ready || 0}/${coverage.total || 0}`;
-    document.getElementById('sourceCoverageMeta').textContent = `${coverage.percent || 0}% 來源可用`;
+    document.getElementById('sourceCoverageMeta').textContent = reconciliation.single_source_dependency
+      ? '僅 1 個來源可用'
+      : `${coverage.percent || 0}% 來源可用`;
     document.getElementById('lastRunValue').textContent = receipt ? formatTime(receipt.completed_at) : '尚無執行';
     document.getElementById('lastRunMeta').textContent = receipt ? receiptStatusLabel(receipt.status) : '尚無執行紀錄';
     document.getElementById('pageRuntimeSummary').textContent = freshness.decision_ready
@@ -191,13 +212,17 @@
       : `正式業績${lag === null || lag === undefined ? '尚未建立' : `已落後 ${lag} 天`}`;
 
     const list = document.getElementById('sourceList');
-    list.innerHTML = (payload.sources || []).map(source => `
+    list.innerHTML = (payload.sources || []).map(source => {
+      const runtime = runtimeSources.get(source.id) || {};
+      const state = runtime.runtime_state || (source.ready ? 'ready' : source.enabled ? 'not_ready' : 'disabled');
+      return `
       <div class="ai-source-row">
-        <span class="ai-source-row__indicator ai-source-row__indicator--${source.ready ? 'ready' : source.enabled ? 'degraded' : 'disabled'}" aria-hidden="true"></span>
+        <span class="ai-source-row__indicator ai-source-row__indicator--${state === 'imported' || state === 'ready' ? 'ready' : state === 'disabled' ? 'disabled' : 'degraded'}" aria-hidden="true"></span>
         <span class="ai-source-row__name">${escapeHtml(SOURCE_LABELS[source.id] || source.id)}</span>
-        <span class="ai-source-row__state">${escapeHtml(sourceStatusLabel(source))}</span>
+        <span class="ai-source-row__state">${escapeHtml(sourceStatusLabel(source, state))}</span>
       </div>
-    `).join('');
+    `;
+    }).join('');
   }
 
   async function loadReadiness() {

tests/test_pchome_sales_acquisition_service.py

@@ -177,12 +177,19 @@ def test_no_candidate_with_missing_freshness_is_blocked_and_receipted(monkeypatc
     _disable_remote_sources(monkeypatch)
     monkeypatch.delenv("PCHOME_SALES_LOCAL_DROP_DIR", raising=False)
     monkeypatch.setattr(import_service, "drive_service", EmptyDrive())
+    monkeypatch.setattr(acquisition, "drive_service", ReadyDrive())
 
     result = acquisition.PChomeSalesAcquisitionService().run(trigger="test")
 
     assert result["success"] is False
     assert result["status"] == "blocked_with_safe_next_action"
-    assert result["safe_next_action"] == "scheduler_retry_authorized_sources"
+    assert result["safe_next_action"] == "scheduled_retry_and_enabled_fallback_probe"
+    assert result["next_retry_at"]
+    assert result["source_reconciliation"]["contract"] == "pchome_sales_source_reconciliation_v1"
+    assert result["source_reconciliation"]["manual_review_required"] is False
+    assert result["providers"][1]["status"] == "disabled"
+    assert result["providers"][2]["status"] == "disabled"
+    assert result["providers"][3]["status"] == "disabled"
     assert result["receipt_persisted"] is True
     with import_service.engine.connect() as conn:
         receipt = conn.execute(text(
@@ -303,6 +310,7 @@ def test_auto_import_api_uses_governed_acquisition_service(monkeypatch):
 def test_scheduler_continues_when_legacy_hitl_pause_is_set(monkeypatch):
     import scheduler
     import services.agent_actions as agent_actions
+    import services.notification_manager as notification_manager
     import services.pchome_sales_acquisition_service as acquisition
 
     calls = []
@@ -322,8 +330,19 @@ def test_scheduler_continues_when_legacy_hitl_pause_is_set(monkeypatch):
             "data_lag_days": 1,
             "decision_ready": True,
             "safe_next_action": "scheduler_retry_authorized_sources",
+            "next_retry_at": "2026-07-14T19:00:00+08:00",
+            "notification_policy": {
+                "send": False,
+                "reason": "duplicate_within_cooldown",
+                "fingerprint": "same-state",
+            },
         },
     )
+    monkeypatch.setattr(
+        notification_manager,
+        "NotificationManager",
+        lambda: (_ for _ in ()).throw(AssertionError("suppressed notification must not send")),
+    )
     monkeypatch.setattr(scheduler, "_save_stats", lambda name, stats: saved.update({"name": name, "stats": stats}))
 
     scheduler.run_auto_import_task()
@@ -332,3 +351,5 @@ def test_scheduler_continues_when_legacy_hitl_pause_is_set(monkeypatch):
     assert saved["name"] == "auto_import_task"
     assert saved["stats"]["status"] == "Degraded"
     assert saved["stats"]["receipt_persisted"] is True
+    assert saved["stats"]["next_retry_at"] == "2026-07-14T19:00:00+08:00"
+    assert saved["stats"]["notification_policy"]["send"] is False

tests/test_pchome_sales_source_reconciliation.py

@@ -0,0 +1,126 @@
+from datetime import datetime, timedelta, timezone
+
+from services.pchome_sales_source_reconciliation import build_source_reconciliation
+
+
+NOW = datetime(2026, 7, 14, 18, 30, tzinfo=timezone(timedelta(hours=8)))
+
+
+def _source(source_id, *, enabled=True, ready=True, status="ready"):
+    return {
+        "id": source_id,
+        "enabled": enabled,
+        "ready": ready,
+        "status": status,
+    }
+
+
+def _receipt(provider, *, completed_at, reconciliation=None):
+    stages = {"providers": [provider]}
+    if reconciliation:
+        stages["source_reconciliation"] = reconciliation
+    return {
+        "status": "blocked_with_safe_next_action",
+        "completed_at": completed_at,
+        "stages": stages,
+    }
+
+
+def test_stale_single_source_schedules_retry_and_reports_fallback_gap():
+    payload = build_source_reconciliation(
+        sources=[
+            _source("google_drive"),
+            _source("authorized_https", enabled=False, ready=False, status="disabled"),
+        ],
+        provider_attempts=[
+            {"source": "google_drive", "status": "upstream_stale", "file_count": 0},
+            {"source": "authorized_https", "status": "disabled"},
+        ],
+        prior_receipts=[],
+        freshness={"latest_sales_date": "2026-06-24", "data_lag_days": 20, "decision_ready": False},
+        now=NOW,
+    )
+
+    assert payload["contract"] == "pchome_sales_source_reconciliation_v1"
+    assert payload["status"] == "degraded"
+    assert payload["fallback_chain"] == ["google_drive"]
+    assert payload["single_source_dependency"] is True
+    assert payload["next_retry_at"] == "2026-07-14T19:00:00+08:00"
+    assert payload["safe_next_action"] == "scheduled_retry_and_enabled_fallback_probe"
+    assert payload["notification_policy"]["severity"] == "critical"
+    assert payload["notification_policy"]["send"] is True
+    assert payload["sources"][0]["consecutive_empty_runs"] == 1
+    assert payload["sources"][1]["runtime_state"] == "disabled"
+
+
+def test_same_state_within_cooldown_suppresses_duplicate_notification():
+    first = build_source_reconciliation(
+        sources=[_source("google_drive")],
+        provider_attempts=[{"source": "google_drive", "status": "upstream_stale"}],
+        prior_receipts=[],
+        freshness={"data_lag_days": 20, "decision_ready": False},
+        now=NOW - timedelta(minutes=10),
+    )
+    history = [_receipt(
+        {"source": "google_drive", "status": "upstream_stale"},
+        completed_at=(NOW - timedelta(minutes=10)).isoformat(),
+        reconciliation=first,
+    )]
+
+    second = build_source_reconciliation(
+        sources=[_source("google_drive")],
+        provider_attempts=[{"source": "google_drive", "status": "upstream_stale"}],
+        prior_receipts=history,
+        freshness={"data_lag_days": 20, "decision_ready": False},
+        now=NOW,
+    )
+
+    assert second["notification_policy"]["fingerprint"] == first["notification_policy"]["fingerprint"]
+    assert second["notification_policy"]["send"] is False
+    assert second["notification_policy"]["reason"] == "duplicate_within_cooldown"
+    assert second["sources"][0]["consecutive_empty_runs"] == 2
+
+
+def test_provider_failures_use_bounded_exponential_backoff():
+    failed = {"source": "authorized_https", "status": "failed", "error_kind": "http_fetch_failed"}
+    history = [
+        _receipt(failed, completed_at=(NOW - timedelta(minutes=30)).isoformat()),
+        _receipt(failed, completed_at=(NOW - timedelta(minutes=60)).isoformat()),
+    ]
+
+    payload = build_source_reconciliation(
+        sources=[_source("authorized_https")],
+        provider_attempts=[failed],
+        prior_receipts=history,
+        freshness={"data_lag_days": 5, "decision_ready": False},
+        now=NOW,
+        retry_interval_minutes=30,
+        max_failure_backoff_minutes=240,
+    )
+
+    source = payload["sources"][0]
+    assert source["runtime_state"] == "failed"
+    assert source["consecutive_failure_runs"] == 3
+    assert source["retry_minutes"] == 120
+    assert source["next_retry_at"] == "2026-07-14T20:30:00+08:00"
+    assert source["action"] == "retry_with_bounded_backoff_and_probe_fallback"
+
+
+def test_duplicate_source_remains_on_scheduled_retry_without_writing():
+    payload = build_source_reconciliation(
+        sources=[_source("google_drive")],
+        provider_attempts=[{
+            "source": "google_drive",
+            "status": "duplicate",
+            "duplicate_count": 1,
+        }],
+        prior_receipts=[],
+        freshness={"data_lag_days": 10, "decision_ready": False},
+        now=NOW,
+    )
+
+    source = payload["sources"][0]
+    assert source["runtime_state"] == "duplicate"
+    assert source["retry_minutes"] == 30
+    assert source["next_retry_at"] == "2026-07-14T19:00:00+08:00"
+    assert payload["safe_next_action"] == "scheduled_retry_and_enabled_fallback_probe"