Merge remote-tracking branch 'origin/dev' into codex/sales-analysis-contract-fix

docs(ai): record V10.816 runtime closure
Merge remote-tracking branch 'origin/main' into codex/sales-analysis-contract-fix-dev-sync
2026-07-23 09:23:27 +08:00 · 2026-07-23 09:22:46 +08:00 · 2026-07-23 08:55:36 +08:00 · 2026-07-22 22:58:35 +08:00 · 2026-07-22 22:58:08 +08:00 · 2026-07-22 22:57:07 +08:00
30 changed files with 5761 additions and 1442 deletions
--- a/.env.example
+++ b/.env.example
@@ -460,7 +460,8 @@ OLLAMA_HOST_HEALTH_MODEL_PROBE_INCLUDE_111=false
 OLLAMA_HOST_HEALTH_EMBED_MODEL=bge-m3:latest
 OLLAMA_HOST_HEALTH_EMBED_TIMEOUT=30
 OLLAMA_HOST_HEALTH_EMBED_KEEP_ALIVE=1m
-# 111 是 Mac final fallback，不承接 7B+ / vision / long-context / 長輸出任務；落到 111 時自動降級與縮短常駐。
+# 111 的一般 OllamaService fallback 不承接 7B+ / vision / long-context / 長輸出任務。
+# NemoTron 是唯一受控特例：只允許下方 exact-digest qwen3:8b，並強制總 deadline、num_ctx 4096、num_predict 512。
 OLLAMA_111_MODEL_FALLBACK=llama3.2:latest
 OLLAMA_111_MODEL_DOWNGRADE_PATTERNS=qwen3:*,deepseek-r1:*,hermes3:*,llama3.1:*,qwen2.5:*,qwen2.5-coder:*,gemma3:*,minicpm-v:*,llava:*,*:7b*,*:8b*,*:14b*,*:32b*,*:70b*
 OLLAMA_111_KEEP_ALIVE=5m
@@ -480,6 +481,11 @@ NEMOTRON_OLLAMA_FIRST=true
 NEMOTRON_OLLAMA_MODEL=qwen3:14b
 NEMOTRON_OLLAMA_TIMEOUT=180
 NEMOTRON_OLLAMA_EXPECTED_DIGEST=bdbd181c33f2ed1b31c972991882db3cf4d192569092138a7d29e973cd9debe8
+NEMOTRON_OLLAMA_FALLBACK_MODEL=qwen3:8b
+NEMOTRON_OLLAMA_FALLBACK_EXPECTED_DIGEST=500a1f067a9f782620b40bee6f7b0c89e17ae61f686b92c24933e4ca4b2b8b41
+NEMOTRON_GCP_ATTEMPT_TIMEOUT_SEC=60
+NEMOTRON_111_ATTEMPT_TIMEOUT_SEC=45
+NEMOTRON_DISPATCH_TOTAL_TIMEOUT_SEC=180
 NEMOTRON_DECISION_CANARY_TIMEOUT_SEC=300
 NEMOTRON_MODEL_IDENTITY_TIMEOUT_SEC=10
 NEMOTRON_DECISION_CANARY_MAX_AGE_HOURS=26
--- a/config.py
+++ b/config.py
@@ -414,7 +414,7 @@ YOUTUBE_API_KEY = os.getenv('YOUTUBE_API_KEY', '')
 # ==========================================
 # 系統版本與路徑
 # ==========================================
-SYSTEM_VERSION = "V10.814"
+SYSTEM_VERSION = "V10.816"
 LOG_FILE_PATH = os.path.join(BASE_DIR, 'logs/system.log')
 public_url = PUBLIC_URL  # 用於模板顯示

--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -107,6 +107,12 @@ services:
      - RAG_EMBED_EXPECTED_DIGEST=${RAG_EMBED_EXPECTED_DIGEST:-7907646426070047a77226ac3e684fbbe8410524f7b4a74d02837e43f2146bab}
      - EMBED_CONSISTENCY_TIMEOUT_SEC=${EMBED_CONSISTENCY_TIMEOUT_SEC:-150}
      - NEMOTRON_OLLAMA_EXPECTED_DIGEST=${NEMOTRON_OLLAMA_EXPECTED_DIGEST:-bdbd181c33f2ed1b31c972991882db3cf4d192569092138a7d29e973cd9debe8}
+      - NEMOTRON_OLLAMA_FALLBACK_MODEL=${NEMOTRON_OLLAMA_FALLBACK_MODEL:-qwen3:8b}
+      - NEMOTRON_OLLAMA_FALLBACK_EXPECTED_DIGEST=${NEMOTRON_OLLAMA_FALLBACK_EXPECTED_DIGEST:-500a1f067a9f782620b40bee6f7b0c89e17ae61f686b92c24933e4ca4b2b8b41}
+      - NEMOTRON_GCP_ATTEMPT_TIMEOUT_SEC=${NEMOTRON_GCP_ATTEMPT_TIMEOUT_SEC:-60}
+      - NEMOTRON_111_ATTEMPT_TIMEOUT_SEC=${NEMOTRON_111_ATTEMPT_TIMEOUT_SEC:-45}
+      - NEMOTRON_DISPATCH_TOTAL_TIMEOUT_SEC=${NEMOTRON_DISPATCH_TOTAL_TIMEOUT_SEC:-180}
+      - NEMOTRON_MODEL_IDENTITY_TIMEOUT_SEC=${NEMOTRON_MODEL_IDENTITY_TIMEOUT_SEC:-10}
      - NEMOTRON_DECISION_CANARY_TIMEOUT_SEC=${NEMOTRON_DECISION_CANARY_TIMEOUT_SEC:-300}
      # Gemini 只能作緊急備援；即使 .env 有 API key，預設也不得產生付費出站。
      - GEMINI_API_HARD_DISABLED=${GEMINI_API_HARD_DISABLED:-true}
@@ -250,6 +256,12 @@ services:
      - RAG_EMBED_EXPECTED_DIGEST=${RAG_EMBED_EXPECTED_DIGEST:-7907646426070047a77226ac3e684fbbe8410524f7b4a74d02837e43f2146bab}
      - EMBED_CONSISTENCY_TIMEOUT_SEC=${EMBED_CONSISTENCY_TIMEOUT_SEC:-150}
      - NEMOTRON_OLLAMA_EXPECTED_DIGEST=${NEMOTRON_OLLAMA_EXPECTED_DIGEST:-bdbd181c33f2ed1b31c972991882db3cf4d192569092138a7d29e973cd9debe8}
+      - NEMOTRON_OLLAMA_FALLBACK_MODEL=${NEMOTRON_OLLAMA_FALLBACK_MODEL:-qwen3:8b}
+      - NEMOTRON_OLLAMA_FALLBACK_EXPECTED_DIGEST=${NEMOTRON_OLLAMA_FALLBACK_EXPECTED_DIGEST:-500a1f067a9f782620b40bee6f7b0c89e17ae61f686b92c24933e4ca4b2b8b41}
+      - NEMOTRON_GCP_ATTEMPT_TIMEOUT_SEC=${NEMOTRON_GCP_ATTEMPT_TIMEOUT_SEC:-60}
+      - NEMOTRON_111_ATTEMPT_TIMEOUT_SEC=${NEMOTRON_111_ATTEMPT_TIMEOUT_SEC:-45}
+      - NEMOTRON_DISPATCH_TOTAL_TIMEOUT_SEC=${NEMOTRON_DISPATCH_TOTAL_TIMEOUT_SEC:-180}
+      - NEMOTRON_MODEL_IDENTITY_TIMEOUT_SEC=${NEMOTRON_MODEL_IDENTITY_TIMEOUT_SEC:-10}
      - NEMOTRON_DECISION_CANARY_TIMEOUT_SEC=${NEMOTRON_DECISION_CANARY_TIMEOUT_SEC:-300}
      # Gemini 只能作緊急備援；即使 .env 有 API key，預設也不得產生付費出站。
      - GEMINI_API_HARD_DISABLED=${GEMINI_API_HARD_DISABLED:-true}
@@ -322,6 +334,12 @@ services:
      - RAG_EMBED_EXPECTED_DIGEST=${RAG_EMBED_EXPECTED_DIGEST:-7907646426070047a77226ac3e684fbbe8410524f7b4a74d02837e43f2146bab}
      - EMBED_CONSISTENCY_TIMEOUT_SEC=${EMBED_CONSISTENCY_TIMEOUT_SEC:-150}
      - NEMOTRON_OLLAMA_EXPECTED_DIGEST=${NEMOTRON_OLLAMA_EXPECTED_DIGEST:-bdbd181c33f2ed1b31c972991882db3cf4d192569092138a7d29e973cd9debe8}
+      - NEMOTRON_OLLAMA_FALLBACK_MODEL=${NEMOTRON_OLLAMA_FALLBACK_MODEL:-qwen3:8b}
+      - NEMOTRON_OLLAMA_FALLBACK_EXPECTED_DIGEST=${NEMOTRON_OLLAMA_FALLBACK_EXPECTED_DIGEST:-500a1f067a9f782620b40bee6f7b0c89e17ae61f686b92c24933e4ca4b2b8b41}
+      - NEMOTRON_GCP_ATTEMPT_TIMEOUT_SEC=${NEMOTRON_GCP_ATTEMPT_TIMEOUT_SEC:-60}
+      - NEMOTRON_111_ATTEMPT_TIMEOUT_SEC=${NEMOTRON_111_ATTEMPT_TIMEOUT_SEC:-45}
+      - NEMOTRON_DISPATCH_TOTAL_TIMEOUT_SEC=${NEMOTRON_DISPATCH_TOTAL_TIMEOUT_SEC:-180}
+      - NEMOTRON_MODEL_IDENTITY_TIMEOUT_SEC=${NEMOTRON_MODEL_IDENTITY_TIMEOUT_SEC:-10}
      - NEMOTRON_DECISION_CANARY_TIMEOUT_SEC=${NEMOTRON_DECISION_CANARY_TIMEOUT_SEC:-300}
      # Gemini 只能作緊急備援；即使 .env 有 API key，預設也不得產生付費出站。
      - GEMINI_API_HARD_DISABLED=${GEMINI_API_HARD_DISABLED:-true}
--- a/docs/AI_INTELLIGENCE_MODULE_SOT.md
+++ b/docs/AI_INTELLIGENCE_MODULE_SOT.md
@@ -1,19 +1,21 @@
 # PChome 業績成長自動化作戰系統 — AI 競價情報模組 Single Source of Truth

-> **最後更新**: 2026-07-17 (台北時間)
-> **狀態**: 🟠 Partial。Production runtime 仍是 V10.813；V10.814 source 已修正 BGE-M3 冷啟動假陰性並加入 NemoTron decision-only canary，但部署與 production canary 尚未在本段 source truth 中宣稱完成。Windows 99 audit run `534c6208-2088-428c-ba6f-f7a63a67f422` 證明舊 project `astral-gateway-484913-d7` 已為 `DELETE_REQUESTED`、GCP-A `22/11434` 不可達；GCP-B Ollama `0.22.1` 可達，`bge-m3:latest` digest `7907646426070047a77226ac3e684fbbe8410524f7b4a74d02837e43f2146bab` 的 1024 維 no-write probe 成功但冷啟動需 `116.475s`，`qwen3:14b` digest 為 `bdbd181c33f2ed1b31c972991882db3cf4d192569092138a7d29e973cd9debe8`。四 Agent runtime、MCP/RAG flags 與 telemetry 仍沿用 V10.813 production readback；程式、測試、tags 或 digest 可見都不能覆蓋尚未完成的 production closure。
-> **適用版本**: V10.813 production runtime；V10.814 source candidate；AI Agent/MCP/RAG full product integration remains partial
+> **最後更新**: 2026-07-23 (台北時間)
+> **狀態**: 🟠 Partial。Production runtime 是 V10.816、exact Gitea runtime marker `2757585e5e3455e96c743ddfd82d59304f32c3f0`。Windows 99 controlled run `1b335e97-fa31-418b-b925-759bc1036267` 已完成 `20/20` changed-source hash、`30/30` runtime env、三應用健康、`momo-db` identity unchanged、獨立 verifier 與公開 readback。PixelRAG receipt -> exact-digest BGE-M3 -> read-only pgvector canary 實際通過，但 `RAG_ENABLED=false` 仍阻擋 activation。NemoTron 實際走完 GCP-A -> GCP-B -> 111 三候選，由 111 exact-digest `qwen3:8b` 回傳合法 decision envelope，狀態為 `canary_passed_degraded_fallback`；工具、DB、價格、insight 與 Telegram 執行數皆為零。完整 AI Agent/MCP/RAG 產品閉環仍未完成，GCP 容量、RAG activation、MCP runtime 與業績 outcome telemetry 都必須繼續列為缺口。
+> **適用版本**: V10.816 production runtime；AI Agent/MCP/RAG full product integration remains partial

 ---

-## 零之負五、AI Agent 產品整合 truth 與 internal RAG/NemoTron canary（V10.814 source）
+## 零之負五、AI Agent 產品整合 truth 與 internal RAG/NemoTron canary（V10.816 production）

 - `/api/ai-automation/agent-product-integration` 與 `scripts/ops/report_ai_agent_product_integration.py` 分開輸出四 Agent source/scheduler wiring、七日 `ai_calls` 實際呼叫與錯誤率、MCP/RAG telemetry、action plan/outcome、AutoHeal incident retry，以及九階段 closure。只有四 Agent 全部有健康 runtime、MCP/RAG 已啟用且有 telemetry、internal RAG canary 已通過、受控執行/驗證/重試/學習都有實證時才可回 `fully_integrated`。
- `/api/ai-automation/internal-rag-candidate-canary` 只有 GET，永遠是 no-model/no-DB-write readback；production execute 由 `momo-scheduler` 每日 04:45 自動跑一次。V10.814 起 consistency probe 會先由 Ollama `/api/tags` 驗證 manifest digest，再以 canary 專用 `150s` timeout 容納 GCP-B 的實測冷啟動；一般 embedding 呼叫仍保留既有短 timeout cap，不把所有 request 放大成 150 秒。至少一台核准 GCP host 必須同時通過 digest 與 1024 維 embedding，111 不可形成 quorum；單一 GCP 可進 degraded read-only pgvector shadow probe，但 `rag_embedding_redundancy_degraded` 仍阻擋正式 activation。
+- `/api/ai-automation/internal-rag-candidate-canary` 只有 GET，永遠是 no-model/no-DB-write readback；production execute 由 `momo-scheduler` 每日 04:45 自動跑一次。V10.815 production 已使用先由 Ollama `/api/tags` 驗證 manifest digest、再以 canary 專用 `150s` timeout 容納 GCP-B 冷啟動的契約；一般 embedding 呼叫仍保留既有短 timeout cap，不把所有 request 放大成 150 秒。至少一台核准 GCP host 必須同時通過 digest 與 1024 維 embedding，111 不可形成 quorum；單一 GCP 可進 degraded read-only pgvector shadow probe，但 `rag_embedding_redundancy_degraded` 仍阻擋正式 activation。
 - RAG canary 不 INSERT/UPDATE `ai_insights`、`competitor_prices`、`external_offers` 或任何正式價格表；artifact 與 scheduler receipt 共用 `trace_id/run_id/work_item_id`，明確輸出 `transaction_read_only`、similarity、embedding signature、expected/observed digest、GCP reachability、Telegram acknowledgement 與 zero-write/rollback terminal。digest drift、零核准 GCP host、向量維度錯誤、pgvector probe 失敗或 semantic threshold 未達都必須 fail closed。
 - `bge-m3:latest` 只可在 expected digest gate 下使用：source contract 固定 digest，runtime 每次 canary 由 `/api/tags` 比對，漂移時在 embedding/DB 前阻擋。Ollama 官方 `/api/tags` 明確提供模型 `digest`，因此 immutable contract 使用「固定 expected digest + runtime verifier」，不是只信任 floating tag。MCP runtime 仍受 localhost-only、read-only tool contract 與 required secret presence preflight 約束，不可因 registry 已存在就宣稱上線。
- V10.814 新增 `/api/ai-automation/nemotron-decision-canary`、`scripts/ops/run_nemotron_decision_canary.py` 與每日 05:00 scheduler lane。execute 只用 production 共用 qwen3 tool-calling payload產生 synthetic decision，驗證 expected model digest、allowed tool、synthetic SKU 與 post-call digest，固定 `tool_execution_count=0`、`database_call_performed=false`、`writes_price_tables=false`、`writes_ai_insights=false`、`telegram_sent=false`；scheduler 只在 model canary 後發生命週期 acknowledgement 並原子回寫 receipt。
- `AI Agent product integration truth` 會讀取 fresh NemoTron decision-only receipt；這份 shadow canary 能證明模型決策路徑真的執行，但不能冒充正式商品 action、agent outcome 或完整 Controlled Apply。舊的 NemoTron/ElephantAlpha class/method smoke 保留為 source guard，也不能單獨代表產品整合完成。
+- V10.816 production 提供 `/api/ai-automation/nemotron-decision-canary`、`scripts/ops/run_nemotron_decision_canary.py` 與每日 05:00 scheduler lane，且 production dispatcher 與 canary 共用 `services/nemotron_runtime_candidate_service.py`：GCP-A/GCP-B 必須使用 exact-digest `qwen3:14b`，每台 chat 最多 60 秒；最終 111 fallback 必須使用 exact-digest `qwen3:8b`，最多 45 秒並固定 `num_ctx=4096`、`num_predict=512`，所有 payload 固定 `think=false`。Production Ollama + NIM 共用單一 180 秒 monotonic deadline，canary 的 `timeout_sec` 也是整次執行上限，不再對每個候選重新計時。每次呼叫前後都驗證 digest；畸形 `/api/tags` payload/model/details schema 必須 fail closed 並轉下一候選，不能中斷 fallback。實際 model/provider/host/fallback/attempts 必須進 logger、通知 footprint 與 receipt。
+- 模型輸出必須在選定候選前通過 deterministic tool contract：工具在 allowlist、必要欄位與型別正確、SKU 必須來自本次輸入、每個 SKU 恰好一個 call 且全數覆蓋。Production 去重使用 `services/nemotron_dispatch_reservation_service.py` 在三容器共用 `/app/data` 上執行 `flock`、file fsync、atomic replace 與 parent-directory fsync 的 process/container-shared ownership-token store；state 只保存 SHA-256 SKU key、owner token、phase 與 expiry，不保存商品名稱或原始資料。In-flight lease 必須涵蓋最大模型 deadline；每個 handler 前須由原 owner 原子寫入 `side_effect_started` 並切換為四小時 crash quarantine，寫入失敗就不得執行副作用。舊 token 不可 release/commit 新 owner，state/lock 無法讀寫、schema 損壞或平台缺少 process-shared lock 時 fail closed。本地非 production 測試才使用記憶體 backend。告警工具必須有 EventRouter delivered/durable-queue acknowledgement，推薦工具必須有 DB write 或 durable notification，KM 工具必須有持久化 insight，才可提交四小時 TTL；若 durable side effect 已完成但 dedupe commit 未驗證，結果仍按已派發計數、輸出 `dedupe_commit_unverified`，並保留 quarantine，絕不可 release 後重送。模型契約錯誤、reservation ownership lost、boundary 未建立或 handler 未回 durable outcome 時才 release 自己持有的 lease。程序若在 boundary 後、side effect 前崩潰，最多抑制該 SKU 四小時；這是避免重複通知/寫入的 at-most-once 取捨，必須由 receipt/告警揭露而非靜默重試。
+- Decision-only canary 只產生 synthetic decision，並共用 production `_validate_tool_call_contract()` 驗證完整 call set，不可只看第一筆；多餘、重複、未知、缺欄位或未綁定 SKU 的 call 都必須 fail closed。Canary 同時驗證 post-call digest，固定 `tool_execution_count=0`、`database_call_performed=false`、`writes_price_tables=false`、`writes_ai_insights=false`、`telegram_sent=false`。使用 111 成功時狀態必須是 `canary_passed_degraded_fallback`，不能偽裝成 GCP healthy；scheduler 只在 model canary 後發生命週期 acknowledgement 並原子回寫 receipt。111 的 `num_ctx=4096`、`num_predict=512` 是不可被 env 弱化的固定 NemoTron 契約；候選失敗鏈只輸出 label/tier/model/status/digest-match/error-class 的 privacy-safe footprint。
+- `AI Agent product integration truth` 會讀取 fresh NemoTron decision-only receipt；這份 shadow canary 能證明模型決策路徑真的執行，但不能冒充正式商品 action、agent outcome 或完整 Controlled Apply。V10.816 run `1b335e97-fa31-418b-b925-759bc1036267` 已取得 fresh degraded-fallback receipt 與獨立零寫入 readback，因此 model-aware dispatcher 可標記為 production verified；四 Agent 業績 outcome、MCP runtime 與 RAG activation 仍保持 partial。
 - V10.812 起四 Agent activity 依角色證據判定：Hermes/OpenClaw 仍以實際 `ai_calls` 為主，NemoTron 可加入其派發/insight durable evidence，ElephantAlpha 可由其 `ai_insights`/action plan 證明 active；沒有成功 call 或 verified executed action 時只能是 `runtime_active_unverified`，不得把 deterministic artifact 數量包裝成 healthy integration。

 ---
@@ -59,7 +61,7 @@
 - `services/security_governance_review_service.py`、`scripts/ops/report_security_governance_review.py` 與 `/api/ai-automation/security-governance-review` 提供 machine-readable checks、completion、release gate 與 ordered work items。
 - Gitea CD 必須執行 `report_security_governance_review.py --strict`，且不得解析 GitHub action/source/image。
 - production 安全與治理完成度必須分開顯示 program、asset coverage、runtime closure；目前結論是 partial，不是 complete。
- 目前 ordered P0 以 `docs/guides/ai_automation_mainline_work_items.md` 為準：SEC-P0-001 access exposure 已關閉，current P0 是 identity/RBAC，之後依序為 webhook trust、supply chain、asset reconciliation、controlled-apply envelope、internal RAG canary 與 MCP/RAG runtime closure。
+- 目前 ordered P0 只以 `docs/guides/ai_automation_mainline_work_items.md` 為準；current P0 是 `GROWTH-P0-001 comparison coverage truth + autonomous refresh`，安全、治理、RAG/MCP 與 release 缺口依該表固定次序推進，不得用單一 canary 或 source-ready 狀態改寫優先順序。
 - V10.786 起 `/metrics` 不再是 public exception；應用中央政策與 Nginx exact-match edge policy 僅允許 loopback / Docker transport，Prometheus 必須以內部 canary 證明 target `up`，且 public/LAN readback 必須拒絕。監控 Compose 不得保存明文 Grafana 管理密碼，Prometheus 對外埠預設只綁 loopback。
 - V10.787 依 live ingress 修正 metrics transport：production ingress 實際位於 188，110 global Prometheus 只能經專用 `172.20.0.1:19191` Docker-only Nginx bridge 抓取；bridge 以系統 CA、`mo.wooo.work` SNI/Host 轉送至 188，應用僅額外允許 `192.168.0.110/32` 單一監控主機，不得放行整段 LAN，也不得把 110 的其他 vhost 誤當 EwoooC production ingress。controlled apply 支援新 include 建立、Prometheus bind-mount reconciliation、`momo_app_info`/health/database product marker canary、精確 scrape URL、worker reload convergence 與失敗自動移除/回滾；production 已讀回 public 404、internal 200、target up、Prometheus identity preserved、8/8 anonymous matrix 與 `momo-db` unchanged，證據在 `governance/evidence/SEC-P0-001-20260711T122758Z.json`。
 - V10.788 把既有 `users/login_history/user_permissions` 正式接入登入：`auto` 模式先以 hybrid 維持回滾能力，只計 active admin 的 database-auth success durable receipts，預設連續兩次後自動轉為 database-only；legacy shared Session 下一次請求自動撤銷。鎖定改由 `login_history` 跨 worker 持久化，Session 綁定 password/role/active identity version，中央 policy 區分 read/operate/admin，且只有 allowlisted direct proxy peer 才能提供 `X-Forwarded-For/X-Real-IP`。帳號建立、身份更新、密碼變更、停用與權限異動會在同一交易寫入不含密碼的 audit event，commit 失敗時兩者一併 rollback。`/api/auth/governance` 與 `report_auth_identity_governance.py` 僅輸出非機密 readback；source ready 不等於 production cutover completed。
@@ -70,6 +72,7 @@
 ## 零、LLM 路由紅線（2026-05-12）

 - 所有 AI Agent、LLM 推理與 embedding 預設必須走 Ollama 三主機級聯：GCP-A `34.87.90.216:11434` → GCP-B `34.21.145.224:11434` → 111 `192.168.0.111:11434`。
+- NemoTron decision runtime 是 model-aware cascade，不得把同一 floating model tag盲送三台：GCP-A/GCP-B 鎖 `qwen3:14b` exact digest，111 最終備援鎖 `qwen3:8b` exact digest與較小輸出上限；GCP capacity degraded 時可維持自架決策可用性，但必須在 receipt/UI 明示 degraded fallback，並持續保留恢復 GCP capacity 的 P0。
 - `services/ollama_service.resolve_ollama_host()` 是主機解析契約；`OLLAMA_HOST`、`HERMES_URL`、`EMBEDDING_HOST`、`OLLAMA_API_BASE` 只接受 GCP-A / GCP-B / 111 或 110 的核准轉發端口。
 - 188 直連 GCP-A / GCP-B timeout 時，resolver 可先使用同順位 110 proxy rescue：GCP-A direct → `192.168.0.110:11435` → GCP-B direct → `192.168.0.110:11436` → 111。proxy rescue 只是同一順位的可用入口，不代表 GCP direct host 已恢復。
 - `OLLAMA_RESOLVE_HOST_HEALTH_SKIP_ENABLED=true` 時，resolver 會讀最近 `host_health_probes`；若 direct GCP-A/GCP-B 在視窗內已被判定不健康，會直接略過該 direct endpoint，先試同順位 proxy rescue，避免每 120 秒 cache refresh 都等待 direct timeout。此 skip 只套用 direct GCP，不套用 110 proxy。
@@ -280,7 +283,7 @@ SQL漏斗(~300筆)
 | 角色 | 模型 | 主機 | 成本 | 每日限額 |
 |------|------|------|------|---------|
 | Hermes 分析師 | hermes3:latest / bge-m3 | GCP-A → GCP-B → 111 Ollama | 零 | 無限 |
-| NemoTron 派發器 | qwen3:14b；111 fallback 降級 llama3.2；NIM fallback | GCP-A → GCP-B → 111；NVIDIA NIM 備援 | Ollama 零；NIM 配額內免費 | NIM 80 |
+| NemoTron 派發器 | GCP exact-digest qwen3:14b；111 exact-digest qwen3:8b；NIM fallback | GCP-A → GCP-B → 111；NVIDIA NIM 備援 | Ollama 零；NIM 配額內免費 | NIM 80 |
 | OpenClaw 策略師 | qwen2.5-coder:7b / qwen3:14b；111 fallback 降級 llama3.2 | Ollama-first；Gemini emergency fallback only | Ollama 零；Gemini 預設封鎖 | — |
 | ElephantAlpha 編排者 | ElephantAlpha | 依部署環境 | 受控 | AI 例外決策 / 任務制 |

--- a/docs/guides/ai_automation_mainline_work_items.md
+++ b/docs/guides/ai_automation_mainline_work_items.md
@@ -1,6 +1,6 @@
 # AI Automation Mainline Work Items

-> Updated: 2026-07-17 10:28 Asia/Taipei
+> Updated: 2026-07-23 Asia/Taipei
 > Governance: `global_product_governance_v2` + ADR-038
 > Current P0: `GROWTH-P0-001 comparison coverage truth + autonomous refresh`

@@ -17,16 +17,16 @@
 | Order | ID | Status | Work item | Exit evidence / next machine action |
 |---:|---|---|---|---|
 | 1 | `SEC-P0-001` | Completed | Deny-by-default route access control | `governance/evidence/SEC-P0-001-20260711T122758Z.json` plus the current runtime receipt prove anonymous matrix 8/8 denied, public `/metrics` 404, exact internal target up, EwoooC product markers present, Prometheus identity preserved and `momo-db` unchanged. |
-| 2 | `GROWTH-P0-001` | In progress (`runtime_partial`) | Comparison coverage truth + autonomous refresh | V10.813 is live and retains the growth runtime introduced at exact Gitea object `2647632660673e9f1533e61922c96e4fb4adcb40`. Controlled run `34999f5314054f09909663b46fc42ee2` scanned 20, verified/wrote/read back one exact offer and raised the fixed cohort to `25 ready + 1 candidate validation + 24 unmatched`, `50%` count and `NT$211,667 / NT$354,062 = 59.782%` revenue. Follow-up run `e1508ec3a49241d1965bc99d428b0325` found 19 candidates but safely wrote zero because none passed strict identity/variant/unit verification; its independent terminal is `degraded_no_safe_candidate`. Yahoo remains durably `active + enabled + write_enabled`, and the durable readback proves `formal_source_activated=true` even when the latest run is verified no-write. Formal runtime remains `2/15`. Next: retry the unresolved revenue-weighted batch only with fresh source evidence, then add the next approved structured marketplace adapter without relaxing promotion gates. |
+| 2 | `GROWTH-P0-001` | In progress (`runtime_partial`) | Comparison coverage truth + autonomous refresh | V10.815 is live and retains the growth runtime introduced at exact Gitea object `2647632660673e9f1533e61922c96e4fb4adcb40`. Controlled run `34999f5314054f09909663b46fc42ee2` scanned 20, verified/wrote/read back one exact offer and raised the fixed cohort to `25 ready + 1 candidate validation + 24 unmatched`, `50%` count and `NT$211,667 / NT$354,062 = 59.782%` revenue. Follow-up run `e1508ec3a49241d1965bc99d428b0325` found 19 candidates but safely wrote zero because none passed strict identity/variant/unit verification; its independent terminal is `degraded_no_safe_candidate`. Yahoo remains durably `active + enabled + write_enabled`, and the durable readback proves `formal_source_activated=true` even when the latest run is verified no-write. Formal runtime remains `2/15`. Next: retry the unresolved revenue-weighted batch only with fresh source evidence, then add the next approved structured marketplace adapter without relaxing promotion gates. |
 | 3 | `SEC-P0-002` | In progress (`canary_ready`) | Database identity + least-privilege RBAC | V10.789 is live and `governance/auth_identity_runtime_receipt.json` verifies required tables, two active admins, durable lockout, session revocation, trusted proxy policy and no-secret mutation audit readiness. Runtime is intentionally hybrid because zero database-admin success receipts have been captured; `auto` retires shared authority after two durable successes without a manual review gate. Next: capture database-admin login receipts and verify automatic database-only cutover. |
 | 4 | `SEC-P0-003` | In progress | Webhook trust and replay protection | Telegram secret-token verification code exists; production secret activation remains unproven. Exit: secret provisioned outside source, required mode enabled, invalid-secret 401 and authorized callback canary pass. |
 | 5 | `SUPPLY-P0-001` | In progress | Gitea-only secure software supply chain | Gitea-native checkout, secret-safe `.dockerignore`, commit-bound source receipt and governance gate are active. Exit: exact dependency lock, internal SAST/SCA/secret scan, SBOM, image digest/provenance, vulnerability SLA and production digest readback. |
 | 6 | `GOV-P0-001` | In progress | Canonical full asset graph + runtime reconciliation | `governance/ewoooc_asset_inventory.json` seeds hosts, services, data, AI, routes, supply chain, observability and recovery. Exit: same-run probe receipt for every asset; drift auto-creates work items. |
 | 7 | `GOV-P0-002` | Not started | Unified controlled-apply envelope | Introduce one `trace_id/run_id/work_item_id` across sensor, identity, SOT diff, decision, risk, dry-run, execution, verifier, rollback/retry and learning acknowledgement. Start with EventRouter + AutoHeal. |
