feat(ai): verify RAG and Nemotron runtime canaries

services/ai_agent_product_integration_service.py

@@ -18,6 +18,9 @@ from services.external_mcp_rag_integration_service import (
 from services.internal_rag_candidate_canary_service import (
     run_internal_rag_candidate_canary,
 )
+from services.nemotron_decision_canary_service import (
+    run_nemotron_decision_canary,
+)
 
 
 POLICY = "runtime_truth_ai_agent_product_integration_v1"
@@ -442,6 +445,10 @@ def build_ai_agent_product_integration_readback(
     rag_runtime = dict((external.get("runtime") or {}).get("rag") or {})
     rag_canary = run_internal_rag_candidate_canary(execute=False)
     latest_canary = dict(rag_canary.get("latest_execution") or {})
+    nemotron_canary = run_nemotron_decision_canary(execute=False)
+    latest_nemotron_canary = dict(
+        nemotron_canary.get("latest_execution") or {}
+    )
     totals = telemetry["totals"]
     active_agents = sum(
         1 for item in agents if item["runtime"]["active_in_window"]
@@ -487,6 +494,11 @@ def build_ai_agent_product_integration_readback(
         blockers.append("rag_runtime_telemetry_empty")
     if latest_canary.get("canary_passed") is not True:
         blockers.append("internal_rag_candidate_canary_not_proven")
+    if not (
+        latest_nemotron_canary.get("canary_passed") is True
+        and latest_nemotron_canary.get("fresh") is True
+    ):
+        blockers.append("nemotron_decision_only_canary_not_proven")
     if stage_passed != len(stages):
         blockers.append("agent_controlled_apply_loop_not_closed")
     if telemetry["read_errors"]:
@@ -502,7 +514,8 @@ def build_ai_agent_product_integration_readback(
         "answer_to_owner": (
             f"四個 AI Agent source/排程 wiring={source_agents}/4；正式環境尚未完整整合："
             f"最近 {window} 小時只有 {active_agents}/4 個 Agent 有實際呼叫，"
-            f"完整閉環 {stage_passed}/{len(stages)} 階段，MCP/RAG runtime 與 canary 必須以實證補齊。"
+            f"完整閉環 {stage_passed}/{len(stages)} 階段，MCP/RAG runtime 與 "
+            "NemoTron decision-only canary 必須以實證補齊。"
             if not full_integration
             else "四個 AI Agent 已有 source、runtime、MCP/RAG 與受控執行閉環實證。"
         ),
@@ -535,6 +548,9 @@ def build_ai_agent_product_integration_readback(
                 "telemetry_hits": totals["rag_query_hits"],
                 "latest_candidate_canary": latest_canary,
             },
+            "agents": {
+                "nemotron_decision_only_canary": latest_nemotron_canary,
+            },
             "pixelrag": (external.get("runtime") or {}).get("pixelrag") or {},
         },
         "telemetry": telemetry,
@@ -547,7 +563,7 @@ def build_ai_agent_product_integration_readback(
             "secret_read": False,
         },
         "next_machine_action": (
-            "execute_internal_rag_candidate_canary_then_activate_shadow_runtime"
+            "execute_nemotron_and_internal_rag_canaries_then_activate_shadow_runtime"
             if not full_integration
             else "continue_scheduled_agent_product_integration_verification"
         ),

services/internal_rag_candidate_canary_service.py

@@ -21,6 +21,7 @@ from services.pixelrag_marketplace_candidate_knowledge_replay_service import (
 )
 from services.rag_service import (
     RAG_EMBED_DIM,
+    RAG_EMBED_EXPECTED_DIGEST,
     RAG_EMBED_MODEL,
     get_embedding_signature,
     is_rag_enabled,
@@ -30,6 +31,10 @@ from services.rag_service import (
 POLICY = "controlled_internal_rag_candidate_canary_v1"
 CANARY_VERSION = "internal_rag_candidate_canary_v1"
 DEFAULT_LIMIT = 1
+DEFAULT_MIN_GCP_HOSTS = max(
+    1,
+    min(int(os.getenv("INTERNAL_RAG_CANDIDATE_CANARY_MIN_GCP_HOSTS", "1")), 2),
+)
 DEFAULT_SIMILARITY_THRESHOLD = float(
     os.getenv("INTERNAL_RAG_CANDIDATE_CANARY_THRESHOLD", "0.70")
 )
@@ -267,20 +272,33 @@ def _verify_embedding_consistency() -> dict[str, Any]:
     result = dict(verify_embedding_consistency())
     required_hosts = {"gcp_ollama", "ollama_secondary"}
     reachable = set(result.get("reachable") or [])
+    approved_reachable = required_hosts.intersection(reachable)
     upstream_ok = result.get("ok") is True
-    required_hosts_ready = required_hosts.issubset(reachable)
+    digest_verified = result.get("digest_verified") is True
+    minimum_hosts_ready = len(approved_reachable) >= DEFAULT_MIN_GCP_HOSTS
     errors = list(result.get("errors") or [])
-    if not required_hosts_ready:
-        missing = sorted(required_hosts - reachable)
-        marker = f"required embedding hosts unavailable: {missing}"
+    if not minimum_hosts_ready:
+        marker = (
+            "minimum approved GCP embedding hosts unavailable: "
+            f"required={DEFAULT_MIN_GCP_HOSTS} reachable={sorted(approved_reachable)}"
+        )
+        if marker not in errors:
+            errors.append(marker)
+    if not digest_verified:
+        marker = "embedding model digest is not verified"
         if marker not in errors:
             errors.append(marker)
     result.update(
         {
-            "ok": upstream_ok and required_hosts_ready,
+            "ok": upstream_ok and minimum_hosts_ready and digest_verified,
             "upstream_ok": upstream_ok,
             "required_hosts": sorted(required_hosts),
-            "required_hosts_ready": required_hosts_ready,
+            "minimum_required_host_count": DEFAULT_MIN_GCP_HOSTS,
+            "approved_reachable_hosts": sorted(approved_reachable),
+            "minimum_hosts_ready": minimum_hosts_ready,
+            "required_hosts_ready": minimum_hosts_ready,
+            "digest_verified": digest_verified,
+            "redundancy_degraded": len(approved_reachable) < len(required_hosts),
             "errors": errors,
         }
     )
@@ -367,7 +385,8 @@ def _execute_item(
     reachable = list(consistency.get("reachable") or [])
     consistency_ready = (
         consistency.get("ok") is True
-        and required_hosts.issubset(set(reachable))
+        and len(required_hosts.intersection(set(reachable))) >= DEFAULT_MIN_GCP_HOSTS
+        and consistency.get("digest_verified") is True
     )
     if not consistency_ready:
         result.update(
@@ -382,13 +401,18 @@ def _execute_item(
                 "reachable_consistency_hosts": reachable,
                 "canary_checks": {
                     "cross_host_embedding_consistent": False,
+                    "embedding_model_digest_verified": (
+                        consistency.get("digest_verified") is True
+                    ),
                     "database_call_blocked_by_preflight": True,
                     "database_write_absent": True,
                     "ai_insights_write_absent": True,
                     "price_table_write_absent": True,
                 },
-                "canary_check_count": 5,
-                "canary_check_pass_count": 4,
+                "canary_check_count": 6,
+                "canary_check_pass_count": 4 + int(
+                    consistency.get("digest_verified") is True
+                ),
                 "rollback_terminal": "no_database_call_due_embedding_host_preflight",
                 "error": (
                     "embedding_host_preflight_failed: "
@@ -408,6 +432,7 @@ def _execute_item(
                 "candidate_embedding_dimension_valid": candidate_dimension_valid,
                 "probe_embedding_dimension_valid": probe_dimension_valid,
                 "cross_host_embedding_consistent": True,
+                "embedding_model_digest_verified": True,
                 "database_call_blocked_by_preflight": True,
                 "database_write_absent": True,
                 "ai_insights_write_absent": True,
@@ -443,7 +468,11 @@ def _execute_item(
             "probe_embedding_dimension_valid": len(probe_embedding) == RAG_EMBED_DIM,
             "cross_host_embedding_consistent": (
                 consistency.get("ok") is True
-                and required_hosts.issubset(set(reachable))
+                and len(required_hosts.intersection(set(reachable)))
+                >= DEFAULT_MIN_GCP_HOSTS
+            ),
+            "embedding_model_digest_verified": (
+                consistency.get("digest_verified") is True
             ),
             "pgvector_transaction_read_only": (
                 pgvector.get("transaction_read_only") is True
@@ -602,8 +631,12 @@ def run_internal_rag_candidate_canary(
     activation_blockers: list[str] = []
     if not is_rag_enabled():
         activation_blockers.append("rag_runtime_disabled")
-    if RAG_EMBED_MODEL.endswith(":latest"):
-        activation_blockers.append("rag_embedding_model_floating_tag")
+    if not RAG_EMBED_EXPECTED_DIGEST:
+        activation_blockers.append("rag_embedding_digest_guard_missing")
+    if consistency and consistency.get("digest_verified") is not True:
+        activation_blockers.append("rag_embedding_digest_unverified")
+    if consistency.get("redundancy_degraded") is True:
+        activation_blockers.append("rag_embedding_redundancy_degraded")
     latest_execution = _latest_execution_receipt(receipt_root)
     historical_canary_passed = latest_execution.get("canary_passed") is True
 
@@ -627,7 +660,7 @@ def run_internal_rag_candidate_canary(
             next_action = "repair_embedding_or_pgvector_canary_failure"
     elif activation_blockers:
         status = "canary_passed_activation_blocked"
-        next_action = "pin_embedding_model_then_enable_rag_controlled_canary"
+        next_action = "restore_embedding_redundancy_then_enable_rag_shadow_runtime"
     else:
         status = "complete"
         next_action = "continue_scheduled_internal_rag_canary"
@@ -655,7 +688,9 @@ def run_internal_rag_candidate_canary(
             "model": RAG_EMBED_MODEL,
             "dimension": RAG_EMBED_DIM,
             "signature": get_embedding_signature(),
-            "immutable_model_reference": not RAG_EMBED_MODEL.endswith(":latest"),
+            "expected_digest": RAG_EMBED_EXPECTED_DIGEST,
+            "immutable_digest_guard_configured": bool(RAG_EMBED_EXPECTED_DIGEST),
+            "immutable_model_reference": bool(RAG_EMBED_EXPECTED_DIGEST),
             "cross_host_consistency": consistency,
         },
         "source_items": source_items,

services/nemoton_dispatcher_service.py

@@ -123,6 +123,10 @@ _nim_call_count = {"date": "", "count": 0}
 NEMOTRON_OLLAMA_FIRST = os.getenv("NEMOTRON_OLLAMA_FIRST", "true").lower() == "true"
 NEMOTRON_OLLAMA_MODEL = os.getenv("NEMOTRON_OLLAMA_MODEL", "qwen3:14b")
 NEMOTRON_OLLAMA_TIMEOUT = int(os.getenv("NEMOTRON_OLLAMA_TIMEOUT", "180"))  # 秒
+NEMOTRON_OLLAMA_EXPECTED_DIGEST = os.getenv(
+    "NEMOTRON_OLLAMA_EXPECTED_DIGEST",
+    "bdbd181c33f2ed1b31c972991882db3cf4d192569092138a7d29e973cd9debe8",
+).strip().lower()
 
 
 def _check_nim_quota() -> bool:
@@ -645,6 +649,74 @@ def _parse_content_fallback(raw_content: str) -> list:
     return results
 
 
+def build_qwen3_dispatch_payload(
+    threats: list,
+    *,
+    mcp_context: str | None = None,
+    num_predict: int = 2048,
+    keep_alive: str | int | None = None,
+) -> dict:
+    """Build the shared production/canary qwen3 tool-calling request."""
+    threat_summary = json.dumps(
+        [
+            {
+                "sku": t.sku,
+                "name": t.name,
+                "momo_price": t.momo_price,
+                "pchome_price": t.pchome_price,
+                "gap_pct": t.gap_pct,
+                "sales_delta": t.sales_7d_delta_pct,
+                "risk": t.risk,
+                "action": t.recommended_action,
+                "confidence": t.confidence,
+                **_threat_match_metadata(t),
+            }
+            for t in threats
+        ],
+        ensure_ascii=False,
+    )
+    resolved_mcp_context = (
+        build_mcp_context() if mcp_context is None else str(mcp_context)
+    )
+    system_prompt = (
+        "你是台灣電商競價情報的行動派發器。"
+        f"當前市場背景 (MCP)：\n{resolved_mcp_context}\n\n"
+        "根據 Hermes 分析師提供的威脅清單，決定對每支商品呼叫哪個工具。\n"
+        "路由鐵律（依序判斷，命中即停）：\n"
+        "1. match_type 不是 exact，或 price_basis 不是 total_price，或 alert_tier 不是 price_alert_exact "
+        "→ 不可直接價格告警，呼叫 flag_for_human_review，concern 說明需覆核身份、包裝或單位價。\n"
+        "2. gap_pct < 5% 且 sales_delta < -30% → 非價格異常，呼叫 flag_for_human_review，"
+        "concern 說明『價差接近 0 但銷量大幅下滑，疑似缺貨/下架/平台流量異常，請 AI 例外走查前台』。\n"
+        "3. gap_pct ≥ 5% 且 risk=HIGH → trigger_price_alert（填入 momo_price, comp_price）。\n"
+        "4. 我方價格低於競品且銷量正成長 → add_to_recommendation。\n"
+        "5. confidence < 0.6 或其他複雜情況 → flag_for_human_review。\n"
+        "每支商品只呼叫一個工具。\n"
+        "【語言鐵律 — 台灣標準正體中文（繁體）】所有文字欄位必須遵守：\n"
+        "  1. 嚴禁簡體字、嚴禁異體字（例：不可用『亊』，必須用『事』）\n"
+        "  2. 嚴禁短語重複（語意坍塌）、嚴禁無意義字元組合\n"
+        "若無法產出合理的繁體中文說明，直接輸出『請人工評估議價空間』。"
+    )
+    payload = {
+        "model": NEMOTRON_OLLAMA_MODEL,
+        "messages": [
+            {"role": "system", "content": system_prompt},
+            {
+                "role": "user",
+                "content": f"請處理以下 {len(threats)} 筆威脅清單：\n{threat_summary}",
+            },
+        ],
+        "tools": TOOLS,
+        "stream": False,
+        "options": {
+            "temperature": 0.2,
+            "num_predict": max(32, min(int(num_predict), 2048)),
+        },
+    }
+    if keep_alive is not None:
+        payload["keep_alive"] = keep_alive
+    return payload
+
+
 def _build_footprint_json(hermes_stats: Optional[dict], nim_stats: Optional[dict]) -> dict:
     """
     建立結構化運算足跡 (用於 DB model_footprint JSONB 欄位)
@@ -889,61 +961,7 @@ class NemotronDispatcher:
             resolve_ollama_host,
         )
 
-        threat_summary = json.dumps(
-            [
-                {
-                    "sku":          t.sku,
-                    "name":         t.name,
-                    "momo_price":   t.momo_price,
-                    "pchome_price": t.pchome_price,
-                    "gap_pct":      t.gap_pct,
-                    "sales_delta":  t.sales_7d_delta_pct,
-                    "risk":         t.risk,
-                    "action":       t.recommended_action,
-                    "confidence":   t.confidence,
-                    **_threat_match_metadata(t),
-                }
-                for t in threats
-            ],
-            ensure_ascii=False,
-        )
-
-        # 注入 MCP 市場上下文（與 NIM 路徑一致）
-        mcp_ctx = build_mcp_context()
-
-        # System prompt 與 NIM 完全一致（避免兩套維護）
-        system_prompt = (
-            "你是台灣電商競價情報的行動派發器。"
-            f"當前市場背景 (MCP)：\n{mcp_ctx}\n\n"
-            "根據 Hermes 分析師提供的威脅清單，決定對每支商品呼叫哪個工具。\n"
-            "路由鐵律（依序判斷，命中即停）：\n"
-            "1. match_type 不是 exact，或 price_basis 不是 total_price，或 alert_tier 不是 price_alert_exact "
-            "→ 不可直接價格告警，呼叫 flag_for_human_review，concern 說明需覆核身份、包裝或單位價。\n"
-            "2. gap_pct < 5% 且 sales_delta < -30% → 非價格異常，呼叫 flag_for_human_review，"
-            "concern 說明『價差接近 0 但銷量大幅下滑，疑似缺貨/下架/平台流量異常，請 AI 例外走查前台』。\n"
-            "3. gap_pct ≥ 5% 且 risk=HIGH → trigger_price_alert（填入 momo_price, comp_price）。\n"
-            "4. 我方價格低於競品且銷量正成長 → add_to_recommendation。\n"
-            "5. confidence < 0.6 或其他複雜情況 → flag_for_human_review。\n"
-            "每支商品只呼叫一個工具。\n"
-            "【語言鐵律 — 台灣標準正體中文（繁體）】所有文字欄位必須遵守：\n"
-            "  1. 嚴禁簡體字、嚴禁異體字（例：不可用「亊」，必須用「事」）\n"
-            "  2. 嚴禁短語重複（語意坍塌）、嚴禁無意義字元組合\n"
-            "若無法產出合理的繁體中文說明，直接輸出「請人工評估議價空間」。"
-        )
-
-        payload = {
-            "model": NEMOTRON_OLLAMA_MODEL,
-            "messages": [
-                {"role": "system", "content": system_prompt},
-                {"role": "user",   "content": f"請處理以下 {len(threats)} 筆威脅清單：\n{threat_summary}"},
-            ],
-            "tools": TOOLS,           # 重用既有 NIM tools schema
-            "stream": False,
-            "options": {
-                "temperature": 0.2,
-                "num_predict": 2048,
-            },
-        }
+        payload = build_qwen3_dispatch_payload(threats)
 
         with log_ai_call(
             caller='nemotron_dispatch',

services/nemotron_decision_canary_scheduler_task.py

@@ -0,0 +1,120 @@
+"""Scheduler adapter for the decision-only Nemotron runtime canary."""
+
+from __future__ import annotations
+
+import logging
+from collections.abc import Callable
+
+
+def run_scheduled_nemotron_decision_canary(
+    *,
+    env_flag: Callable[[str, bool], bool],
+    save_stats: Callable[[str, dict[str, object]], None],
+    notify_failure: Callable[..., None],
+) -> dict[str, object]:
+    """Run one model decision, persist proof, and send lifecycle acknowledgement."""
+    if not env_flag("NEMOTRON_DECISION_CANARY_SCHEDULED_ENABLED", True):
+        payload: dict[str, object] = {
+            "success": True,
+            "status": "skipped",
+            "terminal_status": "no_write_terminal",
+            "reason": "scheduled_canary_disabled_by_policy",
+            "next_machine_action": "enable_nemotron_decision_canary_after_policy_review",
+        }
+        save_stats("nemotron_decision_canary", payload)
+        return payload
+
+    try:
+        from services.nemotron_decision_canary_service import (
+            acknowledge_nemotron_decision_canary,
+            run_nemotron_decision_canary,
+        )
+        from services.telegram_templates import send_telegram_with_result
+
+        payload = run_nemotron_decision_canary(
+            execute=True,
+            write_receipt=True,
+        )
+        execution = payload.get("execution") or {}
+        telegram = send_telegram_with_result(
+            "\n".join([
+                "Nemotron decision-only canary",
+                f"status: {payload.get('status')}",
+                f"host: {execution.get('selected_host')}",
+                f"model: {execution.get('model')}",
+                f"decision: {execution.get('decision_tool')}",
+                f"elapsed_ms: {execution.get('model_elapsed_ms')}",
+                "side effects: tool=0, database=0, Telegram action=0",
+                f"run_id: {(payload.get('run_identity') or {}).get('run_id')}",
+            ]),
+            parse_mode=None,
+        )
+        telegram_status = "acknowledged" if telegram.get("ok") else "failed"
+        receipt_path = str(execution.get("receipt_path") or "")
+        receipt_acknowledged = False
+        if receipt_path:
+            acknowledged = acknowledge_nemotron_decision_canary(
+                receipt_path,
+                telegram_status=telegram_status,
+                telegram_sent=int(telegram.get("sent") or 0),
+                telegram_failed=int(telegram.get("failed") or 0),
+            )
+            payload["execution"] = acknowledged
+            payload["latest_execution"] = run_nemotron_decision_canary(
+                execute=False
+            ).get("latest_execution") or {}
+            receipt_acknowledged = True
+        else:
+            payload["receipt_error"] = "durable_execution_receipt_missing"
+
+        verified = bool(
+            payload.get("success")
+            and telegram_status == "acknowledged"
+            and receipt_acknowledged
+        )
+        payload["terminal_status"] = (
+            "verified_decision_only_no_write" if verified else "partial"
+        )
+        save_stats("nemotron_decision_canary", payload)
+        logging.info(
+            "[NemotronCanary] status=%s terminal=%s model_ms=%s telegram=%s receipt=%s",
+            payload.get("status"),
+            payload.get("terminal_status"),
+            execution.get("model_elapsed_ms"),
+            telegram_status,
+            "acknowledged" if receipt_acknowledged else "missing",
+        )
+        if not verified:
+            notify_failure(
+                "run_nemotron_decision_canary_task",
+                RuntimeError(
+                    "Nemotron decision canary closure or acknowledgement failed"
+                ),
+                source="Scheduler.NemotronCanary",
+                event_type="nemotron_decision_canary_failure",
+                title="Nemotron decision-only canary failed",
+                dedup_ttl_sec=86400,
+            )
+        return payload
+    except Exception as error:
+        logging.error("[NemotronCanary] task failed: %s", error, exc_info=True)
+        notify_failure(
+            "run_nemotron_decision_canary_task",
+            error,
+            source="Scheduler.NemotronCanary",
+            event_type="nemotron_decision_canary_task_failure",
+            title="Nemotron decision-only canary task failed",
+            dedup_ttl_sec=86400,
+        )
+        payload = {
+            "success": False,
+            "status": "failed",
+            "terminal_status": "partial",
+            "error": f"{type(error).__name__}: {str(error)[:300]}",
+            "next_machine_action": "inspect_nemotron_canary_receipt_and_retry",
+        }
+        save_stats("nemotron_decision_canary", payload)
+        return payload
+
+
+__all__ = ["run_scheduled_nemotron_decision_canary"]

services/nemotron_decision_canary_service.py

@@ -0,0 +1,459 @@
+"""Decision-only NemoTron runtime canary.
+
+The canary exercises the production qwen3 tool-calling payload against an
+approved GCP Ollama host, validates the returned decision, and stops before any
+tool, Telegram, database, price, or insight write is allowed.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import time
+import uuid
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any, Mapping
+
+import requests
+
+from services.ollama_service import OLLAMA_HOST_PRIMARY, OLLAMA_HOST_SECONDARY
+
+
+POLICY = "controlled_nemotron_decision_only_canary_v1"
+CANARY_VERSION = "nemotron_decision_only_canary_v1"
+NEMOTRON_OLLAMA_MODEL = os.getenv("NEMOTRON_OLLAMA_MODEL", "qwen3:14b")
+NEMOTRON_OLLAMA_EXPECTED_DIGEST = os.getenv(
+    "NEMOTRON_OLLAMA_EXPECTED_DIGEST",
+    "bdbd181c33f2ed1b31c972991882db3cf4d192569092138a7d29e973cd9debe8",
+).strip().lower()
+DEFAULT_TIMEOUT_SEC = max(
+    30,
+    min(int(os.getenv("NEMOTRON_DECISION_CANARY_TIMEOUT_SEC", "300")), 600),
+)
+DEFAULT_MODEL_IDENTITY_TIMEOUT_SEC = max(
+    2,
+    min(int(os.getenv("NEMOTRON_MODEL_IDENTITY_TIMEOUT_SEC", "10")), 30),
+)
+DEFAULT_MAX_AGE_HOURS = max(
+    1,
+    min(int(os.getenv("NEMOTRON_DECISION_CANARY_MAX_AGE_HOURS", "26")), 168),
+)
+DEFAULT_OUTPUT_ROOT = os.getenv(
+    "NEMOTRON_DECISION_CANARY_RECEIPT_ROOT",
+    "/app/data/ai_automation/nemotron_decision_canary_receipts"
+    if Path("/app/data").exists()
+    else "runtime_artifacts/nemotron_decision_canary_receipts",
+)
+
+APPROVED_TOOLS = {
+    "trigger_price_alert",
+    "add_to_recommendation",
+    "flag_for_human_review",
+}
+
+
+@dataclass(frozen=True)
+class _SyntheticThreat:
+    sku: str = "CANARY-NEMO-001"
+    name: str = "NemoTron 受控決策驗證品"
+    momo_price: float = 1200.0
+    pchome_price: float = 980.0
+    gap_pct: float = 22.4
+    sales_7d_delta_pct: float = -35.0
+    risk: str = "HIGH"
+    recommended_action: str = "依競價證據建立受控價格告警候選"
+    confidence: float = 0.91
+    match_type: str = "exact"
+    price_basis: str = "total_price"
+    alert_tier: str = "price_alert_exact"
+    match_score: float = 0.99
+    competitor_product_id: str = "CANARY-COMP-001"
+    competitor_product_name: str = "NemoTron 受控決策驗證品"
+
+
+def _parse_datetime(value: Any) -> datetime | None:
+    if not value:
+        return None
+    try:
+        parsed = datetime.fromisoformat(str(value).replace("Z", "+00:00"))
+    except ValueError:
+        return None
+    if parsed.tzinfo is None:
+        parsed = parsed.replace(tzinfo=timezone.utc)
+    return parsed.astimezone(timezone.utc)
+
+
+def _model_identity(host: str) -> dict[str, Any]:
+    started = time.monotonic()
+    try:
+        response = requests.get(
+            f"{host.rstrip('/')}/api/tags",
+            timeout=DEFAULT_MODEL_IDENTITY_TIMEOUT_SEC,
+        )
+        response.raise_for_status()
+        models = response.json().get("models") or []
+    except Exception as exc:
+        return {
+            "ok": False,
+            "host": host,
+            "model": NEMOTRON_OLLAMA_MODEL,
+            "digest": None,
+            "digest_matches": False,
+            "elapsed_ms": round((time.monotonic() - started) * 1000),
+            "error": f"{type(exc).__name__}: {str(exc)[:180]}",
+        }
+
+    target = NEMOTRON_OLLAMA_MODEL
+    target_without_latest = target.removesuffix(":latest")
+    matched: Mapping[str, Any] = {}
+    for item in models:
+        names = {
+            str(item.get("name") or "").strip(),
+            str(item.get("model") or "").strip(),
+        }
+        if target in names or target_without_latest in names:
+            matched = item
+            break
+    digest = str(matched.get("digest") or "").strip().lower()
+    digest_matches = bool(digest) and digest == NEMOTRON_OLLAMA_EXPECTED_DIGEST
+    return {
+        "ok": bool(matched) and digest_matches,
+        "host": host,
+        "model": target,
+        "digest": digest or None,
+        "expected_digest": NEMOTRON_OLLAMA_EXPECTED_DIGEST,
+        "digest_matches": digest_matches,
+        "parameter_size": str((matched.get("details") or {}).get("parameter_size") or ""),
+        "quantization_level": str(
+            (matched.get("details") or {}).get("quantization_level") or ""
+        ),
+        "elapsed_ms": round((time.monotonic() - started) * 1000),
+        "error": None if matched else f"model_not_found:{target}",
+    }
+
+
+def _latest_execution(root: Path, *, now: datetime | None = None) -> dict[str, Any]:
+    if not root.exists():
+        return {}
+    candidates = sorted(
+        root.glob("*/nemotron_decision_canary_receipt.json"),
+        key=lambda path: path.stat().st_mtime,
+        reverse=True,
+    )
+    if not candidates:
+        return {}
+    path = candidates[0]
+    try:
+        payload = json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, json.JSONDecodeError):
+        return {}
+    generated_at = _parse_datetime(payload.get("generated_at"))
+    current = now or datetime.now(timezone.utc)
+    age_hours = (
+        (current - generated_at).total_seconds() / 3600
+        if generated_at is not None
+        else None
+    )
+    return {
+        "receipt_path": str(path),
+        "generated_at": payload.get("generated_at"),
+        "age_hours": round(age_hours, 3) if age_hours is not None else None,
+        "fresh": age_hours is not None and age_hours <= DEFAULT_MAX_AGE_HOURS,
+        "status": payload.get("status"),
+        "canary_passed": payload.get("canary_passed") is True,
+        "selected_host": payload.get("selected_host"),
+        "model": payload.get("model"),
+        "observed_digest": payload.get("observed_digest"),
+        "decision_tool": payload.get("decision_tool"),
+        "decision_sku": payload.get("decision_sku"),
+        "model_elapsed_ms": payload.get("model_elapsed_ms"),
+        "tool_execution_count": int(payload.get("tool_execution_count") or 0),
+        "database_call_performed": payload.get("database_call_performed") is True,
+        "writes_database": payload.get("writes_database") is True,
+        "writes_price_tables": payload.get("writes_price_tables") is True,
+        "writes_ai_insights": payload.get("writes_ai_insights") is True,
+        "telegram_sent": payload.get("telegram_sent") is True,
+        "rollback_terminal": payload.get("rollback_terminal"),
+        "terminal_status": payload.get("terminal_status"),
+        "acknowledgements": payload.get("acknowledgements") or {},
+        "run_identity": payload.get("run_identity") or {},
+        "error": payload.get("error"),
+    }
+
+
+def _write_receipt(root: Path, run_id: str, payload: Mapping[str, Any]) -> str:
+    target = root / run_id / "nemotron_decision_canary_receipt.json"
+    target.parent.mkdir(parents=True, exist_ok=True)
+    target.write_text(
+        json.dumps(dict(payload), ensure_ascii=False, indent=2, sort_keys=True),
+        encoding="utf-8",
+    )
+    return str(target)
+
+
+def acknowledge_nemotron_decision_canary(
+    receipt_path: str | Path,
+    *,
+    telegram_status: str,
+    telegram_sent: int = 0,
+    telegram_failed: int = 0,
+) -> dict[str, Any]:
+    """Atomically append the scheduler Telegram acknowledgement to a receipt."""
+    target = Path(receipt_path)
+    payload = json.loads(target.read_text(encoding="utf-8"))
+    acknowledgement = {
+        "status": str(telegram_status),
+        "sent": int(telegram_sent),
+        "failed": int(telegram_failed),
+    }
+    payload["acknowledgements"] = {
+        "telegram": acknowledgement,
+        "km": "not_applicable_decision_only_canary",
+        "rag": "not_applicable_decision_only_canary",
+        "mcp": "not_applicable_decision_only_canary",
+        "playbook": "nemotron_decision_canary_receipt_written",
+    }
+    closure = payload.setdefault("closure_receipt", {})
+    closure["telegram_acknowledgement"] = str(telegram_status)
+    payload["terminal_status"] = (
+        "verified_decision_only_no_write"
+        if payload.get("canary_passed") is True and telegram_status == "acknowledged"
+        else "partial"
+    )
+    temporary = target.with_suffix(target.suffix + ".tmp")
+    temporary.write_text(
+        json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True),
+        encoding="utf-8",
+    )
+    temporary.replace(target)
+    return payload
+
+
+def run_nemotron_decision_canary(
+    *,
+    output_root: str | Path | None = None,
+    execute: bool = False,
+    write_receipt: bool = False,
+    timeout_sec: int | None = None,
+    trace_id: str | None = None,
+    run_id: str | None = None,
+    work_item_id: str = "AI-RUNTIME-NEMOTRON-CANARY-001",
+) -> dict[str, Any]:
+    """Read the latest receipt or execute one bounded decision-only model call."""
+    now = datetime.now(timezone.utc)
+    receipt_root = Path(output_root or DEFAULT_OUTPUT_ROOT)
+    resolved_run_id = str(run_id or f"nemotron-canary-{uuid.uuid4()}")
+    run_identity = {
+        "trace_id": str(trace_id or f"trace-{resolved_run_id}"),
+        "run_id": resolved_run_id,
+        "work_item_id": str(work_item_id or "AI-RUNTIME-NEMOTRON-CANARY-001"),
+    }
+    latest_before = _latest_execution(receipt_root, now=now)
+    if not execute:
+        verified = (
+            latest_before.get("canary_passed") is True
+            and latest_before.get("fresh") is True
+        )
+        return {
+            "success": True,
+            "policy": POLICY,
+            "canary_version": CANARY_VERSION,
+            "generated_at": now.isoformat(),
+            "status": "verified" if verified else "warning",
+            "execute": False,
+            "model": NEMOTRON_OLLAMA_MODEL,
+            "expected_digest": NEMOTRON_OLLAMA_EXPECTED_DIGEST,
+            "latest_execution": latest_before,
+            "controlled_apply": {
+                "network_call": False,
+                "model_call": False,
+                "tool_execution_count": 0,
+                "database_call_performed": False,
+                "database_write": False,
+                "price_table_write": False,
+                "ai_insights_write": False,
+                "telegram_send": False,
+            },
+            "next_machine_action": (
+                "continue_scheduled_nemotron_decision_canary"
+                if verified
+                else "execute_nemotron_decision_only_canary"
+            ),
+        }
+
+    host_preflights = [
+        _model_identity(OLLAMA_HOST_PRIMARY),
+        _model_identity(OLLAMA_HOST_SECONDARY),
+    ]
+    selected = next((item for item in host_preflights if item.get("ok")), None)
+    model_call_performed = False
+    model_elapsed_ms = 0
+    decisions: list[dict[str, Any]] = []
+    decision_format = "none"
+    error: str | None = None
+    if selected is None:
+        error = "no_approved_gcp_host_with_expected_nemotron_digest"
+    else:
+        from services.nemoton_dispatcher_service import (
+            _parse_content_fallback,
+            _parse_tool_calls_struct,
+            build_qwen3_dispatch_payload,
+        )
+
+        threat = _SyntheticThreat()
+        payload = build_qwen3_dispatch_payload(
+            [threat],
+            mcp_context="controlled decision-only canary; no live market data",
+            num_predict=160,
+            keep_alive=0,
+        )
+        payload["think"] = False
+        started = time.monotonic()
+        model_call_performed = True
+        try:
+            response = requests.post(
+                f"{str(selected['host']).rstrip('/')}/api/chat",
+                json=payload,
+                timeout=max(30, min(int(timeout_sec or DEFAULT_TIMEOUT_SEC), 600)),
+            )
+            response.raise_for_status()
+            body = response.json()
+            message = body.get("message") or {}
+            decisions = _parse_tool_calls_struct(message.get("tool_calls") or [])
+            if decisions:
+                decision_format = "tool_calls"
+            else:
+                decisions = _parse_content_fallback(str(message.get("content") or ""))
+                if decisions:
+                    decision_format = "content_fallback"
+        except Exception as exc:
+            error = f"{type(exc).__name__}: {str(exc)[:240]}"
+        finally:
+            model_elapsed_ms = round((time.monotonic() - started) * 1000)
+
+    decision = decisions[0] if decisions else {}
+    decision_tool = str(decision.get("tool") or "")
+    decision_args = decision.get("args") if isinstance(decision.get("args"), dict) else {}
+    decision_sku = str((decision_args or {}).get("sku") or "")
+    expected_tool = "trigger_price_alert"
+    post_identity = _model_identity(str(selected["host"])) if selected else {}
+    checks = {
+        "approved_host_selected": selected is not None,
+        "expected_digest_verified_before": bool(selected and selected.get("digest_matches")),
+        "model_call_completed": model_call_performed and error is None,
+        "decision_present": bool(decisions),
+        "decision_tool_allowed": decision_tool in APPROVED_TOOLS,
+        "expected_decision_tool_selected": decision_tool == expected_tool,
+        "synthetic_sku_preserved": decision_sku == _SyntheticThreat.sku,
+        "expected_digest_stable_after": bool(post_identity.get("digest_matches")),
+        "tool_execution_absent": True,
+        "database_call_absent": True,
+        "telegram_send_absent": True,
+    }
+    canary_passed = all(checks.values())
+    status = "canary_passed" if canary_passed else "canary_failed"
+    execution_receipt: dict[str, Any] = {
+        "policy": POLICY,
+        "canary_version": CANARY_VERSION,
+        "generated_at": now.isoformat(),
+        "run_identity": run_identity,
+        "status": status,
+        "canary_passed": canary_passed,
+        "model": NEMOTRON_OLLAMA_MODEL,
+        "expected_digest": NEMOTRON_OLLAMA_EXPECTED_DIGEST,
+        "observed_digest": selected.get("digest") if selected else None,
+        "selected_host": selected.get("host") if selected else None,
+        "host_preflights": host_preflights,
+        "post_model_identity": post_identity,
+        "model_call_performed": model_call_performed,
+        "model_elapsed_ms": model_elapsed_ms,
+        "decision_format": decision_format,
+        "decision_count": len(decisions),
+        "decision_tool": decision_tool or None,
+        "decision_sku": decision_sku or None,
+        "expected_decision_tool": expected_tool,
+        "decision_argument_keys": sorted((decision_args or {}).keys()),
+        "checks": checks,
+        "check_count": len(checks),
+        "check_pass_count": sum(checks.values()),
+        "decision_only": True,
+        "tool_execution_count": 0,
+        "database_call_performed": False,
+        "writes_database": False,
+        "writes_price_tables": False,
+        "writes_ai_insights": False,
+        "telegram_sent": False,
+        "error": error,
+        "rollback_terminal": "decision_only_no_tool_or_data_write",
+        "source_of_truth_diff": {
+            "expected_model": NEMOTRON_OLLAMA_MODEL,
+            "expected_digest": NEMOTRON_OLLAMA_EXPECTED_DIGEST,
+            "observed_digest": selected.get("digest") if selected else None,
+            "expected_decision_tool": expected_tool,
+            "observed_decision_tool": decision_tool or None,
+        },
+        "closure_receipt": {
+            "sensor_source_receipt": bool(host_preflights),
+            "normalized_asset_identity": selected.get("host") if selected else None,
+            "source_of_truth_diff_recorded": True,
+            "ai_candidate_decision_recorded": bool(decisions),
+            "risk_policy_decision": "medium_decision_only_no_tool_execution",
+            "check_mode_passed": bool(selected),
+            "bounded_execution_performed": model_call_performed,
+            "independent_verifier_passed": canary_passed,
+            "rollback_or_no_write_terminal": "decision_only_no_tool_or_data_write",
+            "telegram_acknowledgement": "pending_scheduler_dispatch",
+            "learning_write_acknowledgement": "pending_receipt_write",
+        },
+    }
+    if write_receipt:
+        receipt_path = str(
+            receipt_root
+            / resolved_run_id
+            / "nemotron_decision_canary_receipt.json"
+        )
+        execution_receipt["receipt_path"] = receipt_path
+        execution_receipt["closure_receipt"][
+            "learning_write_acknowledgement"
+        ] = "nemotron_decision_canary_receipt_written"
+        _write_receipt(receipt_root, resolved_run_id, execution_receipt)
+
+    latest_after = _latest_execution(receipt_root)
+    return {
+        "success": canary_passed,
+        "policy": POLICY,
+        "canary_version": CANARY_VERSION,
+        "generated_at": now.isoformat(),
+        "run_identity": run_identity,
+        "status": status,
+        "execute": True,
+        "execution": execution_receipt,
+        "latest_execution": latest_after,
+        "controlled_apply": {
+            "risk": "medium",
+            "network_call": True,
+            "model_call": model_call_performed,
+            "tool_execution_count": 0,
+            "database_call_performed": False,
+            "database_write": False,
+            "price_table_write": False,
+            "ai_insights_write": False,
+            "telegram_send": False,
+            "rollback_terminal": "decision_only_no_tool_or_data_write",
+        },
+        "next_machine_action": (
+            "continue_scheduled_nemotron_decision_canary"
+            if canary_passed
+            else "repair_nemotron_model_runtime_then_retry_decision_only_canary"
+        ),
+    }
+
+
+__all__ = [
+    "CANARY_VERSION",
+    "POLICY",
+    "acknowledge_nemotron_decision_canary",
+    "run_nemotron_decision_canary",
+]

services/ollama_service.py

@@ -1281,7 +1281,8 @@ class OllamaService:
 
     def generate_embedding(self, text: str, model: str = "bge-m3:latest",
                            host: str = None, timeout: int = None,
-                           allow_111_fallback: bool = True) -> List[float]:
+                           allow_111_fallback: bool = True,
+                           timeout_cap: int = None) -> List[float]:
         """
         [ADR-007] Embedding — 含三主機自動 retry（HOTFIX 2026-05-04）
 
@@ -1302,7 +1303,8 @@ class OllamaService:
                 model,
             )
             clean_text = clean_text[:EMBED_MAX_CHARS]
-        request_timeout = min(timeout or EMBED_TIMEOUT, EMBED_MAX_TIMEOUT)
+        resolved_timeout_cap = max(1, int(timeout_cap or EMBED_MAX_TIMEOUT))
+        request_timeout = min(timeout or EMBED_TIMEOUT, resolved_timeout_cap)
 
         def _embed_one(target_host: str) -> List[float]:
             """單次 embedding 嘗試 — 成功回 vec，失敗回 [] + mark_unhealthy"""

services/rag_service.py

@@ -64,6 +64,10 @@ RAG_EMBED_DIM = int(os.getenv('RAG_EMBED_DIM', '1024'))
 RAG_EMBED_NORMALIZE = os.getenv('RAG_EMBED_NORMALIZE', 'true').strip().lower() in (
     'true', '1', 'yes', 'on',
 )
+RAG_EMBED_EXPECTED_DIGEST = os.getenv(
+    'RAG_EMBED_EXPECTED_DIGEST',
+    '7907646426070047a77226ac3e684fbbe8410524f7b4a74d02837e43f2146bab',
+).strip().lower()
 
 # query_text 寫入長度上限（與 027 CHECK octet_length<=4096 對齊；中文 1 字 3 byte → ~1300 字）
 _QUERY_TEXT_MAX_BYTES = 4096
@@ -128,7 +132,12 @@ def get_embedding_signature(
 # ─────────────────────────────────────────────────────────────────────────────
 EMBED_CONSISTENCY_TEST_TEXT = "momo電商競品分析測試向量一致性檢查"
 EMBED_CONSISTENCY_MAX_DIFF = 1e-4    # cosine 距離上限（浮點誤差容忍）
-EMBED_CONSISTENCY_TIMEOUT_SEC = 10.0  # 各主機 embedding 探測 timeout
+EMBED_CONSISTENCY_TIMEOUT_SEC = float(
+    os.getenv('EMBED_CONSISTENCY_TIMEOUT_SEC', '150')
+)
+EMBED_MODEL_IDENTITY_TIMEOUT_SEC = float(
+    os.getenv('EMBED_MODEL_IDENTITY_TIMEOUT_SEC', '10')
+)
 EMBED_CONSISTENCY_INCLUDE_111 = os.getenv(
     'EMBED_CONSISTENCY_INCLUDE_111',
     'false',
@@ -147,6 +156,38 @@ def _cosine_distance(vec_a: List[float], vec_b: List[float]) -> float:
     return max(0.0, 1.0 - dot / (norm_a * norm_b))
 
 
+def _fetch_ollama_model_digest(
+    host: str,
+    model: str = RAG_EMBED_MODEL,
+) -> tuple[str | None, str | None]:
+    """Read the local Ollama model manifest digest without loading the model."""
+    import requests
+
+    try:
+        response = requests.get(
+            f"{host.rstrip('/')}/api/tags",
+            timeout=EMBED_MODEL_IDENTITY_TIMEOUT_SEC,
+        )
+        response.raise_for_status()
+        models = response.json().get('models') or []
+    except Exception as exc:
+        return None, f"{type(exc).__name__}: {str(exc)[:160]}"
+
+    target = str(model or '').strip()
+    target_without_latest = target.removesuffix(':latest')
+    for item in models:
+        names = {
+            str(item.get('name') or '').strip(),
+            str(item.get('model') or '').strip(),
+        }
+        if target in names or target_without_latest in names:
+            digest = str(item.get('digest') or '').strip().lower()
+            if digest:
+                return digest, None
+            return None, 'model_manifest_digest_missing'
+    return None, f"model_not_found:{target}"
+
+
 def verify_embedding_consistency(
     test_text: str = EMBED_CONSISTENCY_TEST_TEXT,
     max_diff: float = EMBED_CONSISTENCY_MAX_DIFF,
@@ -182,9 +223,35 @@ def verify_embedding_consistency(
         hosts['ollama_111'] = OLLAMA_HOST_FALLBACK
 
     embeddings: Dict[str, List[float]] = {}
+    model_digests: Dict[str, str] = {}
+    digest_verified_hosts: List[str] = []
     errors: List[str] = []
 
     for label, host in hosts.items():
+        digest, digest_error = _fetch_ollama_model_digest(host, RAG_EMBED_MODEL)
+        if digest:
+            model_digests[label] = digest
+        if digest_error:
+            errors.append(f"{label}: model identity: {digest_error}")
+            logger.warning("[EmbedVerify] %s model identity failed: %s", label, digest_error)
+            continue
+        if not RAG_EMBED_EXPECTED_DIGEST:
+            errors.append(f"{label}: expected model digest is not configured")
+            logger.error("[EmbedVerify] expected model digest is not configured")
+            continue
+        if digest != RAG_EMBED_EXPECTED_DIGEST:
+            errors.append(
+                f"{label}: digest mismatch expected={RAG_EMBED_EXPECTED_DIGEST[:12]} "
+                f"observed={str(digest or '')[:12]}"
+            )
+            logger.error(
+                "[EmbedVerify] %s digest mismatch expected=%s observed=%s",
+                label,
+                RAG_EMBED_EXPECTED_DIGEST[:12],
+                str(digest or '')[:12],
+            )
+            continue
+        digest_verified_hosts.append(label)
         try:
             t0 = time.monotonic()
             vec = ollama_service.generate_embedding(
@@ -193,6 +260,7 @@ def verify_embedding_consistency(
                 host=host,  # 顯式指定（避免 retry 鏈干擾驗證）
                 timeout=int(EMBED_CONSISTENCY_TIMEOUT_SEC),
                 allow_111_fallback=(label == 'ollama_111'),
+                timeout_cap=int(EMBED_CONSISTENCY_TIMEOUT_SEC),
             )
             elapsed = time.monotonic() - t0
             if vec and len(vec) == RAG_EMBED_DIM:
@@ -207,16 +275,24 @@ def verify_embedding_consistency(
 
     signature = get_embedding_signature()
     reachable = list(embeddings.keys())
+    digest_verified = bool(reachable) and set(reachable).issubset(
+        set(digest_verified_hosts)
+    )
 
     if len(embeddings) < 2:
         msg = f"only {len(embeddings)} host reachable, cannot cross-verify"
         logger.warning(f"[EmbedVerify] {msg}")
         return {
-            'ok': True,  # fail-safe：1 主機可達不算錯（戰時可能 2 主機暫斷）
+            'ok': bool(embeddings) and digest_verified,
             'signature': signature,
             'reachable': reachable,
             'max_diff': 0.0,
             'errors': errors + [msg],
+            'expected_digest': RAG_EMBED_EXPECTED_DIGEST,
+            'model_digests': model_digests,
+            'digest_verified_hosts': digest_verified_hosts,
+            'digest_verified': digest_verified,
+            'redundancy_degraded': len(embeddings) == 1,
         }
 
     # 兩兩比對 cosine 距離
@@ -241,11 +317,16 @@ def verify_embedding_consistency(
         )
 
     return {
-        'ok': consistent,
+        'ok': consistent and digest_verified,
         'signature': signature,
         'reachable': reachable,
         'max_diff': max_diff_observed,
         'errors': errors,
+        'expected_digest': RAG_EMBED_EXPECTED_DIGEST,
+        'model_digests': model_digests,
+        'digest_verified_hosts': digest_verified_hosts,
+        'digest_verified': digest_verified,
+        'redundancy_degraded': False,
     }