新增市場情報佇列寫入後 smoke

services/market_intel/candidate_queue_writer_postwrite_smoke.py

@@ -0,0 +1,203 @@
+"""候選審核 queue writer post-write 只讀 smoke。
+
+本模組只依 transaction preview 的 dedupe key 查詢
+market_alert_review_queue，確認 CLI 寫入後的 row 是否存在；
+不建立 ORM session、不寫 DB、不 commit、不掛 scheduler。
+"""
+
+from sqlalchemy import bindparam, create_engine, text
+
+from services.market_intel.candidate_queue_writer_preflight import QUEUE_TABLE
+
+
+def _dedupe_keys_from_transaction(transaction_preview):
+    keys = []
+    for statement in transaction_preview.get("statements", []):
+        lookup = statement.get("lookup") or {}
+        key = lookup.get("dedupe_key")
+        if key:
+            keys.append(str(key))
+    return sorted(set(keys))
+
+
+def _safe_row(row):
+    payload = dict(row._mapping)
+    for key, value in list(payload.items()):
+        if value is not None and not isinstance(value, (bool, int, float, str)):
+            payload[key] = str(value)
+    return payload
+
+
+def _safety_contract():
+    return {
+        "read_only_query_only": True,
+        "does_not_open_transaction": True,
+        "does_not_commit": True,
+        "does_not_insert_queue_rows": True,
+        "does_not_attach_scheduler": True,
+        "target_table": QUEUE_TABLE,
+    }
+
+
+def _planned_result(dedupe_keys):
+    return {
+        "mode": "candidate_queue_writer_postwrite_smoke_planned",
+        "target_table": QUEUE_TABLE,
+        "execute_requested": False,
+        "read_only_query_executed": False,
+        "database_connection_opened": False,
+        "database_session_created": False,
+        "explicit_transaction_opened": False,
+        "database_write_executed": False,
+        "database_commit_executed": False,
+        "database_rollback_executed": False,
+        "external_network_executed": False,
+        "scheduler_attached": False,
+        "expected_dedupe_key_count": len(dedupe_keys),
+        "found_count": 0,
+        "missing_count": len(dedupe_keys),
+        "dedupe_keys": dedupe_keys,
+        "found_dedupe_keys": [],
+        "missing_dedupe_keys": dedupe_keys,
+        "row_summaries": [],
+        "postwrite_smoke_passed": False,
+        "ready_for_operator_review": False,
+        "blocked_reasons": [
+            "execute_false_planned_only",
+            "candidate_queue_writer_postwrite_smoke_not_loaded",
+        ],
+        "safety_contract": _safety_contract(),
+    }
+
+
+def build_candidate_queue_writer_postwrite_smoke(
+    *,
+    transaction_preview,
+    execute_requested=False,
+    database_url=None,
+    database_type=None,
+    engine=None,
+):
+    """建立 post-write smoke；execute=true 仍只讀查詢 queue rows。"""
+    dedupe_keys = _dedupe_keys_from_transaction(transaction_preview)
+    if not execute_requested:
+        return _planned_result(dedupe_keys)
+
+    from config import DATABASE_PATH, DATABASE_TYPE
+
+    effective_database_type = (database_type or DATABASE_TYPE or "").lower()
+    effective_database_url = database_url or DATABASE_PATH
+    created_engine = False
+    connection_opened = False
+
+    try:
+        if engine is None:
+            connect_args = {}
+            if effective_database_type == "postgresql":
+                connect_args = {
+                    "connect_timeout": 8,
+                    "options": "-c statement_timeout=15000",
+                }
+            engine = create_engine(
+                effective_database_url,
+                isolation_level="AUTOCOMMIT",
+                pool_pre_ping=True,
+                connect_args=connect_args,
+            )
+            created_engine = True
+
+        rows = []
+        if dedupe_keys:
+            query = text(
+                """
+                SELECT
+                    alert_candidate_id,
+                    review_state,
+                    priority_lane,
+                    threshold_level,
+                    total_score,
+                    evidence_bundle_id,
+                    dedupe_key,
+                    source_batch_id,
+                    created_at,
+                    updated_at
+                FROM market_alert_review_queue
+                WHERE dedupe_key IN :dedupe_keys
+                ORDER BY dedupe_key
+                """
+            ).bindparams(bindparam("dedupe_keys", expanding=True))
+            with engine.connect() as conn:
+                connection_opened = True
+                rows = conn.execute(
+                    query,
+                    {"dedupe_keys": tuple(dedupe_keys)},
+                ).fetchall()
+        else:
+            with engine.connect():
+                connection_opened = True
+
+        row_summaries = [_safe_row(row) for row in rows]
+        found_keys = sorted(row["dedupe_key"] for row in row_summaries)
+        missing_keys = [key for key in dedupe_keys if key not in set(found_keys)]
+        passed = bool(dedupe_keys and not missing_keys)
+        blocked_reasons = []
+        if not dedupe_keys:
+            blocked_reasons.append("transaction_statement_dedupe_keys_missing")
+        if missing_keys:
+            blocked_reasons.append("queue_rows_missing_after_write")
+
+        return {
+            "mode": "candidate_queue_writer_postwrite_smoke_read_only",
+            "target_table": QUEUE_TABLE,
+            "execute_requested": True,
+            "read_only_query_executed": True,
+            "database_connection_opened": connection_opened,
+            "database_session_created": False,
+            "explicit_transaction_opened": False,
+            "database_write_executed": False,
+            "database_commit_executed": False,
+            "database_rollback_executed": False,
+            "external_network_executed": False,
+            "scheduler_attached": False,
+            "expected_dedupe_key_count": len(dedupe_keys),
+            "found_count": len(found_keys),
+            "missing_count": len(missing_keys),
+            "dedupe_keys": dedupe_keys,
+            "found_dedupe_keys": found_keys,
+            "missing_dedupe_keys": missing_keys,
+            "row_summaries": row_summaries,
+            "postwrite_smoke_passed": passed,
+            "ready_for_operator_review": passed,
+            "blocked_reasons": blocked_reasons,
+            "safety_contract": _safety_contract(),
+        }
+    except Exception as exc:
+        return {
+            "mode": "candidate_queue_writer_postwrite_smoke_error",
+            "target_table": QUEUE_TABLE,
+            "execute_requested": True,
+            "read_only_query_executed": False,
+            "database_connection_opened": connection_opened,
+            "database_session_created": False,
+            "explicit_transaction_opened": False,
+            "database_write_executed": False,
+            "database_commit_executed": False,
+            "database_rollback_executed": False,
+            "external_network_executed": False,
+            "scheduler_attached": False,
+            "expected_dedupe_key_count": len(dedupe_keys),
+            "found_count": 0,
+            "missing_count": len(dedupe_keys),
+            "dedupe_keys": dedupe_keys,
+            "found_dedupe_keys": [],
+            "missing_dedupe_keys": dedupe_keys,
+            "row_summaries": [],
+            "postwrite_smoke_passed": False,
+            "ready_for_operator_review": False,
+            "blocked_reasons": ["candidate_queue_writer_postwrite_smoke_error"],
+            "error_message": str(exc),
+            "safety_contract": _safety_contract(),
+        }
+    finally:
+        if created_engine:
+            engine.dispose()

