fix(growth): make active source canary idempotent

services/external_marketplace_canary_service.py

@@ -287,14 +287,13 @@ def activate_external_market_source_after_canary(
         and bool(item["same_item_reconciliation"].get("independent_verifier_passed"))
         for item in candidates
     )
-    can_activate = bool(
+    canary_evidence_verified = bool(
         independent_readback.get("success")
-        and len(candidates) >= threshold
         and readback_count == len(candidates)
         and guards_passed
         and independent_verifiers_passed
     )
-    if not can_activate:
+    if not canary_evidence_verified:
         return {
             "success": True,
             "status": "canary_threshold_not_met",
@@ -305,6 +304,72 @@ def activate_external_market_source_after_canary(
             "guards_passed": guards_passed,
             "independent_verifiers_passed": independent_verifiers_passed,
             "applied": False,
+            "state_changed": False,
+            "writes_database_count": 0,
+        }
+
+    with engine.begin() as conn:
+        _ensure_external_market_source_seeds(conn)
+        current = (
+            conn.execute(
+                text(
+                    """
+            SELECT status, enabled, write_enabled
+            FROM external_market_sources
+            WHERE code = :source_code
+        """
+                ),
+                {"source_code": source_code},
+            )
+            .mappings()
+            .first()
+        )
+    if not current:
+        return {
+            "success": False,
+            "status": "source_registry_missing",
+            "source_code": source_code,
+            "applied": False,
+            "state_changed": False,
+            "writes_database_count": 0,
+        }
+
+    already_active = bool(
+        str(current.get("status") or "") == "active"
+        and bool(current.get("enabled"))
+        and bool(current.get("write_enabled"))
+    )
+    if already_active:
+        return {
+            "success": True,
+            "status": "already_active_verified",
+            "source_code": source_code,
+            "candidate_count": len(candidates),
+            "readback_pass_count": readback_count,
+            "minimum_verified_count": threshold,
+            "guards_passed": guards_passed,
+            "independent_verifiers_passed": independent_verifiers_passed,
+            "applied": True,
+            "state_changed": False,
+            "previous_state": dict(current),
+            "observed_state": dict(current),
+            "identity": dict(identity or {}),
+            "writes_database_count": 0,
+        }
+
+    if len(candidates) < threshold:
+        return {
+            "success": True,
+            "status": "canary_threshold_not_met",
+            "source_code": source_code,
+            "candidate_count": len(candidates),
+            "readback_pass_count": readback_count,
+            "minimum_verified_count": threshold,
+            "guards_passed": guards_passed,
+            "independent_verifiers_passed": independent_verifiers_passed,
+            "applied": False,
+            "state_changed": False,
+            "observed_state": dict(current),
             "writes_database_count": 0,
         }
 
@@ -330,6 +395,7 @@ def activate_external_market_source_after_canary(
                 "status": "source_registry_missing",
                 "source_code": source_code,
                 "applied": False,
+                "state_changed": False,
                 "writes_database_count": 0,
             }
         conn.execute(
@@ -382,6 +448,7 @@ def activate_external_market_source_after_canary(
         "observed_state": dict(observed or {}),
         "identity": dict(identity or {}),
         "writes_database_count": 1 if applied else 0,
+        "state_changed": applied,
     }
 
 
@@ -452,7 +519,13 @@ def rollback_external_market_source_canary(
                 {"offer_id": offer_id},
             )
 
-        if activation_receipt.get("applied") and previous:
+        state_changed = bool(
+            activation_receipt.get(
+                "state_changed",
+                activation_receipt.get("applied"),
+            )
+        )
+        if state_changed and previous:
             conn.execute(
                 text(
                     """
@@ -478,10 +551,10 @@ def rollback_external_market_source_canary(
         "status": "rolled_back_and_quarantined",
         "source_code": source_code,
         "quarantined_offer_count": len(quarantined_ids),
-        "source_state_restored": bool(activation_receipt.get("applied") and previous),
+        "source_state_restored": bool(state_changed and previous),
         "identity": dict(identity or {}),
         "writes_database_count": len(quarantined_ids)
-        + (1 if activation_receipt.get("applied") and previous else 0),
+        + (1 if state_changed and previous else 0),
     }