-| 8 | `RAG-P0-001` | In progress (`source_fix_ready_runtime_canary_pending`) | Internal RAG candidate canary + NemoTron decision-only proof | V10.813 production still preserves the prior fail-closed receipt. Windows 99 audit run `534c6208-2088-428c-ba6f-f7a63a67f422` now proves the old GCP project is `DELETE_REQUESTED`, GCP-A is unreachable, and GCP-B Ollama `0.22.1` has exact `bge-m3` / `qwen3:14b` digests. Its bounded BGE-M3 probe returned 1024 dimensions with zero DB/app writes but required `116.475s`, proving the 10-second consistency timeout was a false-negative source defect. V10.814 source raises only the canary cold-start cap to 150 seconds, verifies `/api/tags` digest before embedding, permits one digest-verified GCP host as degraded shadow quorum while excluding 111, and adds a daily 05:00 NemoTron model decision with zero tool/DB/price/insight writes. Exit: deploy V10.814, pass both production canaries with Telegram/durable receipts, retain redundancy blocker until a canonical replacement GCP-A asset exists, then controlled `RAG_ENABLED` shadow activation plus query/hit/feedback telemetry. |
+| 8 | `RAG-P0-001` | In progress (`runtime_verified_degraded_fallback_activation_blocked`) | Internal RAG candidate canary + NemoTron decision-only proof | V10.816 is live at exact runtime marker `2757585e5e3455e96c743ddfd82d59304f32c3f0`. Windows 99 run `1b335e97-fa31-418b-b925-759bc1036267` verified PixelRAG receipt -> exact-digest BGE-M3 -> read-only pgvector with `ready_count=1`, `canary_passed_activation_blocked` and zero business writes. The same run exercised all three bounded NemoTron candidates and returned `canary_passed_degraded_fallback` from 111 exact-digest `qwen3:8b`; model identity, deterministic one-call-per-SKU envelope and zero tool/DB/price/insight/Telegram execution passed. App and scheduler independently passed the shared reservation canary, cross-container visibility passed, and the four-hour `side_effect_started` quarantine is active. Final terminal: `v10816_source_runtime_rag_nemotron_verified_no_db_mutation`. Next: restore GCP-A reachability and GCP-B bounded inference capacity, then run controlled `RAG_ENABLED` shadow activation with query/hit/feedback telemetry before any formal product write. |
 | 9 | `MCP-P0-001` | In progress (`federation_source_ready`) | MCP/RAG production runtime closure | V10.796 source adds a strict public aggregate receipt for canonical `ewoooc` and `momo-pro-system` identities without opening authenticated internal APIs or exposing endpoint/tool payload data. Exit still requires V10.796 production `/health`, two fresh AWOOOI durable receipts with fingerprint recompute, live MCP servers/router/RAG, approved caller/tool boundary and production query canary. Current source readiness must not be reported as runtime closure. |
 | 10 | `SEC-P0-004` | Not started | Security operations lifecycle and metrics | Add durable security incident state and publish MTTA, MTTR, recurrence, false positive, human intervention, verifier pass, rollback and freshness. Exit: detect-to-learn production receipt. |
-| 11 | `REL-P0-001` | In progress (`runtime_verified_cd_degraded`) | Formal deploy and visible proof discipline | Production V10.813 runs exact Gitea object `a193acc407ef61b9cc6340e8e11aa9d0fb9a39c5`; dev merge `715f90133101e7fdcac243bf999aef05f80e7401` carries the same source. Host 110 still has no matching EwoooC runner, so formal CD remains unavailable and is not replaced by fallback evidence. Windows 99 controlled run `2ef0a084-7aa8-4748-a0db-127fb623d75d` deployed exact archive SHA-256 `65369d5f78c56376bffe8606adc790c046274dc070dfef9b933a6140cd83cc66`; rollback/evidence is retained at `/home/ollama/momo-deploy-backups/ewoooc-20260716T190911Z-a193acc-2ef0a084`. Internal/external `/health` is healthy at V10.813, all 8 source hashes match, app/scheduler/bot identities and immutable `momo-db` ID `cd092451cb5fd555d0ffff70642e109f3b742882c418beeab631793d1e9dc55d` remained unchanged. Public login visibly reports V10.813; authenticated cockpit visual proof remains unavailable in the current browser session. |
+| 11 | `REL-P0-001` | In progress (`v10816_runtime_verified_cd_runner_gap`) | Formal deploy and visible proof discipline | Production V10.816 runs exact Gitea runtime marker `2757585e5e3455e96c743ddfd82d59304f32c3f0`; its runtime tree was integrated to `main`, feature branch and `dev` before controlled deployment. Windows 99 run `1b335e97-fa31-418b-b925-759bc1036267` verified `20/20` changed source hashes and `30/30` runtime env contracts, recreated only app/scheduler/bot, preserved `momo-db`, passed independent runtime/public readback and finalized the RAG/NemoTron canaries. Rollback is retained at `/home/ollama/momo-deploy-backups/ewoooc-20260723T011250Z-2757585-v10816-1b335e97`; public `/health` reports V10.816 and the exact sales-analysis URL reaches the normal `/login` boundary. Host 110 still lacks a matching EwoooC runner, so formal Gitea CD remains an explicit gap rather than being conflated with the verified Windows 99 release. |

 ### GROWTH-P0-001 Fixed Execution Lanes

@@ -36,7 +36,7 @@ These lanes are one ordered current P0, not optional side work. They must advanc
 |---|---|---|---|
 | A. Sales freshness | In progress (`sla_runtime_closed_source_redundancy_partial`) | Latest sales date `2026-07-13`; before the `2026-07-15 20:00` cutoff, raw lag is `2` but SLA lag is `0`, state is `grace` and decisions remain released. Scheduler receipt `0f24219e0bbb4740b7ad6645e7952296` and explicit canary receipt `6e2df008f65544fe8e557c11ff0dbb93` both persisted `completed_no_write`; durable decision is `no_candidate_fresh_no_write`. Live and persisted readiness now agree at Google Drive `1/4`; HTTPS, IMAP and local remain disabled. | Continue automatic report-arrival reconciliation. At/after 20:00 require `2026-07-14` or automatically block decision use, emit the bounded upstream action and verify the next arrival receipt; keep source redundancy partial until another approved source is live. |
 | B. Verified same-item evidence | In progress | TOP50 fixed cohort: `25` verified, `1` candidate/source validation, `24` unmatched. Count coverage is `50%`; revenue-weighted coverage is `NT$211,667 / NT$354,062 = 59.782%`. V10.810 preserves fingerprint `7b74504fcc1e1801c2ca2b42`; run `34999f5314054f09909663b46fc42ee2` added one independently verified offer (`+2.0pp` count, `+1.89pp` revenue), while run `e1508ec3a49241d1965bc99d428b0325` correctly ended no-write. Their durable artifact SHA-256 values are `994beba6dfa70ef8c50031f130916ee155657dd77ac8a2f59b6530b491d45d1e` and `e58be7bb07599871a8f318ad63885bd868c786b0144b81ba0980ae80cf556541`. | Retry unresolved candidates only after fresh evidence arrives, preserve deterministic identity/unit/variant gates, and publish count/revenue deltas against the same fingerprint per run. |
-| C. Platform runtime coverage | In progress (`runtime_canary_activated`) | V10.813 is live. Yahoo is durably active and exact-offer canary readback remains valid; an already-active canary now returns `already_active_verified`, `state_changed=false` and `writes_database_count=0`, while latest no-write receipts no longer erase durable activation truth. Formal runtime is `2/15`; PixelRAG remains evidence-only. | Continue bounded refresh on schedule, monitor expiry/recurrence/rollback signals, then implement the next approved structured source contract for Shopee, Coupang, ETMall, Friday or Rakuten without treating blocked pages as product data. |
+| C. Platform runtime coverage | In progress (`runtime_canary_activated`) | V10.815 is live. Yahoo is durably active and exact-offer canary readback remains valid; an already-active canary returns `already_active_verified`, `state_changed=false` and `writes_database_count=0`, while latest no-write receipts do not erase durable activation truth. Formal runtime is `2/15`; PixelRAG remains evidence-only. | Continue bounded refresh on schedule, monitor expiry/recurrence/rollback signals, then implement the next approved structured source contract for Shopee, Coupang, ETMall, Friday or Rakuten without treating blocked pages as product data. |

 ### AI Agent Product Integration Acceptance

@@ -45,8 +45,8 @@ This is an acceptance surface inside the current growth P0; it does not reorder
 | Layer | Current status | Exit evidence |
 |---|---|---|
 | Source and scheduler wiring | Source ready (`4/4`) | Hermes, NemoTron, OpenClaw and ElephantAlpha source markers plus scheduler ownership are machine-read and reported separately from runtime. |
-| Agent runtime activity | Production partial (`3/4` role-active; `0/4` healthy); V10.814 source canary ready | V10.813 live readback preserves Hermes/OpenClaw call evidence, zero NemoTron runtime activity and ElephantAlpha durable role-owned insights. V10.814 adds a qwen3 decision-only canary that uses the production payload and exact digest but stops before every tool/action/data write; a fresh pass proves shadow model execution only, not formal product action. Exit requires all four role-active and healthy in the bounded window, without treating class presence, configured fallback or shadow canary as an executed business outcome. |
-| MCP/RAG dependency | Runtime disabled / telemetry empty | Production currently reports `MCP_ROUTER_ENABLED=false`, `RAG_ENABLED=false`, zero `mcp_calls` and zero `rag_query_log` activity. Exit requires enabled approved routes, live health, non-zero agent/product telemetry and the internal RAG candidate canary. |
+| Agent runtime activity | Production partial; bounded fallback runtime verified | V10.816 production exercised the actual model-aware dispatcher: GCP-A -> GCP-B -> 111, with 111 exact-digest `qwen3:8b` returning a valid decision envelope after three attempts. This proves decision-only fallback execution and shared dedupe boundaries, not a formal business action or four-Agent closure. Exit requires all four Agents role-active and healthy in the bounded window with linked product outcomes. |
+| MCP/RAG dependency | RAG canary verified; activation and MCP runtime still blocked | Production RAG executed one read-only candidate canary successfully, but `RAG_ENABLED=false` continues to block activation; MCP router remains disabled and non-zero product telemetry is still absent. Exit requires approved routes enabled, live MCP health, non-zero agent/product telemetry, rollback coverage and a fresh internal RAG shadow receipt. |
 | Controlled automation closure | Runtime partial | `/api/ai-automation/agent-product-integration`, CLI and smoke must report Detect -> Normalize -> Correlate -> Decide -> Check -> Controlled Apply -> Verify -> Retry/Rollback -> Learn/Writeback. Completion requires bounded execution, linked outcome/incident verification and durable learning evidence; aggregate source presence is insufficient. |

 ### Analytics Period-Linkage Closure
@@ -57,8 +57,8 @@ This bounded interruption is closed and control returns to `GROWTH-P0-001` witho
 |---|---|---|
 | Program | Completed for the four primary analysis tabs | Daily sales, sales analysis, growth analysis and monthly summary now share one canonical day/month/range contract; cross-tab links preserve the selected period. |
 | Asset coverage | `4/4` pages and `2/2` sales async APIs verified | Daily KPI/calendar/charts/Top 10, growth KPI/series, monthly KPI/tables/charts and sales KPI/charts/YoY/detail table all use the active period. Historical current-snapshot mixing and blank monthly charts are replaced by explicit honest states. |
-| Runtime closure | Completed in V10.810 at `911393190ded015e384e438c26b68faf50ec260c` | `2026-04` daily stays inside April and bounds extreme DoD/WoW values with traceable markers; growth single-point series remain visible; sales consumes the real `chart_values` payload and aggregates repeated SKU rows before ranking; monthly requests are abortable/deduplicated and publish explicit loading/error/ready states. External HTTPS serves the exact tested JS/CSS checksums and internal/external `/health` report V10.810. |
-| Verification | Passed; formal CD still degraded | Final regression: `2,123 passed / 9 skipped / 0 failed`; focused analytics contracts: `49 passed`. The automated visual guard passed `12/12` combinations (`4` pages x desktop/tablet/mobile), covering `126` chart-target observations, period links, actual pixels/elements, honest empty states, date-tick density, console errors and overflow. Formal CD remains degraded because no matching runner executed the release. |
+| Runtime closure | Completed in V10.815 at `283c8c80c631f5d97315708885413b62ee5a34ea` | The exact `/sales_analysis` URL with blank date fields now resolves one canonical period and applies the same metric/range/filter contract to KPI, charts, YoY and detail rows. Query, chart and export logic are split into dedicated services; repeated SKU rows are aggregated before ranking, date/range links preserve state, and invalid filters fail safely. Windows 99 independent readback verified all 36 target files, runtime env, public `/health` and exact JS/CSS hashes. |
+| Verification | Passed; authenticated visual proof remains pending | V10.815 broad regression: `2,174 passed / 9 skipped / 0 failed`; focused analytics batches: `173 passed` and `70 passed`, plus Python compile, Node syntax, Jinja render and diff checks. Public target route returns the expected login redirect and public assets match SHA-256 `6bb4a985744917ef2243d6d055c85b6803b2baf5c571580f49ef69c536ccdf04` / `d98654ddaac10a49af2351750f350bbbcfc3491d62b14c8a25d02eb4d7803186`. The current browser session cannot provide authenticated chart screenshots, so that evidence is not claimed. |

 ## P1

@@ -69,10 +69,10 @@ This bounded interruption is closed and control returns to `GROWTH-P0-001` witho
 | 14 | `APPSEC-P1-001` | In progress | CSP and DOM/XSS hardening | Security headers are present and CSP is report-only. Collect violations, remove high-risk `innerHTML`/inline sinks, then enforce CSP by canary. |
 | 15 | `APPSEC-P1-002` | Not started | Unsafe shared-cache serialization removal | Replace writable pickle caches in dashboard/daily-sales/EDM/sales with constrained JSON or signed schema. |
 | 16 | `ARCH-P1-001` | In progress | Split oversized policy/executor/verifier modules | Current top debts include 44k-line PChome mapping and 14k-line smoke service. Split by bounded family and independent tests. |
-| 17 | `UX-P1-001` | In progress | Professional full-site UI/UX | V10.810 closes chart rendering defects across all four primary analysis tabs: real payload values render, single points stay visible, extreme percentages no longer flatten normal data, dense date labels auto-skip, Top 50 vendor content uses bounded zoom, and loading/error/empty states are explicit. The `12/12` desktop/tablet/mobile visual matrix has zero horizontal overflow. The broader site-wide first-viewport, progressive-disclosure, accessibility and loading/error/degraded-state audit remains in progress. |
+| 17 | `UX-P1-001` | In progress | Professional full-site UI/UX | V10.815 closes the exact sales-analysis period/filter linkage and keeps the prior four-tab rendering guards: real payload values, visible single points, bounded extreme percentages, date-label auto-skip, zoom bounds and explicit loading/error/empty states. Public assets are hash-verified; authenticated V10.815 visual proof and the broader site-wide first-viewport, progressive-disclosure, accessibility and loading/error/degraded-state audit remain in progress. |
 | 18 | `PIXELRAG-P1-001` | Not started | Ollama-first multimodal embedding benchmark | Verify approved visual embedding on GCP-A -> GCP-B -> 111 and design pgvector-compatible visual metadata; FAISS remains disallowed without ADR. |
 | 19 | `MARKET-P1-001` | In progress | Marketplace source contracts | Yahoo Shopping remains active in V10.810 production with public-boundary allowlists, bounded streaming/rate, provenance, current product-detail readback, stock/spec/variant guards, source-specific promotion partition, idempotent exact canary activation and durable activation readback across no-write runs. Four fresh verified Yahoo offers now contribute formal evidence across completed batches; non-exact and unit-price candidates do not. Shopee, Coupang, ETMall, Friday and Rakuten still require equivalent structured contracts, and blocked pages remain non-product data. |
-| 20 | `QA-P1-001` | In progress | Deterministic test and CI governance | V10.810 final broad regression is `2,123 passed / 9 skipped / 0 failed`; analytics contracts are `49 passed`, and the new runtime chart guard passes `12/12` page/viewport cases with actual pixel/element and console assertions. Production host parity is `18/18` files and public HTTPS parity is `9/9` key JS/CSS assets. No matching EwoooC runner executed this release, so CI/CD parity is not claimed and exact-object fallback receipts remain separate evidence. |
+| 20 | `QA-P1-001` | In progress | Deterministic test and CI governance | V10.816 broad regression is `2,223 passed / 15 skipped / 0 failed`; focused model/dedupe regression is `93 passed`, and independent ninth-round review found no material issue. Production parity is verified for all `20/20` changed source hashes and `30/30` runtime env contracts, with an independent Windows 99 verifier plus RAG/NemoTron canaries. No matching EwoooC runner executed V10.816, so formal Gitea CI/CD remains missing even though the bounded production release is verified. |

 ## P2

@@ -97,7 +97,7 @@ These are reusable foundations, not proof that the full program is complete.
 | Completed | PromotionGate replay | No production write. |
 | Completed | Embedding-signature guard replay | Signature readiness only. |
 | Completed | Candidate knowledge replay | Internal RAG preview only; no DB/model call. |
-| Source ready; production pending | Internal RAG + NemoTron decision canaries | V10.814 source separates cold-start timeout from normal embedding latency, enforces exact BGE-M3/qwen3 digest readback, permits one approved GCP host only as degraded shadow quorum, and adds zero-tool/zero-data-write NemoTron runtime proof. Production canary receipts, replacement GCP-A control-plane identity, RAG shadow activation and query/outcome telemetry remain P0. |
+| Production verified; degraded fallback | Model-aware NemoTron dispatcher fallback | V10.816 production uses one modular exact-digest candidate registry for production and canary: GCP-A/GCP-B `qwen3:14b` with 60-second attempts, then 111 `qwen3:8b` with 45 seconds, `think=false`, fixed `num_ctx=4096` / `num_predict=512` and one total deadline. Run `1b335e97-fa31-418b-b925-759bc1036267` passed the shared app/scheduler and cross-container reservation canaries, then produced a valid decision from `ollama_111_fallback` after three attempts with no tool or business-data writes. Privacy-safe state hashes SKU values and the four-hour `side_effect_started` quarantine remains fail-closed after durable-effect uncertainty. GCP-A replacement, GCP-B capacity and controlled RAG activation remain unresolved and are not hidden by the fallback. |
 | Completed | PixelRAG application portfolio | Commerce/RAG/UX/ops/marketing/governance inventory. |
 | Completed | Ollama-first VLM route readiness and replay worker | Evidence-bound artifact output; no direct price write. |
 | Completed | Platform probe worker | Shopee/Coupang barriers become structured fallback/backoff receipts. |
--- a/docs/memory/code_modularization_inventory_20260430.md
+++ b/docs/memory/code_modularization_inventory_20260430.md
@@ -74,6 +74,8 @@
 - 2026-07-14 追記：`services/pchome_revenue_growth_service.py` 因新增可稽核比價覆蓋契約與 active-catalog scope 增至 1,366 行；此輪先完成 P0 正確性，後續 `ARCH-P1-001` 應拆出 catalog coverage query 與 metric-contract builder，主 service 保留 orchestration。
 - 2026-07-15 追記：`services/pchome_growth_same_item_reconciliation.py` 已達 880 行；同商品 identity verifier、exact DB readback、coverage post-verifier 與 durable receipt persistence 應在 `ARCH-P1-001` 拆成獨立 policy/verifier/repository，主模組只保留 bounded orchestration。
 - 2026-07-17 追記：Nemotron decision-only canary 的排程執行、Telegram acknowledgement 與 durable receipt 終局已移至 `services/nemotron_decision_canary_scheduler_task.py`（120 行）；`run_scheduler.py` 僅保留薄委派與排程註冊，清冊同步為 1,684 行，下一步仍依序拆 task registration 與 runtime startup。
+- 2026-07-22 追記：業績分析的 canonical query、metric aggregate、Other-category contract 與 period-linked Excel export 已抽到 `services/sales_analysis_query_service.py`（391 行）及 `services/sales_analysis_export_service.py`（207 行）；`routes/sales_routes.py` 降為 2,954 行，後續繼續拆 page context 與 legacy pandas API。
+- 2026-07-22 追記：V10.816 將 exact-digest model candidate/identity policy 抽成 `services/nemotron_runtime_candidate_service.py`（234 行）、process/container-shared ownership-token lease 與 side-effect crash quarantine 抽成 `services/nemotron_dispatch_reservation_service.py`（376 行），decision-only runtime proof 維持在 `services/nemotron_decision_canary_service.py`（604 行）；`services/nemoton_dispatcher_service.py` 因補齊共用 tool contract、durable outcome、commit-failure quarantine 與 privacy-safe fallback footprint增至 2,864 行。此輪先完成 P0 自動化正確性；後續 `ARCH-P1-001` 應依序抽出 delivery outcome adapter、tool execution coordinator 與 footprint renderer，dispatcher 只保留 orchestration。

 ## 達到或超過 800 行檔案清單

@@ -131,12 +133,12 @@
 | 4158 | `routes/admin_observability_routes.py` | P0 觀測台巨型 Blueprint | query、action、render context 與 route glue 分離 |
 | 3763 | `services/competitor_price_feeder.py` | P0 feeder 巨型 service | acquisition、matching、refresh、persistence、receipt 分離 |
 | 3322 | `routes/dashboard_routes.py` | P0 Dashboard Blueprint | query、decision projection、review action 分離 |
-| 3290 | `routes/sales_routes.py` | P0 Sales Blueprint | chart、query、calendar、export 分離 |
+| 2954 | `routes/sales_routes.py` | P0 Sales Blueprint、拆分進行中 | canonical query 與 export 已外移；下一步拆 page context、calendar 與 legacy pandas API |
 | 3112 | `scheduler.py` | P0 排程總管 | task registry 與 domain jobs 分離 |
 | 2961 | `services/openclaw_strategist_service.py` | P1 strategist | prompt、query、report、notification 分離 |
 | 2383 | `services/competitor_intel_repository.py` | P1 repository | query、decision envelope、UI projection、cache 分離 |
 | 2332 | `services/elephant_alpha_autonomous_engine.py` | P1 engine | trigger、planner、executor、notification 分離 |
-| 2166 | `services/nemoton_dispatcher_service.py` | P1 dispatcher | model client、tool parser、decision、delivery 分離 |
+| 2864 | `services/nemoton_dispatcher_service.py` | P1 dispatcher | 已抽 exact-digest candidate registry 與 process-shared lease/crash-quarantine store；下一步拆 delivery outcome adapter、tool execution coordinator、footprint renderer |
 | 2011 | `services/external_market_offer_service.py` | P1 market offers | source registry、normalizer、sync、review、readiness 分離 |
 | 1917 | `services/market_intel/deployment_readiness.py` | P1 readiness | check registry、runtime probe、projection 分離 |
 | 1658 | `routes/market_intel_review_report_routes.py` | P1 review route | report query、export、route glue 分離 |
--- a/routes/sales_routes.py
+++ b/routes/sales_routes.py
--- a/scripts/ops/run_nemotron_dispatch_reservation_canary.py
+++ b/scripts/ops/run_nemotron_dispatch_reservation_canary.py
@@ -0,0 +1,32 @@
+#!/usr/bin/env python3
+"""Run the no-business-side-effect NemoTron shared reservation canary."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+from pathlib import Path
+
+
+ROOT = Path(__file__).resolve().parents[2]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+from services.nemotron_dispatch_reservation_service import (  # noqa: E402
+    PRODUCTION_STATE_PATH,
+    run_shared_reservation_canary,
+)
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--state-path", default=str(PRODUCTION_STATE_PATH))
+    args = parser.parse_args()
+    payload = run_shared_reservation_canary(state_path=args.state_path)
+    print(json.dumps(payload, ensure_ascii=False, sort_keys=True))
+    return 0 if payload["success"] else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
--- a/services/nemoton_dispatcher_service.py
+++ b/services/nemoton_dispatcher_service.py
--- a/services/nemotron_decision_canary_service.py
+++ b/services/nemotron_decision_canary_service.py
@@ -1,8 +1,9 @@
 """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.
