fix(growth): preserve durable source activation readback

config.py

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

services/yahoo_shopping_public_offer_service.py

@@ -93,6 +93,36 @@ def _receipt_summary(data: Mapping[str, Any]) -> dict[str, Any]:
     }
 
 
+def _formal_source_state_transition(receipt: Mapping[str, Any]) -> bool | None:
+    activation = receipt.get("source_activation")
+    activation = activation if isinstance(activation, Mapping) else {}
+    rollback = receipt.get("source_rollback")
+    rollback = rollback if isinstance(rollback, Mapping) else {}
+
+    if rollback.get("status") == "rolled_back_and_quarantined" and rollback.get(
+        "source_state_restored"
+    ):
+        previous = activation.get("previous_state")
+        previous = previous if isinstance(previous, Mapping) else {}
+        return bool(
+            str(previous.get("status") or "") == "active"
+            and previous.get("enabled")
+            and previous.get("write_enabled")
+        )
+
+    if activation.get("applied") and activation.get("status") in {
+        "activated",
+        "already_active_verified",
+    }:
+        return True
+
+    summary = receipt.get("summary")
+    summary = summary if isinstance(summary, Mapping) else {}
+    if summary.get("formal_source_activated") is True:
+        return True
+    return None
+
+
 def _write_compact_receipt(
     result: Mapping[str, Any],
     *,
@@ -359,6 +389,17 @@ def build_yahoo_shopping_runtime_readback(
     )
     summary = receipt.get("summary")
     summary = summary if isinstance(summary, dict) else {}
+    formal_source_activated = None
+    for path in paths:
+        try:
+            historical = json.loads(path.read_text(encoding="utf-8"))
+        except (OSError, json.JSONDecodeError):
+            continue
+        formal_source_activated = _formal_source_state_transition(historical)
+        if formal_source_activated is not None:
+            break
+    if formal_source_activated is None:
+        formal_source_activated = bool(summary.get("formal_source_activated"))
     candidate_count = int(summary.get("source_candidate_count") or 0)
     has_candidate = bool(candidate_count and not stale)
     return {
@@ -372,7 +413,7 @@ def build_yahoo_shopping_runtime_readback(
         "candidate_offer_count": candidate_count if has_candidate else 0,
         "verified_candidate_count": int(summary.get("verified_candidate_count") or 0),
         "written_count": int(summary.get("written_count") or 0),
-        "formal_source_activated": bool(summary.get("formal_source_activated")),
+        "formal_source_activated": formal_source_activated,
         "last_seen_at": generated_at,
         "age_hours": round(age_hours, 3) if age_hours is not None else None,
         "stale": stale,

tests/test_yahoo_shopping_public_offer_service.py

@@ -388,6 +388,55 @@ def test_yahoo_runtime_readback_uses_fresh_compact_receipt(tmp_path):
     assert readback["writes_database_count"] == 0
 
 
+def test_yahoo_runtime_readback_preserves_activation_across_no_write_run(tmp_path):
+    from services.yahoo_shopping_public_offer_service import (
+        _write_compact_receipt,
+        build_yahoo_shopping_runtime_readback,
+    )
+
+    activated = {
+        "success": True,
+        "data": {
+            "identity": {"run_id": "activated-run"},
+            "terminal_status": "verified_with_coverage_gain",
+            "next_machine_action": "continue_next_revenue_weighted_batch",
+            "source_activation": {
+                "status": "activated",
+                "applied": True,
+                "writes_database_count": 1,
+            },
+            "source_rollback": {"status": "not_required"},
+            "post_verifier": {"readback_pass_count": 2},
+            "receipt_persistence": {"success": True},
+            "reconciliation_receipt": {},
+        },
+    }
+    no_write = {
+        "success": True,
+        "data": {
+            "identity": {"run_id": "no-write-run"},
+            "terminal_status": "degraded_no_safe_candidate",
+            "next_machine_action": "retry_unresolved_batch_with_fresh_source_evidence",
+            "source_activation": {
+                "status": "not_requested",
+                "applied": False,
+                "writes_database_count": 0,
+            },
+            "source_rollback": {"status": "not_required"},
+            "post_verifier": {"readback_pass_count": 0},
+            "receipt_persistence": {"success": True},
+            "reconciliation_receipt": {},
+        },
+    }
+
+    _write_compact_receipt(activated, receipt_root=tmp_path)
+    _write_compact_receipt(no_write, receipt_root=tmp_path)
+    readback = build_yahoo_shopping_runtime_readback(receipt_root=tmp_path)
+
+    assert readback["terminal_status"] == "degraded_no_safe_candidate"
+    assert readback["formal_source_activated"] is True
+
+
 def test_yahoo_runtime_forces_check_mode_before_controlled_apply(monkeypatch):
     from services import external_marketplace_canary_service as canary_service
     from services import pchome_growth_momo_backfill_service as backfill_service