services/market_intel/deployment_readiness.py

@@ -9,6 +9,9 @@ from services.market_intel.candidate_queue_writer_cli import (
 from services.market_intel.candidate_queue_writer_preflight import (
     build_candidate_queue_writer_preflight,
 )
+from services.market_intel.candidate_queue_writer_postwrite_smoke import (
+    build_candidate_queue_writer_postwrite_smoke,
+)
 
 
 def build_deployment_readiness_preview(
@@ -51,6 +54,12 @@ def build_deployment_readiness_preview(
         transaction_preview=manual_sample_candidate_queue_transaction,
         writer_preflight=candidate_queue_writer_preflight,
     )
+    candidate_queue_writer_postwrite_smoke = (
+        build_candidate_queue_writer_postwrite_smoke(
+            transaction_preview=manual_sample_candidate_queue_transaction,
+            execute_requested=False,
+        )
+    )
     match_review_plan = service.build_match_review_plan()
     opportunity_plan = service.build_opportunity_plan()
     opportunity_scoring_plan = service.build_opportunity_scoring_plan()
@@ -227,6 +236,15 @@ def build_deployment_readiness_preview(
             and not candidate_queue_writer_preflight["database_commit_executed"]
             and not candidate_queue_writer_preflight["scheduler_attached"]
         ),
+        "candidate_queue_writer_postwrite_smoke_planned_safe": bool(
+            candidate_queue_writer_postwrite_smoke["mode"]
+            == "candidate_queue_writer_postwrite_smoke_planned"
+            and not candidate_queue_writer_postwrite_smoke["read_only_query_executed"]
+            and not candidate_queue_writer_postwrite_smoke["database_connection_opened"]
+            and not candidate_queue_writer_postwrite_smoke["database_write_executed"]
+            and not candidate_queue_writer_postwrite_smoke["database_commit_executed"]
+            and not candidate_queue_writer_postwrite_smoke["scheduler_attached"]
+        ),
         "match_review_plan_preview_safe": bool(
             match_review_plan["mode"] == "match_review_plan_preview"
             and not match_review_plan["review_queue_created"]
@@ -441,6 +459,7 @@ def build_deployment_readiness_preview(
             "/api/market_intel/migration_catalog_review",
             "/api/market_intel/migration_live_smoke",
             "/api/market_intel/live_db_inventory",
+            "/api/market_intel/manual_sample_review/candidate_queue_writer_postwrite_smoke",
         ],
         "status": status.to_dict(),
         "schema_smoke": schema_smoke,
@@ -479,6 +498,9 @@ def build_deployment_readiness_preview(
         ),
         "candidate_queue_writer_preflight": candidate_queue_writer_preflight,
         "candidate_queue_writer_cli_status": candidate_queue_writer_cli_status,
+        "candidate_queue_writer_postwrite_smoke": (
+            candidate_queue_writer_postwrite_smoke
+        ),
         "match_review_plan": match_review_plan,
         "opportunity_plan": opportunity_plan,
         "opportunity_scoring_plan": opportunity_scoring_plan,

services/market_intel/service.py

@@ -108,7 +108,7 @@ class MarketIntelRuntimeStatus:
 class MarketIntelService:
     """市場情報入口服務，先集中 feature gate 與安全狀態。"""
 
-    phase = "phase_57_candidate_queue_writer_cli_transaction"
+    phase = "phase_58_candidate_queue_writer_postwrite_smoke"
 
     def get_runtime_status(self) -> MarketIntelRuntimeStatus:
         return MarketIntelRuntimeStatus(