+exact-digest GCP-first candidate chain with a bounded 111 fallback, validates
+the returned decision, and stops before any tool, Telegram, database, price, or
+insight write is allowed.
 """

 from __future__ import annotations
@@ -18,24 +19,30 @@ from typing import Any, Mapping

 import requests

+from services.nemotron_runtime_candidate_service import (
+    FALLBACK_EXPECTED_DIGEST,
+    FALLBACK_MODEL,
+    IDENTITY_TIMEOUT_SEC,
+    PRIMARY_EXPECTED_DIGEST,
+    PRIMARY_MODEL,
+    NemotronRuntimeCandidate,
+    build_nemotron_runtime_candidates,
+    inspect_nemotron_model_identity,
+)
 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()
+NEMOTRON_OLLAMA_MODEL = PRIMARY_MODEL
+NEMOTRON_OLLAMA_EXPECTED_DIGEST = PRIMARY_EXPECTED_DIGEST
+NEMOTRON_OLLAMA_FALLBACK_MODEL = FALLBACK_MODEL
+NEMOTRON_OLLAMA_FALLBACK_EXPECTED_DIGEST = FALLBACK_EXPECTED_DIGEST
 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_MODEL_IDENTITY_TIMEOUT_SEC = IDENTITY_TIMEOUT_SEC
 DEFAULT_MAX_AGE_HOURS = max(
    1,
    min(int(os.getenv("NEMOTRON_DECISION_CANARY_MAX_AGE_HOURS", "26")), 168),
@@ -86,52 +93,21 @@ def _parse_datetime(value: Any) -> datetime | None:


 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}",
-    }
+    candidate = NemotronRuntimeCandidate(
+        label="compatibility_identity",
+        tier="primary",
+        host=host.rstrip("/"),
+        model=NEMOTRON_OLLAMA_MODEL,
+        expected_digest=NEMOTRON_OLLAMA_EXPECTED_DIGEST,
+        request_timeout_sec=DEFAULT_TIMEOUT_SEC,
+        num_predict=160,
+        num_ctx=None,
+    )
+    return inspect_nemotron_model_identity(
+        candidate,
+        request_get=requests.get,
+        timeout_sec=DEFAULT_MODEL_IDENTITY_TIMEOUT_SEC,
+    )


 def _latest_execution(root: Path, *, now: datetime | None = None) -> dict[str, Any]:
@@ -165,7 +141,12 @@ def _latest_execution(root: Path, *, now: datetime | None = None) -> dict[str, A
        "canary_passed": payload.get("canary_passed") is True,
        "selected_host": payload.get("selected_host"),
        "model": payload.get("model"),
+        "expected_digest": payload.get("expected_digest"),
        "observed_digest": payload.get("observed_digest"),
+        "selected_candidate_label": payload.get("selected_candidate_label"),
+        "fallback_used": payload.get("fallback_used") is True,
+        "degraded_runtime": payload.get("degraded_runtime") is True,
+        "attempt_count": int(payload.get("attempt_count") or 0),
        "decision_tool": payload.get("decision_tool"),
        "decision_sku": payload.get("decision_sku"),
        "model_elapsed_ms": payload.get("model_elapsed_ms"),
@@ -283,67 +264,204 @@ def run_nemotron_decision_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)
+    from services.nemoton_dispatcher_service import (
+        ToolCallContractError,
+        _parse_content_fallback,
+        _parse_tool_calls_struct,
+        _remaining_timeout,
+        _validate_tool_call_contract,
+        build_qwen3_dispatch_payload,
+    )
+
+    candidates = build_nemotron_runtime_candidates(
+        primary_model=NEMOTRON_OLLAMA_MODEL,
+        primary_expected_digest=NEMOTRON_OLLAMA_EXPECTED_DIGEST,
+        fallback_model=NEMOTRON_OLLAMA_FALLBACK_MODEL,
+        fallback_expected_digest=NEMOTRON_OLLAMA_FALLBACK_EXPECTED_DIGEST,
+    )
+    timeout_cap = max(30, min(int(timeout_sec or DEFAULT_TIMEOUT_SEC), 600))
+    canary_started = time.monotonic()
+    deadline_monotonic = canary_started + timeout_cap
+    host_preflights: list[dict[str, Any]] = []
+    runtime_attempts: list[dict[str, Any]] = []
+    selected: dict[str, Any] | None = None
+    selected_candidate: NemotronRuntimeCandidate | None = None
+    post_identity: dict[str, Any] = {}
    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,
+    expected_tool = "trigger_price_alert"
+    threat = _SyntheticThreat()
+    for candidate in candidates:
+        attempt: dict[str, Any] = {
+            "candidate": candidate.as_public_dict(),
+            "identity": {},
+            "model_call_performed": False,
+            "model_elapsed_ms": 0,
+            "decision_format": "none",
+            "decision_count": 0,
+            "status": "identity_failed",
+            "error": None,
+        }
+        try:
+            identity_timeout = _remaining_timeout(
+                deadline_monotonic,
+                DEFAULT_MODEL_IDENTITY_TIMEOUT_SEC,
+            )
+        except requests.Timeout as exc:
+            attempt["status"] = "deadline_exhausted"
+            attempt["error"] = str(exc)
+            runtime_attempts.append(attempt)
+            error = str(exc)
+            break
+        identity = inspect_nemotron_model_identity(
+            candidate,
+            request_get=requests.get,
+            timeout_sec=identity_timeout,
        )
+        host_preflights.append(identity)
+        attempt["identity"] = identity
+        attempt["error"] = identity.get("error")
+        if identity.get("ok") is not True:
+            runtime_attempts.append(attempt)
+            error = str(identity.get("error") or "model identity failed")
+            continue

-        threat = _SyntheticThreat()
        payload = build_qwen3_dispatch_payload(
            [threat],
            mcp_context="controlled decision-only canary; no live market data",
-            num_predict=160,
+            model=candidate.model,
+            num_predict=candidate.num_predict,
+            num_ctx=candidate.num_ctx,
            keep_alive=0,
        )
-        payload["think"] = False
        started = time.monotonic()
        model_call_performed = True
+        attempt["model_call_performed"] = True
+        candidate_decisions: list[dict[str, Any]] = []
+        candidate_decision_format = "none"
+        candidate_error: str | None = None
+        candidate_contract_error: str | None = None
+        candidate_contract_valid = False
        try:
+            request_timeout = _remaining_timeout(
+                deadline_monotonic,
+                candidate.request_timeout_sec,
+            )
+            attempt["request_timeout_sec"] = round(request_timeout, 3)
            response = requests.post(
-                f"{str(selected['host']).rstrip('/')}/api/chat",
+                f"{candidate.host}/api/chat",
                json=payload,
-                timeout=max(30, min(int(timeout_sec or DEFAULT_TIMEOUT_SEC), 600)),
+                timeout=request_timeout,
            )
            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"
+            candidate_decisions = _parse_tool_calls_struct(
+                message.get("tool_calls") or []
+            )
+            if candidate_decisions:
+                candidate_decision_format = "tool_calls"
            else:
-                decisions = _parse_content_fallback(str(message.get("content") or ""))
-                if decisions:
-                    decision_format = "content_fallback"
+                candidate_decisions = _parse_content_fallback(
+                    str(message.get("content") or "")
+                )
+                if candidate_decisions:
+                    candidate_decision_format = "content_fallback"
+            try:
+                candidate_decisions = _validate_tool_call_contract(
+                    candidate_decisions,
+                    [threat],
+                )
+                candidate_contract_valid = True
+            except ToolCallContractError as exc:
+                candidate_contract_error = str(exc)[:240]
        except Exception as exc:
-            error = f"{type(exc).__name__}: {str(exc)[:240]}"
+            candidate_error = f"{type(exc).__name__}: {str(exc)[:240]}"
        finally:
-            model_elapsed_ms = round((time.monotonic() - started) * 1000)
+            elapsed_ms = round((time.monotonic() - started) * 1000)
+            model_elapsed_ms += elapsed_ms
+            attempt["model_elapsed_ms"] = elapsed_ms
+
+        candidate_decision = candidate_decisions[0] if candidate_decisions else {}
+        candidate_args = (
+            candidate_decision.get("args")
+            if isinstance(candidate_decision.get("args"), dict)
+            else {}
+        )
+        candidate_tool = str(candidate_decision.get("tool") or "")
+        candidate_sku = str((candidate_args or {}).get("sku") or "")
+        candidate_post_identity: dict[str, Any] = {}
+        if candidate_error is None:
+            try:
+                post_identity_timeout = _remaining_timeout(
+                    deadline_monotonic,
+                    DEFAULT_MODEL_IDENTITY_TIMEOUT_SEC,
+                )
+                candidate_post_identity = inspect_nemotron_model_identity(
+                    candidate,
+                    request_get=requests.get,
+                    timeout_sec=post_identity_timeout,
+                )
+            except requests.Timeout as exc:
+                candidate_error = f"{type(exc).__name__}: {str(exc)[:240]}"
+        candidate_checks = {
+            "approved_host_selected": True,
+            "expected_digest_verified_before": identity.get("digest_matches") is True,
+            "model_call_completed": candidate_error is None,
+            "decision_present": bool(candidate_decisions),
+            "decision_contract_valid": candidate_contract_valid,
+            "decision_tool_allowed": candidate_tool in APPROVED_TOOLS,
+            "expected_decision_tool_selected": candidate_tool == expected_tool,
+            "synthetic_sku_preserved": candidate_sku == _SyntheticThreat.sku,
+            "expected_digest_stable_after": (
+                candidate_post_identity.get("digest_matches") is True
+            ),
+            "tool_execution_absent": True,
+            "database_call_absent": True,
+            "telegram_send_absent": True,
+        }
+        attempt.update({
+            "decision_format": candidate_decision_format,
+            "decision_count": len(candidate_decisions),
+            "decision_tool": candidate_tool or None,
+            "decision_sku": candidate_sku or None,
+            "post_model_identity": candidate_post_identity,
+            "checks": candidate_checks,
+            "status": "passed" if all(candidate_checks.values()) else "failed",
+            "error": candidate_error or candidate_contract_error,
+            "contract_error": candidate_contract_error,
+        })
+        runtime_attempts.append(attempt)
+        if all(candidate_checks.values()):
+            selected = identity
+            selected_candidate = candidate
+            post_identity = candidate_post_identity
+            decisions = candidate_decisions
+            decision_format = candidate_decision_format
+            error = None
+            break
+        error = candidate_error or "candidate decision contract failed"
+
+    if selected_candidate is None:
+        error = (
+            "decision_total_deadline_exhausted"
+            if time.monotonic() >= deadline_monotonic
+            else "all_approved_model_candidates_failed"
+        )

    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,
+        "approved_host_selected": selected_candidate is not None,
        "expected_digest_verified_before": bool(selected and selected.get("digest_matches")),
-        "model_call_completed": model_call_performed and error is None,
+        "model_call_completed": selected_candidate is not None and error is None,
        "decision_present": bool(decisions),
+        "decision_contract_valid": selected_candidate is not None,
        "decision_tool_allowed": decision_tool in APPROVED_TOOLS,
        "expected_decision_tool_selected": decision_tool == expected_tool,
        "synthetic_sku_preserved": decision_sku == _SyntheticThreat.sku,
@@ -353,7 +471,22 @@ def run_nemotron_decision_canary(
        "telegram_send_absent": True,
    }
    canary_passed = all(checks.values())
-    status = "canary_passed" if canary_passed else "canary_failed"
+    fallback_used = bool(selected_candidate and selected_candidate.is_fallback)
+    status = (
+        "canary_passed_degraded_fallback"
+        if canary_passed and fallback_used
+        else "canary_passed"
+        if canary_passed
+        else "canary_failed"
+    )
+    selected_model = (
+        selected_candidate.model if selected_candidate else NEMOTRON_OLLAMA_MODEL
+    )
+    selected_expected_digest = (
+        selected_candidate.expected_digest
+        if selected_candidate
+        else NEMOTRON_OLLAMA_EXPECTED_DIGEST
+    )
    execution_receipt: dict[str, Any] = {
        "policy": POLICY,
        "canary_version": CANARY_VERSION,
@@ -361,14 +494,24 @@ def run_nemotron_decision_canary(
        "run_identity": run_identity,
        "status": status,
        "canary_passed": canary_passed,
-        "model": NEMOTRON_OLLAMA_MODEL,
-        "expected_digest": NEMOTRON_OLLAMA_EXPECTED_DIGEST,
+        "model": selected_model,
+        "expected_digest": selected_expected_digest,
        "observed_digest": selected.get("digest") if selected else None,
        "selected_host": selected.get("host") if selected else None,
+        "selected_candidate_label": (
+            selected_candidate.label if selected_candidate else None
+        ),
+        "fallback_used": fallback_used,
+        "degraded_runtime": fallback_used,
        "host_preflights": host_preflights,
+        "runtime_attempts": runtime_attempts,
+        "attempt_count": len(runtime_attempts),
        "post_model_identity": post_identity,
        "model_call_performed": model_call_performed,
        "model_elapsed_ms": model_elapsed_ms,
+        "timeout_budget_sec": timeout_cap,
+        "total_elapsed_ms": round((time.monotonic() - canary_started) * 1000),
+        "deadline_exhausted": time.monotonic() >= deadline_monotonic,
        "decision_format": decision_format,
        "decision_count": len(decisions),
        "decision_tool": decision_tool or None,
@@ -388,8 +531,8 @@ def run_nemotron_decision_canary(
        "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,
+            "expected_model": selected_model,
+            "expected_digest": selected_expected_digest,
            "observed_digest": selected.get("digest") if selected else None,
            "expected_decision_tool": expected_tool,
            "observed_decision_tool": decision_tool or None,
@@ -400,7 +543,7 @@ def run_nemotron_decision_canary(
            "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),
+            "check_mode_passed": canary_passed,
            "bounded_execution_performed": model_call_performed,
            "independent_verifier_passed": canary_passed,
            "rollback_or_no_write_terminal": "decision_only_no_tool_or_data_write",
@@ -444,7 +587,9 @@ def run_nemotron_decision_canary(
            "rollback_terminal": "decision_only_no_tool_or_data_write",
        },
        "next_machine_action": (
-            "continue_scheduled_nemotron_decision_canary"
+            "restore_gcp_decision_runtime_capacity"
+            if canary_passed and fallback_used
+            else "continue_scheduled_nemotron_decision_canary"
            if canary_passed
            else "repair_nemotron_model_runtime_then_retry_decision_only_canary"
        ),
--- a/services/nemotron_dispatch_reservation_service.py
+++ b/services/nemotron_dispatch_reservation_service.py
@@ -0,0 +1,376 @@
+"""Process-shared reservation store for NemoTron dispatch side effects."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import logging
+import os
+import time
+import uuid
+from pathlib import Path
+from typing import Callable
+
+try:
+    import fcntl
+except ImportError:  # pragma: no cover - exercised by compatibility monkeypatch
+    fcntl = None
+
+
+logger = logging.getLogger(__name__)
+
+STATE_VERSION = 1
+PRODUCTION_STATE_PATH = Path(
+    "/app/data/ai_automation/nemotron_dispatch_dedupe_state.json"
+)
+
+
+def production_shared_reservation_required() -> bool:
+    return (
+        os.getenv("FLASK_ENV", "").strip().lower() in {"prod", "production"}
+        or os.getenv("USE_POSTGRESQL", "").strip().lower() == "true"
+    )
+
+
+def _sku_key(sku: str) -> str:
+    value = str(sku or "").strip()
+    if not value:
+        return ""
+    return hashlib.sha256(f"nemotron-dispatch:{value}".encode("utf-8")).hexdigest()
+
+
+class ReservationStateError(RuntimeError):
+    pass
+
+
+class SharedFileReservationStore:
+    """Atomic lease/commit state shared by processes using the same bind mount."""
+
+    def __init__(
+        self,
+        state_path: str | Path,
+        *,
+        committed_ttl_sec: int,
+        lease_ttl_sec: int,
+        clock: Callable[[], float] | None = None,
+    ):
+        self.state_path = Path(state_path)
+        self.lock_path = self.state_path.with_suffix(self.state_path.suffix + ".lock")
+        self.committed_ttl_sec = max(1, int(committed_ttl_sec))
+        self.lease_ttl_sec = max(1, int(lease_ttl_sec))
+        self.clock = clock or time.time
+
+    @staticmethod
+    def _empty_state() -> dict:
+        return {"version": STATE_VERSION, "leases": {}, "committed": {}}
+
+    def _read_state(self) -> dict:
+        if not self.state_path.exists():
+            return self._empty_state()
+        try:
+            payload = json.loads(self.state_path.read_text(encoding="utf-8"))
+        except (OSError, json.JSONDecodeError) as exc:
+            raise ReservationStateError("reservation_state_unreadable") from exc
+        if not isinstance(payload, dict) or payload.get("version") != STATE_VERSION:
+            raise ReservationStateError("reservation_state_schema_invalid")
+        leases = payload.get("leases")
+        committed = payload.get("committed")
+        if not isinstance(leases, dict) or not isinstance(committed, dict):
+            raise ReservationStateError("reservation_state_schema_invalid")
+        for lease in leases.values():
+            if (
+                not isinstance(lease, dict)
+                or not isinstance(lease.get("token"), str)
+                or not isinstance(lease.get("expires_at"), (int, float))
+                or lease.get("phase", "reserved") not in {
+                    "reserved",
+                    "side_effect_started",
+                }
+            ):
+                raise ReservationStateError("reservation_lease_schema_invalid")
+        for marker in committed.values():
+            if (
+                not isinstance(marker, dict)
+                or not isinstance(marker.get("until"), (int, float))
+            ):
+                raise ReservationStateError("reservation_commit_schema_invalid")
+        return payload
+
+    def _write_state(self, payload: dict) -> None:
+        temporary = self.state_path.with_name(
+            f".{self.state_path.name}.{os.getpid()}.{uuid.uuid4().hex}.tmp"
+        )
+        data = json.dumps(payload, sort_keys=True, separators=(",", ":"))
+        try:
+            with temporary.open("w", encoding="utf-8") as handle:
+                handle.write(data)
+                handle.flush()
+                os.fsync(handle.fileno())
+            os.chmod(temporary, 0o640)
+            os.replace(temporary, self.state_path)
+            directory_fd = os.open(str(self.state_path.parent), os.O_RDONLY)
+            try:
+                os.fsync(directory_fd)
+            finally:
+                os.close(directory_fd)
+        finally:
+            try:
+                temporary.unlink(missing_ok=True)
+            except OSError:
+                pass
+
+    @staticmethod
+    def _cleanup(payload: dict, now: float) -> bool:
+        changed = False
+        for key in [
+            key
+            for key, lease in payload["leases"].items()
+            if float(lease["expires_at"]) <= now
+        ]:
+            del payload["leases"][key]
+            changed = True
+        for key in [
+            key
+            for key, marker in payload["committed"].items()
+            if float(marker["until"]) <= now
+        ]:
+            del payload["committed"][key]
+            changed = True
+        return changed
+
+    def _locked(self, operation):
+        if fcntl is None:
+            raise ReservationStateError("process_shared_lock_unavailable")
+        self.state_path.parent.mkdir(mode=0o750, parents=True, exist_ok=True)
+        with self.lock_path.open("a+", encoding="utf-8") as lock_handle:
+            os.chmod(self.lock_path, 0o640)
+            fcntl.flock(lock_handle.fileno(), fcntl.LOCK_EX)
+            try:
+                payload = self._read_state()
+                now = float(self.clock())
+                cleanup_changed = self._cleanup(payload, now)
+                result, operation_changed = operation(payload, now)
+                if cleanup_changed or operation_changed:
+                    self._write_state(payload)
+                return result
+            finally:
+                fcntl.flock(lock_handle.fileno(), fcntl.LOCK_UN)
+
+    def reserve(self, sku: str) -> str | None:
+        key = _sku_key(sku)
+        if not key:
+            return None
+
+        def operation(payload, now):
+            if key in payload["leases"] or key in payload["committed"]:
+                return None, False
+            token = uuid.uuid4().hex
+            payload["leases"][key] = {
+                "token": token,
+                "expires_at": now + self.lease_ttl_sec,
+                "phase": "reserved",
+            }
+            return token, True
+
+        try:
+            return self._locked(operation)
+        except Exception as exc:
+            logger.error(
+                "[NemotronReservation] shared reserve failed closed: %s",
+                type(exc).__name__,
+            )
+            return None
+
+    def refresh(self, sku: str, token: str) -> bool:
+        key = _sku_key(sku)
+
+        def operation(payload, now):
+            lease = payload["leases"].get(key)
+            if not lease or lease.get("token") != str(token):
+                return False, False
+            ttl = (
+                self.committed_ttl_sec
+                if lease.get("phase") == "side_effect_started"
+                else self.lease_ttl_sec
+            )
+            lease["expires_at"] = now + ttl
+            return True, True
+
+        try:
+            return bool(key and self._locked(operation))
+        except Exception as exc:
+            logger.error(
+                "[NemotronReservation] shared refresh failed closed: %s",
+                type(exc).__name__,
+            )
+            return False
+
+    def begin_side_effect(self, sku: str, token: str) -> bool:
+        """Persist the at-most-once boundary before invoking an external effect."""
+        key = _sku_key(sku)
+
+        def operation(payload, now):
+            lease = payload["leases"].get(key)
+            if not lease or lease.get("token") != str(token):
+                return False, False
+            lease["phase"] = "side_effect_started"
+            lease["expires_at"] = now + self.committed_ttl_sec
+            return True, True
+
+        try:
+            return bool(key and self._locked(operation))
+        except Exception as exc:
+            logger.error(
+                "[NemotronReservation] shared side-effect boundary failed closed: %s",
+                type(exc).__name__,
+            )
+            return False
+
+    def commit(self, sku: str, token: str) -> bool:
+        key = _sku_key(sku)
+
+        def operation(payload, now):
+            lease = payload["leases"].get(key)
+            if not lease or lease.get("token") != str(token):
+                return False, False
+            del payload["leases"][key]
+            payload["committed"][key] = {
+                "until": now + self.committed_ttl_sec,
+            }
+            return True, True
+
+        try:
+            return bool(key and self._locked(operation))
+        except Exception as exc:
+            logger.error(
+                "[NemotronReservation] shared commit failed closed: %s",
+                type(exc).__name__,
+            )
+            return False
+
+    def release(self, sku: str, token: str | None) -> bool:
+        key = _sku_key(sku)
+
+        def operation(payload, _now):
+            lease = payload["leases"].get(key)
+            if not lease or lease.get("token") != str(token or ""):
+                return False, False
+            del payload["leases"][key]
+            return True, True
+
+        try:
+            return bool(key and self._locked(operation))
+        except Exception as exc:
+            logger.error(
+                "[NemotronReservation] shared release failed closed: %s",
+                type(exc).__name__,
+            )
+            return False
+
+    def is_duplicate(self, sku: str) -> bool:
+        key = _sku_key(sku)
+        if not key:
+            return True
+
+        def operation(payload, _now):
+            return (
+                key in payload["leases"] or key in payload["committed"],
+                False,
+            )
+
+        try:
+            return bool(self._locked(operation))
+        except Exception as exc:
+            logger.error(
+                "[NemotronReservation] shared read failed closed: %s",
+                type(exc).__name__,
+            )
+            return True
+
+
+def build_production_shared_reservation_store(
+    *,
+    committed_ttl_sec: int,
+    lease_ttl_sec: int,
+) -> SharedFileReservationStore | None:
+    if not production_shared_reservation_required():
+        return None
+    return SharedFileReservationStore(
+        PRODUCTION_STATE_PATH,
+        committed_ttl_sec=committed_ttl_sec,
+        lease_ttl_sec=lease_ttl_sec,
+    )
+
+
+def run_shared_reservation_canary(
+    *,
+    state_path: str | Path = PRODUCTION_STATE_PATH,
+    committed_ttl_sec: int = 4 * 3600,
+    lease_ttl_sec: int = 15 * 60,
+) -> dict:
+    """Exercise shared ownership semantics without business side effects."""
+    first = SharedFileReservationStore(
+        state_path,
+        committed_ttl_sec=committed_ttl_sec,
+        lease_ttl_sec=lease_ttl_sec,
+    )
+    second = SharedFileReservationStore(
+        state_path,
+        committed_ttl_sec=committed_ttl_sec,
+        lease_ttl_sec=lease_ttl_sec,
+    )
+    synthetic_sku = f"CANARY-NEMOTRON-RESERVATION-{uuid.uuid4().hex}"
+    token = first.reserve(synthetic_sku)
+    competing_token = second.reserve(synthetic_sku) if token else None
+    stale_release_rejected = (
+        second.release(synthetic_sku, "stale-canary-token") is False
+        if token
+        else False
+    )
+    side_effect_boundary_persisted = (
+        first.begin_side_effect(synthetic_sku, token) if token else False
+    )
+    owner_release_succeeded = (
+        first.release(synthetic_sku, token) if token else False
+    )
+    terminal_clear = (
+        first.is_duplicate(synthetic_sku) is False
+        if owner_release_succeeded
+        else False
+    )
+    checks = {
+        "first_owner_reserved": bool(token),
+        "competing_owner_rejected": competing_token is None,
+        "stale_token_release_rejected": stale_release_rejected,
+        "side_effect_boundary_persisted": side_effect_boundary_persisted,
+        "owner_release_succeeded": owner_release_succeeded,
+        "terminal_state_clear": terminal_clear,
+    }
+    success = all(checks.values())
+    return {
+        "success": success,
+        "status": "shared_reservation_canary_passed" if success else "shared_reservation_canary_failed",
+        "policy": "nemotron_process_shared_reservation_v1",
+        "state_path": str(Path(state_path)),
+        "checks": checks,
+        "check_count": len(checks),
+        "check_pass_count": sum(checks.values()),
+        "controlled_apply": {
+            "writes_shared_reservation_state": True,
+            "terminal_state_clear": terminal_clear,
+            "writes_database": False,
+            "sends_telegram": False,
+            "executes_business_tool": False,
+            "reads_secret": False,
+        },
+    }
+
+
+__all__ = [
+    "PRODUCTION_STATE_PATH",
+    "ReservationStateError",
+    "SharedFileReservationStore",
+    "build_production_shared_reservation_store",
+    "production_shared_reservation_required",
+    "run_shared_reservation_canary",
+]
--- a/services/nemotron_runtime_candidate_service.py
+++ b/services/nemotron_runtime_candidate_service.py
@@ -0,0 +1,234 @@
+"""Digest-locked Ollama candidates for the decision agent runtime."""
+
+from __future__ import annotations
+
+import os
+import time
+from dataclasses import asdict, dataclass
+from typing import Any, Callable, Mapping
+
+import requests
+
+from services.ollama_service import (
+    OLLAMA_HOST_FALLBACK,
+    OLLAMA_HOST_PRIMARY,
+    OLLAMA_HOST_SECONDARY,
+)
+
+
+PRIMARY_MODEL = os.getenv("NEMOTRON_OLLAMA_MODEL", "qwen3:14b").strip()
+PRIMARY_EXPECTED_DIGEST = os.getenv(
+    "NEMOTRON_OLLAMA_EXPECTED_DIGEST",
+    "bdbd181c33f2ed1b31c972991882db3cf4d192569092138a7d29e973cd9debe8",
+).strip().lower()
+FALLBACK_MODEL = os.getenv(
+    "NEMOTRON_OLLAMA_FALLBACK_MODEL",
+    "qwen3:8b",
+).strip()
+FALLBACK_EXPECTED_DIGEST = os.getenv(
+    "NEMOTRON_OLLAMA_FALLBACK_EXPECTED_DIGEST",
+    "500a1f067a9f782620b40bee6f7b0c89e17ae61f686b92c24933e4ca4b2b8b41",
+).strip().lower()
+NEMOTRON_FALLBACK_NUM_CTX = 4096
+NEMOTRON_FALLBACK_NUM_PREDICT = 512
+
+
+def _bounded_int_env(name: str, default: int, minimum: int, maximum: int) -> int:
+    try:
+        value = int(os.getenv(name, str(default)))
+    except (TypeError, ValueError):
+        value = default
+    return max(minimum, min(value, maximum))
+
+
+IDENTITY_TIMEOUT_SEC = _bounded_int_env(
+    "NEMOTRON_MODEL_IDENTITY_TIMEOUT_SEC", 10, 2, 30
+)
+GCP_ATTEMPT_TIMEOUT_SEC = _bounded_int_env(
+    "NEMOTRON_GCP_ATTEMPT_TIMEOUT_SEC", 60, 30, 180
+)
+FALLBACK_ATTEMPT_TIMEOUT_SEC = _bounded_int_env(
+    "NEMOTRON_111_ATTEMPT_TIMEOUT_SEC", 45, 20, 120
+)
+
+
+@dataclass(frozen=True)
+class NemotronRuntimeCandidate:
+    label: str
+    tier: str
+    host: str
+    model: str
+    expected_digest: str
+    request_timeout_sec: int
+    num_predict: int
+    num_ctx: int | None
+
+    @property
+    def is_fallback(self) -> bool:
+        return self.tier == "fallback"
+
+    def as_public_dict(self) -> dict[str, Any]:
+        payload = asdict(self)
+        payload["is_fallback"] = self.is_fallback
+        return payload
+
+
+def build_nemotron_runtime_candidates(
+    *,
+    primary_model: str | None = None,
+    primary_expected_digest: str | None = None,
+    fallback_model: str | None = None,
+    fallback_expected_digest: str | None = None,
+) -> tuple[NemotronRuntimeCandidate, ...]:
+    """Return the approved GCP-first, 111-final candidate chain."""
+    resolved_primary_model = str(primary_model or PRIMARY_MODEL).strip()
+    resolved_primary_digest = str(
+        primary_expected_digest or PRIMARY_EXPECTED_DIGEST
+    ).strip().lower()
+    resolved_fallback_model = str(fallback_model or FALLBACK_MODEL).strip()
+    resolved_fallback_digest = str(
+        fallback_expected_digest or FALLBACK_EXPECTED_DIGEST
+    ).strip().lower()
+    return (
+        NemotronRuntimeCandidate(
+            label="gcp_primary",
+            tier="primary",
+            host=OLLAMA_HOST_PRIMARY.rstrip("/"),
+            model=resolved_primary_model,
+            expected_digest=resolved_primary_digest,
+            request_timeout_sec=GCP_ATTEMPT_TIMEOUT_SEC,
+            num_predict=2048,
+            num_ctx=None,
+        ),
+        NemotronRuntimeCandidate(
+            label="gcp_secondary",
+            tier="secondary",
+            host=OLLAMA_HOST_SECONDARY.rstrip("/"),
+            model=resolved_primary_model,
+            expected_digest=resolved_primary_digest,
+            request_timeout_sec=GCP_ATTEMPT_TIMEOUT_SEC,
+            num_predict=2048,
+            num_ctx=None,
+        ),
+        NemotronRuntimeCandidate(
+            label="ollama_111_fallback",
+            tier="fallback",
+            host=OLLAMA_HOST_FALLBACK.rstrip("/"),
+            model=resolved_fallback_model,
+            expected_digest=resolved_fallback_digest,
+            request_timeout_sec=FALLBACK_ATTEMPT_TIMEOUT_SEC,
+            num_predict=NEMOTRON_FALLBACK_NUM_PREDICT,
+            num_ctx=NEMOTRON_FALLBACK_NUM_CTX,
+        ),
+    )
+
+
+def inspect_nemotron_model_identity(
+    candidate: NemotronRuntimeCandidate,
+    *,
+    request_get: Callable[..., Any] | None = None,
+    timeout_sec: int | None = None,
+) -> dict[str, Any]:
+    """Verify an exact model digest without loading the model."""
+    started = time.monotonic()
+    getter = request_get or requests.get
+    timeout = max(0.1, min(float(timeout_sec or IDENTITY_TIMEOUT_SEC), 30.0))
+    try:
+        response = getter(f"{candidate.host}/api/tags", timeout=timeout)
+        response.raise_for_status()
+        payload = response.json()
+        if not isinstance(payload, Mapping):
+            raise ValueError("malformed_tags_payload")
+        models = payload.get("models", [])
+        if models is None:
+            models = []
+        if not isinstance(models, list):
+            raise ValueError("malformed_models_schema")
+    except Exception as exc:
+        return {
+            **candidate.as_public_dict(),
+            "ok": False,
+            "digest": None,
+            "digest_matches": False,
+            "elapsed_ms": round((time.monotonic() - started) * 1000),
+            "error": f"{type(exc).__name__}: {str(exc)[:180]}",
+        }
+
+    target = candidate.model
+    target_without_latest = target.removesuffix(":latest")
+    matched: Mapping[str, Any] = {}
+    if any(not isinstance(item, Mapping) for item in models):
+        return {
+            **candidate.as_public_dict(),
+            "ok": False,
+            "digest": None,
+            "digest_matches": False,
+            "parameter_size": "",
+            "quantization_level": "",
+            "elapsed_ms": round((time.monotonic() - started) * 1000),
+            "error": "malformed_models_schema",
+        }
+    if any(
+        "details" in item and not isinstance(item.get("details"), Mapping)
+        for item in models
+    ):
+        return {
+            **candidate.as_public_dict(),
+            "ok": False,
+            "digest": None,
+            "digest_matches": False,
+            "parameter_size": "",
+            "quantization_level": "",
+            "elapsed_ms": round((time.monotonic() - started) * 1000),
+            "error": "malformed_model_details",
+        }
+    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()
+    raw_details = matched.get("details", {}) if matched else {}
+    malformed_details = bool(matched) and not isinstance(raw_details, Mapping)
+    details = raw_details if isinstance(raw_details, Mapping) else {}
+    digest_matches = (
+        bool(digest)
+        and digest == candidate.expected_digest
+        and not malformed_details
+    )
+    error = None
+    if not matched:
+        error = f"model_not_found:{target}"
+    elif malformed_details:
+        error = "malformed_model_details"
+    elif not digest_matches:
+        error = "model_digest_mismatch"
+    return {
+        **candidate.as_public_dict(),
+        "ok": bool(matched) and digest_matches,
+        "digest": digest or None,
+        "digest_matches": digest_matches,
+        "parameter_size": str(details.get("parameter_size") or ""),
+        "quantization_level": str(details.get("quantization_level") or ""),
+        "elapsed_ms": round((time.monotonic() - started) * 1000),
+        "error": error,
+    }
+
+
+__all__ = [
+    "FALLBACK_ATTEMPT_TIMEOUT_SEC",
+    "FALLBACK_EXPECTED_DIGEST",
+    "FALLBACK_MODEL",
+    "GCP_ATTEMPT_TIMEOUT_SEC",
+    "IDENTITY_TIMEOUT_SEC",
+    "NemotronRuntimeCandidate",
+    "NEMOTRON_FALLBACK_NUM_CTX",
+    "NEMOTRON_FALLBACK_NUM_PREDICT",
+    "PRIMARY_EXPECTED_DIGEST",
+    "PRIMARY_MODEL",
+    "build_nemotron_runtime_candidates",
+    "inspect_nemotron_model_identity",
+]
--- a/services/ollama_service.py
+++ b/services/ollama_service.py
@@ -47,6 +47,14 @@ def approved_ollama_env(name: str, default: str = '') -> str:
    return default


+def _bounded_int_env(name: str, default: int, minimum: int, maximum: int) -> int:
+    try:
+        value = int(os.getenv(name, str(default)))
+    except (TypeError, ValueError):
+        value = default
+    return max(minimum, min(value, maximum))
+
+
 # Ollama 設定 - 僅允許 GCP-A → GCP-B → 111 三主機
 OLLAMA_HOST_PRIMARY   = approved_ollama_env('OLLAMA_HOST_PRIMARY',   'http://34.87.90.216:11434')
 OLLAMA_HOST_SECONDARY = approved_ollama_env('OLLAMA_HOST_SECONDARY', 'http://34.21.145.224:11434')
