新增市場情報 queue writer closeout
All checks were successful
CD Pipeline / deploy (push) Successful in 1m10s

This commit is contained in:
OoO
2026-05-19 12:21:58 +08:00
parent 2d4a3a4bf9
commit 04844099d9
14 changed files with 937 additions and 156 deletions

View File

@@ -0,0 +1,290 @@
"""候選審核 queue writer operator run closeout preview。
本模組只在 receipt review 之後整理 closeout gate不回吐原始 receipt、
不讀 approval token、不執行 CLI、不開 DB connection、不寫 queue、不掛 scheduler。
"""
FORBIDDEN_TOKEN_KEYWORDS = (
"approval_token",
"approval-token",
"market_intel_queue_write_approval",
)
SAFE_TOKEN_METADATA_KEYS = {
"approval_token_present",
"approval_token_valid",
"approval_token_secret_configured",
}
SAFE_APPROVAL_ENV_VAR = "MARKET_INTEL_QUEUE_WRITE_APPROVAL"
REQUIRED_RECEIPT_BOUNDARIES = {
"do_not_execute_cli_from_receipt_preview",
"do_not_read_approval_token_from_api",
"do_not_echo_operator_receipt_payload",
"do_not_open_database_connection_from_receipt_preview",
"do_not_write_queue_from_receipt_preview",
"do_not_attach_scheduler_from_receipt_preview",
}
def _as_dict(value):
return value if isinstance(value, dict) else {}
def _as_list(value):
if value is None:
return []
if isinstance(value, (list, tuple, set)):
return list(value)
return [value]
def _has_text(value):
return bool(isinstance(value, str) and value.strip())
def _contains_forbidden_token_key(value):
if isinstance(value, dict):
for key, nested in value.items():
normalized_key = str(key).lower()
if normalized_key in SAFE_TOKEN_METADATA_KEYS and isinstance(nested, bool):
continue
if normalized_key == "approval_env_var" and nested == SAFE_APPROVAL_ENV_VAR:
continue
if any(token_key in normalized_key for token_key in FORBIDDEN_TOKEN_KEYWORDS):
return True
if _contains_forbidden_token_key(nested):
return True
elif isinstance(value, list):
return any(_contains_forbidden_token_key(item) for item in value)
return False
def _receipt_summary(run_receipt):
run_receipt = _as_dict(run_receipt)
writer = _as_dict(run_receipt.get("writer_output_summary"))
smoke = _as_dict(run_receipt.get("postwrite_smoke_summary"))
expected_keys = _as_list(run_receipt.get("expected_dedupe_keys"))
blocked_reasons = _as_list(run_receipt.get("blocked_reasons"))
safe_boundaries = set(str(item) for item in _as_list(run_receipt.get("safe_boundaries")))
missing_boundaries = sorted(REQUIRED_RECEIPT_BOUNDARIES - safe_boundaries)
return {
"provided": bool(run_receipt),
"mode": run_receipt.get("mode"),
"receipt_passed": bool(run_receipt.get("receipt_passed")),
"ready_for_next_manual_review": bool(
run_receipt.get("ready_for_next_manual_review")
),
"expected_dedupe_key_count": len(expected_keys),
"blocked_count": len(blocked_reasons),
"writer_mode": writer.get("mode"),
"writer_dedupe_keys_match_expected": bool(
writer.get("dedupe_keys_match_expected")
),
"postwrite_smoke_passed": bool(smoke.get("postwrite_smoke_passed")),
"postwrite_smoke_dedupe_keys_match_expected": bool(
smoke.get("dedupe_keys_match_expected")
),
"safe_boundaries_complete": not missing_boundaries,
"missing_safe_boundaries": missing_boundaries,
"ready_for_api_database_write": bool(
run_receipt.get("ready_for_api_database_write")
),
"ready_for_scheduler_attach": bool(run_receipt.get("ready_for_scheduler_attach")),
"api_executes_cli": bool(run_receipt.get("api_executes_cli")),
"api_reads_approval_token": bool(run_receipt.get("api_reads_approval_token")),
"api_writes_file": bool(run_receipt.get("api_writes_file")),
"api_writes_database": bool(run_receipt.get("api_writes_database")),
"database_connection_opened": bool(
run_receipt.get("database_connection_opened")
),
"database_write_executed": bool(run_receipt.get("database_write_executed")),
"database_commit_executed": bool(run_receipt.get("database_commit_executed")),
"scheduler_attached": bool(run_receipt.get("scheduler_attached")),
}
def _operator_closeout_summary(operator_evidence):
operator_evidence = _as_dict(operator_evidence)
return {
"provided_keys": sorted(operator_evidence.keys()),
"closeout_artifact_path_recorded": _has_text(
operator_evidence.get("closeout_artifact_path")
),
"operator_confirmed_queue_review_next": bool(
operator_evidence.get("operator_confirmed_queue_review_next")
),
"operator_confirmed_no_scheduler_attach": bool(
operator_evidence.get("operator_confirmed_no_scheduler_attach")
),
"operator_confirmed_no_api_db_write": bool(
operator_evidence.get("operator_confirmed_no_api_db_write")
),
"closeout_notes_recorded": _has_text(operator_evidence.get("closeout_notes")),
"approval_token_submitted_to_api": _contains_forbidden_token_key(
operator_evidence
),
}
def _closeout_gates(receipt_summary, closeout_summary):
return [
{
"key": "receipt_preview_provided",
"label": "必須提供上一階段 receipt preview 摘要",
"passed": bool(
receipt_summary["provided"]
and receipt_summary["mode"]
== "candidate_queue_writer_run_receipt_preview"
),
},
{
"key": "receipt_passed",
"label": "receipt 必須已通過 writer output 與 post-write smoke 審核",
"passed": receipt_summary["receipt_passed"],
},
{
"key": "receipt_ready_for_next_manual_review",
"label": "receipt 只能放行到下一個人工審核階段",
"passed": receipt_summary["ready_for_next_manual_review"],
},
{
"key": "receipt_expected_dedupe_keys_present",
"label": "receipt 必須保有 transaction dedupe key 摘要",
"passed": bool(receipt_summary["expected_dedupe_key_count"]),
},
{
"key": "writer_and_smoke_dedupe_match",
"label": "writer output 與 smoke 查詢都必須符合預期 dedupe key",
"passed": bool(
receipt_summary["writer_dedupe_keys_match_expected"]
and receipt_summary["postwrite_smoke_dedupe_keys_match_expected"]
and receipt_summary["postwrite_smoke_passed"]
),
},
{
"key": "receipt_safe_boundaries_complete",
"label": "receipt 必須保留不讀 token、不執行 CLI、不寫 queue 的安全邊界",
"passed": receipt_summary["safe_boundaries_complete"],
},
{
"key": "receipt_no_api_database_write",
"label": "closeout 前不得出現 API/UI DB 寫入或 commit",
"passed": bool(
not receipt_summary["ready_for_api_database_write"]
and not receipt_summary["api_writes_database"]
and not receipt_summary["database_write_executed"]
and not receipt_summary["database_commit_executed"]
),
},
{
"key": "receipt_no_scheduler_or_cli_attach",
"label": "closeout 前不得從 API/UI 執行 CLI 或掛 scheduler",
"passed": bool(
not receipt_summary["ready_for_scheduler_attach"]
and not receipt_summary["api_executes_cli"]
and not receipt_summary["scheduler_attached"]
),
},
{
"key": "receipt_no_token_or_file_side_effect",
"label": "closeout 前 API/UI 不得讀 token 或寫 artifact 檔案",
"passed": bool(
not receipt_summary["api_reads_approval_token"]
and not receipt_summary["api_writes_file"]
),
},
{
"key": "closeout_artifact_path_recorded",
"label": "操作員需記錄 closeout artifact 路徑",
"passed": closeout_summary["closeout_artifact_path_recorded"],
},
{
"key": "operator_confirmed_queue_review_next",
"label": "操作員確認下一步只進人工 queue review / read-only inventory",
"passed": closeout_summary["operator_confirmed_queue_review_next"],
},
{
"key": "operator_confirmed_no_scheduler_attach",
"label": "操作員確認本階段不掛 scheduler",
"passed": closeout_summary["operator_confirmed_no_scheduler_attach"],
},
{
"key": "operator_confirmed_no_api_db_write",
"label": "操作員確認本階段不由 API/UI 寫 DB",
"passed": closeout_summary["operator_confirmed_no_api_db_write"],
},
{
"key": "closeout_no_approval_token_submitted_to_api",
"label": "closeout payload 不得包含一次性 approval token key",
"passed": not closeout_summary["approval_token_submitted_to_api"],
},
]
def build_candidate_queue_writer_run_closeout(
*,
transaction_preview,
run_receipt,
operator_evidence=None,
):
"""建立 CLI 寫入後 closeout gate不執行任何副作用。"""
receipt_summary = _receipt_summary(run_receipt)
closeout_summary = _operator_closeout_summary(operator_evidence)
gates = _closeout_gates(receipt_summary, closeout_summary)
blocked_reasons = [gate["key"] for gate in gates if not gate["passed"]]
closeout_passed = bool(not blocked_reasons)
statement_count = len(_as_list(_as_dict(transaction_preview).get("statements")))
return {
"mode": "candidate_queue_writer_run_closeout_preview",
"target_table": "market_alert_review_queue",
"closeout_reviewed": True,
"closeout_passed": closeout_passed,
"ready_for_next_manual_phase": closeout_passed,
"ready_for_api_database_write": False,
"ready_for_scheduler_attach": False,
"api_executes_cli": False,
"api_reads_approval_token": False,
"api_writes_file": False,
"api_writes_database": 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,
"writes_executed": False,
"would_write_database": False,
"statement_count": statement_count,
"blocked_reasons": blocked_reasons,
"gates": gates,
"receipt_summary": receipt_summary,
"operator_closeout_summary": closeout_summary,
"promotion_gate": {
"allowed": closeout_passed,
"next_manual_phase": "manual_queue_review_and_live_inventory_read_only",
"requires_real_db_write": False,
"requires_scheduler_attach": False,
"requires_operator_approval": True,
},
"next_operator_steps": [
"保存 closeout artifact 與 receipt 摘要",
"以人工方式執行 live DB inventory read-only smoke",
"在審核頁只檢查 needs_review queue row不從 UI/API 補寫資料",
"若任何 closeout gate 阻擋,停在 receipt 階段並保留 feature flags 關閉",
],
"safe_boundaries": [
"do_not_echo_operator_receipt_payload",
"do_not_execute_cli_from_closeout_preview",
"do_not_read_approval_token_from_api",
"do_not_write_closeout_artifact_from_api",
"do_not_open_database_connection_from_closeout_preview",
"do_not_write_queue_from_closeout_preview",
"do_not_attach_scheduler_from_closeout_preview",
"no_remove_orphans",
"no_momo_db_lifecycle_change",
],
}