@@ -66,8 +74,8 @@ EMBED_GCP_FAILURE_COOLDOWN_SEC = int(os.getenv('OLLAMA_EMBED_GCP_FAILURE_COOLDOW
 EMBED_GCP_FAILURE_NOTICE_SEC = int(os.getenv('OLLAMA_EMBED_GCP_FAILURE_NOTICE_SEC', '30'))
 FALLBACK_111_KEEP_ALIVE = os.getenv('OLLAMA_111_KEEP_ALIVE', '5m')
 FALLBACK_111_MAX_TIMEOUT = int(os.getenv('OLLAMA_111_MAX_TIMEOUT', '20'))
-FALLBACK_111_NUM_CTX = int(os.getenv('OLLAMA_111_NUM_CTX', '4096'))
-FALLBACK_111_NUM_PREDICT = int(os.getenv('OLLAMA_111_NUM_PREDICT', '512'))
+FALLBACK_111_NUM_CTX = _bounded_int_env('OLLAMA_111_NUM_CTX', 4096, 512, 4096)
+FALLBACK_111_NUM_PREDICT = _bounded_int_env('OLLAMA_111_NUM_PREDICT', 512, 32, 512)
 FALLBACK_111_MODEL = os.getenv('OLLAMA_111_MODEL_FALLBACK', 'llama3.2:latest')
 FALLBACK_111_CIRCUIT_CACHE_SEC = int(os.getenv('OLLAMA_111_CIRCUIT_CACHE_SEC', '60'))
 FALLBACK_111_MODEL_PATTERNS = tuple(
--- a/services/sales_analysis_chart_service.py
+++ b/services/sales_analysis_chart_service.py
@@ -0,0 +1,318 @@
+"""Pure builders for the sales-analysis chart payload contract."""
+
+from __future__ import annotations
+
+from datetime import datetime
+from typing import Any, Iterable
+
+import numpy as np
+import pandas as pd
+
+
+DOW_LABELS = ["週一", "週二", "週三", "週四", "週五", "週六", "週日"]
+HOUR_LABELS = [f"{hour:02d}:00" for hour in range(24)]
+
+
+def _has_columns(frame: pd.DataFrame, *columns: str | None) -> bool:
+    return bool(columns) and all(column and column in frame.columns for column in columns)
+
+
+def _to_number(value: Any) -> float:
+    try:
+        number = float(value)
+    except (TypeError, ValueError):
+        return 0.0
+    return number if np.isfinite(number) else 0.0
+
+
+def build_sales_filter_options(
+    frame: pd.DataFrame,
+    columns: dict[str, str | None],
+) -> dict[str, list[str]]:
+    """Build period-scoped filter choices without another full-table query."""
+    result = {
+        "categories": [],
+        "brands": [],
+        "vendors": [],
+        "activities": [],
+        "payments": [],
+        "months": [],
+    }
+    if frame is None or frame.empty:
+        return result
+
+    column_map = {
+        "categories": columns.get("category"),
+        "brands": columns.get("brand"),
+        "vendors": columns.get("vendor"),
+        "activities": columns.get("activity"),
+        "payments": columns.get("payment"),
+    }
+    for key, column in column_map.items():
+        if not column or column not in frame.columns:
+            continue
+        values = frame[column].dropna().astype(str).str.strip()
+        result[key] = sorted(value for value in values.unique().tolist() if value)
+
+    if "_month_str" in frame.columns:
+        result["months"] = sorted({
+            month
+            for value in frame["_month_str"].dropna().tolist()
+            if (month := _normalise_month(value))
+        })
+    return result
+
+
+def build_analysis_scope(
+    period: dict[str, Any],
+    *,
+    month: Any = "all",
+    dow: Any = "all",
+    hour: Any = "all",
+) -> dict[str, Any]:
+    """Build the concise, visible scope label shared by every chart and table."""
+    parts = [str(period.get("label") or "全部資料")]
+    month_value = _normalise_month(month) if str(month or "").lower() != "all" else None
+    if month_value:
+        parts.append(f"月份 {month_value}")
+
+    dow_raw = str(dow or "all")
+    if dow_raw != "all" and dow_raw.isdigit() and 0 <= int(dow_raw) < len(DOW_LABELS):
+        parts.append(DOW_LABELS[int(dow_raw)])
+
+    hour_raw = str(hour or "all")
+    if hour_raw != "all" and hour_raw.isdigit() and 0 <= int(hour_raw) <= 23:
+        parts.append(f"{int(hour_raw):02d}:00")
+
+    return {
+        "label": " · ".join(parts),
+        "start_date": str(period.get("start_date") or ""),
+        "end_date": str(period.get("end_date") or ""),
+    }
+
+
+def build_treemap_chart_data(
+    frame: pd.DataFrame,
+    category_col: str | None,
+    product_name_col: str | None,
+    amount_col: str | None,
+    *,
+    category_limit: int = 10,
+    product_limit: int = 5,
+) -> list[dict[str, Any]]:
+    """Return the category/name/value contract consumed by Chart.js treemap."""
+    if frame is None or frame.empty or not _has_columns(frame, category_col, product_name_col, amount_col):
+        return []
+
+    grouped = frame[[category_col, product_name_col, amount_col]].copy()
+    grouped[amount_col] = pd.to_numeric(grouped[amount_col], errors="coerce").fillna(0)
+    grouped[category_col] = grouped[category_col].fillna("未分類").astype(str)
+    grouped[product_name_col] = grouped[product_name_col].fillna("未命名商品").astype(str)
+    grouped = (
+        grouped.groupby([category_col, product_name_col], as_index=False, dropna=False)[amount_col]
+        .sum()
+    )
+    grouped = grouped[grouped[amount_col] > 0]
+    if grouped.empty:
+        return []
+
+    category_order = (
+        grouped.groupby(category_col)[amount_col]
+        .sum()
+        .nlargest(category_limit)
+        .index.tolist()
+    )
+    result: list[dict[str, Any]] = []
+    for category in category_order:
+        products = grouped[grouped[category_col] == category].nlargest(product_limit, amount_col)
+        result.extend(
+            {
+                "category": str(row[category_col]),
+                "name": str(row[product_name_col]),
+                "value": _to_number(row[amount_col]),
+            }
+            for _, row in products.iterrows()
+        )
+    return result
+
+
+def build_bcg_chart_data(
+    frame: pd.DataFrame,
+    product_id_col: str | None,
+    product_name_col: str | None,
+    qty_col: str | None,
+    amount_col: str | None,
+    profit_col: str = "calculated_profit",
+    *,
+    limit: int = 400,
+) -> dict[str, Any]:
+    """Aggregate SKU rows and return points plus weighted-margin medians."""
+    empty = {"points": [], "x_median": 0.0, "y_median": 0.0}
+    if frame is None or frame.empty or not _has_columns(
+        frame, product_name_col, qty_col, amount_col, profit_col
+    ):
+        return empty
+
+    group_cols = [product_name_col]
+    if product_id_col and product_id_col in frame.columns:
+        group_cols.insert(0, product_id_col)
+    columns = group_cols + [qty_col, amount_col, profit_col]
+    working = frame[columns].copy()
+    for column in (qty_col, amount_col, profit_col):
+        working[column] = pd.to_numeric(working[column], errors="coerce").fillna(0)
+
+    grouped = working.groupby(group_cols, as_index=False, dropna=False).agg(
+        _qty=(qty_col, "sum"),
+        _amount=(amount_col, "sum"),
+        _profit=(profit_col, "sum"),
+    )
+    grouped = grouped[(grouped["_qty"] > 0) & (grouped["_amount"] > 0)]
+    if grouped.empty:
+        return empty
+    grouped["_margin"] = grouped["_profit"] * 100.0 / grouped["_amount"]
+    grouped["_margin"] = grouped["_margin"].replace([np.inf, -np.inf], np.nan).fillna(0)
+    grouped = grouped.nlargest(limit, "_amount")
+
+    names = grouped[product_name_col].fillna("未命名商品").astype(str)
+    duplicate_names = names.value_counts()
+    points = []
+    for _, row in grouped.iterrows():
+        raw_name = row.get(product_name_col)
+        name = str(raw_name) if raw_name is not None and not pd.isna(raw_name) else "未命名商品"
+        product_id = row.get(product_id_col) if product_id_col else None
+        if duplicate_names.get(name, 0) > 1 and product_id is not None and not pd.isna(product_id):
+            name = f"{name} [{product_id}]"
+        points.append(
+            {
+                "x": _to_number(row["_qty"]),
+                "y": _to_number(row["_margin"]),
+                "name": name,
+                "amount": _to_number(row["_amount"]),
+            }
+        )
+
+    return {
+        "points": points,
+        "x_median": _to_number(grouped["_qty"].median()),
+        "y_median": _to_number(grouped["_margin"].median()),
+    }
+
+
+def build_heatmap_chart_data(
+    frame: pd.DataFrame,
+    amount_col: str | None,
+    *,
+    dow_col: str = "_dow",
+    hour_col: str = "_hour",
+) -> dict[str, Any]:
+    """Return a stable 7 x 24 weekday/hour matrix."""
+    empty = {"dows": DOW_LABELS, "hours": HOUR_LABELS, "matrix": []}
+    if frame is None or frame.empty or not _has_columns(frame, dow_col, hour_col, amount_col):
+        return empty
+
+    working = frame[[dow_col, hour_col, amount_col]].copy()
+    for column in (dow_col, hour_col, amount_col):
+        working[column] = pd.to_numeric(working[column], errors="coerce")
+    working = working[
+        working[dow_col].between(0, 6)
+        & working[hour_col].between(0, 23)
+        & working[amount_col].notna()
+    ]
+    if working.empty:
+        return empty
+
+    matrix = [[0.0 for _ in range(24)] for _ in range(7)]
+    grouped = working.groupby([dow_col, hour_col])[amount_col].sum()
+    for (dow, hour), value in grouped.items():
+        matrix[int(dow)][int(hour)] = _to_number(value)
+    return {"dows": DOW_LABELS, "hours": HOUR_LABELS, "matrix": matrix}
+
+
+def build_seasonality_chart_data(
+    frame: pd.DataFrame,
+    category_col: str | None,
+    amount_col: str | None,
+    *,
+    month_col: str = "_month_str",
+    category_limit: int = 10,
+) -> dict[str, Any]:
+    """Return categories x months matrix for the selected period only."""
+    empty = {"categories": [], "months": [], "matrix": []}
+    if frame is None or frame.empty or not _has_columns(frame, month_col, category_col, amount_col):
+        return empty
+
+    working = frame[[month_col, category_col, amount_col]].copy()
+    working[month_col] = working[month_col].map(_normalise_month)
+    working[category_col] = working[category_col].fillna("未分類").astype(str)
+    working[amount_col] = pd.to_numeric(working[amount_col], errors="coerce").fillna(0)
+    working = working[working[month_col].notna() & (working[amount_col] > 0)]
+    if working.empty:
+        return empty
+
+    categories = (
+        working.groupby(category_col)[amount_col]
+        .sum()
+        .nlargest(category_limit)
+        .index.tolist()
+    )
+    months = sorted(working[month_col].dropna().unique().tolist())
+    pivot = (
+        working[working[category_col].isin(categories)]
+        .pivot_table(index=category_col, columns=month_col, values=amount_col, aggfunc="sum", fill_value=0)
+        .reindex(index=categories, columns=months, fill_value=0)
+    )
+    return {
+        "categories": [str(category) for category in categories],
+        "months": months,
+        "matrix": [[_to_number(value) for value in row] for row in pivot.to_numpy().tolist()],
+    }
+
+
+def build_marketing_chart_data(summary: dict[str, Any] | None, metric: str) -> dict[str, Any]:
+    """Convert marketing summary records into the chart labels/values contract."""
+    value_key = {"amount": "revenue", "qty": "qty", "profit": "profit"}.get(metric, "revenue")
+    result: dict[str, Any] = {}
+    for key in ("discount", "coupon", "bonus", "click"):
+        records = summary.get(key, []) if isinstance(summary, dict) else []
+        if not isinstance(records, list):
+            records = []
+        labels = []
+        values = []
+        for record in records:
+            if not isinstance(record, dict):
+                continue
+            labels.append(str(record.get("name") or "未命名活動"))
+            values.append(_to_number(record.get(value_key)))
+        result[key] = {"labels": labels, "values": values}
+    result["metric"] = value_key
+    return result
+
+
+def resolve_yoy_year_options(
+    months: Iterable[Any],
+    analysis_period: dict[str, Any] | None,
+    *,
+    fallback_year: int | None = None,
+) -> dict[str, Any]:
+    """Derive comparison-year controls from live data and the active period."""
+    years = {
+        int(month[:4])
+        for value in (months if months is not None else [])
+        if (month := _normalise_month(value))
+    }
+    period_end = str((analysis_period or {}).get("end_date") or "")
+    current_year = int(period_end[:4]) if period_end[:4].isdigit() else (fallback_year or datetime.now().year)
+    years.update({current_year - 1, current_year})
+    return {
+        "years": sorted(years, reverse=True),
+        "year1": current_year - 1,
+        "year2": current_year,
+    }
+
+
+def _normalise_month(value: Any) -> str | None:
+    raw = str(value or "").strip().replace("/", "-")[:7]
+    try:
+        return datetime.strptime(raw, "%Y-%m").strftime("%Y-%m")
+    except ValueError:
+        return None
--- a/services/sales_analysis_export_service.py
+++ b/services/sales_analysis_export_service.py
@@ -0,0 +1,207 @@
+"""Period-linked, Excel-safe export queries for the sales analysis page."""
+
+from __future__ import annotations
+
+import re
+from typing import Any, Mapping
+
+import pandas as pd
+from sqlalchemy import text
+
+from services.sales_analysis_query_service import (
+    build_sales_metric_aggregate_sql,
+    build_sales_where_clause,
+    prepare_sales_query_context,
+    quote_identifier,
+)
+from utils.security import validate_table_name
+
+
+_EXCEL_ILLEGAL_CHAR_RE = re.compile(r"[\x00-\x08\x0B-\x0C\x0E-\x1F]")
+_FORMULA_PREFIXES = ("=", "+", "-", "@")
+_MARKETING_COLUMNS = {
+    "coupon": ("coupon_activity", "折價券活動"),
+    "discount": ("discount_activity", "折扣活動"),
+    "bonus": ("bonus_activity", "滿額再折扣"),
+    "click": ("click_activity", "點我再折扣"),
+}
+
+
+def _sanitize_excel_cell(value: Any) -> Any:
+    if not isinstance(value, str):
+        return value
+    cleaned = _EXCEL_ILLEGAL_CHAR_RE.sub("", value)
+    if cleaned.lstrip().startswith(_FORMULA_PREFIXES):
+        return f"'{cleaned}"
+    return cleaned
+
+
+def sanitize_excel_dataframe(frame: pd.DataFrame) -> pd.DataFrame:
+    """Prevent control-character failures and spreadsheet formula injection."""
+    cleaned = frame.copy()
+    for column in cleaned.columns:
+        dtype = cleaned[column].dtype
+        if pd.api.types.is_object_dtype(dtype) or pd.api.types.is_string_dtype(dtype):
+            cleaned[column] = cleaned[column].map(_sanitize_excel_cell)
+    return cleaned
+
+
+def query_sales_vendor_export_frame(
+    engine,
+    args: Mapping[str, Any],
+    *,
+    table_name: str = "realtime_sales_monthly",
+) -> pd.DataFrame:
+    """Return the same filtered vendor ranking represented on the page."""
+    table_name = validate_table_name(table_name)
+    filters, columns = prepare_sales_query_context(
+        engine,
+        table_name,
+        args,
+        default_data_range=1,
+    )
+    vendor = quote_identifier(engine, columns.get("vendor"))
+    amount_sql, _ = build_sales_metric_aggregate_sql(engine, columns, "amount")
+    try:
+        profit_sql, _ = build_sales_metric_aggregate_sql(engine, columns, "profit")
+    except ValueError:
+        profit_sql = "0"
+    qty_sql = (
+        build_sales_metric_aggregate_sql(engine, columns, "qty")[0]
+        if columns.get("qty")
+        else "0"
+    )
+    where_sql, params = build_sales_where_clause(engine, columns, filters)
+    where_suffix = f" AND {where_sql}" if where_sql else ""
+    table = quote_identifier(engine, table_name)
+    query = text(f"""
+        SELECT {vendor} AS vendor,
+               {amount_sql} AS amount,
+               {qty_sql} AS qty,
+               {profit_sql} AS profit,
+               CASE WHEN {amount_sql} > 0
+                    THEN ({profit_sql}) * 100.0 / ({amount_sql}) ELSE 0 END AS margin_rate
+        FROM {table}
+        WHERE {vendor} IS NOT NULL
+          AND TRIM(CAST({vendor} AS TEXT)) <> '' {where_suffix}
+        GROUP BY {vendor}
+        ORDER BY amount DESC
+    """)
+    frame = pd.read_sql(query, engine, params=params)
+    if frame.empty:
+        return frame
+    frame = frame.rename(columns={
+        "vendor": "廠商",
+        "amount": "銷售金額",
+        "qty": "銷售數量",
+        "profit": "毛利金額",
+        "margin_rate": "毛利率(%)",
+    })
+    return sanitize_excel_dataframe(frame)
+
+
+def query_sales_marketing_export_frames(
+    engine,
+    args: Mapping[str, Any],
+    *,
+    activity_type: str = "all",
+    table_name: str = "realtime_sales_monthly",
+) -> dict[str, pd.DataFrame]:
+    """Aggregate every requested marketing dimension with the active page filters."""
+    if activity_type != "all" and activity_type not in _MARKETING_COLUMNS:
+        raise ValueError("type 僅允許 all、coupon、discount、bonus、click")
+    table_name = validate_table_name(table_name)
+    filters, columns = prepare_sales_query_context(
+        engine,
+        table_name,
+        args,
+        default_data_range=1,
+    )
+    metric = str(args.get("metric", "amount") or "amount").strip().lower()
+    if metric not in {"amount", "qty", "profit"}:
+        raise ValueError("metric 僅允許 amount、qty、profit")
+
+    amount_sql, _ = build_sales_metric_aggregate_sql(engine, columns, "amount")
+    qty_sql = (
+        build_sales_metric_aggregate_sql(engine, columns, "qty")[0]
+        if columns.get("qty")
+        else "0"
+    )
+    try:
+        profit_sql, _ = build_sales_metric_aggregate_sql(engine, columns, "profit")
+    except ValueError:
+        profit_sql = "0"
+    if metric == "qty" and qty_sql == "0":
+        raise ValueError("目前資料來源沒有銷量欄位")
+    if metric == "profit" and profit_sql == "0":
+        raise ValueError("目前資料來源沒有毛利或成本欄位")
+
+    metric_alias = {"amount": "revenue", "qty": "qty", "profit": "profit"}[metric]
+    where_sql, params = build_sales_where_clause(engine, columns, filters)
+    where_suffix = f" AND {where_sql}" if where_sql else ""
+    table = quote_identifier(engine, table_name)
+    selected_types = (
+        list(_MARKETING_COLUMNS)
+        if activity_type == "all"
+        else [activity_type]
+    )
+    branches = []
+    for key in selected_types:
+        column_key, sheet_name = _MARKETING_COLUMNS[key]
+        activity_column = columns.get(column_key)
+        if not activity_column:
+            continue
+        activity = quote_identifier(engine, activity_column)
+        type_param = f"activity_type_{key}"
+        params[type_param] = sheet_name
+        branches.append(f"""
+            SELECT :{type_param} AS activity_type,
+                   {activity} AS activity_name,
+                   {amount_sql} AS revenue,
+                   {qty_sql} AS qty,
+                   {profit_sql} AS profit,
+                   COUNT(*) AS item_count
+            FROM {table}
+            WHERE {activity} IS NOT NULL
+              AND TRIM(CAST({activity} AS TEXT)) NOT IN ('', '0') {where_suffix}
+            GROUP BY {activity}
+        """)
+    if not branches:
+        return {}
+
+    combined = pd.read_sql(text(" UNION ALL ".join(branches)), engine, params=params)
+    frames: dict[str, pd.DataFrame] = {}
+    for key in selected_types:
+        _column_key, sheet_name = _MARKETING_COLUMNS[key]
+        frame = combined[combined["activity_type"] == sheet_name].copy()
+        if frame.empty:
+            continue
+        frame = frame.sort_values(metric_alias, ascending=False)
+        frame = frame.rename(columns={
+            "activity_type": "活動類型",
+            "activity_name": "活動名稱",
+            "revenue": "銷售金額",
+            "qty": "銷售數量",
+            "profit": "毛利金額",
+            "item_count": "項目筆數",
+        })
+        frames[sheet_name] = sanitize_excel_dataframe(frame)
+    return frames
+
+
+def combine_sales_marketing_export_frames(
+    frames: Mapping[str, pd.DataFrame],
+    metric: str,
+) -> pd.DataFrame:
+    """Build the combined sheet in the same order as the active page metric."""
+    if metric not in {"amount", "qty", "profit"}:
+        raise ValueError("metric 僅允許 amount、qty、profit")
+    if not frames:
+        return pd.DataFrame()
+    combined = pd.concat(frames.values(), ignore_index=True)
+    sort_column = {
+        "amount": "銷售金額",
+        "qty": "銷售數量",
+        "profit": "毛利金額",
+    }[metric]
+    return combined.sort_values(sort_column, ascending=False)
--- a/services/sales_analysis_query_service.py
+++ b/services/sales_analysis_query_service.py
@@ -0,0 +1,391 @@
+"""Canonical, parameterized query contract for sales-analysis APIs."""
+
+from __future__ import annotations
+
+import math
+from datetime import date, datetime, timedelta, timezone
+from typing import Any, Mapping
+
+from sqlalchemy import inspect, text
+
+from services.analysis_period_service import parse_iso_date, parse_month
+from utils.df_helpers import find_col
+from utils.security import validate_table_name
+
+
+ALLOWED_DATA_RANGES = {0, 1, 3, 6, 12}
+TAIPEI_TZ = timezone(timedelta(hours=8))
+SALES_TIME_TEXT_PATTERN = (
+    r"^([01][0-9]|2[0-3]):[0-5][0-9](:[0-5][0-9](\.[0-9]+)?)?$"
+)
+
+
+def _get_value(args: Mapping[str, Any], key: str, default: Any = "") -> Any:
+    getter = getattr(args, "get", None)
+    return getter(key, default) if callable(getter) else default
+
+
+def _bounded_text(value: Any, field: str, *, max_length: int = 500) -> str:
+    result = str(value or "").strip()
+    if len(result) > max_length:
+        raise ValueError(f"{field} 長度不可超過 {max_length} 字元")
+    return result
+
+
+def _optional_number(args: Mapping[str, Any], key: str) -> float | None:
+    raw = str(_get_value(args, key, "") or "").strip()
+    if not raw:
+        return None
+    try:
+        result = float(raw)
+    except (TypeError, ValueError) as exc:
+        raise ValueError(f"{key} 必須是數字") from exc
+    if not math.isfinite(result):
+        raise ValueError(f"{key} 必須是有限數字")
+    return result
+
+
+def _optional_dimension(args: Mapping[str, Any], key: str) -> str | None:
+    raw = _bounded_text(_get_value(args, key, "all"), key)
+    return None if not raw or raw.lower() == "all" else raw
+
+
+def _optional_bounded_int(
+    args: Mapping[str, Any],
+    key: str,
+    *,
+    minimum: int,
+    maximum: int,
+) -> int | None:
+    raw = str(_get_value(args, key, "all") or "all").strip().lower()
+    if raw == "all":
+        return None
+    if not raw.isdigit() or not minimum <= int(raw) <= maximum:
+        raise ValueError(f"{key} 必須介於 {minimum} 到 {maximum}")
+    return int(raw)
+
+
+def normalize_sales_query_args(
+    args: Mapping[str, Any],
+    *,
+    default_data_range: int = 1,
+) -> dict[str, Any]:
+    """Validate and canonicalize all filter values shared by page APIs."""
+    range_raw = str(_get_value(args, "data_range", "") or "").strip()
+    if not range_raw:
+        data_range = default_data_range
+    elif not range_raw.isdigit() or int(range_raw) not in ALLOWED_DATA_RANGES:
+        raise ValueError("data_range 必須是 0、1、3、6 或 12")
+    else:
+        data_range = int(range_raw)
+
+    raw_start = str(_get_value(args, "start_date", "") or "").strip()
+    raw_end = str(_get_value(args, "end_date", "") or "").strip()
+    start = parse_iso_date(raw_start)
+    end = parse_iso_date(raw_end)
+    if raw_start and not start:
+        raise ValueError("start_date 必須是 YYYY-MM-DD")
+    if raw_end and not end:
+        raise ValueError("end_date 必須是 YYYY-MM-DD")
+    if start and end and start > end:
+        start, end = end, start
+
+    month_raw = str(_get_value(args, "month", "all") or "all").strip().lower()
+    month = None
+    if month_raw != "all":
+        parsed_month = parse_month(month_raw)
+        if not parsed_month:
+            raise ValueError("month 必須是 YYYY-MM")
+        month = parsed_month.strftime("%Y-%m")
+
+    return {
+        "data_range": data_range,
+        "start_date": start,
+        "end_date": end,
+        "category": _optional_dimension(args, "category"),
+        "brand": _optional_dimension(args, "brand"),
+        "vendor": _optional_dimension(args, "vendor"),
+        "activity": _optional_dimension(args, "activity"),
+        "payment": _optional_dimension(args, "payment"),
+        "month": month,
+        "dow": _optional_bounded_int(args, "dow", minimum=0, maximum=6),
+        "hour": _optional_bounded_int(args, "hour", minimum=0, maximum=23),
+        "min_price": _optional_number(args, "min_price"),
+        "max_price": _optional_number(args, "max_price"),
+        "min_margin": _optional_number(args, "min_margin"),
+        "max_margin": _optional_number(args, "max_margin"),
+        "keyword": _bounded_text(_get_value(args, "keyword", ""), "keyword", max_length=200),
+    }
+
+
+def resolve_sales_query_columns(engine, table_name: str) -> dict[str, str | None]:
+    """Resolve identifiers from the live table schema rather than request input."""
+    validate_table_name(table_name)
+    names = [column["name"] for column in inspect(engine).get_columns(table_name)]
+    return {
+        "date": find_col(names, ["日期", "交易日期", "Date", "Day"]),
+        "time": find_col(names, ["訂單時間", "成立時間", "下單時間", "購買時間", "時間", "Time", "Created"]),
+        "pid": find_col(names, ["商品ID", "Product ID", "i_code", "Item Code", "ID"]),
+        "name": find_col(names, ["商品名稱", "品名", "Name", "Product"]),
+        "brand": find_col(names, ["品牌", "Brand"]),
+        "vendor": find_col(names, ["廠商名稱", "Vendor Name", "廠商", "供應商", "Vendor", "Supplier"]),
+        "category": find_col(names, ["商品館", "館別", "分類", "Category"]),
+        "activity": find_col(names, ["折扣活動名稱", "折價券活動名稱", "滿額再折扣活動名稱", "活動", "Activity", "Campaign", "Promotion", "專案"]),
+        "discount_activity": find_col(names, ["折扣活動名稱"]),
+        "coupon_activity": find_col(names, ["折價券活動名稱"]),
+        "bonus_activity": find_col(names, ["滿額再折扣活動名稱"]),
+        "click_activity": find_col(names, ["點我再折扣"]),
+        "payment": find_col(names, ["付款方式", "付款", "Payment", "Pay"]),
+        "amount": find_col(names, ["銷售金額", "總業績", "業績", "金額", "Amount", "Sales", "Total"]),
+        "qty": find_col(names, ["銷售數量", "銷量", "數量", "Qty", "Quantity"]),
+        "cost": find_col(names, ["總成本", "成本", "Cost", "進價", "Cost Price", "Wholesale"]),
+        "profit": find_col(names, ["毛利", "Profit", "利潤"]),
+        "return_qty": find_col(names, ["退貨數量", "Return Qty", "退貨"]),
+    }
+
+
+def quote_identifier(engine, identifier: str | None) -> str:
+    if not identifier:
+        raise ValueError("查詢所需欄位不存在")
+    return engine.dialect.identifier_preparer.quote(str(identifier))
+
+
+def build_sales_metric_aggregate_sql(
+    engine,
+    columns: Mapping[str, str | None],
+    metric: str,
+) -> tuple[str, str]:
+    """Build one allowlisted aggregate shared by APIs and exports."""
+    amount = quote_identifier(engine, columns.get("amount"))
+    if metric in {"amount", "revenue"}:
+        return f"COALESCE(SUM(CAST({amount} AS REAL)), 0)", "銷售金額"
+    if metric == "qty":
+        qty = quote_identifier(engine, columns.get("qty"))
+        return f"COALESCE(SUM(CAST({qty} AS REAL)), 0)", "銷售數量"
+    if metric == "profit":
+        if columns.get("profit"):
+            profit = quote_identifier(engine, columns.get("profit"))
+            return f"COALESCE(SUM(CAST({profit} AS REAL)), 0)", "毛利金額"
+        cost = quote_identifier(engine, columns.get("cost"))
+        return (
+            f"COALESCE(SUM(CAST({amount} AS REAL)), 0) "
+            f"- COALESCE(SUM(CAST({cost} AS REAL)), 0)",
+            "毛利金額",
+        )
+    raise ValueError("metric 僅允許 amount、revenue、qty、profit")
+
+
+def _normalised_date_sql(engine, quoted_date: str) -> tuple[str, str]:
+    if engine.dialect.name == "postgresql":
+        date_sql = f"LEFT(REPLACE(CAST({quoted_date} AS TEXT), '/', '-'), 10)"
+        return date_sql, f"LEFT({date_sql}, 7)"
+    date_sql = f"substr(replace(CAST({quoted_date} AS TEXT), '/', '-'), 1, 10)"
+    return date_sql, f"substr({date_sql}, 1, 7)"
+
+
+def resolve_sales_date_bounds(
+    engine,
+    table_name: str,
+    columns: Mapping[str, str | None] | None = None,
+) -> tuple[date | None, date | None]:
+    """Return canonical source bounds for closing one-sided page ranges."""
+    table_name = validate_table_name(table_name)
+    resolved_columns = columns or resolve_sales_query_columns(engine, table_name)
+    quoted_date = quote_identifier(engine, resolved_columns.get("date"))
+    date_sql, _month_sql = _normalised_date_sql(engine, quoted_date)
+    table = quote_identifier(engine, table_name)
+    query = text(f"SELECT MIN({date_sql}), MAX({date_sql}) FROM {table}")
+    with engine.connect() as connection:
+        row = connection.execute(query).fetchone()
+    if not row:
+        return None, None
+    return parse_iso_date(row[0]), parse_iso_date(row[1])
+
+
+def build_sales_where_clause(
+    engine,
+    columns: Mapping[str, str | None],
+    filters: Mapping[str, Any],
+    *,
+    period_start: date | str | None = None,
+    period_end: date | str | None = None,
+    apply_month: bool = True,
+    include_numeric: bool = True,
+    now: date | None = None,
+) -> tuple[str, dict[str, Any]]:
+    """Return a SQL fragment and bind parameters for the canonical filters."""
+    conditions: list[str] = []
+    params: dict[str, Any] = {}
+
+    explicit_period = period_start is not None or period_end is not None
+    start = parse_iso_date(period_start) if period_start is not None else filters.get("start_date")
+    end = parse_iso_date(period_end) if period_end is not None else filters.get("end_date")
+    if start and end and start > end:
+        start, end = end, start
+    if not explicit_period and not start and not end and int(filters.get("data_range") or 0) > 0:
+        end = now or datetime.now(TAIPEI_TZ).date()
+        start = end - timedelta(days=int(filters["data_range"]) * 30)
+
+    needs_date = bool(start or end or (apply_month and filters.get("month")) or filters.get("dow") is not None)
+    date_sql = month_sql = None
+    if needs_date:
+        quoted_date = quote_identifier(engine, columns.get("date"))
+        date_sql, month_sql = _normalised_date_sql(engine, quoted_date)
+    if start:
+        conditions.append(f"{date_sql} >= :filter_start_date")
+        params["filter_start_date"] = start.isoformat()
+    if end:
+        conditions.append(f"{date_sql} <= :filter_end_date")
+        params["filter_end_date"] = end.isoformat()
+    if apply_month and filters.get("month"):
+        conditions.append(f"{month_sql} = :filter_month")
+        params["filter_month"] = filters["month"]
+
+    dimension_columns = {
+        "category": "category",
+        "brand": "brand",
+        "vendor": "vendor",
+        "activity": "activity",
+        "payment": "payment",
+    }
+    for filter_name, column_name in dimension_columns.items():
+        value = filters.get(filter_name)
+        if value is None:
+            continue
+        quoted = quote_identifier(engine, columns.get(column_name))
+        if filter_name == "category" and value == "其他":
+            exclusions = filters.get("category_other_exclusions")
+            if exclusions is None:
+                raise ValueError("其他分類缺少排行範圍契約")
+            if not exclusions:
+                conditions.append("1 = 0")
+                continue
+            placeholders = []
+            for index, excluded in enumerate(exclusions):
+                param_name = f"filter_category_other_{index}"
+                placeholders.append(f":{param_name}")
+                params[param_name] = excluded
+            conditions.append(f"{quoted} NOT IN ({', '.join(placeholders)})")
+            continue
+        conditions.append(f"{quoted} = :filter_{filter_name}")
+        params[f"filter_{filter_name}"] = value
+
+    if filters.get("dow") is not None:
+        database_dow = (int(filters["dow"]) + 1) % 7
+        if engine.dialect.name == "postgresql":
+            conditions.append(
+                f"EXTRACT(DOW FROM TO_DATE({date_sql}, 'YYYY-MM-DD')) = :filter_db_dow"
+            )
+            params["filter_db_dow"] = database_dow
+        else:
+            conditions.append(f"strftime('%w', {date_sql}) = :filter_db_dow")
+            params["filter_db_dow"] = str(database_dow)
+
+    if filters.get("hour") is not None:
+        quoted_time = quote_identifier(engine, columns.get("time"))
+        if engine.dialect.name == "postgresql":
+            time_text = f"TRIM(CAST({quoted_time} AS TEXT))"
+            conditions.append(
+                "CASE WHEN "
+                f"{time_text} ~ '{SALES_TIME_TEXT_PATTERN}' "
+                f"THEN CAST(SUBSTRING({time_text} FROM 1 FOR 2) AS INTEGER) END "
+                "= :filter_hour"
+            )
+        else:
+            time_text = f"TRIM(CAST({quoted_time} AS TEXT))"
+            conditions.append(
+                f"CASE WHEN {time_text} GLOB '[0-2][0-9]:[0-5][0-9]*' "
+                "THEN CAST(strftime('%H', '2000-01-01 ' || "
+                f"{time_text}) AS INTEGER) END = :filter_hour"
+            )
+        params["filter_hour"] = int(filters["hour"])
+
+    keyword = filters.get("keyword")
+    if keyword:
+        keyword_columns = [columns.get(key) for key in ("name", "pid", "brand", "vendor")]
+        keyword_conditions = [
+            f"LOWER(CAST({quote_identifier(engine, column)} AS TEXT)) "
+            "LIKE LOWER(:filter_keyword) ESCAPE '\\'"
+            for column in keyword_columns
+            if column
+        ]
+        if not keyword_conditions:
+            raise ValueError("目前資料來源沒有可搜尋欄位")
+        conditions.append(f"({' OR '.join(keyword_conditions)})")
+        escaped_keyword = (
+            keyword.replace("\\", "\\\\")
+            .replace("%", "\\%")
+            .replace("_", "\\_")
+        )
+        params["filter_keyword"] = f"%{escaped_keyword}%"
+
+    if include_numeric:
+        amount = columns.get("amount")
+        qty = columns.get("qty")
+        if filters.get("min_price") is not None or filters.get("max_price") is not None:
+            price_sql = (
+                f"CAST({quote_identifier(engine, amount)} AS REAL) / "
+                f"NULLIF(CAST({quote_identifier(engine, qty)} AS REAL), 0)"
+            )
+            for bound in ("min_price", "max_price"):
+                value = filters.get(bound)
+                if value is not None:
+                    operator = ">=" if bound == "min_price" else "<="
+                    conditions.append(f"{price_sql} {operator} :filter_{bound}")
+                    params[f"filter_{bound}"] = value
+
+        if filters.get("min_margin") is not None or filters.get("max_margin") is not None:
+            quoted_amount = quote_identifier(engine, amount)
+            if columns.get("profit"):
+                profit_sql = f"CAST({quote_identifier(engine, columns['profit'])} AS REAL)"
+            elif columns.get("cost"):
+                profit_sql = (
+                    f"CAST({quoted_amount} AS REAL) - "
+                    f"CAST({quote_identifier(engine, columns['cost'])} AS REAL)"
+                )
+            else:
+                raise ValueError("目前資料來源沒有毛利或成本欄位")
+            margin_sql = f"({profit_sql} * 100.0 / NULLIF(CAST({quoted_amount} AS REAL), 0))"
+            for bound in ("min_margin", "max_margin"):
+                value = filters.get(bound)
+                if value is not None:
+                    operator = ">=" if bound == "min_margin" else "<="
+                    conditions.append(f"{margin_sql} {operator} :filter_{bound}")
+                    params[f"filter_{bound}"] = value
+
+    return " AND ".join(conditions), params
+
+
+def prepare_sales_query_context(
+    engine,
+    table_name: str,
+    args: Mapping[str, Any],
+    *,
+    default_data_range: int = 1,
+) -> tuple[dict[str, Any], dict[str, str | None]]:
+    """Resolve canonical filters, live columns and the dynamic Other bucket."""
+    table_name = validate_table_name(table_name)
+    filters = normalize_sales_query_args(args, default_data_range=default_data_range)
+    columns = resolve_sales_query_columns(engine, table_name)
+    if filters.get("category") != "其他":
+        return filters, columns
+
+    ranking_filters = dict(filters)
+    ranking_filters["category"] = None
+    where_sql, params = build_sales_where_clause(engine, columns, ranking_filters)
+    category = quote_identifier(engine, columns.get("category"))
+    amount = quote_identifier(engine, columns.get("amount"))
+    table = quote_identifier(engine, table_name)
+    where_suffix = f" AND {where_sql}" if where_sql else ""
+    query = text(f"""
+        SELECT {category} AS category_name
+        FROM {table}
+        WHERE {category} IS NOT NULL {where_suffix}
+        GROUP BY {category}
+        ORDER BY SUM(CAST({amount} AS REAL)) DESC, CAST({category} AS TEXT) ASC
+        LIMIT 12
+    """)
+    with engine.connect() as connection:
+        rows = connection.execute(query, params).fetchall()
+    filters["category_other_exclusions"] = [str(row[0]) for row in rows]
+    return filters, columns
--- a/templates/sales_analysis.html
+++ b/templates/sales_analysis.html
@@ -13,6 +13,12 @@
  - Marketing chart titles：text-primary 藍 / text-success 綠 → page-accent / olive
 #}

+{% macro period_badge(period) -%}
+  <span class="sa-period-badge" data-analysis-period="{{ period.start_date }}:{{ period.end_date }}">
+    <i class="fas fa-calendar-alt" aria-hidden="true"></i>{{ period.label }}
+  </span>
+{%- endmacro %}
+
 {% block extra_css %}
  <link rel="stylesheet" href="https://cdn.datatables.net/1.11.5/css/dataTables.bootstrap5.min.css">
  <link rel="stylesheet" href="https://cdn.jsdelivr.net/npm/flatpickr@4.6.13/dist/flatpickr.min.css">
@@ -141,7 +147,7 @@
                  <label class="form-label">
                    <i class="fas fa-calendar-week me-1" aria-hidden="true"></i>期間快速選擇
                  </label>
-                  <select name="data_range" class="form-select" onchange="handleDataRangeChange(this)">
+                  <select id="data_range" name="data_range" class="form-select" onchange="handleDataRangeChange(this)">
                    <option value="" {% if not request.args.get('data_range') %}selected{% endif %}>選擇期間</option>
                    {% for opt in [(1,'最近 1 個月 (推薦)'), (3,'最近 3 個月'), (6,'最近 6 個月'), (12,'最近 12 個月'), (0,'全部資料')] %}
                    <option value="{{ opt.0 }}" {% if request.args.get('data_range') == opt.0|string %}selected{% endif %}>{{ opt.1 }}</option>
@@ -279,7 +285,7 @@
              </h6>
              <div class="row g-2">
                <!-- 月份 -->
-                <div class="col-md-4">
+                <div class="{{ 'col-md-4' if cols.hour else 'col-md-6' }}">
                  <label class="form-label"><i class="fas fa-calendar-alt me-1" aria-hidden="true"></i>月份</label>
                  <div class="btn-group w-100 sa-dropdown-filter">
                    <button type="button" class="btn btn-outline-secondary dropdown-toggle w-100 text-start text-truncate" data-bs-toggle="dropdown" aria-expanded="false">
@@ -298,7 +304,7 @@
                </div>
                <!-- 星期 -->
                {% set dow_labels = {'all':'全部星期','0':'週一','1':'週二','2':'週三','3':'週四','4':'週五','5':'週六','6':'週日'} %}
-                <div class="col-md-4">
+                <div class="{{ 'col-md-4' if cols.hour else 'col-md-6' }}">
                  <label class="form-label"><i class="fas fa-calendar-day me-1" aria-hidden="true"></i>星期</label>
                  <div class="btn-group w-100 sa-dropdown-filter">
                    <button type="button" class="btn btn-outline-secondary dropdown-toggle w-100 text-start text-truncate" data-bs-toggle="dropdown" aria-expanded="false">
@@ -315,6 +321,7 @@
                  </div>
                </div>
                <!-- 時段 -->
+                {% if cols.hour %}
                <div class="col-md-4">
                  <label class="form-label"><i class="fas fa-clock me-1" aria-hidden="true"></i>時段</label>
                  <div class="btn-group w-100 sa-dropdown-filter">
@@ -332,6 +339,7 @@
                    <input type="hidden" name="hour" value="{{ selected_hour }}">
                  </div>
                </div>
+                {% endif %}
              </div>
            </div>
          </div>
@@ -479,21 +487,18 @@
                <i class="fas fa-chart-line me-2" aria-hidden="true"></i>年度成長對照
              </h6>
              <div class="sa-yoy-controls">
+                {{ period_badge(analysis_scope) }}
                <select id="yoy-year1" class="form-select form-select-sm" aria-label="基準年">
-                  <option value="2025" selected>2025</option>
-                  <option value="2024">2024</option>
-                  <option value="2023">2023</option>
+                  {% for year in yoy_years %}<option value="{{ year }}" {% if year == yoy_year1 %}selected{% endif %}>{{ year }}</option>{% endfor %}
                </select>
                <span class="sa-yoy-controls__sep">vs</span>
                <select id="yoy-year2" class="form-select form-select-sm" aria-label="對比年">
-                  <option value="2026" selected>2026</option>
-                  <option value="2025">2025</option>
-                  <option value="2024">2024</option>
+                  {% for year in yoy_years %}<option value="{{ year }}" {% if year == yoy_year2 %}selected{% endif %}>{{ year }}</option>{% endfor %}
                </select>
                <select id="yoy-metric" class="form-select form-select-sm" aria-label="指標">
                  <option value="revenue" selected>銷售金額</option>
                  <option value="qty">銷售數量</option>
-                  <option value="profit">毛利金額</option>
+                  {% if cols.cost or cols.profit %}<option value="profit">毛利金額</option>{% endif %}
                </select>
                <button class="btn btn-sm btn-outline-primary" onclick="loadYoYData()" title="重新載入">
                  <i class="fas fa-sync-alt" aria-hidden="true"></i>
@@ -505,7 +510,7 @@
              <div class="col-md-4">
                <div class="card sa-yoy-card">
                  <div class="card-body text-center py-3">
-                    <small id="yoy-year1-label">2024年</small>
+                    <small id="yoy-year1-label">{{ yoy_year1 }}年</small>
                    <h4 class="sa-yoy-card__value sa-yoy-card__value--neutral" id="yoy-year1-value">$0</h4>
                  </div>
                </div>
@@ -513,7 +518,7 @@
              <div class="col-md-4">
                <div class="card sa-yoy-card">
                  <div class="card-body text-center py-3">
-                    <small id="yoy-year2-label">2025年</small>
+                    <small id="yoy-year2-label">{{ yoy_year2 }}年</small>
                    <h4 class="sa-yoy-card__value sa-yoy-card__value--accent" id="yoy-year2-value">$0</h4>
                  </div>
                </div>
@@ -542,11 +547,14 @@
    <section class="row mb-4">
      <div class="col-12">
        <div class="card">
-          <div class="card-header d-flex justify-content-between align-items-center">
+          <div class="card-header d-flex justify-content-between align-items-center gap-2 flex-wrap">
            <span><i class="fas fa-industry me-2" aria-hidden="true"></i>廠商毛利與主推優先序</span>
-            <a href="/api/export/excel/vendor?{{ request.query_string.decode() }}" class="btn btn-sm btn-outline-success">
-              <i class="fas fa-file-excel me-1" aria-hidden="true"></i>匯出清單
-            </a>
+            <div class="d-flex align-items-center gap-2">
+              {{ period_badge(analysis_scope) }}
+              <a href="/api/sales_analysis/export_vendor?{{ request.query_string.decode() }}" class="btn btn-sm btn-outline-success">
+                <i class="fas fa-file-excel me-1" aria-hidden="true"></i>匯出清單
+              </a>
+            </div>
          </div>
          <div class="card-body">
            <div class="table-responsive sa-vendor-table">
@@ -570,7 +578,8 @@
                  <tr>
                    <td class="text-center text-muted">{{ loop.index }}</td>
                    <td class="fw-bold">
-                      <a href="javascript:setFilter('vendor', '{{ v.name }}')" class="sa-vendor-link" title="篩選此廠商商品">
+                      <a href="#" class="sa-vendor-link" title="篩選此廠商商品"
+                         data-sales-filter-key="vendor" data-sales-filter-value="{{ v.name }}">
                        {{ v.name }}
                        <i class="fas fa-filter ms-1 sa-vendor-link__icon" aria-hidden="true"></i>
                      </a>
@@ -684,13 +693,19 @@
    <div class="row">
      <div class="col-lg-8">
        <div class="card">
-          <div class="card-header"><i class="fas fa-chart-bar me-2" aria-hidden="true"></i>Top 20 熱銷排行（{{ '銷售金額' if selected_metric == 'amount' else '銷售數量' }}）</div>
+          <div class="card-header d-flex justify-content-between align-items-center gap-2 flex-wrap">
+            <span><i class="fas fa-chart-bar me-2" aria-hidden="true"></i>Top 20 {{ {'amount':'業績','qty':'銷量','profit':'毛利'}.get(selected_metric, '業績') }}排行</span>
+            {{ period_badge(analysis_scope) }}
+          </div>
          <div class="card-body"><div class="sa-chart-shell" style="height: 600px;"><canvas id="barChart"></canvas></div></div>
        </div>
      </div>
      <div class="col-lg-4">
        <div class="card">
-          <div class="card-header"><i class="fas fa-chart-pie me-2" aria-hidden="true"></i>類別業績佔比</div>
+          <div class="card-header d-flex justify-content-between align-items-center gap-2 flex-wrap">
+            <span><i class="fas fa-chart-pie me-2" aria-hidden="true"></i>類別業績佔比</span>
+            {{ period_badge(analysis_scope) }}
+          </div>
          <div class="card-body"><div class="sa-chart-shell" style="height: 400px;"><canvas id="categoryChart"></canvas></div></div>
        </div>
      </div>
@@ -699,7 +714,10 @@
    <div class="row mt-4">
      <div class="col-12">
        <div class="card">
-          <div class="card-header"><i class="fas fa-th-large me-2" aria-hidden="true"></i>業績板塊分佈：分類到商品</div>
+          <div class="card-header d-flex justify-content-between align-items-center gap-2 flex-wrap">
+            <span><i class="fas fa-th-large me-2" aria-hidden="true"></i>業績板塊分佈：分類到商品</span>
+            {{ period_badge(analysis_scope) }}
+          </div>
          <div class="card-body"><div class="sa-chart-shell" style="height: 400px;"><canvas id="treemapChart"></canvas></div></div>
        </div>
      </div>
@@ -708,13 +726,19 @@
    <div class="row mt-4">
      <div class="col-lg-6">
        <div class="card">
-          <div class="card-header"><i class="fas fa-chart-column me-2" aria-hidden="true"></i>價格帶業績貢獻</div>
+          <div class="card-header d-flex justify-content-between align-items-center gap-2 flex-wrap">
+            <span><i class="fas fa-chart-column me-2" aria-hidden="true"></i>價格帶業績貢獻</span>
+            {{ period_badge(analysis_scope) }}
+          </div>
          <div class="card-body"><div class="sa-chart-shell" style="height: 350px;"><canvas id="priceDistChart"></canvas></div></div>
        </div>
      </div>
      <div class="col-lg-6">
        <div class="card">
-          <div class="card-header"><i class="fas fa-braille me-2" aria-hidden="true"></i>價格與銷量分佈</div>
+          <div class="card-header d-flex justify-content-between align-items-center gap-2 flex-wrap">
+            <span><i class="fas fa-braille me-2" aria-hidden="true"></i>價格與銷量分佈</span>
+            {{ period_badge(analysis_scope) }}
+          </div>
          <div class="card-body"><div class="sa-chart-shell" style="height: 350px;"><canvas id="scatterChart"></canvas></div></div>
        </div>
      </div>
@@ -724,7 +748,10 @@
    <div class="row mt-4">
      <div class="col-12">
        <div class="card">
-          <div class="card-header"><i class="fas fa-chess-board me-2" aria-hidden="true"></i>商品策略矩陣</div>
+          <div class="card-header d-flex justify-content-between align-items-center gap-2 flex-wrap">
+            <span><i class="fas fa-chess-board me-2" aria-hidden="true"></i>商品策略矩陣</span>
+            {{ period_badge(analysis_scope) }}
+          </div>
          <div class="card-body">
            <p class="text-muted small mb-2"><i class="fas fa-info-circle me-1" aria-hidden="true"></i>右上優先主推；右下先守價或調整組合。</p>
            <div class="sa-chart-shell" style="height: 500px;"><canvas id="bcgChart"></canvas></div>
@@ -738,7 +765,10 @@
    <div class="row mt-4">
      <div class="col-12">
        <div class="card">
-          <div class="card-header"><i class="fas fa-sun me-2" aria-hidden="true"></i>淡旺季熱力圖：Top 10 分類</div>
+          <div class="card-header d-flex justify-content-between align-items-center gap-2 flex-wrap">
+            <span><i class="fas fa-sun me-2" aria-hidden="true"></i>淡旺季趨勢：Top 10 分類</span>
+            {{ period_badge(analysis_scope) }}
+          </div>
          <div class="card-body"><div class="sa-chart-shell" style="height: 400px;"><canvas id="seasonalityChart"></canvas></div></div>
        </div>
      </div>
@@ -749,11 +779,14 @@
    <div class="row mt-4">
      <div class="col-12">
        <div class="card">
-          <div class="card-header d-flex justify-content-between align-items-center">
+          <div class="card-header d-flex justify-content-between align-items-center gap-2 flex-wrap">
            <span><i class="fas fa-bullhorn me-2" aria-hidden="true"></i>活動業績貢獻</span>
-            <button class="btn btn-sm btn-outline-success" onclick="exportMarketingExcel('all')">
-              <i class="fas fa-file-excel me-1" aria-hidden="true"></i>匯出全部
-            </button>
+            <div class="d-flex align-items-center gap-2">
+              {{ period_badge(analysis_scope) }}
+              <button class="btn btn-sm btn-outline-success" onclick="exportMarketingExcel('all')">
+                <i class="fas fa-file-excel me-1" aria-hidden="true"></i>匯出全部
+              </button>
+            </div>
          </div>
          <div class="card-body">
            <div class="row">
@@ -776,46 +809,64 @@
    <div class="row mt-4">
      <div class="col-lg-6">
        <div class="card">
-          <div class="card-header"><i class="fas fa-calendar-alt me-2" aria-hidden="true"></i>每月業績趨勢</div>
+          <div class="card-header d-flex justify-content-between align-items-center gap-2 flex-wrap">
+            <span><i class="fas fa-calendar-alt me-2" aria-hidden="true"></i>每月業績趨勢</span>
+            {{ period_badge(analysis_scope) }}
+          </div>
          <div class="card-body"><div class="sa-chart-shell" style="height: 300px;"><canvas id="monthlyChart"></canvas></div></div>
        </div>
      </div>
      <div class="col-lg-6">
        <div class="card">
-          <div class="card-header"><i class="fas fa-chart-line me-2" aria-hidden="true"></i>每週業績趨勢（{{ analysis_period.label }}）</div>
+          <div class="card-header d-flex justify-content-between align-items-center gap-2 flex-wrap">
+            <span><i class="fas fa-chart-line me-2" aria-hidden="true"></i>每週業績趨勢</span>
+            {{ period_badge(analysis_scope) }}
+          </div>
          <div class="card-body"><div class="sa-chart-shell" style="height: 300px;"><canvas id="weeklyChart"></canvas></div></div>
        </div>
      </div>
    </div>

    <div class="row mt-4">
-      <div class="col-lg-4">
+      <div class="{{ 'col-lg-4' if cols.hour else 'col-12' }}">
        <div class="card">
-          <div class="card-header"><i class="fas fa-calendar-week me-2" aria-hidden="true"></i>每日業績趨勢 (週一至週日)</div>
+          <div class="card-header d-flex justify-content-between align-items-center gap-2 flex-wrap">
+            <span><i class="fas fa-calendar-week me-2" aria-hidden="true"></i>星期別業績分佈</span>
+            {{ period_badge(analysis_scope) }}
+          </div>
          <div class="card-body"><div class="sa-chart-shell" style="height: 300px;"><canvas id="dowChart"></canvas></div></div>
        </div>
      </div>
+      {% if cols.hour %}
      <div class="col-lg-4">
        <div class="card">
-          <div class="card-header"><i class="fas fa-clock me-2" aria-hidden="true"></i>每小時業績熱點</div>
+          <div class="card-header d-flex justify-content-between align-items-center gap-2 flex-wrap">
+            <span><i class="fas fa-clock me-2" aria-hidden="true"></i>每小時業績熱點</span>
+            {{ period_badge(analysis_scope) }}
+          </div>
          <div class="card-body"><div class="sa-chart-shell" style="height: 300px;"><canvas id="hourlyChart"></canvas></div></div>
        </div>
      </div>
      <div class="col-lg-4">
        <div class="card">
-          <div class="card-header">
-            <i class="fas fa-th me-2" aria-hidden="true"></i>多維度熱點 (星期 × 小時)
-            <i class="fas fa-info-circle text-muted ms-1" data-bs-toggle="tooltip" title="此圖表用於識別每週的「黃金銷售時段」。顏色越深代表該時段業績越好。" aria-hidden="true"></i>
+          <div class="card-header d-flex justify-content-between align-items-center gap-2 flex-wrap">
+            <span>
+              <i class="fas fa-th me-2" aria-hidden="true"></i>星期 × 小時業績熱點
+              <i class="fas fa-info-circle text-muted ms-1" data-bs-toggle="tooltip" title="顏色越深代表該時段業績越高。" aria-hidden="true"></i>
+            </span>
+            {{ period_badge(analysis_scope) }}
          </div>
          <div class="card-body"><div class="sa-chart-shell" style="height: 300px;"><canvas id="heatmapChart"></canvas></div></div>
        </div>
      </div>
+      {% endif %}
    </div>
    {% endif %}

    <div class="card mt-4">
-      <div class="card-header d-flex justify-content-between align-items-center">
-        <span><i class="fas fa-list-ol me-2" aria-hidden="true"></i>商品作戰清單（{{ analysis_period.label }}）</span>
+      <div class="card-header d-flex justify-content-between align-items-center gap-2 flex-wrap">
+        <span><i class="fas fa-list-ol me-2" aria-hidden="true"></i>商品作戰清單</span>
+        {{ period_badge(analysis_scope) }}
      </div>
      <div class="card-body">
        <table id="dataTable" class="table table-hover align-middle mb-0" style="width:100%">
@@ -840,6 +891,43 @@
      </div>
    </div>

+    <div class="modal fade" id="topDetailModal" tabindex="-1" aria-labelledby="topDetailModalLabel" aria-hidden="true">
+      <div class="modal-dialog modal-xl modal-dialog-scrollable">
+        <div class="modal-content">
+          <div class="modal-header">
+            <div>
+              <h5 class="modal-title" id="topDetailModalLabel">商品排行明細</h5>
+              <div class="sa-modal-period">{{ period_badge(analysis_scope) }}</div>
+            </div>
+            <button type="button" class="btn-close" data-bs-dismiss="modal" aria-label="關閉"></button>
+          </div>
+          <div class="modal-body">
+            <div id="topDetailStatus" class="sa-modal-status" role="status">正在載入排行...</div>
+            <div class="table-responsive">
+              <table class="table table-hover align-middle mb-0">
+                <thead>
+                  <tr>
+                    <th>排名</th>
+                    <th>商品</th>
+                    <th>分類</th>
+                    <th class="text-end">數值</th>
+                    <th class="text-end">毛利率</th>
+                  </tr>
+                </thead>
+                <tbody id="topDetailRows"></tbody>
+              </table>
+            </div>
+          </div>
+          <div class="modal-footer">
+            <a id="topDetailExport" class="btn btn-outline-success" href="#">
+              <i class="fas fa-file-excel me-1" aria-hidden="true"></i>匯出明細
+            </a>
+            <button type="button" class="btn btn-secondary" data-bs-dismiss="modal">關閉</button>
+          </div>
+        </div>
+      </div>
+    </div>
+
    {% endif %}{# /no_filter else #}
  </div>{# /sales-analysis-page #}
  {% endif %}{# /error #}
--- a/tests/test_analysis_period_linkage.py
+++ b/tests/test_analysis_period_linkage.py
@@ -1,4 +1,4 @@
-from datetime import date
+from datetime import date, datetime as real_datetime
 from pathlib import Path

 import pandas as pd
@@ -296,12 +296,27 @@ def test_sales_frontend_uses_live_period_linked_api_routes():
    assert "new URLSearchParams(window.location.search)" in script
    assert "/api/sales_analysis/yoy_comparison" in script
    assert "/api/sales_analysis/table_data" in script
+    assert "/api/sales_analysis/top_detail" in script
+    assert "/api/sales_analysis/export_vendor" in template
+    assert "/api/sales_analysis/export_marketing" in script
+    assert "/api/export/excel/marketing" not in script
+    assert "topDetailModal" in script
+    assert "window.open(url.toString()" not in script
+    assert "form.requestSubmit()" in script
+    assert "if (start) start.value = '';" in script
+    assert "if (end) end.value = '';" in script
    assert "d.growth_rate" in script
    assert "d.monthly_breakdown" in script
+    assert "marketingMetric === 'qty' ? fmtNum : fmtMoney" in script
+    assert "order: []" in script
    assert "columns," in script
    assert 'data-field="product_id"' in template
    assert 'data-field="amount"' in template
-    assert "每週業績趨勢（{{ analysis_period.label }}）" in template
+    assert "data-analysis-period" in template
+    assert 'id="topDetailModal"' in template
+    assert template.count("period_badge(analysis_scope)") >= 10
+    assert "yoy_years" in template
+    assert '<option value="2026" selected>' not in template
    assert "全年度" not in template
    assert "fetch(`/api/yoy?" not in script
    assert "'/api/sales_table?'" not in script
@@ -315,6 +330,48 @@ def test_yoy_period_projects_selected_dates_to_each_comparison_year():
    assert _period_month_keys(start, end) == ["2025-04"]


+def test_yoy_period_projects_rolling_range_instead_of_falling_back_to_full_year():
+    start, end = _project_yoy_period(
+        2025,
+        data_range_value="6",
+        anchor_date="2026-07-22",
+    )
+
+    assert start == date(2025, 1, 23)
+    assert end == date(2025, 7, 22)
+
+
+def test_yoy_period_accepts_the_year_month_value_sent_by_the_page_filter():
+    start, end = _project_yoy_period(2025, month_value="2026-04")
+
+    assert start == date(2025, 4, 1)
+    assert end == date(2025, 4, 30)
+
+
+def test_yoy_period_intersects_month_with_the_selected_custom_range():
+    start, end = _project_yoy_period(
+        2025,
+        start_value="2026-03-15",
+        end_value="2026-05-10",
+        month_value="2026-04",
+    )
+
+    assert start == date(2025, 4, 1)
+    assert end == date(2025, 4, 30)
+
+
+def test_yoy_period_intersects_month_with_a_cross_year_custom_range():
+    start, end = _project_yoy_period(
+        2025,
+        start_value="2025-11-15",
+        end_value="2026-02-10",
+        month_value="2025-12",
+    )
+
+    assert start == date(2025, 12, 1)
+    assert end == date(2025, 12, 31)
+
+
 def test_yoy_api_uses_only_the_selected_period_for_totals_and_chart(monkeypatch):
    engine = create_engine("sqlite:///:memory:")
    with engine.begin() as conn:
@@ -361,3 +418,107 @@ def test_yoy_api_uses_only_the_selected_period_for_totals_and_chart(monkeypatch)
        "year2_value": 150,
        "growth_rate": 50,
    }]
+
+
+def test_sales_page_builds_every_chart_from_the_same_custom_period(monkeypatch):
+    engine = create_engine("sqlite:///:memory:")
+    with engine.begin() as conn:
+        conn.exec_driver_sql(
+            '''
+            CREATE TABLE realtime_sales_monthly (
+                "日期" TEXT,
+                "時間" TEXT,
+                "商品ID" TEXT,
+                "商品名稱" TEXT,
+                "商品館" TEXT,
+                "品牌" TEXT,
+                "廠商名稱" TEXT,
+                "總業績" REAL,
+                "數量" REAL,
+                "總成本" REAL,
+                "折扣活動名稱" TEXT,
+                "折價券活動名稱" TEXT
+            )
+            '''
+        )
+        conn.exec_driver_sql(
+            '''
+            INSERT INTO realtime_sales_monthly
+                ("日期", "時間", "商品ID", "商品名稱", "商品館", "品牌", "廠商名稱",
+                 "總業績", "數量", "總成本", "折扣活動名稱", "折價券活動名稱")
+            VALUES
+                ('2026/04/10', '10:00:00', 'P1', '商品 A', '美妝', '品牌 A', '廠商 A', 100, 2, 60, '母親節', ''),
+                ('2026/04/11', '21:00:00', 'P2', '商品 B', '3C', '品牌 B', '廠商 B', 300, 3, 210, '', '折價券'),
+                ('2026/07/01', '12:00:00', 'P3', '區間外商品', '家電', '品牌 C', '廠商 C', 9999, 1, 1, '暑期', '')
+            '''
+        )
+
+    class FakeDatabaseManager:
+        def __init__(self):
+            self.engine = engine
+
+    monkeypatch.setattr(sales_routes, "DatabaseManager", FakeDatabaseManager)
+    monkeypatch.setattr(sales_routes, "DATABASE_TYPE", "sqlite")
+    monkeypatch.setattr(sales_routes, "_get_sales_shared_page_context_cache", lambda _key: None)
+    monkeypatch.setattr(sales_routes, "_set_sales_shared_page_context_cache", lambda _key, _context: None)
+    monkeypatch.setattr(sales_routes, "render_template", lambda _name, **context: context)
+    sales_routes._SALES_PROCESSED_CACHE.clear()
+    sales_routes._SALES_OPTIONS_CACHE.clear()
+    sales_routes._SALES_ANALYSIS_RESULT_CACHE.clear()
+
+    app = Flask(__name__)
+    with app.test_request_context(
+        "/sales_analysis?metric=profit&start_date=2026-04-01&end_date=2026-04-30"
+    ):
+        context = sales_routes.sales_analysis.__wrapped__()
+
+    assert context["analysis_period"]["label"] == "2026 年 04 月"
+    assert context["analysis_scope"]["label"] == "2026 年 04 月"
+    assert context["total_records"] == 2
+    assert context["kpi"]["revenue"] == 400
+    assert context["bar_data"]["metric_label"] == "毛利金額 ($)"
+    assert context["bar_data"]["chart_values"] == [90.0, 40.0]
+    assert {item["name"] for item in context["treemap_data"]} == {"商品 A", "商品 B"}
+    assert len(context["bcg_data"]["points"]) == 2
+    assert context["heatmap_data"]["matrix"][4][10] == 100
+    assert context["heatmap_data"]["matrix"][5][21] == 300
+    assert context["seasonality_data"]["months"] == ["2026-04"]
+    assert context["all_months"] == ["2026-04"]
+    assert context["marketing_data"]["discount"] == {
+        "labels": ["母親節"],
+        "values": [40.0],
+    }
+    assert context["marketing_data"]["coupon"] == {
+        "labels": ["折價券"],
+        "values": [90.0],
+    }
+    assert context["yoy_year1"] == 2025
+    assert context["yoy_year2"] == 2026
+
+    sales_routes._TABLE_DATA_CACHE.clear()
+    with app.test_request_context(
+        "/api/sales_analysis/table_data"
+        "?metric=profit&start_date=2026-04-01&end_date=2026-04-30&category=美妝"
+    ):
+        table_response = sales_routes.api_sales_table_data.__wrapped__()
+    table_payload = table_response.get_json()
+
+    assert len(table_payload["data"]) == 1
+    assert table_payload["data"][0]["product_id"] == "P1"
+    assert table_payload["data"][0]["amount"] == 100
+
+    class FixedDateTime(real_datetime):
+        @classmethod
+        def now(cls, tz=None):
+            value = cls(2026, 7, 22, 12, 0, 0)
+            return value.replace(tzinfo=tz) if tz else value
+
+    monkeypatch.setattr(sales_routes, "datetime", FixedDateTime)
+    sales_routes._SALES_PROCESSED_CACHE.clear()
+    sales_routes._SALES_ANALYSIS_RESULT_CACHE.clear()
+    with app.test_request_context("/sales_analysis?metric=amount&data_range=6"):
+        rolling_context = sales_routes.sales_analysis.__wrapped__()
+
+    assert rolling_context["analysis_period"]["mode"] == "rolling_months"
+    assert rolling_context["analysis_period"]["label"] == "最近 6 個月"
+    assert rolling_context["total_records"] == 3
--- a/tests/test_nemotron_decision_canary_service.py
+++ b/tests/test_nemotron_decision_canary_service.py
@@ -3,6 +3,8 @@ import json
 import subprocess
 import sys

+import pytest
+

 class _Response:
    def __init__(self, payload, status_code=200):
@@ -19,12 +21,18 @@ class _Response:
        return self._payload


-def _tags(service, *, digest=None):
+def _tags(service, *, model=None, digest=None):
+    resolved_model = model or service.NEMOTRON_OLLAMA_MODEL
+    resolved_digest = digest or (
+        service.NEMOTRON_OLLAMA_FALLBACK_EXPECTED_DIGEST
+        if resolved_model == service.NEMOTRON_OLLAMA_FALLBACK_MODEL
+        else service.NEMOTRON_OLLAMA_EXPECTED_DIGEST
+    )
    return {
        "models": [{
-            "name": service.NEMOTRON_OLLAMA_MODEL,
-            "model": service.NEMOTRON_OLLAMA_MODEL,
-            "digest": digest or service.NEMOTRON_OLLAMA_EXPECTED_DIGEST,
+            "name": resolved_model,
+            "model": resolved_model,
+            "digest": resolved_digest,
            "details": {
                "parameter_size": "14.8B",
                "quantization_level": "Q4_K_M",
@@ -33,6 +41,22 @@ def _tags(service, *, digest=None):
    }


+def _valid_canary_tool_call():
+    return {
+        "function": {
+            "name": "trigger_price_alert",
+            "arguments": {
+                "sku": "CANARY-NEMO-001",
+                "name": "NemoTron 受控決策驗證品",
+                "gap_pct": 22.4,
+                "sales_delta": -35.0,
+                "action": "建立價格告警候選",
+                "confidence": 0.91,
+            },
+        }
+    }
+
+
 def test_nemotron_canary_readback_never_calls_network(tmp_path, monkeypatch):
    from services import nemotron_decision_canary_service as service

@@ -113,7 +137,8 @@ def test_nemotron_canary_executes_decision_only_on_verified_secondary(
    assert posted[0][0].endswith("/api/chat")
    assert posted[0][1]["keep_alive"] == 0
    assert posted[0][1]["think"] is False
-    assert posted[0][2] == 45
+    assert posted[0][1]["options"]["num_predict"] == 2048
+    assert posted[0][2] == pytest.approx(45, abs=0.01)
    assert payload["latest_execution"]["canary_passed"] is True

    acknowledged = service.acknowledge_nemotron_decision_canary(
@@ -157,12 +182,263 @@ def test_nemotron_canary_blocks_model_digest_drift_before_model_call(

    assert payload["status"] == "canary_failed"
    assert payload["execution"]["model_call_performed"] is False
-    assert payload["execution"]["error"] == (
-        "no_approved_gcp_host_with_expected_nemotron_digest"
+    assert payload["execution"]["error"] == "all_approved_model_candidates_failed"
+    assert payload["execution"]["attempt_count"] == 3
+    assert all(
+        attempt["status"] == "identity_failed"
+        for attempt in payload["execution"]["runtime_attempts"]
    )
    assert model_calls == []


+def test_nemotron_canary_malformed_primary_tags_falls_through_secondary(
+    tmp_path, monkeypatch
+):
+    from services import nemotron_decision_canary_service as service
+
+    def fake_get(url, timeout):
+        if service.OLLAMA_HOST_PRIMARY in url:
+            return _Response({
+                "models": [{
+                    "name": service.NEMOTRON_OLLAMA_MODEL,
+                    "model": service.NEMOTRON_OLLAMA_MODEL,
+                    "digest": service.NEMOTRON_OLLAMA_EXPECTED_DIGEST,
+                    "details": [],
+                }]
+            })
+        return _Response(_tags(service))
+
+    monkeypatch.setattr(service.requests, "get", fake_get)
+    monkeypatch.setattr(
+        service.requests,
+        "post",
+        lambda *_args, **_kwargs: _Response({
+            "message": {
+                "role": "assistant",
+                "content": "",
+                "tool_calls": [_valid_canary_tool_call()],
+            }
+        }),
+    )
+
+    payload = service.run_nemotron_decision_canary(
+        output_root=tmp_path,
+        execute=True,
+        write_receipt=False,
+        timeout_sec=45,
+    )
+
+    execution = payload["execution"]
+    assert payload["status"] == "canary_passed"
+    assert execution["selected_host"] == service.OLLAMA_HOST_SECONDARY
+    assert execution["runtime_attempts"][0]["status"] == "identity_failed"
+    assert execution["runtime_attempts"][0]["error"] == "malformed_model_details"
+
+
+def test_nemotron_canary_uses_exact_digest_111_fallback_after_gcp_timeout(
+    tmp_path, monkeypatch
+):
+    import requests
+    from services import nemotron_decision_canary_service as service
+    from services.ollama_service import OLLAMA_HOST_FALLBACK
+
+    def fake_get(url, timeout):
+        assert timeout == service.DEFAULT_MODEL_IDENTITY_TIMEOUT_SEC
+        if service.OLLAMA_HOST_PRIMARY in url:
+            raise requests.ConnectionError("primary unavailable")
+        if service.OLLAMA_HOST_SECONDARY in url:
+            return _Response(_tags(service))
+        if OLLAMA_HOST_FALLBACK in url:
+            return _Response(
+                _tags(service, model=service.NEMOTRON_OLLAMA_FALLBACK_MODEL)
+            )
+        raise AssertionError(f"unexpected identity URL: {url}")
+
+    posted = []
+
+    def fake_post(url, json, timeout):
+        posted.append((url, json, timeout))
+        if service.OLLAMA_HOST_SECONDARY in url:
+            raise requests.Timeout("secondary saturated")
+        return _Response({
+            "message": {
+                "role": "assistant",
+                "content": "",
+                "tool_calls": [{
+                    "function": {
+                        "name": "trigger_price_alert",
+                        "arguments": {
+                            "sku": "CANARY-NEMO-001",
+                            "name": "NemoTron 受控決策驗證品",
+                            "gap_pct": 22.4,
+                            "sales_delta": -35.0,
+                            "action": "建立價格告警候選",
+                            "confidence": 0.91,
+                        },
+                    }
+                }],
+            },
+            "prompt_eval_count": 100,
+            "eval_count": 20,
+        })
+
+    monkeypatch.setattr(service.requests, "get", fake_get)
+    monkeypatch.setattr(service.requests, "post", fake_post)
+
+    payload = service.run_nemotron_decision_canary(
+        output_root=tmp_path,
+        execute=True,
+        write_receipt=True,
+        timeout_sec=90,
+        run_id="nemotron-canary-fallback-test-001",
+    )
+
+    assert payload["status"] == "canary_passed_degraded_fallback"
+    execution = payload["execution"]
+    assert execution["selected_host"] == OLLAMA_HOST_FALLBACK
+    assert execution["model"] == service.NEMOTRON_OLLAMA_FALLBACK_MODEL
+    assert execution["fallback_used"] is True
+    assert execution["degraded_runtime"] is True
+    assert execution["selected_candidate_label"] == "ollama_111_fallback"
+    assert execution["attempt_count"] == 3
+    assert posted[0][2] == 60
+    assert posted[1][2] == 45
+    assert posted[1][1]["model"] == service.NEMOTRON_OLLAMA_FALLBACK_MODEL
+    assert posted[1][1]["options"]["num_ctx"] == 4096
+    assert posted[1][1]["options"]["num_predict"] == 512
+    assert payload["next_machine_action"] == "restore_gcp_decision_runtime_capacity"
+    assert payload["latest_execution"]["fallback_used"] is True
+
+
+def test_nemotron_canary_timeout_is_one_total_deadline(tmp_path, monkeypatch):
+    import requests
+    from services import nemotron_decision_canary_service as service
+
+    clock = {"now": 50.0}
+    monkeypatch.setattr(service.time, "monotonic", lambda: clock["now"])
+    monkeypatch.setattr(
+        service,
+        "inspect_nemotron_model_identity",
+        lambda candidate, **_kwargs: {
+            **candidate.as_public_dict(),
+            "ok": True,
+            "digest": candidate.expected_digest,
+            "digest_matches": True,
+            "elapsed_ms": 0,
+            "error": None,
+        },
+    )
+    posted = []
+
+    def timeout_post(url, json, timeout):
+        posted.append((url, timeout))
+        clock["now"] += float(timeout)
+        raise requests.Timeout("simulated timeout")
+
+    monkeypatch.setattr(service.requests, "post", timeout_post)
+
+    payload = service.run_nemotron_decision_canary(
+        output_root=tmp_path,
+        execute=True,
+        write_receipt=False,
+        timeout_sec=30,
+    )
+
+    execution = payload["execution"]
+    assert payload["status"] == "canary_failed"
+    assert execution["error"] == "decision_total_deadline_exhausted"
+    assert execution["deadline_exhausted"] is True
+    assert execution["total_elapsed_ms"] == 30000
+    assert len(posted) == 1
+    assert posted[0][1] == 30
+    assert execution["attempt_count"] == 2
+
+
+@pytest.mark.parametrize(
+    ("tool_calls", "error_fragment"),
+    [
+        (
+            [
+                _valid_canary_tool_call(),
+                {
+                    "function": {
+                        "name": "delete_product",
+                        "arguments": {"sku": "CANARY-NEMO-001"},
+                    }
+                },
+            ],
+            "tool_call_count_mismatch",
+        ),
+        (
+            [_valid_canary_tool_call(), _valid_canary_tool_call()],
+            "tool_call_count_mismatch",
+        ),
+        (
+            [{
+                "function": {
+                    "name": "delete_product",
+                    "arguments": {"sku": "CANARY-NEMO-001"},
+                }
+            }],
+            "unknown_tool",
+        ),
+    ],
+)
+def test_nemotron_canary_rejects_extra_duplicate_or_unknown_tool_calls(
+    tmp_path, monkeypatch, tool_calls, error_fragment
+):
+    from services import nemotron_decision_canary_service as service
+    from services.ollama_service import OLLAMA_HOST_FALLBACK
+
+    def fake_get(url, timeout):
+        model = (
+            service.NEMOTRON_OLLAMA_FALLBACK_MODEL
+            if OLLAMA_HOST_FALLBACK in url
+            else service.NEMOTRON_OLLAMA_MODEL
+        )
+        return _Response(_tags(service, model=model))
+
+    monkeypatch.setattr(service.requests, "get", fake_get)
+    monkeypatch.setattr(
+        service.requests,
+        "post",
+        lambda *_args, **_kwargs: _Response({
+            "message": {
+                "role": "assistant",
+                "content": "",
+                "tool_calls": tool_calls,
+            }
+        }),
+    )
+
+    payload = service.run_nemotron_decision_canary(
+        output_root=tmp_path,
+        execute=True,
+        write_receipt=False,
+        timeout_sec=90,
+    )
+
+    execution = payload["execution"]
+    assert payload["status"] == "canary_failed"
+    assert execution["canary_passed"] is False
+    assert execution["tool_execution_count"] == 0
+    assert execution["database_call_performed"] is False
+    model_attempts = [
+        attempt
+        for attempt in execution["runtime_attempts"]
+        if attempt["model_call_performed"]
+    ]
+    assert len(model_attempts) == 3
+    assert all(
+        attempt["checks"]["decision_contract_valid"] is False
+        for attempt in model_attempts
+    )
+    assert all(
+        error_fragment in str(attempt.get("contract_error") or "")
+        for attempt in model_attempts
+    )
+
+
 def test_nemotron_canary_route_is_read_only():
    from routes import system_public_routes as routes

--- a/tests/test_nemotron_decision_envelope.py
+++ b/tests/test_nemotron_decision_envelope.py
@@ -62,7 +62,10 @@ def test_send_telegram_attaches_decision_envelope_to_event_router(monkeypatch):
        "severity": "P2",
        "guardrails": {"can_auto_execute": False},
    }
-    module.NemotronDispatcher()._send_telegram("hello price alert", decision_envelope=envelope)
+    outcome = module.NemotronDispatcher()._send_telegram(
+        "hello price alert",
+        decision_envelope=envelope,
+    )

    assert len(captured) == 1
    event = captured[0]
@@ -71,6 +74,25 @@ def test_send_telegram_attaches_decision_envelope_to_event_router(monkeypatch):
    assert event["decision_envelope"] == envelope
    assert event["payload"]["decision_envelope"] == envelope
    assert event["payload"]["raw_message"] == "hello price alert"
+    assert outcome["durable"] is False
+
+
+def test_send_telegram_requires_delivery_or_durable_queue(monkeypatch):
+    import services.nemoton_dispatcher_service as module
+
+    fake_event_router = types.ModuleType("services.event_router")
+    fake_event_router.dispatch_sync = lambda event: {
+        "delivered": False,
+        "queued": False,
+        "errors": ["telegram unavailable"],
+    }
+    monkeypatch.setitem(sys.modules, "services.event_router", fake_event_router)
+
+    outcome = module.NemotronDispatcher()._send_telegram("test")
+
+    assert outcome["durable"] is False
+    assert outcome["delivered"] is False
+    assert outcome["queued"] is False


 def test_exec_trigger_price_alert_persists_decision_envelope(monkeypatch):
@@ -82,15 +104,18 @@ def test_exec_trigger_price_alert_persists_decision_envelope(monkeypatch):
    monkeypatch.setattr(
        dispatcher,
        "_send_telegram",
-        lambda message, decision_envelope=None: sent.append((message, decision_envelope)),
+        lambda message, decision_envelope=None: (
+            sent.append((message, decision_envelope))
+            or {"durable": True, "delivered": True, "queued": False}
+        ),
    )
    monkeypatch.setattr(
        dispatcher,
        "_sink_insight_to_km",
-        lambda *args, **kwargs: stored.append((args, kwargs)),
+        lambda *args, **kwargs: (stored.append((args, kwargs)) or True),
    )

-    dispatcher._exec_trigger_price_alert(
+    outcome = dispatcher._exec_trigger_price_alert(
        sku="SKU-1",
        name="測試商品",
        gap_pct=12.5,
@@ -115,3 +140,4 @@ def test_exec_trigger_price_alert_persists_decision_envelope(monkeypatch):
    assert envelope["subject"]["competitor_product_id"] == "PC-1"
    assert stored
    assert stored[0][1]["metadata"]["decision_envelope"] == envelope
+    assert outcome["durable"] is True
--- a/tests/test_nemotron_dispatch_reservation_service.py
+++ b/tests/test_nemotron_dispatch_reservation_service.py
@@ -0,0 +1,281 @@
+import json
+import multiprocessing
+from pathlib import Path
+import subprocess
+import sys
+
+
+def _process_reserve(state_path, start_event, result_queue):
+    from services.nemotron_dispatch_reservation_service import (
+        SharedFileReservationStore,
+    )
+
+    store = SharedFileReservationStore(
+        state_path,
+        committed_ttl_sec=3600,
+        lease_ttl_sec=60,
+    )
+    start_event.wait(timeout=5)
+    result_queue.put(store.reserve("SKU-PROCESS-SHARED"))
+
+
+def test_shared_store_coordinates_independent_instances_and_ttl(tmp_path):
+    from services.nemotron_dispatch_reservation_service import (
+        SharedFileReservationStore,
+    )
+
+    clock = {"now": 1000.0}
+    path = tmp_path / "dedupe" / "state.json"
+    first = SharedFileReservationStore(
+        path,
+        committed_ttl_sec=100,
+        lease_ttl_sec=20,
+        clock=lambda: clock["now"],
+    )
+    second = SharedFileReservationStore(
+        path,
+        committed_ttl_sec=100,
+        lease_ttl_sec=20,
+        clock=lambda: clock["now"],
+    )
+
+    token = first.reserve("SKU-SHARED")
+    assert token
+    assert second.reserve("SKU-SHARED") is None
+    assert second.is_duplicate("SKU-SHARED") is True
+    assert "SKU-SHARED" not in path.read_text(encoding="utf-8")
+    assert second.release("SKU-SHARED", "stale-token") is False
+    assert first.refresh("SKU-SHARED", token) is True
+    assert first.commit("SKU-SHARED", token) is True
+    assert second.reserve("SKU-SHARED") is None
+
+    clock["now"] += 101
+    next_token = second.reserve("SKU-SHARED")
+    assert next_token and next_token != token
+    assert second.release("SKU-SHARED", next_token) is True
+
+
+def test_begin_side_effect_persists_extended_quarantine(tmp_path):
+    from services.nemotron_dispatch_reservation_service import (
+        SharedFileReservationStore,
+    )
+
+    clock = {"now": 1000.0}
+    path = tmp_path / "state.json"
+    first = SharedFileReservationStore(
+        path,
+        committed_ttl_sec=100,
+        lease_ttl_sec=20,
+        clock=lambda: clock["now"],
+    )
+    second = SharedFileReservationStore(
+        path,
+        committed_ttl_sec=100,
+        lease_ttl_sec=20,
+        clock=lambda: clock["now"],
+    )
+
+    token = first.reserve("SKU-SIDE-EFFECT")
+    assert token
+    assert first.begin_side_effect("SKU-SIDE-EFFECT", token) is True
+
+    payload = json.loads(path.read_text(encoding="utf-8"))
+    lease = next(iter(payload["leases"].values()))
+    assert lease == {
+        "token": token,
+        "expires_at": 1100.0,
+        "phase": "side_effect_started",
+    }
+
+    clock["now"] += 21
+    assert second.reserve("SKU-SIDE-EFFECT") is None
+    clock["now"] += 80
+    assert second.reserve("SKU-SIDE-EFFECT")
+
+
+def test_shared_store_is_atomic_across_processes(tmp_path):
+    state_path = str(tmp_path / "process" / "state.json")
+    context = multiprocessing.get_context("fork")
+    start_event = context.Event()
+    result_queue = context.Queue()
+    processes = [
+        context.Process(
+            target=_process_reserve,
+            args=(state_path, start_event, result_queue),
+        )
+        for _index in range(6)
+    ]
+    for process in processes:
+        process.start()
+    start_event.set()
+    tokens = [result_queue.get(timeout=10) for _process in processes]
+    for process in processes:
+        process.join(timeout=10)
+        assert process.exitcode == 0
+
+    owners = [token for token in tokens if token]
+    assert len(owners) == 1
+
+
+def test_shared_store_corruption_fails_closed(tmp_path):
+    from services.nemotron_dispatch_reservation_service import (
+        SharedFileReservationStore,
+    )
+
+    path = tmp_path / "state.json"
+    path.write_text("{not-json", encoding="utf-8")
+    store = SharedFileReservationStore(
+        path,
+        committed_ttl_sec=3600,
+        lease_ttl_sec=900,
+    )
+
+    assert store.reserve("SKU-CORRUPT") is None
+    assert store.refresh("SKU-CORRUPT", "token") is False
+    assert store.commit("SKU-CORRUPT", "token") is False
+    assert store.release("SKU-CORRUPT", "token") is False
+    assert store.is_duplicate("SKU-CORRUPT") is True
+    assert path.read_text(encoding="utf-8") == "{not-json"
+
+
+def test_atomic_write_fsyncs_file_and_parent_directory(tmp_path, monkeypatch):
+    from services import nemotron_dispatch_reservation_service as service
+
+    fsynced = []
+    real_fsync = service.os.fsync
+
+    def record_fsync(fd):
+        fsynced.append(fd)
+        return real_fsync(fd)
+
+    monkeypatch.setattr(service.os, "fsync", record_fsync)
+    store = service.SharedFileReservationStore(
+        tmp_path / "nested" / "state.json",
+        committed_ttl_sec=3600,
+        lease_ttl_sec=900,
+    )
+
+    assert store.reserve("SKU-FSYNC")
+    assert len(fsynced) >= 2
+
+
+def test_missing_process_shared_lock_fails_closed(monkeypatch, tmp_path):
+    from services import nemotron_dispatch_reservation_service as service
+
+    monkeypatch.setattr(service, "fcntl", None)
+    monkeypatch.delenv("FLASK_ENV", raising=False)
+    monkeypatch.delenv("USE_POSTGRESQL", raising=False)
+
+    assert service.build_production_shared_reservation_store(
+        committed_ttl_sec=3600,
+        lease_ttl_sec=900,
+    ) is None
+
+    store = service.SharedFileReservationStore(
+        tmp_path / "state.json",
+        committed_ttl_sec=3600,
+        lease_ttl_sec=900,
+    )
+    assert store.reserve("SKU-NO-FCNTL") is None
+    assert store.is_duplicate("SKU-NO-FCNTL") is True
+
+
+def test_production_builder_forces_canonical_shared_data_path(monkeypatch):
+    from services import nemotron_dispatch_reservation_service as service
+
+    monkeypatch.setenv("FLASK_ENV", "production")
+    store = service.build_production_shared_reservation_store(
+        committed_ttl_sec=3600,
+        lease_ttl_sec=900,
+    )
+
+    assert store is not None
+    assert store.state_path == Path(
+        "/app/data/ai_automation/nemotron_dispatch_dedupe_state.json"
+    )
+    assert store.lock_path == Path(
+        "/app/data/ai_automation/nemotron_dispatch_dedupe_state.json.lock"
+    )
+
+
+def test_postgres_runtime_cannot_fall_back_to_process_local_memory(monkeypatch):
+    from services import nemotron_dispatch_reservation_service as service
+
+    monkeypatch.delenv("FLASK_ENV", raising=False)
+    monkeypatch.setenv("USE_POSTGRESQL", "true")
+
+    assert service.production_shared_reservation_required() is True
+    assert service.build_production_shared_reservation_store(
+        committed_ttl_sec=3600,
+        lease_ttl_sec=900,
+    ) is not None
+
+
+def test_shared_state_contains_only_hashed_sku_and_timing_metadata(tmp_path):
+    from services.nemotron_dispatch_reservation_service import (
+        SharedFileReservationStore,
+    )
+
+    path = tmp_path / "state.json"
+    store = SharedFileReservationStore(
+        path,
+        committed_ttl_sec=3600,
+        lease_ttl_sec=900,
+    )
+    token = store.reserve("RAW-SKU-MUST-NOT-APPEAR")
+    payload = json.loads(path.read_text(encoding="utf-8"))
+
+    assert token
+    assert "RAW-SKU-MUST-NOT-APPEAR" not in path.read_text(encoding="utf-8")
+    assert set(payload) == {"version", "leases", "committed"}
+    assert len(next(iter(payload["leases"]))) == 64
+
+
+def test_production_containers_share_data_mount_and_force_shared_backend():
+    compose = (
+        Path(__file__).resolve().parents[1] / "docker-compose.yml"
+    ).read_text(encoding="utf-8")
+
+    assert compose.count("- ./data:/app/data") >= 3
+    assert compose.count("- FLASK_ENV=production") >= 3
+
+
+def test_shared_reservation_canary_has_no_business_side_effects(tmp_path):
+    from services.nemotron_dispatch_reservation_service import (
+        run_shared_reservation_canary,
+    )
+
+    payload = run_shared_reservation_canary(
+        state_path=tmp_path / "canary" / "state.json",
+    )
+
+    assert payload["success"] is True
+    assert payload["check_pass_count"] == payload["check_count"] == 6
+    assert payload["controlled_apply"] == {
+        "writes_shared_reservation_state": True,
+        "terminal_state_clear": True,
+        "writes_database": False,
+        "sends_telegram": False,
+        "executes_business_tool": False,
+        "reads_secret": False,
+    }
+
+
+def test_shared_reservation_canary_cli_is_machine_readable(tmp_path):
+    completed = subprocess.run(
+        [
+            sys.executable,
+            "scripts/ops/run_nemotron_dispatch_reservation_canary.py",
+            "--state-path",
+            str(tmp_path / "cli" / "state.json"),
+        ],
+        cwd=Path(__file__).resolve().parents[1],
+        capture_output=True,
+        check=False,
+        text=True,
+    )
+
+    assert completed.returncode == 0, completed.stderr
+    payload = json.loads(completed.stdout)
+    assert payload["status"] == "shared_reservation_canary_passed"
+    assert payload["controlled_apply"]["writes_database"] is False
--- a/tests/test_nemotron_fallback.py
+++ b/tests/test_nemotron_fallback.py
@@ -9,9 +9,11 @@ def _force_legacy_nim_first(monkeypatch):
    import services.nemoton_dispatcher_service as module

    module._ALERT_CACHE.clear()
+    module._ALERT_INFLIGHT.clear()
    monkeypatch.setattr(module, "NEMOTRON_OLLAMA_FIRST", False)
    yield
    module._ALERT_CACHE.clear()
+    module._ALERT_INFLIGHT.clear()


@dataclass
@@ -33,6 +35,7 @@ def _patch_execution_methods(monkeypatch, dispatcher):
    def record(kind):
        def _inner(*args, **kwargs):
            calls.append({"kind": kind, "args": args, "kwargs": kwargs})
+            return {"durable": True}
        return _inner

    monkeypatch.setattr(dispatcher, "_exec_trigger_price_alert", record("price_alert"))
@@ -56,6 +59,207 @@ def test_dispatch_falls_back_to_hermes_rules_without_nim_api_key(monkeypatch):
    assert calls[0]["kind"] == "price_alert"


+def test_forced_review_failure_is_counted_as_skipped(monkeypatch):
+    import services.nemoton_dispatcher_service as module
+
+    monkeypatch.setattr(module, "NIM_API_KEY", "")
+    dispatcher = module.NemotronDispatcher()
+    monkeypatch.setattr(
+        dispatcher,
+        "_exec_flag_for_human_review",
+        lambda *_args, **_kwargs: {"durable": False},
+    )
+    threat = FakeThreat("SKU-FORCED", "斷崖品")
+    threat.sales_7d_delta_pct = -100.0
+
+    result = dispatcher.dispatch([threat], hermes_stats={"duration_sec": 1})
+
+    assert result["dispatched"] == 0
+    assert result["skipped"] == 1
+    assert len(result["errors"]) == 1
+    assert "durable_side_effect_not_confirmed" in result["errors"][0]
+
+
+def test_hermes_failure_is_counted_as_skipped(monkeypatch):
+    import services.nemoton_dispatcher_service as module
+
+    monkeypatch.setattr(module, "NIM_API_KEY", "")
+    dispatcher = module.NemotronDispatcher()
+    monkeypatch.setattr(
+        dispatcher,
+        "_exec_trigger_price_alert",
+        lambda *_args, **_kwargs: {"durable": False},
+    )
+
+    result = dispatcher.dispatch(
+        [FakeThreat("SKU-HERMES-FAIL", "告警品")],
+        hermes_stats={"duration_sec": 1},
+    )
+
+    assert result["dispatched"] == 0
+    assert result["skipped"] == 1
+    assert len(result["errors"]) == 1
+    assert "durable_side_effect_not_confirmed" in result["errors"][0]
+
+
+def test_forced_review_commit_failure_keeps_durable_result_quarantined(monkeypatch):
+    import services.nemoton_dispatcher_service as module
+
+    monkeypatch.setattr(module, "NIM_API_KEY", "")
+    dispatcher = module.NemotronDispatcher()
+    monkeypatch.setattr(
+        dispatcher,
+        "_exec_flag_for_human_review",
+        lambda *_args, **_kwargs: {"durable": True},
+    )
+    monkeypatch.setattr(module, "_mark_alert_dispatched", lambda *_args: False)
+    released = []
+    monkeypatch.setattr(
+        module,
+        "_release_alert_reservation",
+        lambda *args: (released.append(args) or True),
+    )
+    threat = FakeThreat("SKU-FORCED-COMMIT", "斷崖品")
+    threat.sales_7d_delta_pct = -100.0
+
+    result = dispatcher.dispatch([threat], hermes_stats={"duration_sec": 1})
+
+    assert result["dispatched"] == 1
+    assert result["skipped"] == 0
+    assert result["errors"] == [
+        "dedupe_commit_unverified:flag_for_human_review(SKU-FORCED-COMMIT)"
+    ]
+    assert released == []
+
+
+def test_hermes_commit_failure_keeps_durable_result_quarantined(monkeypatch):
+    import services.nemoton_dispatcher_service as module
+
+    monkeypatch.setattr(module, "NIM_API_KEY", "")
+    dispatcher = module.NemotronDispatcher()
+    _patch_execution_methods(monkeypatch, dispatcher)
+    monkeypatch.setattr(module, "_mark_alert_dispatched", lambda *_args: False)
+    released = []
+    monkeypatch.setattr(
+        module,
+        "_release_alert_reservation",
+        lambda *args: (released.append(args) or True),
+    )
+
+    result = dispatcher.dispatch(
+        [FakeThreat("SKU-HERMES-COMMIT", "告警品")],
+        hermes_stats={"duration_sec": 1},
+    )
+
+    assert result["dispatched"] == 1
+    assert result["skipped"] == 0
+    assert result["errors"] == [
+        "dedupe_commit_unverified:hermes_rule_fallback(SKU-HERMES-COMMIT)"
+    ]
+    assert released == []
+
+
+def test_mixed_batch_cleanup_never_releases_forced_durable_quarantine(monkeypatch):
+    import services.nemoton_dispatcher_service as module
+
+    monkeypatch.setattr(module, "NIM_API_KEY", "test-key")
+    monkeypatch.setattr(module, "_check_nim_quota", lambda: True)
+    dispatcher = module.NemotronDispatcher()
+    _patch_execution_methods(monkeypatch, dispatcher)
+    forced = FakeThreat("SKU-FORCED-MIXED", "斷崖品")
+    forced.sales_7d_delta_pct = -100.0
+    candidate = FakeThreat("SKU-NIM-MIXED", "模型候選")
+    monkeypatch.setattr(
+        dispatcher,
+        "_call_nim",
+        lambda _threats, **_kwargs: (
+            [{
+                "tool": "trigger_price_alert",
+                "args": {
+                    "sku": candidate.sku,
+                    "name": candidate.name,
+                    "gap_pct": candidate.gap_pct,
+                    "sales_delta": candidate.sales_7d_delta_pct,
+                    "action": candidate.recommended_action,
+                    "confidence": candidate.confidence,
+                },
+            }],
+            {"provider": "nim", "total_tokens": 1},
+        ),
+    )
+    real_mark = module._mark_alert_dispatched
+    monkeypatch.setattr(
+        module,
+        "_mark_alert_dispatched",
+        lambda sku, token: (
+            False if str(sku) == forced.sku else real_mark(sku, token)
+        ),
+    )
+
+    result = dispatcher.dispatch(
+        [forced, candidate],
+        hermes_stats={"duration_sec": 1},
+    )
+
+    assert result["dispatched"] == 2
+    assert result["skipped"] == 0
+    assert result["errors"] == [
+        "dedupe_commit_unverified:flag_for_human_review(SKU-FORCED-MIXED)"
+    ]
+    assert module._is_duplicate_alert(forced.sku) is True
+    assert module._is_duplicate_alert(candidate.sku) is True
+
+
+def test_mixed_batch_counts_failed_model_candidate_as_skipped(monkeypatch):
+    import services.nemoton_dispatcher_service as module
+
+    monkeypatch.setattr(module, "NIM_API_KEY", "test-key")
+    monkeypatch.setattr(module, "_check_nim_quota", lambda: True)
+    dispatcher = module.NemotronDispatcher()
+    forced = FakeThreat("SKU-FORCED-COUNT", "斷崖品")
+    forced.sales_7d_delta_pct = -100.0
+    candidate = FakeThreat("SKU-NIM-COUNT", "模型候選")
+    monkeypatch.setattr(
+        dispatcher,
+        "_exec_flag_for_human_review",
+        lambda *_args, **_kwargs: {"durable": True},
+    )
+    monkeypatch.setattr(
+        dispatcher,
+        "_exec_trigger_price_alert",
+        lambda *_args, **_kwargs: {"durable": False},
+    )
+    monkeypatch.setattr(
+        dispatcher,
+        "_call_nim",
+        lambda _threats, **_kwargs: (
+            [{
+                "tool": "trigger_price_alert",
+                "args": {
+                    "sku": candidate.sku,
+                    "name": candidate.name,
+                    "gap_pct": candidate.gap_pct,
+                    "sales_delta": candidate.sales_7d_delta_pct,
+                    "action": candidate.recommended_action,
+                    "confidence": candidate.confidence,
+                },
+            }],
+            {"provider": "nim", "total_tokens": 1},
+        ),
+    )
+
+    result = dispatcher.dispatch(
+        [forced, candidate],
+        hermes_stats={"duration_sec": 1},
+    )
+
+    assert result["dispatched"] == 1
+    assert result["skipped"] == 1
+    assert len(result["errors"]) == 1
+    assert "durable_side_effect_not_confirmed" in result["errors"][0]
+    assert result["dispatched"] + result["skipped"] == 2
+
+
 def test_dispatch_routes_non_exact_match_to_human_review(monkeypatch):
    import services.nemoton_dispatcher_service as module

@@ -83,7 +287,7 @@ def test_dispatch_falls_back_to_hermes_rules_on_nim_timeout(monkeypatch):
    monkeypatch.setattr(module, "_check_nim_quota", lambda: True)
    dispatcher = module.NemotronDispatcher()
    calls = _patch_execution_methods(monkeypatch, dispatcher)
-    monkeypatch.setattr(dispatcher, "_call_nim", lambda threats: (_ for _ in ()).throw(requests.Timeout("timeout")))
+    monkeypatch.setattr(dispatcher, "_call_nim", lambda threats, **_kwargs: (_ for _ in ()).throw(requests.Timeout("timeout")))

    result = dispatcher.dispatch([FakeThreat("SKU-2", "測試品")], hermes_stats={"duration_sec": 1})

@@ -101,7 +305,7 @@ def test_dispatch_falls_back_to_hermes_rules_on_zero_tool_calls(monkeypatch):
    monkeypatch.setattr(module, "_check_nim_quota", lambda: True)
    dispatcher = module.NemotronDispatcher()
    calls = _patch_execution_methods(monkeypatch, dispatcher)
-    monkeypatch.setattr(dispatcher, "_call_nim", lambda threats: ([], {"total_tokens": 10}))
+    monkeypatch.setattr(dispatcher, "_call_nim", lambda threats, **_kwargs: ([], {"total_tokens": 10}))

    result = dispatcher.dispatch([FakeThreat("SKU-3", "測試品")], hermes_stats={"duration_sec": 1})

--- a/tests/test_nemotron_qwen3_compat.py
+++ b/tests/test_nemotron_qwen3_compat.py
@@ -39,6 +39,20 @@ class FakeThreat:
    sales_7d_prev_amount: float = 120000.0


+def _valid_price_alert_call(sku="SKU-Q1"):
+    return {
+        "tool": "trigger_price_alert",
+        "args": {
+            "sku": sku,
+            "name": "qwen3 測試品",
+            "gap_pct": 22.4,
+            "sales_delta": -35.0,
+            "action": "跟進降價",
+            "confidence": 0.85,
+        },
+    }
+
+
 class _FakeResp:
    def __init__(self, payload: dict, status: int = 200):
        self._payload = payload
@@ -59,6 +73,7 @@ def _noop_log_ai_call(*args, **kwargs):
    class _Ctx:
        def set_tokens(self, **_kw): pass
        def set_provider(self, *_a, **_kw): pass
+        def set_model(self, *_a, **_kw): pass
        def set_error(self, *_a, **_kw): pass
        def fallback_to_caller(self, *_a, **_kw): pass
        def set_cache_hit(self, *_a, **_kw): pass
@@ -76,11 +91,13 @@ def _reset_global_state():
    import services.nemoton_dispatcher_service as _nem
    import services.ollama_service as _oss
    _nem._ALERT_CACHE.clear()
+    _nem._ALERT_INFLIGHT.clear()
    _oss._unhealthy_marks.clear()
    _oss._resolved_host_cache['host'] = None
    _oss._resolved_host_cache['ts'] = 0
    yield
    _nem._ALERT_CACHE.clear()
+    _nem._ALERT_INFLIGHT.clear()
    _oss._unhealthy_marks.clear()
    _oss._resolved_host_cache['host'] = None
    _oss._resolved_host_cache['ts'] = 0
@@ -93,6 +110,7 @@ def _patch_execution_methods(monkeypatch, dispatcher):
    def record(kind):
        def _inner(*args, **kwargs):
            calls.append({"kind": kind, "args": args, "kwargs": kwargs})
+            return {"durable": True}
        return _inner

    monkeypatch.setattr(dispatcher, "_exec_trigger_price_alert", record("price_alert"))
@@ -102,14 +120,27 @@ def _patch_execution_methods(monkeypatch, dispatcher):


 def _enable_qwen3_path(monkeypatch, module):
-    """打開 NEMOTRON_OLLAMA_FIRST + 旁路 mcp/log_ai_call/resolve_host 等副作用"""
+    """打開 Ollama-first，並隔離 MCP、logger 與 model identity 網路副作用。"""
    monkeypatch.setattr(module, "NEMOTRON_OLLAMA_FIRST", True)
    monkeypatch.setattr(module, "log_ai_call", _noop_log_ai_call)
    monkeypatch.setattr(module, "build_mcp_context", lambda: "MCP-MOCK")
    # 確保即使未被呼叫，import 路徑可解析
    import services.ollama_service as ollama_module
+    import services.nemotron_runtime_candidate_service as runtime_module
+
    monkeypatch.setattr(ollama_module, "resolve_ollama_host", lambda: "http://34.87.90.216:11434")
    monkeypatch.setattr(ollama_module, "mark_unhealthy", lambda *a, **kw: None)
+    monkeypatch.setattr(
+        runtime_module,
+        "inspect_nemotron_model_identity",
+        lambda candidate, **_kwargs: {
+            **candidate.as_public_dict(),
+            "ok": True,
+            "digest": candidate.expected_digest,
+            "digest_matches": True,
+            "error": None,
+        },
+    )


 # ─────────────────────────────────────────────────────────────
@@ -178,12 +209,7 @@ def test_qwen3_retries_secondary_when_primary_chat_fails(monkeypatch):
    import services.ollama_service as ollama_module

    _enable_qwen3_path(monkeypatch, module)
-    hosts = iter([
-        "http://34.87.90.216:11434",
-        "http://34.21.145.224:11434",
-    ])
    marked = []
-    monkeypatch.setattr(ollama_module, "resolve_ollama_host", lambda: next(hosts))
    monkeypatch.setattr(ollama_module, "mark_unhealthy", lambda host: marked.append(host))

    fake_body = {
@@ -227,6 +253,519 @@ def test_qwen3_retries_secondary_when_primary_chat_fails(monkeypatch):
    assert calls[0]["kind"] == "price_alert"


+def test_qwen3_uses_bounded_111_model_after_both_gcp_candidates_fail(monkeypatch):
+    """兩台 GCP 都逾時時，正式 dispatcher 應改用 exact-digest 111 qwen3:8b。"""
+    import requests
+    import services.nemoton_dispatcher_service as module
+
+    _enable_qwen3_path(monkeypatch, module)
+    posted = []
+    fake_body = {
+        "message": {
+            "role": "assistant",
+            "content": "",
+            "tool_calls": [{
+                "function": {
+                    "name": "trigger_price_alert",
+                    "arguments": {
+                        "sku": "SKU-Q1",
+                        "name": "qwen3 測試品",
+                        "gap_pct": 22.4,
+                        "sales_delta": -35.0,
+                        "action": "跟進降價",
+                        "confidence": 0.85,
+                    },
+                }
+            }],
+        },
+        "prompt_eval_count": 180,
+        "eval_count": 36,
+    }
+
+    def fake_post(url, json, timeout):
+        posted.append((url, json, timeout))
+        if len(posted) <= 2:
+            raise requests.Timeout("GCP saturated")
+        return _FakeResp(fake_body)
+
+    monkeypatch.setattr(module.requests, "post", fake_post)
+    dispatcher = module.NemotronDispatcher()
+    calls = _patch_execution_methods(monkeypatch, dispatcher)
+
+    result = dispatcher.dispatch([FakeThreat()], hermes_stats={"duration_sec": 1.0})
+
+    stats = result["nim_stats"]
+    assert result["dispatched"] == 1
+    assert stats["provider"] == "ollama_111"
+    assert stats["model"] == "qwen3:8b"
+    assert stats["candidate_label"] == "ollama_111_fallback"
+    assert stats["fallback_used"] is True
+    assert [attempt["status"] for attempt in stats["attempted_candidates"]] == [
+        "failed",
+        "failed",
+        "selected",
+    ]
+    assert posted[2][1]["think"] is False
+    assert posted[2][1]["options"]["num_predict"] == 512
+    assert posted[2][2] == 45
+    assert calls[0]["kind"] == "price_alert"
+    assert "qwen3:8b (111 備援)" in calls[0]["kwargs"]["footprint"]
+    assert "降級備援" in calls[0]["kwargs"]["footprint"]
+
+    structured = module._build_footprint_json(None, stats)
+    assert structured["dispatcher"]["platform"] == "Ollama"
+    assert structured["dispatcher"]["model"] == "qwen3:8b"
+    assert structured["dispatcher"]["fallback_used"] is True
+    assert [
+        attempt["label"]
+        for attempt in structured["dispatcher"]["candidate_attempts"]
+    ] == ["gcp_primary", "gcp_secondary", "ollama_111_fallback"]
+    assert all(
+        "host" not in attempt and "error" not in attempt
+        for attempt in structured["dispatcher"]["candidate_attempts"]
+    )
+
+
+def test_invalid_primary_tool_contract_retries_verified_secondary(monkeypatch):
+    import services.nemoton_dispatcher_service as module
+
+    _enable_qwen3_path(monkeypatch, module)
+    valid_body = {
+        "message": {
+            "tool_calls": [{"function": {
+                "name": "trigger_price_alert",
+                "arguments": _valid_price_alert_call()["args"],
+            }}],
+            "content": "",
+        },
+        "prompt_eval_count": 100,
+        "eval_count": 20,
+    }
+    invalid_body = {
+        "message": {
+            "tool_calls": [{"function": {
+                "name": "delete_product",
+                "arguments": {"sku": "SKU-Q1"},
+            }}],
+            "content": "",
+        },
+    }
+    responses = [_FakeResp(invalid_body), _FakeResp(valid_body)]
+    monkeypatch.setattr(
+        module.requests,
+        "post",
+        lambda *_args, **_kwargs: responses.pop(0),
+    )
+    dispatcher = module.NemotronDispatcher()
+    calls = _patch_execution_methods(monkeypatch, dispatcher)
+
+    result = dispatcher.dispatch([FakeThreat()], hermes_stats={"duration_sec": 1.0})
+
+    attempts = result["nim_stats"]["attempted_candidates"]
+    assert [attempt["status"] for attempt in attempts] == ["failed", "selected"]
+    assert "unknown_tool:delete_product" in attempts[0]["error"]
+    assert result["nim_stats"]["provider"] == "ollama_secondary"
+    assert result["dispatched"] == 1
+    assert calls[0]["kind"] == "price_alert"
+
+
+def test_malformed_primary_tags_schema_falls_through_to_verified_secondary(
+    monkeypatch,
+):
+    import services.nemoton_dispatcher_service as module
+    import services.nemotron_runtime_candidate_service as runtime_module
+    import services.ollama_service as ollama_module
+
+    monkeypatch.setattr(module, "NEMOTRON_OLLAMA_FIRST", True)
+    monkeypatch.setattr(module, "log_ai_call", _noop_log_ai_call)
+    monkeypatch.setattr(module, "build_mcp_context", lambda: "MCP-MOCK")
+    monkeypatch.setattr(ollama_module, "mark_unhealthy", lambda *_args: None)
+
+    def fake_get(url, timeout):
+        if runtime_module.OLLAMA_HOST_PRIMARY in url:
+            return _FakeResp({
+                "models": [
+                    None,
+                    {
+                        "name": runtime_module.PRIMARY_MODEL,
+                        "model": runtime_module.PRIMARY_MODEL,
+                        "digest": runtime_module.PRIMARY_EXPECTED_DIGEST,
+                        "details": {},
+                    },
+                ]
+            })
+        return _FakeResp({
+            "models": [{
+                "name": runtime_module.PRIMARY_MODEL,
+                "model": runtime_module.PRIMARY_MODEL,
+                "digest": runtime_module.PRIMARY_EXPECTED_DIGEST,
+                "details": {
+                    "parameter_size": "14.8B",
+                    "quantization_level": "Q4_K_M",
+                },
+            }]
+        })
+
+    valid_body = {
+        "message": {
+            "tool_calls": [{"function": {
+                "name": "trigger_price_alert",
+                "arguments": _valid_price_alert_call()["args"],
+            }}],
+            "content": "",
+        },
+        "prompt_eval_count": 100,
+        "eval_count": 20,
+    }
+    monkeypatch.setattr(module.requests, "get", fake_get)
+    monkeypatch.setattr(
+        module.requests,
+        "post",
+        lambda *_args, **_kwargs: _FakeResp(valid_body),
+    )
+    dispatcher = module.NemotronDispatcher()
+    calls = _patch_execution_methods(monkeypatch, dispatcher)
+
+    result = dispatcher.dispatch([FakeThreat()], hermes_stats={"duration_sec": 1})
+
+    attempts = result["nim_stats"]["attempted_candidates"]
+    assert [attempt["status"] for attempt in attempts] == [
+        "identity_failed",
+        "selected",
+    ]
+    assert attempts[0]["error"] == "malformed_models_schema"
+    assert result["nim_stats"]["provider"] == "ollama_secondary"
+    assert result["dispatched"] == 1
+    assert calls[0]["kind"] == "price_alert"
+
+
+@pytest.mark.parametrize(
+    ("tool_calls", "threats", "error_fragment"),
+    [
+        ([{"tool": "delete_product", "args": {"sku": "SKU-Q1"}}], [FakeThreat()], "unknown_tool"),
+        ([_valid_price_alert_call("OTHER")], [FakeThreat()], "sku_not_in_request"),
+        ([{"tool": "trigger_price_alert", "args": {"sku": "SKU-Q1"}}], [FakeThreat()], "missing_required_args"),
+        ([_valid_price_alert_call()], [FakeThreat(), FakeThreat(sku="SKU-Q2")], "tool_call_count_mismatch"),
+        (
+            [_valid_price_alert_call(), _valid_price_alert_call()],
+            [FakeThreat(), FakeThreat(sku="SKU-Q2")],
+            "duplicate_sku_tool_call",
+        ),
+    ],
+)
+def test_tool_contract_rejects_unbound_incomplete_or_duplicate_calls(
+    tool_calls, threats, error_fragment
+):
+    import services.nemoton_dispatcher_service as module
+
+    with pytest.raises(module.ToolCallContractError, match=error_fragment):
+        module._validate_tool_call_contract(tool_calls, threats)
+
+
+def test_failed_handler_does_not_poison_alert_dedupe(monkeypatch):
+    import services.nemoton_dispatcher_service as module
+
+    dispatcher = module.NemotronDispatcher()
+    monkeypatch.setattr(
+        dispatcher,
+        "_exec_trigger_price_alert",
+        lambda **_kwargs: (_ for _ in ()).throw(RuntimeError("send failed")),
+    )
+    first_token = module._reserve_alert("SKU-Q1")
+    assert first_token
+    first = dispatcher._execute_tool_calls(
+        tool_calls=[_valid_price_alert_call()],
+        threats=[FakeThreat()],
+        hermes_stats=None,
+        nim_stats={"provider": "nim", "total_tokens": 1},
+        reservation_tokens={"SKU-Q1": first_token},
+    )
+
+    assert first["dispatched"] == 0
+    assert module._is_duplicate_alert("SKU-Q1") is False
+
+    sent = []
+    monkeypatch.setattr(
+        dispatcher,
+        "_exec_trigger_price_alert",
+        lambda **kwargs: (sent.append(kwargs) or {"durable": True}),
+    )
+    second_token = module._reserve_alert("SKU-Q1")
+    assert second_token
+    second = dispatcher._execute_tool_calls(
+        tool_calls=[_valid_price_alert_call()],
+        threats=[FakeThreat()],
+        hermes_stats=None,
+        nim_stats={"provider": "nim", "total_tokens": 1},
+        reservation_tokens={"SKU-Q1": second_token},
+    )
+
+    assert second["dispatched"] == 1
+    assert sent
+    assert module._is_duplicate_alert("SKU-Q1") is True
+
+
+def test_non_durable_alert_outcome_is_failed_and_releases_reservation(monkeypatch):
+    import services.nemoton_dispatcher_service as module
+
+    dispatcher = module.NemotronDispatcher()
+    monkeypatch.setattr(
+        dispatcher,
+        "_send_telegram",
+        lambda *_args, **_kwargs: {
+            "durable": False,
+            "delivered": False,
+            "queued": False,
+        },
+    )
+    monkeypatch.setattr(dispatcher, "_sink_insight_to_km", lambda **_kwargs: True)
+    token = module._reserve_alert("SKU-Q1")
+    assert token
+
+    result = dispatcher._execute_tool_calls(
+        tool_calls=[_valid_price_alert_call()],
+        threats=[FakeThreat()],
+        hermes_stats=None,
+        nim_stats={"provider": "nim", "total_tokens": 1},
+        reservation_tokens={"SKU-Q1": token},
+    )
+
+    assert result["dispatched"] == 0
+    assert result["skipped"] == 1
+    assert "durable_side_effect_not_confirmed" in result["errors"][0]
+    assert module._is_duplicate_alert("SKU-Q1") is False
+
+
+def test_recommendation_requires_db_write_or_durable_notification(monkeypatch):
+    import services.nemoton_dispatcher_service as module
+
+    dispatcher = module.NemotronDispatcher(engine=None)
+    monkeypatch.setattr(
+        dispatcher,
+        "_send_telegram",
+        lambda *_args, **_kwargs: {"durable": False},
+    )
+    monkeypatch.setattr(dispatcher, "_sink_insight_to_km", lambda **_kwargs: True)
+    token = module._reserve_alert("SKU-Q1")
+    assert token
+
+    result = dispatcher._execute_tool_calls(
+        tool_calls=[{
+            "tool": "add_to_recommendation",
+            "args": {
+                "sku": "SKU-Q1",
+                "name": "qwen3 測試品",
+                "reason": "具價格競爭力",
+                "confidence": 0.85,
+            },
+        }],
+        threats=[FakeThreat()],
+        hermes_stats=None,
+        nim_stats={"provider": "nim", "total_tokens": 1},
+        reservation_tokens={"SKU-Q1": token},
+    )
+
+    assert result["dispatched"] == 0
+    assert "durable_side_effect_not_confirmed" in result["errors"][0]
+    assert module._is_duplicate_alert("SKU-Q1") is False
+
+
+def test_km_route_requires_durable_insight_write(monkeypatch):
+    import services.nemoton_dispatcher_service as module
+
+    dispatcher = module.NemotronDispatcher()
+    monkeypatch.setattr(dispatcher, "_sink_insight_to_km", lambda **_kwargs: False)
+    token = module._reserve_alert("SKU-Q1")
+    assert token
+
+    result = dispatcher._execute_tool_calls(
+        tool_calls=[{
+            "tool": "route_to_km",
+            "args": {
+                "sku": "SKU-Q1",
+                "name": "qwen3 測試品",
+                "km_domain": "price_competition",
+                "summary": "競價洞察",
+                "confidence": 0.85,
+            },
+        }],
+        threats=[FakeThreat()],
+        hermes_stats=None,
+        nim_stats={"provider": "nim", "total_tokens": 1},
+        reservation_tokens={"SKU-Q1": token},
+    )
+
+    assert result["dispatched"] == 0
+    assert "durable_side_effect_not_confirmed" in result["errors"][0]
+    assert module._is_duplicate_alert("SKU-Q1") is False
+
+
+def test_alert_reservation_is_atomic_and_releasable():
+    import services.nemoton_dispatcher_service as module
+
+    first_token = module._reserve_alert("SKU-Q1")
+    assert first_token
+    assert module._reserve_alert("SKU-Q1") is None
+    assert module._release_alert_reservation("SKU-Q1", first_token) is True
+    second_token = module._reserve_alert("SKU-Q1")
+    assert second_token and second_token != first_token
+    assert module._mark_alert_dispatched("SKU-Q1", second_token) is True
+    assert module._reserve_alert("SKU-Q1") is None
+
+
+def test_expired_reservation_token_cannot_release_or_commit_new_owner(monkeypatch):
+    import services.nemoton_dispatcher_service as module
+
+    clock = {"now": 1000.0}
+    monkeypatch.setattr(module.time, "time", lambda: clock["now"])
+    first_token = module._reserve_alert("SKU-Q1")
+    assert first_token
+
+    clock["now"] += module._ALERT_INFLIGHT_TTL_SEC + 1
+    assert module._mark_alert_dispatched("SKU-Q1", first_token) is False
+    second_token = module._reserve_alert("SKU-Q1")
+    assert second_token and second_token != first_token
+
+    assert module._release_alert_reservation("SKU-Q1", first_token) is False
+    assert module._mark_alert_dispatched("SKU-Q1", first_token) is False
+    assert module._refresh_alert_reservation("SKU-Q1", first_token) is False
+    assert module._refresh_alert_reservation("SKU-Q1", second_token) is True
+    assert module._mark_alert_dispatched("SKU-Q1", second_token) is True
+    assert module._is_duplicate_alert("SKU-Q1") is True
+
+
+def test_concurrent_reservation_has_exactly_one_owner():
+    from concurrent.futures import ThreadPoolExecutor
+    from threading import Barrier
+
+    import services.nemoton_dispatcher_service as module
+
+    barrier = Barrier(8)
+
+    def reserve_once():
+        barrier.wait()
+        return module._reserve_alert("SKU-CONCURRENT")
+
+    with ThreadPoolExecutor(max_workers=8) as pool:
+        tokens = list(pool.map(lambda _index: reserve_once(), range(8)))
+
+    owners = [token for token in tokens if token is not None]
+    assert len(owners) == 1
+    assert module._release_alert_reservation("SKU-CONCURRENT", owners[0]) is True
+
+
+def test_side_effect_boundary_extends_lease_through_slow_handler(monkeypatch):
+    import services.nemoton_dispatcher_service as module
+
+    clock = {"now": 5000.0}
+    monkeypatch.setattr(module.time, "time", lambda: clock["now"])
+    old_token = module._reserve_alert("SKU-Q1")
+    assert old_token
+    new_owner = {}
+    dispatcher = module.NemotronDispatcher()
+
+    def delayed_handler(**_kwargs):
+        clock["now"] += module._ALERT_INFLIGHT_TTL_SEC + 1
+        new_owner["token"] = module._reserve_alert("SKU-Q1")
+        return {"durable": True}
+
+    monkeypatch.setattr(dispatcher, "_exec_trigger_price_alert", delayed_handler)
+    result = dispatcher._execute_tool_calls(
+        tool_calls=[_valid_price_alert_call()],
+        threats=[FakeThreat()],
+        hermes_stats=None,
+        nim_stats={"provider": "nim", "total_tokens": 1},
+        reservation_tokens={"SKU-Q1": old_token},
+    )
+
+    assert new_owner["token"] is None
+    assert result["dispatched"] == 1
+    assert result["errors"] == []
+    assert module._is_duplicate_alert("SKU-Q1") is True
+
+
+def test_durable_tool_commit_failure_never_releases_quarantine(monkeypatch):
+    import services.nemoton_dispatcher_service as module
+
+    dispatcher = module.NemotronDispatcher()
+    monkeypatch.setattr(
+        dispatcher,
+        "_exec_trigger_price_alert",
+        lambda **_kwargs: {"durable": True},
+    )
+    token = module._reserve_alert("SKU-Q1")
+    released = []
+    monkeypatch.setattr(module, "_mark_alert_dispatched", lambda *_args: False)
+    monkeypatch.setattr(
+        module,
+        "_release_alert_reservation",
+        lambda *args: (released.append(args) or True),
+    )
+
+    result = dispatcher._execute_tool_calls(
+        tool_calls=[_valid_price_alert_call()],
+        threats=[FakeThreat()],
+        hermes_stats=None,
+        nim_stats={"provider": "nim", "total_tokens": 1},
+        reservation_tokens={"SKU-Q1": token},
+    )
+
+    assert result["dispatched"] == 1
+    assert result["skipped"] == 0
+    assert result["errors"] == [
+        "dedupe_commit_unverified:trigger_price_alert(SKU-Q1)"
+    ]
+    assert released == []
+    assert module._is_duplicate_alert("SKU-Q1") is True
+
+
+def test_dispatch_total_deadline_bounds_ollama_and_nim_chain(monkeypatch):
+    import requests
+    import services.nemoton_dispatcher_service as module
+
+    _enable_qwen3_path(monkeypatch, module)
+    monkeypatch.setattr(module, "NEMOTRON_DISPATCH_TOTAL_TIMEOUT_SEC", 30)
+    monkeypatch.setattr(module, "NIM_API_KEY", "fake-key")
+    monkeypatch.setattr(module, "_check_nim_quota", lambda: True)
+    clock = {"now": 100.0}
+    monkeypatch.setattr(module.time, "monotonic", lambda: clock["now"])
+    posted = []
+
+    def timeout_post(url, json=None, timeout=None, **_kwargs):
+        posted.append((url, timeout))
+        clock["now"] += float(timeout or 0)
+        raise requests.Timeout("simulated timeout")
+
+    monkeypatch.setattr(module.requests, "post", timeout_post)
+    dispatcher = module.NemotronDispatcher()
+    calls = _patch_execution_methods(monkeypatch, dispatcher)
+
+    result = dispatcher.dispatch([FakeThreat()], hermes_stats={"duration_sec": 1.0})
+
+    upstream = result["nim_stats"]["upstream_ollama_failure"]
+    assert len(posted) == 1
+    assert posted[0][1] == 30
+    assert clock["now"] == 130.0
+    assert upstream["deadline_exhausted"] is True
+    assert upstream["attempt_count"] == 2
+    assert result["nim_stats"]["provider"] == "hermes_rule_engine"
+    assert "Ollama 候選鏈" in calls[0]["kwargs"]["footprint"]
+    assert "gcp_primary | primary | qwen3:14b | failed | digest=ok | Timeout" in (
+        calls[0]["kwargs"]["footprint"]
+    )
+    structured = module._build_footprint_json(None, result["nim_stats"])
+    attempts = structured["dispatcher"]["candidate_attempts"]
+    assert attempts[0] == {
+        "label": "gcp_primary",
+        "tier": "primary",
+        "model": "qwen3:14b",
+        "status": "failed",
+        "digest_matches": True,
+        "error_class": "Timeout",
+    }
+    assert result["dispatched"] == 1
+
+
 # ─────────────────────────────────────────────────────────────
 # T2. qwen3 沒回 tool_calls 但 content 含 JSON list → fallback 解析
 # ─────────────────────────────────────────────────────────────
@@ -252,7 +791,7 @@ def test_qwen3_content_only_fallback_parsing(monkeypatch):
    calls = _patch_execution_methods(monkeypatch, dispatcher)
    monkeypatch.setattr(
        dispatcher, "_call_nim",
-        lambda threats: (_ for _ in ()).throw(AssertionError("NIM 不應被呼叫")),
+        lambda threats, **_kwargs: (_ for _ in ()).throw(AssertionError("NIM 不應被呼叫")),
    )

    result = dispatcher.dispatch([FakeThreat(confidence=0.45)], hermes_stats={"duration_sec": 1.0})
@@ -325,7 +864,7 @@ def test_qwen3_connection_error_falls_back_to_nim(monkeypatch):
    calls = _patch_execution_methods(monkeypatch, dispatcher)
    nim_invoked = {"v": False}

-    def _fake_nim(threats):
+    def _fake_nim(threats, **_kwargs):
        nim_invoked["v"] = True
        return (
            [{
@@ -369,7 +908,7 @@ def test_qwen3_and_nim_both_fail_falls_back_to_hermes_rules(monkeypatch):
    # 攔 _call_nim 也擲 timeout
    monkeypatch.setattr(
        dispatcher, "_call_nim",
-        lambda threats: (_ for _ in ()).throw(requests.Timeout("NIM timeout")),
+        lambda threats, **_kwargs: (_ for _ in ()).throw(requests.Timeout("NIM timeout")),
    )

    # 攔住規則引擎內部呼叫的 _exec_*，記錄 concern / reason 文字驗證 🟡 標記
@@ -392,6 +931,7 @@ def test_qwen3_and_nim_both_fail_falls_back_to_hermes_rules(monkeypatch):
             'revenue_loss_7d', 'recommended_price'],
            args, kwargs)
        captured.append(("human_review", merged))
+        return {"durable": True}

    def record_alert(*args, **kwargs):
        merged = _merge_positional(
@@ -400,9 +940,11 @@ def test_qwen3_and_nim_both_fail_falls_back_to_hermes_rules(monkeypatch):
             'revenue_loss_7d', 'recommended_price'],
            args, kwargs)
        captured.append(("price_alert", merged))
+        return {"durable": True}

    def record_reco(*args, **kwargs):
        captured.append(("recommendation", kwargs))
+        return {"durable": True}

    monkeypatch.setattr(dispatcher, "_exec_flag_for_human_review", record_review)
    monkeypatch.setattr(dispatcher, "_exec_trigger_price_alert", record_alert)
@@ -448,7 +990,7 @@ def test_flag_false_preserves_nim_first_emergency_path(monkeypatch):
    calls = _patch_execution_methods(monkeypatch, dispatcher)
    monkeypatch.setattr(
        dispatcher, "_call_nim",
-        lambda threats: (
+        lambda threats, **_kwargs: (
            [{
                "tool": "trigger_price_alert",
                "args": {
--- a/tests/test_nemotron_runtime_candidate_service.py
+++ b/tests/test_nemotron_runtime_candidate_service.py
@@ -0,0 +1,205 @@
+import json
+import os
+import subprocess
+import sys
+
+import pytest
+import requests
+
+
+class _Response:
+    def __init__(self, payload):
+        self._payload = payload
+
+    def raise_for_status(self):
+        return None
+
+    def json(self):
+        return self._payload
+
+
+def test_candidate_chain_is_gcp_first_with_exact_digest_111_fallback():
+    from services import nemotron_runtime_candidate_service as service
+
+    candidates = service.build_nemotron_runtime_candidates()
+
+    assert [candidate.label for candidate in candidates] == [
+        "gcp_primary",
+        "gcp_secondary",
+        "ollama_111_fallback",
+    ]
+    assert [candidate.model for candidate in candidates] == [
+        "qwen3:14b",
+        "qwen3:14b",
+        "qwen3:8b",
+    ]
+    assert candidates[0].is_fallback is False
+    assert candidates[2].is_fallback is True
+    assert candidates[0].request_timeout_sec == 60
+    assert candidates[2].request_timeout_sec == 45
+    assert candidates[0].num_predict == 2048
+    assert candidates[2].num_predict == 512
+    assert candidates[0].num_ctx is None
+    assert candidates[2].num_ctx == 4096
+
+
+def test_model_identity_fails_closed_on_digest_drift():
+    from services import nemotron_runtime_candidate_service as service
+
+    candidate = service.build_nemotron_runtime_candidates()[0]
+    identity = service.inspect_nemotron_model_identity(
+        candidate,
+        request_get=lambda *_args, **_kwargs: _Response({
+            "models": [{
+                "name": candidate.model,
+                "model": candidate.model,
+                "digest": "deadbeef",
+                "details": {},
+            }]
+        }),
+    )
+
+    assert identity["ok"] is False
+    assert identity["digest_matches"] is False
+    assert identity["error"] == "model_digest_mismatch"
+
+
+def test_model_identity_reports_network_failure_without_loading_model():
+    from services import nemotron_runtime_candidate_service as service
+
+    candidate = service.build_nemotron_runtime_candidates()[2]
+
+    def fail(*_args, **_kwargs):
+        raise requests.Timeout("offline")
+
+    identity = service.inspect_nemotron_model_identity(
+        candidate,
+        request_get=fail,
+    )
+
+    assert identity["ok"] is False
+    assert identity["digest"] is None
+    assert identity["error"].startswith("Timeout: offline")
+
+
+@pytest.mark.parametrize(
+    ("payload", "error_fragment"),
+    [
+        (None, "malformed_tags_payload"),
+        ({"models": {}}, "malformed_models_schema"),
+        ({"models": [None]}, "malformed_models_schema"),
+        ({"models": [{"name": "other", "details": None}]}, "malformed_model_details"),
+    ],
+)
+def test_model_identity_fails_closed_on_malformed_tags_schema(
+    payload, error_fragment
+):
+    from services import nemotron_runtime_candidate_service as service
+
+    candidate = service.build_nemotron_runtime_candidates()[0]
+    identity = service.inspect_nemotron_model_identity(
+        candidate,
+        request_get=lambda *_args, **_kwargs: _Response(payload),
+    )
+
+    assert identity["ok"] is False
+    assert identity["digest_matches"] is False
+    assert error_fragment in str(identity["error"])
+
+
+def test_model_identity_rejects_malformed_entry_mixed_with_valid_target():
+    from services import nemotron_runtime_candidate_service as service
+
+    candidate = service.build_nemotron_runtime_candidates()[0]
+    identity = service.inspect_nemotron_model_identity(
+        candidate,
+        request_get=lambda *_args, **_kwargs: _Response({
+            "models": [
+                None,
+                {
+                    "name": candidate.model,
+                    "model": candidate.model,
+                    "digest": candidate.expected_digest,
+                    "details": {},
+                },
+            ]
+        }),
+    )
+
+    assert identity["ok"] is False
+    assert identity["digest_matches"] is False
+    assert identity["error"] == "malformed_models_schema"
+
+
+def test_candidate_attempt_timeout_honors_env_but_111_resources_stay_fixed():
+    env = os.environ.copy()
+    env.update({
+        "NEMOTRON_111_ATTEMPT_TIMEOUT_SEC": "33",
+        "OLLAMA_111_NUM_CTX": "2048",
+        "OLLAMA_111_NUM_PREDICT": "256",
+    })
+    completed = subprocess.run(
+        [
+            sys.executable,
+            "-c",
+            (
+                "import json; "
+                "from services.nemotron_runtime_candidate_service import "
+                "build_nemotron_runtime_candidates; "
+                "c=build_nemotron_runtime_candidates()[-1]; "
+                "print(json.dumps(c.as_public_dict(), sort_keys=True))"
+            ),
+        ],
+        cwd=os.path.dirname(os.path.dirname(__file__)),
+        env=env,
+        capture_output=True,
+        check=False,
+        text=True,
+    )
+
+    assert completed.returncode == 0, completed.stderr
+    fallback = json.loads(completed.stdout.strip())
+    assert fallback["request_timeout_sec"] == 33
+    assert fallback["num_ctx"] == 4096
+    assert fallback["num_predict"] == 512
+
+
+@pytest.mark.parametrize(
+    ("num_ctx", "num_predict"),
+    [
+        ("99999", "99999"),
+        ("-7", "-9"),
+        ("not-a-number", "invalid"),
+    ],
+)
+def test_candidate_resource_contract_cannot_be_weakened_by_environment(
+    num_ctx, num_predict
+):
+    env = os.environ.copy()
+    env.update({
+        "OLLAMA_111_NUM_CTX": num_ctx,
+        "OLLAMA_111_NUM_PREDICT": num_predict,
+    })
+    completed = subprocess.run(
+        [
+            sys.executable,
+            "-c",
+            (
+                "import json; "
+                "from services.nemotron_runtime_candidate_service import "
+                "build_nemotron_runtime_candidates; "
+                "c=build_nemotron_runtime_candidates()[-1]; "
+                "print(json.dumps(c.as_public_dict(), sort_keys=True))"
+            ),
+        ],
+        cwd=os.path.dirname(os.path.dirname(__file__)),
+        env=env,
+        capture_output=True,
+        check=False,
+        text=True,
+    )
+
+    assert completed.returncode == 0, completed.stderr
+    fallback = json.loads(completed.stdout.strip())
+    assert fallback["num_ctx"] == 4096
+    assert fallback["num_predict"] == 512
--- a/tests/test_phase3f_cleanup_contracts.py
+++ b/tests/test_phase3f_cleanup_contracts.py
@@ -132,8 +132,15 @@ def test_env_example_documents_runtime_and_ai_automation_variables():
        "N8N_USER",
        "N8N_WEBHOOK_BASE_URL",
        "NEMOTRON_OLLAMA_FIRST",
+        "NEMOTRON_OLLAMA_EXPECTED_DIGEST",
+        "NEMOTRON_OLLAMA_FALLBACK_EXPECTED_DIGEST",
+        "NEMOTRON_OLLAMA_FALLBACK_MODEL",
        "NEMOTRON_OLLAMA_MODEL",
        "NEMOTRON_OLLAMA_TIMEOUT",
+        "NEMOTRON_GCP_ATTEMPT_TIMEOUT_SEC",
+        "NEMOTRON_111_ATTEMPT_TIMEOUT_SEC",
+        "NEMOTRON_DISPATCH_TOTAL_TIMEOUT_SEC",
+        "NEMOTRON_MODEL_IDENTITY_TIMEOUT_SEC",
        "OLLAMA_EMBED_TIMEOUT",
        "OPENCLAW_ADMIN_USER_IDS",
        "OPENCLAW_AGENT_DISPATCH",
--- a/tests/test_qwen3_runtime_usage.py
+++ b/tests/test_qwen3_runtime_usage.py
@@ -7,6 +7,7 @@ ROOT = Path(__file__).resolve().parents[1]
 def test_qwen3_is_active_runtime_model_not_unused_ollama_weight():
    openclaw_source = (ROOT / "services" / "openclaw_strategist_service.py").read_text(encoding="utf-8")
    nemotron_source = (ROOT / "services" / "nemoton_dispatcher_service.py").read_text(encoding="utf-8")
+    candidate_source = (ROOT / "services" / "nemotron_runtime_candidate_service.py").read_text(encoding="utf-8")
    router_source = (ROOT / "services" / "llm_model_router.py").read_text(encoding="utf-8")

    assert "OPENCLAW_QA_OLLAMA_MODEL = os.getenv('OPENCLAW_QA_OLLAMA_MODEL', 'qwen3:14b')" in openclaw_source
@@ -16,7 +17,11 @@ def test_qwen3_is_active_runtime_model_not_unused_ollama_weight():
    assert "OPENCLAW_STRATEGY_OLLAMA_KEEP_ALIVE = os.getenv('OPENCLAW_STRATEGY_OLLAMA_KEEP_ALIVE', '5m')" in openclaw_source
    assert "keep_alive=OPENCLAW_STRATEGY_OLLAMA_KEEP_ALIVE" in openclaw_source
    assert 'keep_alive="24h"' not in openclaw_source
-    assert 'NEMOTRON_OLLAMA_MODEL = os.getenv("NEMOTRON_OLLAMA_MODEL", "qwen3:14b")' in nemotron_source
+    assert "NEMOTRON_OLLAMA_MODEL = PRIMARY_MODEL" in nemotron_source
    assert "def _call_qwen3_dispatch(" in nemotron_source
-    assert "for _attempt in range(3):" in nemotron_source
+    assert "for candidate in build_nemotron_runtime_candidates(" in nemotron_source
+    assert "inspect_nemotron_model_identity(" in nemotron_source
+    assert '"qwen3:14b"' in candidate_source
+    assert '"qwen3:8b"' in candidate_source
+    assert 'label="ollama_111_fallback"' in candidate_source
    assert "'qwen3:14b'" in router_source
--- a/tests/test_sales_analysis_query_contract.py
+++ b/tests/test_sales_analysis_query_contract.py
@@ -0,0 +1,423 @@
+from io import BytesIO
+import re
+
+from flask import Flask
+from openpyxl import load_workbook
+import pandas as pd
+from sqlalchemy import create_engine, event, text
+from sqlalchemy.dialects import postgresql
+
+import routes.sales_routes as sales_routes
+from services.sales_analysis_export_service import (
+    combine_sales_marketing_export_frames,
+    query_sales_marketing_export_frames,
+    query_sales_vendor_export_frame,
+    sanitize_excel_dataframe,
+)
+from services.sales_analysis_query_service import (
+    SALES_TIME_TEXT_PATTERN,
+    build_sales_where_clause,
+    normalize_sales_query_args,
+    prepare_sales_query_context,
+    resolve_sales_query_columns,
+)
+
+
+def _build_sales_engine():
+    engine = create_engine("sqlite:///:memory:")
+    with engine.begin() as conn:
+        conn.exec_driver_sql(
+            '''
+            CREATE TABLE realtime_sales_monthly (
+                "日期" TEXT,
+                "時間" TEXT,
+                "商品ID" TEXT,
+                "商品名稱" TEXT,
+                "商品館" TEXT,
+                "品牌" TEXT,
+                "廠商名稱" TEXT,
+                "總業績" REAL,
+                "數量" REAL,
+                "總成本" REAL,
+                "折扣活動名稱" TEXT,
+                "折價券活動名稱" TEXT
+            )
+            '''
+        )
+        conn.exec_driver_sql(
+            '''
+            INSERT INTO realtime_sales_monthly
+                ("日期", "時間", "商品ID", "商品名稱", "商品館", "品牌",
+                 "廠商名稱", "總業績", "數量", "總成本", "折扣活動名稱", "折價券活動名稱")
+            VALUES
+                ('2025/04/07', '10:00:00', 'P1', '商品 A', '美妝', '品牌 A', '廠商 A', 100, 2, 60, '母親節', ''),
+                ('2025/04/07', '20:00:00', 'P2', '商品 B', '美妝', '品牌 B', '廠商 B', 300, 3, 200, '', '折價券'),
+                ('2025/04/08', '10:00:00', 'P3', '商品 C', '美妝', '品牌 C', '廠商 C', 600, 4, 300, '會員日', ''),
+                ('2026/04/06', '10:00:00', 'P1', '商品 A', '美妝', '品牌 A', '廠商 A', 150, 3, 80, '母親節', ''),
+                ('2026/04/06', '20:00:00', 'P2', '商品 B', '美妝', '品牌 B', '廠商 B', 400, 4, 250, '', '折價券'),
+                ('2026/04/07', '10:00:00', 'P3', '商品 C', '美妝', '品牌 C', '廠商 C', 800, 5, 400, '會員日', ''),
+                ('2026/04/06', '10:00:00', 'P4', '商品 D', '3C', '品牌 D', '廠商 D', 9999, 1, 1, '清倉', '')
+            '''
+        )
+    return engine
+
+
+def test_query_filter_values_are_bound_instead_of_interpolated():
+    engine = _build_sales_engine()
+    attack = "美妝' OR 1=1 --"
+    filters = normalize_sales_query_args({
+        "data_range": "0",
+        "category": attack,
+        "start_date": "2026-04-30",
+        "end_date": "2026-04-01",
+    })
+    columns = resolve_sales_query_columns(engine, "realtime_sales_monthly")
+
+    clause, params = build_sales_where_clause(engine, columns, filters)
+
+    assert attack not in clause
+    assert params["filter_category"] == attack
+    assert params["filter_start_date"] == "2026-04-01"
+    assert params["filter_end_date"] == "2026-04-30"
+
+
+def test_keyword_search_is_case_insensitive_and_treats_wildcards_as_text():
+    engine = _build_sales_engine()
+    with engine.begin() as connection:
+        connection.exec_driver_sql(
+            "UPDATE realtime_sales_monthly SET \"商品名稱\" = 'Alpha%_Beta' "
+            "WHERE \"商品ID\" = 'P1'"
+        )
+    filters = normalize_sales_query_args({
+        "data_range": "0",
+        "keyword": "alpha%_beta",
+    })
+    columns = resolve_sales_query_columns(engine, "realtime_sales_monthly")
+    where_sql, params = build_sales_where_clause(engine, columns, filters)
+
+    with engine.connect() as connection:
+        product_ids = connection.execute(
+            text(f'SELECT "商品ID" FROM realtime_sales_monthly WHERE {where_sql}'),
+            params,
+        ).scalars().all()
+
+    assert product_ids == ["P1", "P1"]
+    assert params["filter_keyword"] == "%alpha\\%\\_beta%"
+
+
+def test_postgres_filter_contract_uses_bound_temporal_dimensions():
+    class PostgresEngine:
+        dialect = postgresql.dialect()
+
+    filters = normalize_sales_query_args({
+        "data_range": "0",
+        "start_date": "2026-04-01",
+        "end_date": "2026-04-30",
+        "dow": "0",
+        "hour": "10",
+        "keyword": "Alpha",
+    })
+    columns = {
+        "date": "日期",
+        "time": "時間",
+        "name": "商品名稱",
+        "pid": "商品ID",
+        "brand": "品牌",
+        "vendor": "廠商名稱",
+    }
+
+    where_sql, params = build_sales_where_clause(PostgresEngine(), columns, filters)
+
+    assert "EXTRACT(DOW FROM TO_DATE" in where_sql
+    assert "SUBSTRING(TRIM(CAST" in where_sql
+    assert "(\\.[0-9]+)?)?$" in where_sql
+    assert "LOWER(CAST" in where_sql
+    assert params["filter_db_dow"] == 1
+    assert params["filter_hour"] == 10
+    assert re.fullmatch(SALES_TIME_TEXT_PATTERN, "10:30:59")
+    assert not re.fullmatch(SALES_TIME_TEXT_PATTERN, "10:30junk")
+
+
+def test_invalid_time_text_never_matches_midnight_filter():
+    engine = _build_sales_engine()
+    with engine.begin() as connection:
+        connection.exec_driver_sql(
+            "UPDATE realtime_sales_monthly SET \"時間\" = '' WHERE \"商品ID\" = 'P1'"
+        )
+    filters = normalize_sales_query_args({"data_range": "0", "hour": "0"})
+    columns = resolve_sales_query_columns(engine, "realtime_sales_monthly")
+    where_sql, params = build_sales_where_clause(engine, columns, filters)
+
+    with engine.connect() as connection:
+        count = connection.execute(
+            text(f"SELECT COUNT(*) FROM realtime_sales_monthly WHERE {where_sql}"),
+            params,
+        ).scalar_one()
+
+    assert count == 0
+
+
+def test_top_detail_rejects_injection_as_data_and_keeps_the_table_intact():
+    engine = _build_sales_engine()
+    frame, _meta = sales_routes._query_sales_top_detail_frame(engine, {
+        "data_range": "0",
+        "start_date": "2026-04-01",
+        "end_date": "2026-04-30",
+        "category": "美妝' OR 1=1 --",
+        "metric": "amount",
+        "view": "product",
+        "type": "revenue",
+    })
+
+    assert frame.empty
+    with engine.connect() as conn:
+        count = conn.exec_driver_sql("SELECT COUNT(*) FROM realtime_sales_monthly").scalar_one()
+    assert count == 7
+
+
+def test_top_detail_rejects_a_mismatched_business_type_and_metric():
+    engine = _build_sales_engine()
+
+    try:
+        sales_routes._query_sales_top_detail_frame(engine, {
+            "data_range": "0",
+            "type": "margin",
+            "metric": "amount",
+            "view": "product",
+        })
+    except ValueError as exc:
+        assert "type 與 metric" in str(exc)
+    else:
+        raise AssertionError("mismatched type/metric must be rejected")
+
+
+def test_yoy_applies_month_weekday_hour_and_category_together(monkeypatch):
+    engine = _build_sales_engine()
+
+    class FakeDatabaseManager:
+        def __init__(self):
+            self.engine = engine
+
+    monkeypatch.setattr(sales_routes, "DatabaseManager", FakeDatabaseManager)
+    app = Flask(__name__)
+    with app.test_request_context(
+        "/api/sales_analysis/yoy_comparison"
+        "?year1=2025&year2=2026&metric=revenue&data_range=0"
+        "&month=2026-04&dow=0&hour=10&category=美妝"
+    ):
+        response = sales_routes.api_yoy_comparison.__wrapped__()
+
+    payload = response.get_json()
+    assert payload["year1"]["total"] == 100
+    assert payload["year2"]["total"] == 150
+    assert payload["growth_rate"] == 50
+
+
+def test_table_api_normalizes_reverse_dates_and_preserves_metric_order(monkeypatch):
+    engine = _build_sales_engine()
+
+    class FakeDatabaseManager:
+        def __init__(self):
+            self.engine = engine
+
+    monkeypatch.setattr(sales_routes, "DatabaseManager", FakeDatabaseManager)
+    sales_routes._TABLE_DATA_CACHE.clear()
+    app = Flask(__name__)
+    with app.test_request_context(
+        "/api/sales_analysis/table_data?metric=qty&data_range=0"
+        "&start_date=2026-04-30&end_date=2026-04-01&category=美妝"
+    ):
+        response = sales_routes.api_sales_table_data.__wrapped__()
+
+    payload = response.get_json()
+    assert [item["product_id"] for item in payload["data"]] == ["P3", "P2", "P1"]
+    assert [item["qty"] for item in payload["data"]] == [5, 4, 3]
+
+
+def test_sales_page_redirects_reverse_dates_to_one_canonical_url(monkeypatch):
+    engine = _build_sales_engine()
+
+    class FakeDatabaseManager:
+        def __init__(self):
+            self.engine = engine
+
+    monkeypatch.setattr(sales_routes, "DatabaseManager", FakeDatabaseManager)
+    sales_routes._SALES_OPTIONS_CACHE.clear()
+    app = Flask(__name__)
+    app.register_blueprint(sales_routes.sales_bp)
+    with app.test_request_context(
+        "/sales_analysis?data_range=0&start_date=2026-04-30&end_date=2026-04-01"
+    ):
+        response = sales_routes.sales_analysis.__wrapped__()
+
+    assert response.status_code == 302
+    assert "start_date=2026-04-01" in response.location
+    assert "end_date=2026-04-30" in response.location
+
+
+def test_sales_page_closes_one_sided_dates_with_live_source_bounds(monkeypatch):
+    engine = _build_sales_engine()
+
+    class FakeDatabaseManager:
+        def __init__(self):
+            self.engine = engine
+
+    monkeypatch.setattr(sales_routes, "DatabaseManager", FakeDatabaseManager)
+    sales_routes._SALES_OPTIONS_CACHE.clear()
+    app = Flask(__name__)
+    app.register_blueprint(sales_routes.sales_bp)
+
+    with app.test_request_context(
+        "/sales_analysis?data_range=0&end_date=2026-04-10"
+    ):
+        response = sales_routes.sales_analysis.__wrapped__()
+
+    assert response.status_code == 302
+    assert "start_date=2025-04-07" in response.location
+    assert "end_date=2026-04-10" in response.location
+
+
+def test_other_category_uses_the_same_filtered_top_twelve_contract():
+    engine = create_engine("sqlite:///:memory:")
+    with engine.begin() as connection:
+        connection.exec_driver_sql(
+            'CREATE TABLE realtime_sales_monthly ('
+            '"日期" TEXT, "商品名稱" TEXT, "商品館" TEXT, "總業績" REAL)'
+        )
+        connection.execute(
+            text(
+                'INSERT INTO realtime_sales_monthly '
+                '("日期", "商品名稱", "商品館", "總業績") '
+                'VALUES (:sale_date, :name, :category, :amount)'
+            ),
+            [
+                {
+                    "sale_date": "2026/04/01",
+                    "name": f"商品 {index}",
+                    "category": f"分類 {index:02d}",
+                    "amount": 100 - index if index <= 10 else 1,
+                }
+                for index in range(14)
+            ] + [{
+                "sale_date": "2026/04/01",
+                "name": "未分類商品",
+                "category": None,
+                "amount": 1,
+            }],
+        )
+
+    filters, columns = prepare_sales_query_context(
+        engine,
+        "realtime_sales_monthly",
+        {"data_range": "0", "category": "其他"},
+    )
+    where_sql, params = build_sales_where_clause(engine, columns, filters)
+    with engine.connect() as connection:
+        categories = connection.execute(
+            text(
+                'SELECT "商品館" FROM realtime_sales_monthly '
+                f'WHERE {where_sql} ORDER BY "商品館"'
+            ),
+            params,
+        ).scalars().all()
+
+    assert categories == ["分類 12", "分類 13"]
+
+    sales_routes._SALES_PROCESSED_CACHE["other-contract"] = {
+        "df": pd.read_sql(text("SELECT * FROM realtime_sales_monthly"), engine),
+        "cols": {
+            "name": "商品名稱",
+            "category": "商品館",
+            "amount": "總業績",
+        },
+    }
+    app = Flask(__name__)
+    with app.test_request_context("/sales_analysis?category=其他"):
+        page_frame, _columns, error = sales_routes._get_filtered_sales_data(
+            "other-contract"
+        )
+    sales_routes._SALES_PROCESSED_CACHE.pop("other-contract", None)
+
+    assert error is None
+    assert page_frame["商品館"].tolist() == ["分類 12", "分類 13"]
+
+
+def test_vendor_and_marketing_exports_apply_the_page_filters():
+    engine = _build_sales_engine()
+    statements = []
+
+    def capture_statement(_conn, _cursor, statement, _params, _context, _many):
+        statements.append(statement)
+
+    event.listen(engine, "before_cursor_execute", capture_statement)
+    args = {
+        "metric": "amount",
+        "data_range": "0",
+        "start_date": "2026-04-01",
+        "end_date": "2026-04-30",
+        "category": "美妝",
+        "hour": "10",
+    }
+
+    vendor_frame = query_sales_vendor_export_frame(engine, args)
+    marketing_frames = query_sales_marketing_export_frames(engine, args)
+
+    assert vendor_frame[["廠商", "銷售金額"]].to_dict("records") == [
+        {"廠商": "廠商 C", "銷售金額": 800.0},
+        {"廠商": "廠商 A", "銷售金額": 150.0},
+    ]
+    assert marketing_frames["折扣活動"][["活動名稱", "銷售金額"]].to_dict("records") == [
+        {"活動名稱": "會員日", "銷售金額": 800.0},
+        {"活動名稱": "母親節", "銷售金額": 150.0},
+    ]
+    assert "折價券活動" not in marketing_frames
+    assert sum("UNION ALL" in statement for statement in statements) == 1
+
+
+def test_excel_export_sanitizer_blocks_formula_cells():
+    frame = sanitize_excel_dataframe(pd.DataFrame({"活動名稱": ["=2+2", "一般活動"]}))
+
+    assert frame["活動名稱"].tolist() == ["'=2+2", "一般活動"]
+
+
+def test_marketing_combined_sheet_follows_the_active_metric():
+    frames = {
+        "折扣活動": pd.DataFrame([
+            {"活動名稱": "高營收", "銷售金額": 1000, "銷售數量": 1, "毛利金額": 20},
+        ]),
+        "折價券活動": pd.DataFrame([
+            {"活動名稱": "高銷量", "銷售金額": 100, "銷售數量": 10, "毛利金額": 50},
+        ]),
+    }
+
+    combined = combine_sales_marketing_export_frames(frames, "qty")
+
+    assert combined["活動名稱"].tolist() == ["高銷量", "高營收"]
+
+
+def test_sales_export_routes_return_valid_period_linked_workbooks(monkeypatch):
+    engine = _build_sales_engine()
+
+    class FakeDatabaseManager:
+        def __init__(self):
+            self.engine = engine
+
+    monkeypatch.setattr(sales_routes, "DatabaseManager", FakeDatabaseManager)
+    app = Flask(__name__)
+    query = (
+        "?metric=amount&data_range=0&start_date=2026-04-01&end_date=2026-04-30"
+        "&category=美妝&hour=10"
+    )
+    routes = (
+        (sales_routes.api_export_sales_vendor, "/api/sales_analysis/export_vendor", "廠商分析"),
+        (sales_routes.api_export_sales_marketing, "/api/sales_analysis/export_marketing", "折扣活動"),
+    )
+
+    for route, path, expected_sheet in routes:
+        with app.test_request_context(f"{path}{query}"):
+            response = route.__wrapped__()
+        response.direct_passthrough = False
+        workbook = load_workbook(BytesIO(response.get_data()), read_only=True)
+
+        assert response.status_code == 200
+        assert expected_sheet in workbook.sheetnames
--- a/tests/test_sales_chart_data_contract.py
+++ b/tests/test_sales_chart_data_contract.py
@@ -1,6 +1,16 @@
 import pandas as pd

 from routes.sales_routes import _build_top_product_chart_data
+from services.sales_analysis_chart_service import (
+    build_analysis_scope,
+    build_bcg_chart_data,
+    build_heatmap_chart_data,
+    build_marketing_chart_data,
+    build_sales_filter_options,
+    build_seasonality_chart_data,
+    build_treemap_chart_data,
+    resolve_yoy_year_options,
+)


 def test_top_product_chart_aggregates_sku_rows_before_ranking():
@@ -34,4 +44,111 @@ def test_top_product_chart_returns_explicit_empty_contract():
        'labels': [],
        'chart_values': [],
        'metric_label': '銷售數量',
+        'is_money': False,
+    }
+
+
+def test_sales_chart_builders_emit_the_contract_consumed_by_frontend():
+    frame = pd.DataFrame([
+        {
+            'pid': 'P1', 'name': '商品 A', 'category': '美妝', 'amount': 100,
+            'qty': 2, 'calculated_profit': 30, '_month_str': '2026-04',
+            '_dow': 2, '_hour': 10,
+        },
+        {
+            'pid': 'P1', 'name': '商品 A', 'category': '美妝', 'amount': 50,
+            'qty': 1, 'calculated_profit': 10, '_month_str': '2026-04',
+            '_dow': 2, '_hour': 10,
+        },
+        {
+            'pid': 'P2', 'name': '商品 B', 'category': '3C', 'amount': 300,
+            'qty': 3, 'calculated_profit': 60, '_month_str': '2026-05',
+            '_dow': 4, '_hour': 21,
+        },
+    ])
+
+    treemap = build_treemap_chart_data(frame, 'category', 'name', 'amount')
+    bcg = build_bcg_chart_data(frame, 'pid', 'name', 'qty', 'amount')
+    heatmap = build_heatmap_chart_data(frame, 'amount')
+    seasonality = build_seasonality_chart_data(frame, 'category', 'amount')
+
+    assert treemap[0].keys() == {'category', 'name', 'value'}
+    assert {item['name'] for item in treemap} == {'商品 A', '商品 B'}
+    assert bcg['points'][0]['name'] == '商品 B'
+    assert bcg['points'][1]['x'] == 3
+    assert bcg['x_median'] == 3
+    assert len(heatmap['matrix']) == 7
+    assert all(len(row) == 24 for row in heatmap['matrix'])
+    assert heatmap['matrix'][2][10] == 150
+    assert heatmap['matrix'][4][21] == 300
+    assert seasonality == {
+        'categories': ['3C', '美妝'],
+        'months': ['2026-04', '2026-05'],
+        'matrix': [[0.0, 300.0], [150.0, 0.0]],
+    }
+
+
+def test_sales_chart_builders_return_explicit_empty_contracts():
+    empty = pd.DataFrame()
+
+    assert build_treemap_chart_data(empty, 'category', 'name', 'amount') == []
+    assert build_bcg_chart_data(empty, 'pid', 'name', 'qty', 'amount') == {
+        'points': [], 'x_median': 0.0, 'y_median': 0.0,
+    }
+    assert build_heatmap_chart_data(empty, 'amount')['matrix'] == []
+    assert build_seasonality_chart_data(empty, 'category', 'amount')['matrix'] == []
+
+
+def test_marketing_and_yoy_controls_follow_active_metric_and_period():
+    marketing = build_marketing_chart_data({
+        'discount': [{'name': '母親節', 'revenue': 500, 'qty': 5, 'profit': 120}],
+        'coupon': [{'name': '折價券', 'revenue': 300, 'qty': 3, 'profit': 80}],
+    }, 'profit')
+    years = resolve_yoy_year_options(
+        ['2024-12', '2026-04', float('nan')],
+        {'end_date': '2026-04-30'},
+    )
+
+    assert marketing['metric'] == 'profit'
+    assert marketing['discount'] == {'labels': ['母親節'], 'values': [120.0]}
+    assert marketing['coupon'] == {'labels': ['折價券'], 'values': [80.0]}
+    assert years == {'years': [2026, 2025, 2024], 'year1': 2025, 'year2': 2026}
+
+
+def test_filter_options_are_scoped_to_the_loaded_period_frame():
+    frame = pd.DataFrame([
+        {'category': '美妝', 'brand': 'A', '_month_str': '2026-04'},
+        {'category': '3C', 'brand': 'B', '_month_str': '2026-04'},
+        {'category': '美妝', 'brand': 'A', '_month_str': None},
+    ])
+
+    options = build_sales_filter_options(frame, {
+        'category': 'category',
+        'brand': 'brand',
+        'vendor': None,
+        'activity': None,
+        'payment': None,
+    })
+
+    assert options['categories'] == ['3C', '美妝']
+    assert options['brands'] == ['A', 'B']
+    assert options['months'] == ['2026-04']
+
+
+def test_analysis_scope_names_every_active_time_filter():
+    scope = build_analysis_scope(
+        {
+            'label': '最近 6 個月',
+            'start_date': '2026-01-23',
+            'end_date': '2026-07-22',
+        },
+        month='2026-04',
+        dow='2',
+        hour='9',
+    )
+
+    assert scope == {
+        'label': '最近 6 個月 · 月份 2026-04 · 週三 · 09:00',
+        'start_date': '2026-01-23',
+        'end_date': '2026-07-22',
    }
--- a/web/static/css/page-sales-analysis-bem.css
+++ b/web/static/css/page-sales-analysis-bem.css
@@ -103,6 +103,24 @@
  color: color-mix(in srgb, var(--momo-page-accent) 60%, var(--momo-text-strong));
 }
 .sa-tag__sep { opacity: 0.5; margin: 0 0.15rem; }
+.sa-period-badge {
+  display: inline-flex;
+  align-items: center;
+  gap: 0.35rem;
+  flex: 0 0 auto;
+  padding: 0.2rem 0.45rem;
+  border: 1px solid var(--momo-border-subtle);
+  border-radius: 4px;
+  background: var(--momo-surface);
+  color: var(--momo-text-muted);
+  font-family: var(--momo-font-mono, ui-monospace, monospace);
+  font-size: 0.72rem;
+  font-weight: 700;
+  letter-spacing: 0;
+  line-height: 1.2;
+  white-space: nowrap;
+}
+.sa-period-badge i { color: var(--momo-page-accent); }
 .sa-alert { margin: 1rem 0; }

 /* ---------- Filter card ---------- */
@@ -602,6 +620,18 @@
 .sa-mkt-title--accent { color: var(--momo-page-accent); }
 .sa-mkt-title--olive  { color: var(--momo-tag-olive); }

+.sa-modal-period { margin-top: 0.35rem; }
+.sa-modal-status {
+  padding: 0.75rem;
+  border: 1px solid var(--momo-border-subtle);
+  border-radius: 4px;
+  background: var(--momo-surface);
+  color: var(--momo-text-muted);
+  font-size: 0.82rem;
+  font-weight: 700;
+}
+.sa-modal-status[hidden] { display: none; }
+
 /* ---------- Responsive trims ---------- */
@media (max-width: 991.98px) {
  .sa-insights__col { border-right: none; border-bottom: 1px dashed var(--momo-border-subtle); }
--- a/web/static/js/page-sales-analysis.js
+++ b/web/static/js/page-sales-analysis.js
@@ -64,7 +64,13 @@
    if (!sel.value) return;
    showLoading('正在載入資料...');
    const form = sel.closest('form');
-    if (form) form.submit();
+    if (form) {
+      const start = form.querySelector('[name="start_date"]');
+      const end = form.querySelector('[name="end_date"]');
+      if (start) start.value = '';
+      if (end) end.value = '';
+      form.requestSubmit();
+    }
  };

  window.clearDateRange = function () {
@@ -85,19 +91,76 @@
  };

  window.showTopDetail = function (kind, metric) {
-    const url = new URL('/top_detail', window.location.origin);
-    url.searchParams.set('kind', kind);
+    const url = new URL('/api/sales_analysis/top_detail', window.location.origin);
+    url.searchParams.set('type', kind);
    url.searchParams.set('metric', metric);
+    url.searchParams.set('view', 'product');
    // forward existing filters
    new URLSearchParams(window.location.search).forEach((v, k) => {
      if (!url.searchParams.has(k)) url.searchParams.set(k, v);
    });
-    window.open(url.toString(), '_blank');
+    const modalElement = document.getElementById('topDetailModal');
+    const rows = document.getElementById('topDetailRows');
+    const status = document.getElementById('topDetailStatus');
+    const title = document.getElementById('topDetailModalLabel');
+    const exportLink = document.getElementById('topDetailExport');
+    if (!modalElement || !rows || !status || !window.bootstrap) return;
+
+    const labels = {
+      revenue: '業績主推商品',
+      margin: '高毛利守價商品',
+      quantity: '人氣引流商品'
+    };
+    title.textContent = labels[kind] || '商品排行明細';
+    rows.replaceChildren();
+    status.hidden = false;
+    status.textContent = '正在載入排行...';
+    const exportUrl = new URL('/api/sales_analysis/export_top_detail', window.location.origin);
+    url.searchParams.forEach((value, key) => exportUrl.searchParams.set(key, value));
+    exportLink.href = exportUrl.toString();
+    bootstrap.Modal.getOrCreateInstance(modalElement).show();
+
+    fetch(url.toString(), { headers: { Accept: 'application/json' } })
+      .then((response) => {
+        if (!response.ok) throw new Error(`HTTP ${response.status}`);
+        return response.json();
+      })
+      .then((payload) => {
+        const items = Array.isArray(payload.items) ? payload.items : [];
+        if (!items.length) {
+          status.textContent = '所選期間沒有符合條件的商品。';
+          return;
+        }
+        const fragment = document.createDocumentFragment();
+        items.forEach((item, index) => {
+          const row = document.createElement('tr');
+          const values = [
+            fmtNum(index + 1),
+            [item.name, item.product_id].filter(Boolean).join(' · '),
+            item.category || '—',
+            metric === 'qty' ? fmtNum(item.value) : fmtMoney(item.value),
+            item.margin_rate == null ? '—' : `${Number(item.margin_rate).toFixed(1)}%`
+          ];
+          values.forEach((value, cellIndex) => {
+            const cell = document.createElement('td');
+            cell.textContent = String(value);
+            if (cellIndex >= 3) cell.classList.add('text-end');
+            row.appendChild(cell);
+          });
+          fragment.appendChild(row);
+        });
+        rows.appendChild(fragment);
+        status.hidden = true;
+      })
+      .catch((error) => {
+        console.warn('[sales-analysis] top detail failed', error);
+        status.textContent = '排行明細暫時無法載入，請稍後重試。';
+      });
  };

  window.exportMarketingExcel = function (campaign) {
-    const url = new URL('/api/export/excel/marketing', window.location.origin);
-    url.searchParams.set('campaign', campaign || 'all');
+    const url = new URL('/api/sales_analysis/export_marketing', window.location.origin);
+    url.searchParams.set('type', campaign || 'all');
    new URLSearchParams(window.location.search).forEach((v, k) => {
      if (!url.searchParams.has(k)) url.searchParams.set(k, v);
    });
@@ -115,8 +178,15 @@
    };
    const start = document.getElementById('start_date');
    const end   = document.getElementById('end_date');
-    if (start) flatpickr(start, { ...opts, onChange: () => end && end.dispatchEvent(new Event('focus')) });
-    if (end)   flatpickr(end,   opts);
+    const clearQuickRange = () => {
+      const range = document.getElementById('data_range');
+      if (range) range.value = '';
+    };
+    if (start) flatpickr(start, { ...opts, onChange: () => {
+      clearQuickRange();
+      if (end && end._flatpickr) end._flatpickr.open();
+    }});
+    if (end) flatpickr(end, { ...opts, onChange: clearQuickRange });
  };

  // ─── Read embedded data ────────────────────────────────
@@ -327,10 +397,10 @@
      // quadrant labels
      ctx.fillStyle = mix(INK, 50);
      ctx.font = 'bold 11px system-ui';
-      ctx.fillText('★ 明星 (Star)',     chartArea.left + 12,  chartArea.top + 18);
-      ctx.fillText('? 問題兒童 (Question)', x + 12,         chartArea.top + 18);
-      ctx.fillText('🐮 金牛 (Cash Cow)', chartArea.left + 12, chartArea.bottom - 8);
-      ctx.fillText('🐶 瘦狗 (Dog)',     x + 12,            chartArea.bottom - 8);
+      ctx.fillText('高毛利潛力', chartArea.left + 12, chartArea.top + 18);
+      ctx.fillText('主推明星', x + 12, chartArea.top + 18);
+      ctx.fillText('待調整', chartArea.left + 12, chartArea.bottom - 8);
+      ctx.fillText('走量守價', x + 12, chartArea.bottom - 8);
      ctx.restore();
    }
  });
@@ -344,7 +414,7 @@
      horizontalBar(
        document.getElementById('barChart'),
        data.barData.labels, chartValues(data.barData),
-        data.barData.label || '銷售',
+        data.barData.metric_label || data.barData.label || '銷售',
        data.barData.is_money ? fmtMoney : fmtNum
      );
    }
@@ -499,17 +569,21 @@
    const discountCanvas = document.getElementById('mktDiscountChart');
    const couponCanvas = document.getElementById('mktCouponChart');
    if (data.marketingData) {
+      const marketingMetric = data.marketingData.metric || 'revenue';
+      const marketingFormatter = marketingMetric === 'qty' ? fmtNum : fmtMoney;
+      const marketingLabel = marketingMetric === 'qty' ? '活動銷量' :
+        (marketingMetric === 'profit' ? '活動毛利' : '活動業績');
      if (data.marketingData.discount) {
        horizontalBar(discountCanvas,
          data.marketingData.discount.labels, data.marketingData.discount.values,
-          '折扣業績', fmtMoney);
+          marketingLabel, marketingFormatter);
      } else {
        renderChartEmpty(discountCanvas, '所選期間沒有折扣活動業績。');
      }
      if (data.marketingData.coupon) {
        horizontalBar(couponCanvas,
          data.marketingData.coupon.labels, data.marketingData.coupon.values,
-          '折價券業績', fmtMoney);
+          marketingLabel, marketingFormatter);
      } else {
        renderChartEmpty(couponCanvas, '所選期間沒有折價券活動業績。');
      }
@@ -659,7 +733,7 @@
      columns,
      processing: true, serverSide: false,
      pageLength: 25, lengthMenu: [10, 25, 50, 100],
-      order: [[$tbl.find('thead th').length - 1, 'desc']],
+      order: [],
      language: typeof window.EwoooCDataTableLanguage === 'function'
        ? window.EwoooCDataTableLanguage()
        : {},
@@ -691,6 +765,13 @@
    }
  };

+  document.addEventListener('click', (event) => {
+    const trigger = event.target.closest('[data-sales-filter-key]');
+    if (!trigger) return;
+    event.preventDefault();
+    window.setFilter(trigger.dataset.salesFilterKey, trigger.dataset.salesFilterValue);
+  });
+
  // Show loading on form submit
  document.addEventListener('submit', (e) => {
    if (e.target.matches('form[action="/sales_analysis"]')) showLoading('正在查詢...');
Author	SHA1	Message	Date
ogt	ddcd85abee	Merge remote-tracking branch 'origin/dev' into codex/sales-analysis-contract-fix	2026-07-23 09:23:27 +08:00
ogt	a0d38af4f1	docs(ai): record V10.816 runtime closure	2026-07-23 09:22:46 +08:00
ogt	93e4cf550b	Merge remote-tracking branch 'origin/main' into codex/sales-analysis-contract-fix-dev-sync	2026-07-23 08:55:36 +08:00
ogt	2757585e5e	chore(cd): deploy `a5a7af6` V10.816 [skip ci]	2026-07-22 22:58:35 +08:00
ogt	001f05477e	Merge remote-tracking branch 'origin/main' into codex/sales-analysis-contract-fix-dev-sync	2026-07-22 22:58:08 +08:00
ogt	a5a7af6e95	feat(ai): add bounded model-aware decision fallback Some checks are pending CD Pipeline / deploy (push) Waiting to run Details	2026-07-22 22:57:07 +08:00
ogt	df83c646cd	chore(dev): sync V10.815 sales analysis [skip ci]	2026-07-22 19:11:14 +08:00
ogt	283c8c80c6	chore(cd): deploy `f5df7bb` V10.815 [skip ci]	2026-07-22 18:40:41 +08:00
ogt	f5df7bb1eb	fix(analytics): harden period-linked sales queries	2026-07-22 17:59:10 +08:00
ogt	9b677ae616	fix(analytics): link sales filters to chart contracts	2026-07-22 16:41:00 +08:00