View File

@@ -10,6 +10,24 @@ from services.market_intel.candidate_queue_writer_operator_drill import build_ca
from services.market_intel.candidate_queue_writer_run_package import build_candidate_queue_writer_run_package
from services.market_intel.candidate_queue_writer_run_readiness import build_candidate_queue_writer_run_readiness
from services.market_intel.candidate_queue_writer_run_receipt import build_candidate_queue_writer_run_receipt
from services.market_intel.candidate_queue_writer_run_closeout import build_candidate_queue_writer_run_closeout
def _run_review_preview_safe(payload, mode):
blocked_keys = (
"ready_for_api_database_write",
"ready_for_scheduler_attach",
"api_executes_cli",
"api_reads_approval_token",
"api_writes_file",
"database_connection_opened",
"database_write_executed",
"database_commit_executed",
"scheduler_attached",
)
return bool(
payload["mode"] == mode and all(not payload.get(key) for key in blocked_keys)
)
def build_deployment_readiness_preview(
@@ -85,6 +103,10 @@ def build_deployment_readiness_preview(
transaction_preview=manual_sample_candidate_queue_transaction,
run_readiness=candidate_queue_writer_run_readiness,
)
candidate_queue_writer_run_closeout = build_candidate_queue_writer_run_closeout(
transaction_preview=manual_sample_candidate_queue_transaction,
run_receipt=candidate_queue_writer_run_receipt,
)
match_review_plan = service.build_match_review_plan()
opportunity_plan = service.build_opportunity_plan()
opportunity_scoring_plan = service.build_opportunity_scoring_plan()
@@ -304,18 +326,13 @@ def build_deployment_readiness_preview(
and not candidate_queue_writer_run_readiness["database_commit_executed"]
and not candidate_queue_writer_run_readiness["scheduler_attached"]
),
"candidate_queue_writer_run_receipt_preview_safe": bool(
candidate_queue_writer_run_receipt["mode"]
== "candidate_queue_writer_run_receipt_preview"
and not candidate_queue_writer_run_receipt["ready_for_api_database_write"]
and not candidate_queue_writer_run_receipt["ready_for_scheduler_attach"]
and not candidate_queue_writer_run_receipt["api_executes_cli"]
and not candidate_queue_writer_run_receipt["api_reads_approval_token"]
and not candidate_queue_writer_run_receipt["api_writes_file"]
and not candidate_queue_writer_run_receipt["database_connection_opened"]
and not candidate_queue_writer_run_receipt["database_write_executed"]
and not candidate_queue_writer_run_receipt["database_commit_executed"]
and not candidate_queue_writer_run_receipt["scheduler_attached"]
"candidate_queue_writer_run_receipt_preview_safe": _run_review_preview_safe(
candidate_queue_writer_run_receipt,
"candidate_queue_writer_run_receipt_preview",
),
"candidate_queue_writer_run_closeout_preview_safe": _run_review_preview_safe(
candidate_queue_writer_run_closeout,
"candidate_queue_writer_run_closeout_preview",
),
"match_review_plan_preview_safe": bool(
match_review_plan["mode"] == "match_review_plan_preview"
@@ -536,6 +553,7 @@ def build_deployment_readiness_preview(
"/api/market_intel/manual_sample_review/candidate_queue_writer_run_package",
"/api/market_intel/manual_sample_review/candidate_queue_writer_run_readiness",
"/api/market_intel/manual_sample_review/candidate_queue_writer_run_receipt",
"/api/market_intel/manual_sample_review/candidate_queue_writer_run_closeout",
],
"status": status.to_dict(),
"schema_smoke": schema_smoke,
@@ -583,6 +601,7 @@ def build_deployment_readiness_preview(
"candidate_queue_writer_run_package": candidate_queue_writer_run_package,
"candidate_queue_writer_run_readiness": candidate_queue_writer_run_readiness,
"candidate_queue_writer_run_receipt": candidate_queue_writer_run_receipt,
"candidate_queue_writer_run_closeout": candidate_queue_writer_run_closeout,
"match_review_plan": match_review_plan,
"opportunity_plan": opportunity_plan,
"opportunity_scoring_plan": opportunity_scoring_plan,

View File

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