diff --git a/.env.example b/.env.example index 0ee9933..086b9df 100644 --- a/.env.example +++ b/.env.example @@ -407,8 +407,12 @@ RAG_DEFAULT_TOP_K=5 RAG_EMBED_MODEL=bge-m3:latest RAG_EMBED_DIM=1024 RAG_EMBED_NORMALIZE=true +RAG_EMBED_EXPECTED_DIGEST=7907646426070047a77226ac3e684fbbe8410524f7b4a74d02837e43f2146bab +EMBED_CONSISTENCY_TIMEOUT_SEC=150 +EMBED_MODEL_IDENTITY_TIMEOUT_SEC=10 EMBED_CONSISTENCY_INCLUDE_111=false INTERNAL_RAG_CANDIDATE_CANARY_THRESHOLD=0.70 +INTERNAL_RAG_CANDIDATE_CANARY_MIN_GCP_HOSTS=1 INTERNAL_RAG_CANDIDATE_CANARY_SCHEDULED_ENABLED=true PPT_VISION_ENABLED=true PPT_VISION_MODEL=minicpm-v:latest @@ -475,6 +479,11 @@ OPENCLAW_IMAGE_GEMINI_MODEL=gemini-1.5-flash NEMOTRON_OLLAMA_FIRST=true NEMOTRON_OLLAMA_MODEL=qwen3:14b NEMOTRON_OLLAMA_TIMEOUT=180 +NEMOTRON_OLLAMA_EXPECTED_DIGEST=bdbd181c33f2ed1b31c972991882db3cf4d192569092138a7d29e973cd9debe8 +NEMOTRON_DECISION_CANARY_TIMEOUT_SEC=300 +NEMOTRON_MODEL_IDENTITY_TIMEOUT_SEC=10 +NEMOTRON_DECISION_CANARY_MAX_AGE_HOURS=26 +NEMOTRON_DECISION_CANARY_SCHEDULED_ENABLED=true OPENCLAW_STRATEGY_OLLAMA_MODEL=qwen3:14b OPENCLAW_STRATEGY_OLLAMA_TIMEOUT=90 OPENCLAW_STRATEGY_OLLAMA_NUM_PREDICT=2048 @@ -687,6 +696,7 @@ PIXELRAG_MARKETPLACE_ADAPTER_DRY_RUN_RECEIPT_ROOT=/app/data/pixelrag/receipts/ad PIXELRAG_MARKETPLACE_ADAPTER_PREFLIGHT_RECEIPT_ROOT=/app/data/pixelrag/receipts/adapter_preflight PIXELRAG_MARKETPLACE_CANDIDATE_KNOWLEDGE_REPLAY_RECEIPT_ROOT=/app/data/pixelrag/receipts/knowledge_replay INTERNAL_RAG_CANDIDATE_CANARY_RECEIPT_ROOT=/app/data/pixelrag/receipts/internal_rag_canary +NEMOTRON_DECISION_CANARY_RECEIPT_ROOT=/app/data/ai_automation/nemotron_decision_canary_receipts PIXELRAG_MARKETPLACE_EMBEDDING_SIGNATURE_GUARD_REPLAY_RECEIPT_ROOT=/app/data/pixelrag/receipts/embedding_guard PIXELRAG_MARKETPLACE_IDENTITY_MATCHER_REPLAY_RECEIPT_ROOT=/app/data/pixelrag/receipts/identity_matcher PIXELRAG_MARKETPLACE_PROMOTION_GATE_REPLAY_RECEIPT_ROOT=/app/data/pixelrag/receipts/promotion_gate diff --git a/config.py b/config.py index 8e1a388..c4bac84 100644 --- a/config.py +++ b/config.py @@ -414,7 +414,7 @@ YOUTUBE_API_KEY = os.getenv('YOUTUBE_API_KEY', '') # ========================================== # 系統版本與路徑 # ========================================== -SYSTEM_VERSION = "V10.813" +SYSTEM_VERSION = "V10.814" LOG_FILE_PATH = os.path.join(BASE_DIR, 'logs/system.log') public_url = PUBLIC_URL # 用於模板顯示 diff --git a/docker-compose.yml b/docker-compose.yml index 2c622f8..77cffbf 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -104,6 +104,10 @@ services: - OLLAMA_HOST_FALLBACK=${OLLAMA_HOST_FALLBACK:-http://192.168.0.111:11434} # EMBEDDING_HOST 若未設定，由 resolve_ollama_host() 自動決定（三主機級聯） - EMBEDDING_HOST=${EMBEDDING_HOST:-} + - 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_DECISION_CANARY_TIMEOUT_SEC=${NEMOTRON_DECISION_CANARY_TIMEOUT_SEC:-300} # Gemini 只能作緊急備援；即使 .env 有 API key，預設也不得產生付費出站。 - GEMINI_API_HARD_DISABLED=${GEMINI_API_HARD_DISABLED:-true} - GEMINI_FALLBACK_ENABLED=${GEMINI_FALLBACK_ENABLED:-false} @@ -243,6 +247,10 @@ services: - OLLAMA_HOST_SECONDARY=${OLLAMA_HOST_SECONDARY:-http://34.21.145.224:11434} - OLLAMA_HOST_FALLBACK=${OLLAMA_HOST_FALLBACK:-http://192.168.0.111:11434} - EMBEDDING_HOST=${EMBEDDING_HOST:-} + - 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_DECISION_CANARY_TIMEOUT_SEC=${NEMOTRON_DECISION_CANARY_TIMEOUT_SEC:-300} # Gemini 只能作緊急備援；即使 .env 有 API key，預設也不得產生付費出站。 - GEMINI_API_HARD_DISABLED=${GEMINI_API_HARD_DISABLED:-true} - GEMINI_FALLBACK_ENABLED=${GEMINI_FALLBACK_ENABLED:-false} @@ -311,6 +319,10 @@ services: - OLLAMA_HOST_SECONDARY=${OLLAMA_HOST_SECONDARY:-http://34.21.145.224:11434} - OLLAMA_HOST_FALLBACK=${OLLAMA_HOST_FALLBACK:-http://192.168.0.111:11434} - EMBEDDING_HOST=${EMBEDDING_HOST:-} + - 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_DECISION_CANARY_TIMEOUT_SEC=${NEMOTRON_DECISION_CANARY_TIMEOUT_SEC:-300} # Gemini 只能作緊急備援；即使 .env 有 API key，預設也不得產生付費出站。 - GEMINI_API_HARD_DISABLED=${GEMINI_API_HARD_DISABLED:-true} - GEMINI_FALLBACK_ENABLED=${GEMINI_FALLBACK_ENABLED:-false} diff --git a/docs/AI_INTELLIGENCE_MODULE_SOT.md b/docs/AI_INTELLIGENCE_MODULE_SOT.md index e888be1..335e45e 100644 --- a/docs/AI_INTELLIGENCE_MODULE_SOT.md +++ b/docs/AI_INTELLIGENCE_MODULE_SOT.md @@ -1,18 +1,19 @@ # PChome 業績成長自動化作戰系統 — AI 競價情報模組 Single Source of Truth > **最後更新**: 2026-07-17 (台北時間) -> **狀態**: 🟠 Partial。Production runtime 已是 V10.813；daily scheduled internal RAG candidate canary 與四 AI Agent 產品整合 truth readback 已上線，但不得標示完整整合。Windows 99 run `2ef0a084-7aa8-4748-a0db-127fb623d75d` 的獨立 post-verifier 已證明 exact Gitea object `a193acc407ef61b9cc6340e8e11aa9d0fb9a39c5`、8 個 source hash、內外 `/health` 與所有容器 identity 一致。GCP-A/GCP-B 無有效 embedding 時，execution、persisted latest 與 top-level receipt 均為 `database_call_performed=false`、`transaction_read_only=null`、`no_database_call_due_embedding_host_preflight`，Telegram acknowledged `1/1`，零 DB/價格/`ai_insights` 寫入。四 Agent source wiring 為 `4/4`，runtime active `3/4`、healthy `0/4`、closure `3/9`；ElephantAlpha 是 `runtime_active_unverified`，NemoTron 尚無 runtime activity，且 `MCP_ROUTER_ENABLED=false`、`RAG_ENABLED=false`、MCP/RAG telemetry 仍為 0。程式存在、版本上線或 smoke 顯示 ok 都不能覆蓋這個 partial runtime truth。 -> **適用版本**: V10.813 production runtime；AI Agent/MCP/RAG full product integration remains partial +> **狀態**: 🟠 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 --- -## 零之負五、AI Agent 產品整合 truth 與 internal RAG canary（V10.811 source） +## 零之負五、AI Agent 產品整合 truth 與 internal RAG/NemoTron canary（V10.814 source） - `/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 自動跑一次，CLI 只保留給 Windows 99 controlled apply / verifier。每次最多一筆 candidate knowledge receipt，使用 Ollama-first BGE-M3 與 GCP-A/GCP-B consistency probe；V10.812 起任一主要 host 不可達或向量維度錯誤會在 DB 前 fail closed，只有完整 1024-dim embedding 才能進 PostgreSQL `SET TRANSACTION READ ONLY` pgvector probe，最後 rollback。V10.813 起 upstream 單節點 fail-safe 不可覆蓋 canary 的雙主機 requirement，latest 與 top-level receipt 必須區分 no-DB-call、read-only transaction 與 rollback terminal。 -- 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、GCP-A/GCP-B reachability、Telegram acknowledgement 與 zero-write/rollback terminal。簽名漂移、任一主要 GCP embedding host 不可達、pgvector probe 失敗或 semantic threshold 未達都必須 fail closed。 -- `bge-m3:latest` 仍是 immutable supply-chain blocker；即使 canary 通過，也只能回 `canary_passed_activation_blocked`，必須先鎖定可重現模型版本，再以 Windows 99 controlled apply 啟用 RAG shadow/canary。MCP runtime 仍受 localhost-only、read-only tool contract與 required secret presence preflight 約束，不可因 registry 已存在就宣稱上線。 -- `AI Agent product integration truth` 與 `Internal RAG candidate canary` 已接入 `/api/ai-automation/smoke`。舊的 NemoTron/ElephantAlpha class/method smoke 保留為 source guard，但不再能單獨代表產品整合完成。 +- `/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。 +- 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.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。 --- diff --git a/docs/adr/ADR-033-rag-three-guardrails.md b/docs/adr/ADR-033-rag-three-guardrails.md index 1d7ba00..a7ea06c 100644 --- a/docs/adr/ADR-033-rag-three-guardrails.md +++ b/docs/adr/ADR-033-rag-three-guardrails.md @@ -2,6 +2,7 @@ - **Status**: Accepted - **Date**: 2026-05-03 +- **Amended**: 2026-07-17 (V10.814 digest gate, degraded GCP quorum and cold-start SLO) - **Decision Maker**: 統帥 - **Author**: Operation Ollama-First v5.0（Owen 三點專業洞察 → v5.0 強化） - **Related**: ADR-032（RAG 自主學習迴圈）、ADR-031（MCP 自建）、ADR-002（pgvector）、ADR-027（Primary Ollama on GCP） @@ -101,7 +102,8 @@ chrome-reaper: #### 風險來源 - `bge-m3:latest` floating tag → Ollama upgrade 跳版本 - normalize / pooling 參數未顯式傳遞 → server-side 預設改變無感知 -- 跨主機（GCP / Secondary / 111）模型版本可能不一致 +- 跨主機（GCP-A / GCP-B）模型版本可能不一致；111 不得承接背景 embedding quorum +- `/api/tags` 成功但模型尚未載入時，冷啟動延遲可能遠高於 warm-call latency #### 簽名鎖定機制 ```python @@ -116,6 +118,11 @@ def get_embedding_signature( return hashlib.sha1(raw.encode()).hexdigest()[:12] ``` +`embedding_signature` 保留資料相容性判定；供應鏈身分另以 +`RAG_EMBED_EXPECTED_DIGEST` 固定 manifest digest。每次 canary 必須先由 +Ollama `/api/tags` 讀回 runtime digest，匹配後才允許產生 embedding。只設定 +`bge-m3:latest` 而沒有 expected digest 不構成 immutable contract。 + #### Schema 強制（migration 026） ```sql ALTER TABLE ai_insights @@ -126,24 +133,38 @@ CREATE INDEX CONCURRENTLY idx_ai_insights_embedding_signature WHERE embedding IS NOT NULL; ``` -#### 啟動時驗證（Phase 11.0 護欄） +#### 啟動 / canary 驗證（Phase 11.0 + V10.814 護欄） ```python def verify_embedding_consistency(): - """RAG service 啟動時跑： - 用同一段測試文字呼叫 GCP / Secondary / 111 三主機， - 驗證 cosine 距離 < 1e-4（浮點誤差），否則拒絕啟動。 + """先驗證 GCP-A/GCP-B manifest digest，再產生測試 embedding。 + + 兩台可達時驗證 cosine 距離 < 1e-4；只有一台 digest-verified GCP + 可達時，只允許 degraded read-only shadow canary，並保留 redundancy + activation blocker。零台 GCP 或只有 111 時 fail closed。 """ test_text = "momo電商競品分析測試向量一致性檢查" - embeddings = { - host: call_ollama(host, 'bge-m3:latest', test_text) - for host in [GCP_PRIMARY, GCP_SECONDARY, OLLAMA_111] - } + embeddings = {} + for host in [GCP_PRIMARY, GCP_SECONDARY]: + assert read_manifest_digest(host, 'bge-m3:latest') == EXPECTED_DIGEST + embeddings[host] = call_ollama( + host, + 'bge-m3:latest', + test_text, + canary_timeout_cap=150, + ) + assert len(embeddings) >= 1 diffs = [cosine_distance(embeddings[a], embeddings[b]) for a, b in itertools.combinations(embeddings, 2)] - if max(diffs) > 1e-4: + if diffs and max(diffs) > 1e-4: raise EmbeddingInconsistencyError(...) ``` +150 秒只適用於已通過 exact digest 的 bounded canary cold start。Windows 99 +run `534c6208-2088-428c-ba6f-f7a63a67f422` 實測 GCP-B BGE-M3 冷啟動 +`116.475s`、1024 維成功；一般查詢與 background worker 不得因此全面放大 +timeout。GCP-A control-plane identity 未 reconciliation 時，不得用 blind start +或新增付費 VM 取代 redundancy blocker。 + #### RAG 查詢時保護 ```python # rag_service.py:_select_hits diff --git a/docs/guides/ai_automation_mainline_work_items.md b/docs/guides/ai_automation_mainline_work_items.md index 800c460..ce16fa8 100644 --- 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 03:13 Asia/Taipei +> Updated: 2026-07-17 10:28 Asia/Taipei > Governance: `global_product_governance_v2` + ADR-038 > Current P0: `GROWTH-P0-001 comparison coverage truth + autonomous refresh` @@ -23,7 +23,7 @@ | 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 (`production_canary_failed_gcp_unreachable`) | Internal RAG candidate canary | V10.813 production runs the default-enabled daily 04:45 scheduler lane. Windows 99 run `2ef0a084-7aa8-4748-a0db-127fb623d75d` independently verifies execution, persisted latest and top-level receipts all fail closed with `database_call_performed=false`, `transaction_read_only=null` and `no_database_call_due_embedding_host_preflight`; Telegram acknowledged `1/1`, zero price/`ai_insights`/DB writes occurred, and no session-id truncation recurred. GCP-A/GCP-B still returned no valid embedding, so canary status correctly remains partial. Exit still requires both hosts healthy, an executed read-only pgvector receipt, immutable model reference, controlled `RAG_ENABLED` shadow activation, query/hit telemetry and feedback readback. | +| 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. | | 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. | @@ -45,7 +45,7 @@ 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.813 live readback preserves Hermes/OpenClaw call evidence, zero NemoTron runtime activity and ElephantAlpha durable role-owned insights. ElephantAlpha is now correctly `runtime_active_unverified`, never falsely `healthy`, until an executed plan/outcome exists. Exit requires all four role-active and healthy in the bounded window, without treating class presence or configured fallback as activity. | +| 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. | | 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. | @@ -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. | -| Production partial | Internal RAG candidate canary | V10.813 production proves unavailable/empty embeddings are blocked before any DB call, receipt semantics are consistent at all three layers, Telegram is acknowledged and auto-rollback remains available. GCP recovery, model pin and RAG shadow activation remain P0. | +| 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. | | 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. | diff --git a/docs/guides/devops_handbook.md b/docs/guides/devops_handbook.md index cbd3e2f..a885507 100644 --- a/docs/guides/devops_handbook.md +++ b/docs/guides/devops_handbook.md @@ -124,18 +124,19 @@ - **修復**: 188 的 `monitoring/prometheus.yml` 與 110 的 `/home/wooo/monitoring/prometheus.yml` blackbox HTTP targets 必須使用 `/health`；Gunicorn 保持 `worker_class=gthread`、`GUNICORN_THREADS=4`、`preload_app=False`。 ### 12. GCP-A Ollama refused / 110:11435 502 -- **快速診斷**: 在 repo 根目錄執行 `scripts/ops/diagnose_ollama_gcp_failover.sh`。此腳本不需要 root，也不會修改 nginx、Docker、GCP 或正式服務。 +- **快速診斷**: 由 Windows 99 dispatch `scripts/ops/diagnose_ollama_gcp_failover.sh` 或等價結構化 probe。此腳本不需要 root，也不會修改 nginx、Docker、GCP 或正式服務；run 必須留下 `trace_id/run_id/work_item_id` 與 no-write receipt。 - **判讀**: - `GCP-A direct /api/version` 失敗且 `GCP-B direct` 成功：primary VM、防火牆或 Ollama 服務異常；應用層會走 GCP-A → GCP-B → 111，但仍需修 primary。 - `110 proxy primary` 502 且 `110 proxy secondary` 成功：110 的 `11435` 固定代理 GCP-A，所以 primary 掛時舊 proxy 入口會失敗；需 110 root 才能改 nginx 或 reload。 - - `GCP-B embed` 成功但耗時接近 30s：表示 `bge-m3` runner 慢但可用；若經常超過 30s，應處理 GCP-B runner/CPU/模型併發，不要把 111 納入背景 embedding。 + - `GCP-B /api/tags` 成功不等於 embedding 可用。Windows 99 run `534c6208-2088-428c-ba6f-f7a63a67f422` 實測固定 digest 的 `bge-m3` 冷啟動為 `116.475s`、1024 維成功；健康指標必須拆成 model identity、cold-start latency 與 warm-call latency，不能再用單一 30 秒門檻把冷啟動誤判為故障。 - **GCP-A 修復方向**: - - 有 GCP/SSH 權限時，先確認 VM 是否開機、Firewall 是否開 `22` 與 `11434`、Ollama process 是否在 listen。 + - 先由 Windows 99 驗證 canonical project、instance、zone、external IP 與 lifecycle。舊 `astral-gateway-484913-d7` 已由 live gcloud readback 證明為 `DELETE_REQUESTED`，不得再用舊腳本直接 start/repair `momo-pro-gcp`。 + - 只有 canonical control-plane identity、成本差異與 critical break-glass authorization 都齊全後，才能確認 VM 開機、Firewall `22/11434` 與 Ollama listen 狀態。 - 110 現況若 `ssh gcp-a` 回 `port 22: Connection refused`，代表目前跳板無法進主機，不能靠 momo-pro app 修復。 - **110 proxy failover 方向**: - 需要 root：`sudo nginx -t`、修改 `/etc/nginx/sites-enabled/110-ollama-proxy.conf`、`sudo systemctl reload nginx`。 - 若要讓 `11435` 在 GCP-A 掛時 fallback 到 GCP-B，必須明確標註這是 proxy failover，不代表 GCP-A 已恢復；host health 仍應以 direct GCP-A 探針為準。 - **紅線**: - 不要把背景 `bge-m3` 任務改落 111。 - - 不要用更長 timeout 掩蓋 GCP-A refused；GCP-A 是 primary infra blocker。 - - 沒有 110 root 或 GCP SSH 權限時，只能完成診斷、文件與應用層降級，不能假裝已修復 primary。 + - canary 專用 150 秒只處理已通過 exact digest 的冷啟動；一般 embedding timeout 不隨之放大，也不能用長 timeout 掩蓋 GCP-A refused。 + - control-plane identity 未 reconciliation 時只能維持 GCP-B degraded shadow quorum、建立 drift work item，不得盲目新增或啟動付費 GCP 資產。 diff --git a/docs/memory/code_modularization_inventory_20260430.md b/docs/memory/code_modularization_inventory_20260430.md index d626c99..7abb5cc 100644 --- a/docs/memory/code_modularization_inventory_20260430.md +++ b/docs/memory/code_modularization_inventory_20260430.md @@ -73,6 +73,7 @@ - 2026-05-29 追記：同步 PChome near-threshold / focused identity 回收系列後的 `services/marketplace_product_matcher.py` 行數；此處只更新 inventory，不變更拆分策略。 - 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。 ## 達到或超過 800 行檔案清單 @@ -144,7 +145,7 @@ | 1417 | `services/telegram_templates.py` | P1 templates | decision、incident、report renderer 分離 | | 1409 | `services/import_service.py` | P1 import | acquisition、validation、writer、verifier 分離 | | 1394 | `services/telegram_bot_service.py` | P1 bot | command、formatter、client 分離 | -| 1440 | `run_scheduler.py` | P1 scheduler entry、拆分進行中 | OpenClaw timed-job registry 已外移；下一步拆 task registration 與 runtime startup | +| 1684 | `run_scheduler.py` | P1 scheduler entry、拆分進行中 | OpenClaw timed-job registry、Nemotron canary task adapter 已外移；下一步拆 task registration 與 runtime startup | | 1306 | `routes/system_public_routes.py` | P1 public routes | health、metrics、public projection 分離 | | 1269 | `app.py` | P1 bootstrap、拆分進行中 | web scheduler 相容層已移到 `services/web_scheduler_compat.py`；下一步拆 extension setup | | 1272 | `services/code_review_pipeline_service.py` | P2 code review | scan、normalize、persist、report 分離 | diff --git a/governance/ewoooc_asset_inventory.json b/governance/ewoooc_asset_inventory.json index a7976e7..dca9fec 100644 --- a/governance/ewoooc_asset_inventory.json +++ b/governance/ewoooc_asset_inventory.json @@ -86,12 +86,15 @@ "canonical_id": "host:gcp:ollama-a", "asset_type": "model_host", "owner_lane": "ai-runtime", - "source_truth": "Ollama host health receipts", + "source_truth": "Windows 99 gcloud control-plane receipt plus Ollama model runtime receipt", "runtime_identity": "34.87.90.216:11434", + "control_plane_identity_state": "drift_unresolved_deleted_project_reference", + "runtime_state": "unreachable", + "last_verified_receipt": "C:\\Wooo\\Agent99\\evidence\\ewoooc-gcp-runtime-audit-534c6208-2088-428c-ba6f-f7a63a67f422\\windows99-gcp-runtime-audit-receipt.json", "signal_freshness": {"source": "approved model probe", "runtime_target": "/api/tags", "max_age_minutes": 20}, - "policy": ["Ollama primary", "no secret payload logging"], - "executor": "services.ollama_service", - "verifier": "host health and model availability readback", + "policy": ["Ollama primary", "no secret payload logging", "no instance mutation until canonical project, instance, zone and cost are reconciled"], + "executor": "Windows 99 Agent99 for infrastructure; services.ollama_service for bounded model calls", + "verifier": "gcloud lifecycle, TCP, Ollama version, model digest and model-call readback", "backup_restore": "fallback to GCP-B then 111", "learning_targets": ["model latency", "fallback cause"] }, @@ -99,12 +102,15 @@ "canonical_id": "host:gcp:ollama-b", "asset_type": "model_host", "owner_lane": "ai-runtime", - "source_truth": "Ollama host health receipts", + "source_truth": "Windows 99 Ollama version, model digest and bounded model-call receipts", "runtime_identity": "34.21.145.224:11434", + "control_plane_identity_state": "project_instance_zone_unreconciled", + "runtime_state": "degraded_single_host_available", + "last_verified_receipt": "C:\\Wooo\\Agent99\\evidence\\ewoooc-gcp-runtime-audit-534c6208-2088-428c-ba6f-f7a63a67f422\\windows99-gcp-runtime-audit-receipt.json", "signal_freshness": {"source": "approved model probe", "runtime_target": "/api/tags", "max_age_minutes": 20}, - "policy": ["Ollama secondary", "embedding fallback"], - "executor": "services.ollama_service", - "verifier": "host health and model availability readback", + "policy": ["Ollama secondary", "embedding fallback", "expected model digest required before canary execution"], + "executor": "Windows 99 Agent99 for infrastructure; services.ollama_service for bounded model calls", + "verifier": "Ollama version, expected digest, embedding dimension and decision-only model readback", "backup_restore": "fallback to 111 for approved non-batch traffic", "learning_targets": ["model latency", "fallback cause"] }, @@ -271,12 +277,25 @@ "source_truth": "pgvector records and RAG query receipts", "runtime_identity": "services.rag_service", "signal_freshness": {"source": "RAG smoke", "runtime_target": "query/feedback logs", "max_age_minutes": 30}, - "policy": ["pgvector only", "BGE-M3 signature", "PromotionGate"], + "policy": ["pgvector only", "BGE-M3 signature and expected digest", "at least one approved GCP host for degraded shadow canary", "111 is not an embedding quorum", "PromotionGate"], "executor": "services.rag_service", "verifier": "candidate canary, signature and retrieval quality checks", "backup_restore": "disable retrieval and preserve structured DB truth", "learning_targets": ["hit rate", "feedback", "recurrence"] }, + { + "canonical_id": "ai:ewoooc:nemotron-decision-canary", + "asset_type": "agent_runtime_canary", + "owner_lane": "ai-platform", + "source_truth": "decision-only model receipt plus scheduler Telegram acknowledgement", + "runtime_identity": "services.nemotron_decision_canary_service", + "signal_freshness": {"source": "artifact receipt", "runtime_target": "qwen3 tool-calling decision", "max_age_minutes": 1560}, + "policy": ["expected qwen3 digest required", "tool execution forbidden", "database and price writes forbidden", "Telegram business action forbidden"], + "executor": "momo-scheduler daily bounded decision-only run", + "verifier": "allowed tool, synthetic SKU, stable digest and zero-side-effect terminal", + "backup_restore": "no-write terminal; disable scheduled canary without changing production dispatch", + "learning_targets": ["model latency", "tool-call validity", "digest drift", "canary recurrence"] + }, { "canonical_id": "ai:ewoooc:pixelrag", "asset_type": "visual_evidence_pipeline", diff --git a/governance/pchome_growth_program_review.json b/governance/pchome_growth_program_review.json index cb80f5a..e6308c4 100644 --- a/governance/pchome_growth_program_review.json +++ b/governance/pchome_growth_program_review.json @@ -1,21 +1,23 @@ { - "version": "pchome_growth_program_review_v1", - "generated_at": "2026-07-11T12:00:00+08:00", + "version": "pchome_growth_program_review_v2", + "generated_at": "2026-07-17T10:28:00+08:00", "supersedes_status_only_completion": true, "goal": "以新鮮 PChome 業績、可信跨平台同款資料與可歸因動作持續提升業績", "production_baseline": { - "runtime_source_sha": "b4284693e4a86057e24daa3aa2fb3494f046edae", - "runtime_version": "V10.773", - "latest_sales_date": "2026-06-24", - "sales_age_days": 16, + "runtime_source_sha": "a193acc407ef61b9cc6340e8e11aa9d0fb9a39c5", + "runtime_version": "V10.813", + "latest_sales_date": "2026-07-13", + "sales_age_days_at_last_verified_readback": 2, "catalog_product_count": 13032, "catalog_products_with_external_offer": 35, "catalog_external_offer_coverage_pct": 0.27, "top_opportunity_count": 20, - "top_opportunity_actionable_mapped_count": 5, - "top_opportunity_mapping_rate_pct": 25.0, + "fixed_top50_verified_count": 25, + "fixed_top50_count_coverage_pct": 50.0, + "fixed_top50_revenue_coverage_pct": 59.782, "external_offer_sources_with_verified_rows": [ - "momo_reference" + "momo_reference", + "yahoo_shopping_tw" ], "verified_external_offer_count": 473, "verified_external_product_count": 394, @@ -23,19 +25,30 @@ "runtime_rag_enabled": false }, "implementation_receipt": { - "feature_source_sha": "b8cc27e4c4ae8cc6a649bc04451b7e94d0611bef", - "dev_integration_sha": "107d8ed", - "runtime_version": "V10.775", - "deployment_mode": "controlled_sync_with_targeted_rollback_archive", - "gitea_cd_run": 1098, - "gitea_cd_status": "waiting_runner_unavailable", + "latest_verified_source_sha": "a193acc407ef61b9cc6340e8e11aa9d0fb9a39c5", + "latest_deploy_marker_sha": "9d2b74376bf8661c7ff80c8bbf71e15d9e4565f1", + "runtime_version": "V10.813", + "next_source_version": "V10.814", + "deployment_mode": "windows99_controlled_apply_with_independent_post_verifier", + "windows99_deploy_run_id": "2ef0a084-7aa8-4748-a0db-127fb623d75d", + "windows99_gcp_runtime_audit_run_id": "534c6208-2088-428c-ba6f-f7a63a67f422", + "gitea_cd_status": "runner_unavailable_controlled_fallback_verified", "application_containers_healthy": 3, "database_container_recreated": false, - "latest_sales_date": "2026-06-24", - "sales_decision_ready": false, - "valid_catalog_mapping_count": 31, - "valid_catalog_mapping_rate_pct": 0.24, - "runtime_platforms_with_data": 1, + "latest_sales_date": "2026-07-13", + "sales_decision_ready_at_last_verified_readback": true, + "fixed_top50_verified_count": 25, + "fixed_top50_count_coverage_pct": 50.0, + "runtime_platforms_with_data": 2, + "ai_runtime_source_test_evidence": "146 focused tests passed; 2149 full-suite tests passed and 9 skipped", + "gcp_runtime_truth": { + "legacy_project_state": "DELETE_REQUESTED", + "gcp_a_ollama_reachable": false, + "gcp_b_ollama_version": "0.22.1", + "gcp_b_bge_m3_dimension": 1024, + "gcp_b_bge_m3_cold_start_ms": 116475, + "database_call_performed": false + }, "catchup_result": { "scanned": 8, "candidates": 3, @@ -100,7 +113,7 @@ "bounded high-sales backfill" ], "remaining": [ - "raise full-catalog coverage from 0.27%", + "raise the fixed TOP50 verified count coverage from 50% and revenue-weighted coverage from 59.782%", "brand, GTIN, MPN, model, size, count, color and seller identity graph", "automatic replay for unresolved high-sales products" ], @@ -209,7 +222,7 @@ "six-column verified-run history" ], "remaining": [ - "production V10.776 visible readback", + "authenticated production cockpit visual readback for the latest deployed version", "per-product cross-platform offer drawer and experiment outcome view" ], "production_acceptance": "first viewport shows freshness, revenue trend, coverage, risk and one executable next action without overflow" @@ -218,21 +231,27 @@ "order": 8, "id": "PCH-GROWTH-P1-002", "priority": "P1", - "status": "in_progress", + "status": "source_fix_ready_runtime_canary_pending", "lane": "ai_mcp_rag_runtime", "objective": "Use Ollama-first agents, MCP and RAG to automate retrieval, matching, validation and learning", "completed": [ "internal MCP and RAG inventory", "PixelRAG receipts and capture manifests", - "text RAG service implementation" + "text RAG service implementation", + "PixelRAG receipt to RAG candidate replay", + "daily internal RAG candidate canary with read-only pgvector rollback terminal", + "Windows 99 GCP runtime and deleted-project drift receipt", + "BGE-M3 expected-digest gate and canary-only cold-start timeout", + "NemoTron decision-only canary CLI, API and daily scheduler source" ], "remaining": [ "enable and verify MCP router runtime", - "enable and verify RAG runtime", + "deploy V10.814 and pass production RAG plus NemoTron canaries", + "enable and verify controlled RAG shadow runtime after canary proof", "Qwen3-VL Ollama-first visual replay", - "PixelRAG receipt to RAG candidate replay", - "restore GCP-A direct runtime and 110 primary proxy", - "recover GCP-B bge-m3 embedding latency below the 30 second verifier SLO", + "replace the deleted-project GCP-A reference with a canonical project, instance, zone and cost-approved asset", + "restore GCP-A direct runtime and 110 primary proxy only after control-plane reconciliation", + "split GCP-B cold-start and warm embedding SLOs and optimize the measured 116.475 second cold path", "restore or formally retire the unreachable 111 fallback route", "agent outcome metrics and durable learning acknowledgement" ], diff --git a/routes/system_public_routes.py b/routes/system_public_routes.py index 77e7db0..0fdfe2d 100644 --- a/routes/system_public_routes.py +++ b/routes/system_public_routes.py @@ -1193,6 +1193,23 @@ def ai_automation_internal_rag_candidate_canary_api(): )) +@system_public_bp.route( + '/api/ai-automation/nemotron-decision-canary', + methods=['GET'], +) +@login_required +def ai_automation_nemotron_decision_canary_api(): + """Read the latest decision-only NemoTron runtime receipt without execution.""" + from services.nemotron_decision_canary_service import ( + run_nemotron_decision_canary, + ) + + return jsonify(run_nemotron_decision_canary( + execute=False, + write_receipt=False, + )) + + @system_public_bp.route('/api/ai-automation/agent-product-integration') @login_required def ai_automation_agent_product_integration_api(): diff --git a/run_scheduler.py b/run_scheduler.py index a710936..a6084f3 100644 --- a/run_scheduler.py +++ b/run_scheduler.py @@ -9,7 +9,7 @@ run_scheduler.py — momo-scheduler 容器入口點 每 6 小時：pchome_growth_yahoo_backfill、quality_rescore、action_plan_hygiene 每 12 小時：dedup_batch 每 10 分鐘：ppt_auto_generation_catchup（補跑被長任務卡過的定期簡報） - 每 1 天 ：db_backup（03:00）、cleanup_agent_context（03:30）、backup_monitor（04:00）、internal_rag_candidate_canary（04:45）、daily_report（09:00）、roi_monthly_report gate（09:05）、ai_smoke_summary（09:10）、observability_daily_summary（09:30）、pchome_match_backfill（10:30）、pchome_growth_momo_backfill（10:45）、openclaw_meta_analysis（12:00, Phase 4 降頻）、ppt_auto_generation_daily（20:30）、ppt_vision_audit（22:00）、daily_token_report（23:55） + 每 1 天 ：db_backup（03:00）、cleanup_agent_context（03:30）、backup_monitor（04:00）、internal_rag_candidate_canary（04:45）、nemotron_decision_canary（05:00）、daily_report（09:00）、roi_monthly_report gate（09:05）、ai_smoke_summary（09:10）、observability_daily_summary（09:30）、pchome_match_backfill（10:30）、pchome_growth_momo_backfill（10:45）、openclaw_meta_analysis（12:00, Phase 4 降頻）、ppt_auto_generation_daily（20:30）、ppt_vision_audit（22:00）、daily_token_report（23:55） 每 1 週 ：weekly_strategy（週一 06:00）、ppt_auto_generation_weekly（週一 20:40） 每 1 月 ：monthly_report（每月1日 07:00）、ppt_auto_generation_monthly（每月1日 20:50） 每 1 季 ：ppt_auto_generation_quarterly（1/4/7/10 月 1 日 21:00） @@ -398,6 +398,22 @@ def run_internal_rag_candidate_canary_task() -> dict[str, object]: return payload +def run_nemotron_decision_canary_task() -> dict[str, object]: + """Run a decision-only NemoTron model canary with no tool or data writes.""" + from services.nemotron_decision_canary_scheduler_task import ( + run_scheduled_nemotron_decision_canary, + ) + + def notify_failure(*args, **kwargs): + return _notify_scheduler_failure(*args, **kwargs) + + return run_scheduled_nemotron_decision_canary( + env_flag=_env_flag, + save_stats=_save_stats, + notify_failure=notify_failure, + ) + + def _register_schedules(): schedule.every(30).minutes.do(run_revenue_automation_cycle_background_task) logger.info("📅 每 30 分鐘：revenue_automation_cycle（業績匯入 → 比價回填，非阻塞）") @@ -582,6 +598,9 @@ def _register_schedules(): schedule.every().day.at("04:45").do(run_internal_rag_candidate_canary_task) logger.info("📅 每日 04:45：internal_rag_candidate_canary（bounded read-only probe）") + schedule.every().day.at("05:00").do(run_nemotron_decision_canary_task) + logger.info("📅 每日 05:00：nemotron_decision_canary（decision-only, no tool/data writes）") + schedule.every().monday.at("06:00").do(run_weekly_strategy_task) logger.info("📅 每週一 06:00：weekly_strategy") diff --git a/scripts/ops/run_nemotron_decision_canary.py b/scripts/ops/run_nemotron_decision_canary.py new file mode 100644 index 0000000..48b335b --- /dev/null +++ b/scripts/ops/run_nemotron_decision_canary.py @@ -0,0 +1,52 @@ +#!/usr/bin/env python3 +"""Read or execute the bounded NemoTron decision-only runtime canary.""" + +from __future__ import annotations + +import argparse +import contextlib +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_decision_canary_service import ( # noqa: E402 + run_nemotron_decision_canary, +) + + +def main() -> int: + parser = argparse.ArgumentParser( + description="執行 NemoTron 模型決策但不執行工具、不寫資料庫的受控 canary。" + ) + parser.add_argument("--output-root") + parser.add_argument("--timeout-sec", type=int) + parser.add_argument("--execute", action="store_true") + parser.add_argument("--write-receipt", action="store_true") + parser.add_argument("--trace-id") + parser.add_argument("--run-id") + parser.add_argument( + "--work-item-id", + default="AI-RUNTIME-NEMOTRON-CANARY-001", + ) + args = parser.parse_args() + with contextlib.redirect_stdout(sys.stderr): + payload = run_nemotron_decision_canary( + output_root=args.output_root, + execute=args.execute, + write_receipt=bool(args.execute and args.write_receipt), + timeout_sec=args.timeout_sec, + trace_id=args.trace_id, + run_id=args.run_id, + work_item_id=args.work_item_id, + ) + print(json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True)) + return 0 if payload.get("success") else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/services/ai_agent_product_integration_service.py b/services/ai_agent_product_integration_service.py index 8759212..c4c88ab 100644 --- a/services/ai_agent_product_integration_service.py +++ b/services/ai_agent_product_integration_service.py @@ -18,6 +18,9 @@ from services.external_mcp_rag_integration_service import ( from services.internal_rag_candidate_canary_service import ( run_internal_rag_candidate_canary, ) +from services.nemotron_decision_canary_service import ( + run_nemotron_decision_canary, +) POLICY = "runtime_truth_ai_agent_product_integration_v1" @@ -442,6 +445,10 @@ def build_ai_agent_product_integration_readback( rag_runtime = dict((external.get("runtime") or {}).get("rag") or {}) rag_canary = run_internal_rag_candidate_canary(execute=False) latest_canary = dict(rag_canary.get("latest_execution") or {}) + nemotron_canary = run_nemotron_decision_canary(execute=False) + latest_nemotron_canary = dict( + nemotron_canary.get("latest_execution") or {} + ) totals = telemetry["totals"] active_agents = sum( 1 for item in agents if item["runtime"]["active_in_window"] @@ -487,6 +494,11 @@ def build_ai_agent_product_integration_readback( blockers.append("rag_runtime_telemetry_empty") if latest_canary.get("canary_passed") is not True: blockers.append("internal_rag_candidate_canary_not_proven") + if not ( + latest_nemotron_canary.get("canary_passed") is True + and latest_nemotron_canary.get("fresh") is True + ): + blockers.append("nemotron_decision_only_canary_not_proven") if stage_passed != len(stages): blockers.append("agent_controlled_apply_loop_not_closed") if telemetry["read_errors"]: @@ -502,7 +514,8 @@ def build_ai_agent_product_integration_readback( "answer_to_owner": ( f"四個 AI Agent source/排程 wiring={source_agents}/4；正式環境尚未完整整合：" f"最近 {window} 小時只有 {active_agents}/4 個 Agent 有實際呼叫，" - f"完整閉環 {stage_passed}/{len(stages)} 階段，MCP/RAG runtime 與 canary 必須以實證補齊。" + f"完整閉環 {stage_passed}/{len(stages)} 階段，MCP/RAG runtime 與 " + "NemoTron decision-only canary 必須以實證補齊。" if not full_integration else "四個 AI Agent 已有 source、runtime、MCP/RAG 與受控執行閉環實證。" ), @@ -535,6 +548,9 @@ def build_ai_agent_product_integration_readback( "telemetry_hits": totals["rag_query_hits"], "latest_candidate_canary": latest_canary, }, + "agents": { + "nemotron_decision_only_canary": latest_nemotron_canary, + }, "pixelrag": (external.get("runtime") or {}).get("pixelrag") or {}, }, "telemetry": telemetry, @@ -547,7 +563,7 @@ def build_ai_agent_product_integration_readback( "secret_read": False, }, "next_machine_action": ( - "execute_internal_rag_candidate_canary_then_activate_shadow_runtime" + "execute_nemotron_and_internal_rag_canaries_then_activate_shadow_runtime" if not full_integration else "continue_scheduled_agent_product_integration_verification" ), diff --git a/services/internal_rag_candidate_canary_service.py b/services/internal_rag_candidate_canary_service.py index 763bf8c..b6a5921 100644 --- a/services/internal_rag_candidate_canary_service.py +++ b/services/internal_rag_candidate_canary_service.py @@ -21,6 +21,7 @@ from services.pixelrag_marketplace_candidate_knowledge_replay_service import ( ) from services.rag_service import ( RAG_EMBED_DIM, + RAG_EMBED_EXPECTED_DIGEST, RAG_EMBED_MODEL, get_embedding_signature, is_rag_enabled, @@ -30,6 +31,10 @@ from services.rag_service import ( POLICY = "controlled_internal_rag_candidate_canary_v1" CANARY_VERSION = "internal_rag_candidate_canary_v1" DEFAULT_LIMIT = 1 +DEFAULT_MIN_GCP_HOSTS = max( + 1, + min(int(os.getenv("INTERNAL_RAG_CANDIDATE_CANARY_MIN_GCP_HOSTS", "1")), 2), +) DEFAULT_SIMILARITY_THRESHOLD = float( os.getenv("INTERNAL_RAG_CANDIDATE_CANARY_THRESHOLD", "0.70") ) @@ -267,20 +272,33 @@ def _verify_embedding_consistency() -> dict[str, Any]: result = dict(verify_embedding_consistency()) required_hosts = {"gcp_ollama", "ollama_secondary"} reachable = set(result.get("reachable") or []) + approved_reachable = required_hosts.intersection(reachable) upstream_ok = result.get("ok") is True - required_hosts_ready = required_hosts.issubset(reachable) + digest_verified = result.get("digest_verified") is True + minimum_hosts_ready = len(approved_reachable) >= DEFAULT_MIN_GCP_HOSTS errors = list(result.get("errors") or []) - if not required_hosts_ready: - missing = sorted(required_hosts - reachable) - marker = f"required embedding hosts unavailable: {missing}" + if not minimum_hosts_ready: + marker = ( + "minimum approved GCP embedding hosts unavailable: " + f"required={DEFAULT_MIN_GCP_HOSTS} reachable={sorted(approved_reachable)}" + ) + if marker not in errors: + errors.append(marker) + if not digest_verified: + marker = "embedding model digest is not verified" if marker not in errors: errors.append(marker) result.update( { - "ok": upstream_ok and required_hosts_ready, + "ok": upstream_ok and minimum_hosts_ready and digest_verified, "upstream_ok": upstream_ok, "required_hosts": sorted(required_hosts), - "required_hosts_ready": required_hosts_ready, + "minimum_required_host_count": DEFAULT_MIN_GCP_HOSTS, + "approved_reachable_hosts": sorted(approved_reachable), + "minimum_hosts_ready": minimum_hosts_ready, + "required_hosts_ready": minimum_hosts_ready, + "digest_verified": digest_verified, + "redundancy_degraded": len(approved_reachable) < len(required_hosts), "errors": errors, } ) @@ -367,7 +385,8 @@ def _execute_item( reachable = list(consistency.get("reachable") or []) consistency_ready = ( consistency.get("ok") is True - and required_hosts.issubset(set(reachable)) + and len(required_hosts.intersection(set(reachable))) >= DEFAULT_MIN_GCP_HOSTS + and consistency.get("digest_verified") is True ) if not consistency_ready: result.update( @@ -382,13 +401,18 @@ def _execute_item( "reachable_consistency_hosts": reachable, "canary_checks": { "cross_host_embedding_consistent": False, + "embedding_model_digest_verified": ( + consistency.get("digest_verified") is True + ), "database_call_blocked_by_preflight": True, "database_write_absent": True, "ai_insights_write_absent": True, "price_table_write_absent": True, }, - "canary_check_count": 5, - "canary_check_pass_count": 4, + "canary_check_count": 6, + "canary_check_pass_count": 4 + int( + consistency.get("digest_verified") is True + ), "rollback_terminal": "no_database_call_due_embedding_host_preflight", "error": ( "embedding_host_preflight_failed: " @@ -408,6 +432,7 @@ def _execute_item( "candidate_embedding_dimension_valid": candidate_dimension_valid, "probe_embedding_dimension_valid": probe_dimension_valid, "cross_host_embedding_consistent": True, + "embedding_model_digest_verified": True, "database_call_blocked_by_preflight": True, "database_write_absent": True, "ai_insights_write_absent": True, @@ -443,7 +468,11 @@ def _execute_item( "probe_embedding_dimension_valid": len(probe_embedding) == RAG_EMBED_DIM, "cross_host_embedding_consistent": ( consistency.get("ok") is True - and required_hosts.issubset(set(reachable)) + and len(required_hosts.intersection(set(reachable))) + >= DEFAULT_MIN_GCP_HOSTS + ), + "embedding_model_digest_verified": ( + consistency.get("digest_verified") is True ), "pgvector_transaction_read_only": ( pgvector.get("transaction_read_only") is True @@ -602,8 +631,12 @@ def run_internal_rag_candidate_canary( activation_blockers: list[str] = [] if not is_rag_enabled(): activation_blockers.append("rag_runtime_disabled") - if RAG_EMBED_MODEL.endswith(":latest"): - activation_blockers.append("rag_embedding_model_floating_tag") + if not RAG_EMBED_EXPECTED_DIGEST: + activation_blockers.append("rag_embedding_digest_guard_missing") + if consistency and consistency.get("digest_verified") is not True: + activation_blockers.append("rag_embedding_digest_unverified") + if consistency.get("redundancy_degraded") is True: + activation_blockers.append("rag_embedding_redundancy_degraded") latest_execution = _latest_execution_receipt(receipt_root) historical_canary_passed = latest_execution.get("canary_passed") is True @@ -627,7 +660,7 @@ def run_internal_rag_candidate_canary( next_action = "repair_embedding_or_pgvector_canary_failure" elif activation_blockers: status = "canary_passed_activation_blocked" - next_action = "pin_embedding_model_then_enable_rag_controlled_canary" + next_action = "restore_embedding_redundancy_then_enable_rag_shadow_runtime" else: status = "complete" next_action = "continue_scheduled_internal_rag_canary" @@ -655,7 +688,9 @@ def run_internal_rag_candidate_canary( "model": RAG_EMBED_MODEL, "dimension": RAG_EMBED_DIM, "signature": get_embedding_signature(), - "immutable_model_reference": not RAG_EMBED_MODEL.endswith(":latest"), + "expected_digest": RAG_EMBED_EXPECTED_DIGEST, + "immutable_digest_guard_configured": bool(RAG_EMBED_EXPECTED_DIGEST), + "immutable_model_reference": bool(RAG_EMBED_EXPECTED_DIGEST), "cross_host_consistency": consistency, }, "source_items": source_items, diff --git a/services/nemoton_dispatcher_service.py b/services/nemoton_dispatcher_service.py index 2094b1d..9376efc 100644 --- a/services/nemoton_dispatcher_service.py +++ b/services/nemoton_dispatcher_service.py @@ -123,6 +123,10 @@ _nim_call_count = {"date": "", "count": 0} NEMOTRON_OLLAMA_FIRST = os.getenv("NEMOTRON_OLLAMA_FIRST", "true").lower() == "true" NEMOTRON_OLLAMA_MODEL = os.getenv("NEMOTRON_OLLAMA_MODEL", "qwen3:14b") NEMOTRON_OLLAMA_TIMEOUT = int(os.getenv("NEMOTRON_OLLAMA_TIMEOUT", "180")) # 秒 +NEMOTRON_OLLAMA_EXPECTED_DIGEST = os.getenv( + "NEMOTRON_OLLAMA_EXPECTED_DIGEST", + "bdbd181c33f2ed1b31c972991882db3cf4d192569092138a7d29e973cd9debe8", +).strip().lower() def _check_nim_quota() -> bool: @@ -645,6 +649,74 @@ def _parse_content_fallback(raw_content: str) -> list: return results +def build_qwen3_dispatch_payload( + threats: list, + *, + mcp_context: str | None = None, + num_predict: int = 2048, + keep_alive: str | int | None = None, +) -> dict: + """Build the shared production/canary qwen3 tool-calling request.""" + threat_summary = json.dumps( + [ + { + "sku": t.sku, + "name": t.name, + "momo_price": t.momo_price, + "pchome_price": t.pchome_price, + "gap_pct": t.gap_pct, + "sales_delta": t.sales_7d_delta_pct, + "risk": t.risk, + "action": t.recommended_action, + "confidence": t.confidence, + **_threat_match_metadata(t), + } + for t in threats + ], + ensure_ascii=False, + ) + resolved_mcp_context = ( + build_mcp_context() if mcp_context is None else str(mcp_context) + ) + system_prompt = ( + "你是台灣電商競價情報的行動派發器。" + f"當前市場背景 (MCP)：\n{resolved_mcp_context}\n\n" + "根據 Hermes 分析師提供的威脅清單，決定對每支商品呼叫哪個工具。\n" + "路由鐵律（依序判斷，命中即停）：\n" + "1. match_type 不是 exact，或 price_basis 不是 total_price，或 alert_tier 不是 price_alert_exact " + "→ 不可直接價格告警，呼叫 flag_for_human_review，concern 說明需覆核身份、包裝或單位價。\n" + "2. gap_pct < 5% 且 sales_delta < -30% → 非價格異常，呼叫 flag_for_human_review，" + "concern 說明『價差接近 0 但銷量大幅下滑，疑似缺貨/下架/平台流量異常，請 AI 例外走查前台』。\n" + "3. gap_pct ≥ 5% 且 risk=HIGH → trigger_price_alert（填入 momo_price, comp_price）。\n" + "4. 我方價格低於競品且銷量正成長 → add_to_recommendation。\n" + "5. confidence < 0.6 或其他複雜情況 → flag_for_human_review。\n" + "每支商品只呼叫一個工具。\n" + "【語言鐵律 — 台灣標準正體中文（繁體）】所有文字欄位必須遵守：\n" + " 1. 嚴禁簡體字、嚴禁異體字（例：不可用『亊』，必須用『事』）\n" + " 2. 嚴禁短語重複（語意坍塌）、嚴禁無意義字元組合\n" + "若無法產出合理的繁體中文說明，直接輸出『請人工評估議價空間』。" + ) + payload = { + "model": NEMOTRON_OLLAMA_MODEL, + "messages": [ + {"role": "system", "content": system_prompt}, + { + "role": "user", + "content": f"請處理以下 {len(threats)} 筆威脅清單：\n{threat_summary}", + }, + ], + "tools": TOOLS, + "stream": False, + "options": { + "temperature": 0.2, + "num_predict": max(32, min(int(num_predict), 2048)), + }, + } + if keep_alive is not None: + payload["keep_alive"] = keep_alive + return payload + + def _build_footprint_json(hermes_stats: Optional[dict], nim_stats: Optional[dict]) -> dict: """ 建立結構化運算足跡 (用於 DB model_footprint JSONB 欄位) @@ -889,61 +961,7 @@ class NemotronDispatcher: resolve_ollama_host, ) - threat_summary = json.dumps( - [ - { - "sku": t.sku, - "name": t.name, - "momo_price": t.momo_price, - "pchome_price": t.pchome_price, - "gap_pct": t.gap_pct, - "sales_delta": t.sales_7d_delta_pct, - "risk": t.risk, - "action": t.recommended_action, - "confidence": t.confidence, - **_threat_match_metadata(t), - } - for t in threats - ], - ensure_ascii=False, - ) - - # 注入 MCP 市場上下文（與 NIM 路徑一致） - mcp_ctx = build_mcp_context() - - # System prompt 與 NIM 完全一致（避免兩套維護） - system_prompt = ( - "你是台灣電商競價情報的行動派發器。" - f"當前市場背景 (MCP)：\n{mcp_ctx}\n\n" - "根據 Hermes 分析師提供的威脅清單，決定對每支商品呼叫哪個工具。\n" - "路由鐵律（依序判斷，命中即停）：\n" - "1. match_type 不是 exact，或 price_basis 不是 total_price，或 alert_tier 不是 price_alert_exact " - "→ 不可直接價格告警，呼叫 flag_for_human_review，concern 說明需覆核身份、包裝或單位價。\n" - "2. gap_pct < 5% 且 sales_delta < -30% → 非價格異常，呼叫 flag_for_human_review，" - "concern 說明『價差接近 0 但銷量大幅下滑，疑似缺貨/下架/平台流量異常，請 AI 例外走查前台』。\n" - "3. gap_pct ≥ 5% 且 risk=HIGH → trigger_price_alert（填入 momo_price, comp_price）。\n" - "4. 我方價格低於競品且銷量正成長 → add_to_recommendation。\n" - "5. confidence < 0.6 或其他複雜情況 → flag_for_human_review。\n" - "每支商品只呼叫一個工具。\n" - "【語言鐵律 — 台灣標準正體中文（繁體）】所有文字欄位必須遵守：\n" - " 1. 嚴禁簡體字、嚴禁異體字（例：不可用「亊」，必須用「事」）\n" - " 2. 嚴禁短語重複（語意坍塌）、嚴禁無意義字元組合\n" - "若無法產出合理的繁體中文說明，直接輸出「請人工評估議價空間」。" - ) - - payload = { - "model": NEMOTRON_OLLAMA_MODEL, - "messages": [ - {"role": "system", "content": system_prompt}, - {"role": "user", "content": f"請處理以下 {len(threats)} 筆威脅清單：\n{threat_summary}"}, - ], - "tools": TOOLS, # 重用既有 NIM tools schema - "stream": False, - "options": { - "temperature": 0.2, - "num_predict": 2048, - }, - } + payload = build_qwen3_dispatch_payload(threats) with log_ai_call( caller='nemotron_dispatch', diff --git a/services/nemotron_decision_canary_scheduler_task.py b/services/nemotron_decision_canary_scheduler_task.py new file mode 100644 index 0000000..cae6951 --- /dev/null +++ b/services/nemotron_decision_canary_scheduler_task.py @@ -0,0 +1,120 @@ +"""Scheduler adapter for the decision-only Nemotron runtime canary.""" + +from __future__ import annotations + +import logging +from collections.abc import Callable + + +def run_scheduled_nemotron_decision_canary( + *, + env_flag: Callable[[str, bool], bool], + save_stats: Callable[[str, dict[str, object]], None], + notify_failure: Callable[..., None], +) -> dict[str, object]: + """Run one model decision, persist proof, and send lifecycle acknowledgement.""" + if not env_flag("NEMOTRON_DECISION_CANARY_SCHEDULED_ENABLED", True): + payload: dict[str, object] = { + "success": True, + "status": "skipped", + "terminal_status": "no_write_terminal", + "reason": "scheduled_canary_disabled_by_policy", + "next_machine_action": "enable_nemotron_decision_canary_after_policy_review", + } + save_stats("nemotron_decision_canary", payload) + return payload + + try: + from services.nemotron_decision_canary_service import ( + acknowledge_nemotron_decision_canary, + run_nemotron_decision_canary, + ) + from services.telegram_templates import send_telegram_with_result + + payload = run_nemotron_decision_canary( + execute=True, + write_receipt=True, + ) + execution = payload.get("execution") or {} + telegram = send_telegram_with_result( + "\n".join([ + "Nemotron decision-only canary", + f"status: {payload.get('status')}", + f"host: {execution.get('selected_host')}", + f"model: {execution.get('model')}", + f"decision: {execution.get('decision_tool')}", + f"elapsed_ms: {execution.get('model_elapsed_ms')}", + "side effects: tool=0, database=0, Telegram action=0", + f"run_id: {(payload.get('run_identity') or {}).get('run_id')}", + ]), + parse_mode=None, + ) + telegram_status = "acknowledged" if telegram.get("ok") else "failed" + receipt_path = str(execution.get("receipt_path") or "") + receipt_acknowledged = False + if receipt_path: + acknowledged = acknowledge_nemotron_decision_canary( + receipt_path, + telegram_status=telegram_status, + telegram_sent=int(telegram.get("sent") or 0), + telegram_failed=int(telegram.get("failed") or 0), + ) + payload["execution"] = acknowledged + payload["latest_execution"] = run_nemotron_decision_canary( + execute=False + ).get("latest_execution") or {} + receipt_acknowledged = True + else: + payload["receipt_error"] = "durable_execution_receipt_missing" + + verified = bool( + payload.get("success") + and telegram_status == "acknowledged" + and receipt_acknowledged + ) + payload["terminal_status"] = ( + "verified_decision_only_no_write" if verified else "partial" + ) + save_stats("nemotron_decision_canary", payload) + logging.info( + "[NemotronCanary] status=%s terminal=%s model_ms=%s telegram=%s receipt=%s", + payload.get("status"), + payload.get("terminal_status"), + execution.get("model_elapsed_ms"), + telegram_status, + "acknowledged" if receipt_acknowledged else "missing", + ) + if not verified: + notify_failure( + "run_nemotron_decision_canary_task", + RuntimeError( + "Nemotron decision canary closure or acknowledgement failed" + ), + source="Scheduler.NemotronCanary", + event_type="nemotron_decision_canary_failure", + title="Nemotron decision-only canary failed", + dedup_ttl_sec=86400, + ) + return payload + except Exception as error: + logging.error("[NemotronCanary] task failed: %s", error, exc_info=True) + notify_failure( + "run_nemotron_decision_canary_task", + error, + source="Scheduler.NemotronCanary", + event_type="nemotron_decision_canary_task_failure", + title="Nemotron decision-only canary task failed", + dedup_ttl_sec=86400, + ) + payload = { + "success": False, + "status": "failed", + "terminal_status": "partial", + "error": f"{type(error).__name__}: {str(error)[:300]}", + "next_machine_action": "inspect_nemotron_canary_receipt_and_retry", + } + save_stats("nemotron_decision_canary", payload) + return payload + + +__all__ = ["run_scheduled_nemotron_decision_canary"] diff --git a/services/nemotron_decision_canary_service.py b/services/nemotron_decision_canary_service.py new file mode 100644 index 0000000..27b8095 --- /dev/null +++ b/services/nemotron_decision_canary_service.py @@ -0,0 +1,459 @@ +"""Decision-only NemoTron runtime canary. + +The canary exercises the production qwen3 tool-calling payload against an +approved GCP Ollama host, validates the returned decision, and stops before any +tool, Telegram, database, price, or insight write is allowed. +""" + +from __future__ import annotations + +import json +import os +import time +import uuid +from dataclasses import dataclass +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Mapping + +import requests + +from services.ollama_service import OLLAMA_HOST_PRIMARY, OLLAMA_HOST_SECONDARY + + +POLICY = "controlled_nemotron_decision_only_canary_v1" +CANARY_VERSION = "nemotron_decision_only_canary_v1" +NEMOTRON_OLLAMA_MODEL = os.getenv("NEMOTRON_OLLAMA_MODEL", "qwen3:14b") +NEMOTRON_OLLAMA_EXPECTED_DIGEST = os.getenv( + "NEMOTRON_OLLAMA_EXPECTED_DIGEST", + "bdbd181c33f2ed1b31c972991882db3cf4d192569092138a7d29e973cd9debe8", +).strip().lower() +DEFAULT_TIMEOUT_SEC = max( + 30, + min(int(os.getenv("NEMOTRON_DECISION_CANARY_TIMEOUT_SEC", "300")), 600), +) +DEFAULT_MODEL_IDENTITY_TIMEOUT_SEC = max( + 2, + min(int(os.getenv("NEMOTRON_MODEL_IDENTITY_TIMEOUT_SEC", "10")), 30), +) +DEFAULT_MAX_AGE_HOURS = max( + 1, + min(int(os.getenv("NEMOTRON_DECISION_CANARY_MAX_AGE_HOURS", "26")), 168), +) +DEFAULT_OUTPUT_ROOT = os.getenv( + "NEMOTRON_DECISION_CANARY_RECEIPT_ROOT", + "/app/data/ai_automation/nemotron_decision_canary_receipts" + if Path("/app/data").exists() + else "runtime_artifacts/nemotron_decision_canary_receipts", +) + +APPROVED_TOOLS = { + "trigger_price_alert", + "add_to_recommendation", + "flag_for_human_review", +} + + +@dataclass(frozen=True) +class _SyntheticThreat: + sku: str = "CANARY-NEMO-001" + name: str = "NemoTron 受控決策驗證品" + momo_price: float = 1200.0 + pchome_price: float = 980.0 + gap_pct: float = 22.4 + sales_7d_delta_pct: float = -35.0 + risk: str = "HIGH" + recommended_action: str = "依競價證據建立受控價格告警候選" + confidence: float = 0.91 + match_type: str = "exact" + price_basis: str = "total_price" + alert_tier: str = "price_alert_exact" + match_score: float = 0.99 + competitor_product_id: str = "CANARY-COMP-001" + competitor_product_name: str = "NemoTron 受控決策驗證品" + + +def _parse_datetime(value: Any) -> datetime | None: + if not value: + return None + try: + parsed = datetime.fromisoformat(str(value).replace("Z", "+00:00")) + except ValueError: + return None + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=timezone.utc) + return parsed.astimezone(timezone.utc) + + +def _model_identity(host: str) -> dict[str, Any]: + started = time.monotonic() + try: + response = requests.get( + f"{host.rstrip('/')}/api/tags", + timeout=DEFAULT_MODEL_IDENTITY_TIMEOUT_SEC, + ) + response.raise_for_status() + models = response.json().get("models") or [] + except Exception as exc: + return { + "ok": False, + "host": host, + "model": NEMOTRON_OLLAMA_MODEL, + "digest": None, + "digest_matches": False, + "elapsed_ms": round((time.monotonic() - started) * 1000), + "error": f"{type(exc).__name__}: {str(exc)[:180]}", + } + + target = NEMOTRON_OLLAMA_MODEL + target_without_latest = target.removesuffix(":latest") + matched: Mapping[str, Any] = {} + for item in models: + names = { + str(item.get("name") or "").strip(), + str(item.get("model") or "").strip(), + } + if target in names or target_without_latest in names: + matched = item + break + digest = str(matched.get("digest") or "").strip().lower() + digest_matches = bool(digest) and digest == NEMOTRON_OLLAMA_EXPECTED_DIGEST + return { + "ok": bool(matched) and digest_matches, + "host": host, + "model": target, + "digest": digest or None, + "expected_digest": NEMOTRON_OLLAMA_EXPECTED_DIGEST, + "digest_matches": digest_matches, + "parameter_size": str((matched.get("details") or {}).get("parameter_size") or ""), + "quantization_level": str( + (matched.get("details") or {}).get("quantization_level") or "" + ), + "elapsed_ms": round((time.monotonic() - started) * 1000), + "error": None if matched else f"model_not_found:{target}", + } + + +def _latest_execution(root: Path, *, now: datetime | None = None) -> dict[str, Any]: + if not root.exists(): + return {} + candidates = sorted( + root.glob("*/nemotron_decision_canary_receipt.json"), + key=lambda path: path.stat().st_mtime, + reverse=True, + ) + if not candidates: + return {} + path = candidates[0] + try: + payload = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + return {} + generated_at = _parse_datetime(payload.get("generated_at")) + current = now or datetime.now(timezone.utc) + age_hours = ( + (current - generated_at).total_seconds() / 3600 + if generated_at is not None + else None + ) + return { + "receipt_path": str(path), + "generated_at": payload.get("generated_at"), + "age_hours": round(age_hours, 3) if age_hours is not None else None, + "fresh": age_hours is not None and age_hours <= DEFAULT_MAX_AGE_HOURS, + "status": payload.get("status"), + "canary_passed": payload.get("canary_passed") is True, + "selected_host": payload.get("selected_host"), + "model": payload.get("model"), + "observed_digest": payload.get("observed_digest"), + "decision_tool": payload.get("decision_tool"), + "decision_sku": payload.get("decision_sku"), + "model_elapsed_ms": payload.get("model_elapsed_ms"), + "tool_execution_count": int(payload.get("tool_execution_count") or 0), + "database_call_performed": payload.get("database_call_performed") is True, + "writes_database": payload.get("writes_database") is True, + "writes_price_tables": payload.get("writes_price_tables") is True, + "writes_ai_insights": payload.get("writes_ai_insights") is True, + "telegram_sent": payload.get("telegram_sent") is True, + "rollback_terminal": payload.get("rollback_terminal"), + "terminal_status": payload.get("terminal_status"), + "acknowledgements": payload.get("acknowledgements") or {}, + "run_identity": payload.get("run_identity") or {}, + "error": payload.get("error"), + } + + +def _write_receipt(root: Path, run_id: str, payload: Mapping[str, Any]) -> str: + target = root / run_id / "nemotron_decision_canary_receipt.json" + target.parent.mkdir(parents=True, exist_ok=True) + target.write_text( + json.dumps(dict(payload), ensure_ascii=False, indent=2, sort_keys=True), + encoding="utf-8", + ) + return str(target) + + +def acknowledge_nemotron_decision_canary( + receipt_path: str | Path, + *, + telegram_status: str, + telegram_sent: int = 0, + telegram_failed: int = 0, +) -> dict[str, Any]: + """Atomically append the scheduler Telegram acknowledgement to a receipt.""" + target = Path(receipt_path) + payload = json.loads(target.read_text(encoding="utf-8")) + acknowledgement = { + "status": str(telegram_status), + "sent": int(telegram_sent), + "failed": int(telegram_failed), + } + payload["acknowledgements"] = { + "telegram": acknowledgement, + "km": "not_applicable_decision_only_canary", + "rag": "not_applicable_decision_only_canary", + "mcp": "not_applicable_decision_only_canary", + "playbook": "nemotron_decision_canary_receipt_written", + } + closure = payload.setdefault("closure_receipt", {}) + closure["telegram_acknowledgement"] = str(telegram_status) + payload["terminal_status"] = ( + "verified_decision_only_no_write" + if payload.get("canary_passed") is True and telegram_status == "acknowledged" + else "partial" + ) + temporary = target.with_suffix(target.suffix + ".tmp") + temporary.write_text( + json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True), + encoding="utf-8", + ) + temporary.replace(target) + return payload + + +def run_nemotron_decision_canary( + *, + output_root: str | Path | None = None, + execute: bool = False, + write_receipt: bool = False, + timeout_sec: int | None = None, + trace_id: str | None = None, + run_id: str | None = None, + work_item_id: str = "AI-RUNTIME-NEMOTRON-CANARY-001", +) -> dict[str, Any]: + """Read the latest receipt or execute one bounded decision-only model call.""" + now = datetime.now(timezone.utc) + receipt_root = Path(output_root or DEFAULT_OUTPUT_ROOT) + resolved_run_id = str(run_id or f"nemotron-canary-{uuid.uuid4()}") + run_identity = { + "trace_id": str(trace_id or f"trace-{resolved_run_id}"), + "run_id": resolved_run_id, + "work_item_id": str(work_item_id or "AI-RUNTIME-NEMOTRON-CANARY-001"), + } + latest_before = _latest_execution(receipt_root, now=now) + if not execute: + verified = ( + latest_before.get("canary_passed") is True + and latest_before.get("fresh") is True + ) + return { + "success": True, + "policy": POLICY, + "canary_version": CANARY_VERSION, + "generated_at": now.isoformat(), + "status": "verified" if verified else "warning", + "execute": False, + "model": NEMOTRON_OLLAMA_MODEL, + "expected_digest": NEMOTRON_OLLAMA_EXPECTED_DIGEST, + "latest_execution": latest_before, + "controlled_apply": { + "network_call": False, + "model_call": False, + "tool_execution_count": 0, + "database_call_performed": False, + "database_write": False, + "price_table_write": False, + "ai_insights_write": False, + "telegram_send": False, + }, + "next_machine_action": ( + "continue_scheduled_nemotron_decision_canary" + if verified + else "execute_nemotron_decision_only_canary" + ), + } + + host_preflights = [ + _model_identity(OLLAMA_HOST_PRIMARY), + _model_identity(OLLAMA_HOST_SECONDARY), + ] + selected = next((item for item in host_preflights if item.get("ok")), None) + model_call_performed = False + model_elapsed_ms = 0 + decisions: list[dict[str, Any]] = [] + decision_format = "none" + error: str | None = None + if selected is None: + error = "no_approved_gcp_host_with_expected_nemotron_digest" + else: + from services.nemoton_dispatcher_service import ( + _parse_content_fallback, + _parse_tool_calls_struct, + build_qwen3_dispatch_payload, + ) + + threat = _SyntheticThreat() + payload = build_qwen3_dispatch_payload( + [threat], + mcp_context="controlled decision-only canary; no live market data", + num_predict=160, + keep_alive=0, + ) + payload["think"] = False + started = time.monotonic() + model_call_performed = True + try: + response = requests.post( + f"{str(selected['host']).rstrip('/')}/api/chat", + json=payload, + timeout=max(30, min(int(timeout_sec or DEFAULT_TIMEOUT_SEC), 600)), + ) + response.raise_for_status() + body = response.json() + message = body.get("message") or {} + decisions = _parse_tool_calls_struct(message.get("tool_calls") or []) + if decisions: + decision_format = "tool_calls" + else: + decisions = _parse_content_fallback(str(message.get("content") or "")) + if decisions: + decision_format = "content_fallback" + except Exception as exc: + error = f"{type(exc).__name__}: {str(exc)[:240]}" + finally: + model_elapsed_ms = round((time.monotonic() - started) * 1000) + + decision = decisions[0] if decisions else {} + decision_tool = str(decision.get("tool") or "") + decision_args = decision.get("args") if isinstance(decision.get("args"), dict) else {} + decision_sku = str((decision_args or {}).get("sku") or "") + expected_tool = "trigger_price_alert" + post_identity = _model_identity(str(selected["host"])) if selected else {} + checks = { + "approved_host_selected": selected is not None, + "expected_digest_verified_before": bool(selected and selected.get("digest_matches")), + "model_call_completed": model_call_performed and error is None, + "decision_present": bool(decisions), + "decision_tool_allowed": decision_tool in APPROVED_TOOLS, + "expected_decision_tool_selected": decision_tool == expected_tool, + "synthetic_sku_preserved": decision_sku == _SyntheticThreat.sku, + "expected_digest_stable_after": bool(post_identity.get("digest_matches")), + "tool_execution_absent": True, + "database_call_absent": True, + "telegram_send_absent": True, + } + canary_passed = all(checks.values()) + status = "canary_passed" if canary_passed else "canary_failed" + execution_receipt: dict[str, Any] = { + "policy": POLICY, + "canary_version": CANARY_VERSION, + "generated_at": now.isoformat(), + "run_identity": run_identity, + "status": status, + "canary_passed": canary_passed, + "model": NEMOTRON_OLLAMA_MODEL, + "expected_digest": NEMOTRON_OLLAMA_EXPECTED_DIGEST, + "observed_digest": selected.get("digest") if selected else None, + "selected_host": selected.get("host") if selected else None, + "host_preflights": host_preflights, + "post_model_identity": post_identity, + "model_call_performed": model_call_performed, + "model_elapsed_ms": model_elapsed_ms, + "decision_format": decision_format, + "decision_count": len(decisions), + "decision_tool": decision_tool or None, + "decision_sku": decision_sku or None, + "expected_decision_tool": expected_tool, + "decision_argument_keys": sorted((decision_args or {}).keys()), + "checks": checks, + "check_count": len(checks), + "check_pass_count": sum(checks.values()), + "decision_only": True, + "tool_execution_count": 0, + "database_call_performed": False, + "writes_database": False, + "writes_price_tables": False, + "writes_ai_insights": False, + "telegram_sent": False, + "error": error, + "rollback_terminal": "decision_only_no_tool_or_data_write", + "source_of_truth_diff": { + "expected_model": NEMOTRON_OLLAMA_MODEL, + "expected_digest": NEMOTRON_OLLAMA_EXPECTED_DIGEST, + "observed_digest": selected.get("digest") if selected else None, + "expected_decision_tool": expected_tool, + "observed_decision_tool": decision_tool or None, + }, + "closure_receipt": { + "sensor_source_receipt": bool(host_preflights), + "normalized_asset_identity": selected.get("host") if selected else None, + "source_of_truth_diff_recorded": True, + "ai_candidate_decision_recorded": bool(decisions), + "risk_policy_decision": "medium_decision_only_no_tool_execution", + "check_mode_passed": bool(selected), + "bounded_execution_performed": model_call_performed, + "independent_verifier_passed": canary_passed, + "rollback_or_no_write_terminal": "decision_only_no_tool_or_data_write", + "telegram_acknowledgement": "pending_scheduler_dispatch", + "learning_write_acknowledgement": "pending_receipt_write", + }, + } + if write_receipt: + receipt_path = str( + receipt_root + / resolved_run_id + / "nemotron_decision_canary_receipt.json" + ) + execution_receipt["receipt_path"] = receipt_path + execution_receipt["closure_receipt"][ + "learning_write_acknowledgement" + ] = "nemotron_decision_canary_receipt_written" + _write_receipt(receipt_root, resolved_run_id, execution_receipt) + + latest_after = _latest_execution(receipt_root) + return { + "success": canary_passed, + "policy": POLICY, + "canary_version": CANARY_VERSION, + "generated_at": now.isoformat(), + "run_identity": run_identity, + "status": status, + "execute": True, + "execution": execution_receipt, + "latest_execution": latest_after, + "controlled_apply": { + "risk": "medium", + "network_call": True, + "model_call": model_call_performed, + "tool_execution_count": 0, + "database_call_performed": False, + "database_write": False, + "price_table_write": False, + "ai_insights_write": False, + "telegram_send": False, + "rollback_terminal": "decision_only_no_tool_or_data_write", + }, + "next_machine_action": ( + "continue_scheduled_nemotron_decision_canary" + if canary_passed + else "repair_nemotron_model_runtime_then_retry_decision_only_canary" + ), + } + + +__all__ = [ + "CANARY_VERSION", + "POLICY", + "acknowledge_nemotron_decision_canary", + "run_nemotron_decision_canary", +] diff --git a/services/ollama_service.py b/services/ollama_service.py index 0447804..f3fc312 100644 --- a/services/ollama_service.py +++ b/services/ollama_service.py @@ -1281,7 +1281,8 @@ class OllamaService: def generate_embedding(self, text: str, model: str = "bge-m3:latest", host: str = None, timeout: int = None, - allow_111_fallback: bool = True) -> List[float]: + allow_111_fallback: bool = True, + timeout_cap: int = None) -> List[float]: """ [ADR-007] Embedding — 含三主機自動 retry（HOTFIX 2026-05-04） @@ -1302,7 +1303,8 @@ class OllamaService: model, ) clean_text = clean_text[:EMBED_MAX_CHARS] - request_timeout = min(timeout or EMBED_TIMEOUT, EMBED_MAX_TIMEOUT) + resolved_timeout_cap = max(1, int(timeout_cap or EMBED_MAX_TIMEOUT)) + request_timeout = min(timeout or EMBED_TIMEOUT, resolved_timeout_cap) def _embed_one(target_host: str) -> List[float]: """單次 embedding 嘗試 — 成功回 vec，失敗回 [] + mark_unhealthy""" diff --git a/services/rag_service.py b/services/rag_service.py index 6146d19..87418d1 100644 --- a/services/rag_service.py +++ b/services/rag_service.py @@ -64,6 +64,10 @@ RAG_EMBED_DIM = int(os.getenv('RAG_EMBED_DIM', '1024')) RAG_EMBED_NORMALIZE = os.getenv('RAG_EMBED_NORMALIZE', 'true').strip().lower() in ( 'true', '1', 'yes', 'on', ) +RAG_EMBED_EXPECTED_DIGEST = os.getenv( + 'RAG_EMBED_EXPECTED_DIGEST', + '7907646426070047a77226ac3e684fbbe8410524f7b4a74d02837e43f2146bab', +).strip().lower() # query_text 寫入長度上限（與 027 CHECK octet_length<=4096 對齊；中文 1 字 3 byte → ~1300 字） _QUERY_TEXT_MAX_BYTES = 4096 @@ -128,7 +132,12 @@ def get_embedding_signature( # ───────────────────────────────────────────────────────────────────────────── EMBED_CONSISTENCY_TEST_TEXT = "momo電商競品分析測試向量一致性檢查" EMBED_CONSISTENCY_MAX_DIFF = 1e-4 # cosine 距離上限（浮點誤差容忍） -EMBED_CONSISTENCY_TIMEOUT_SEC = 10.0 # 各主機 embedding 探測 timeout +EMBED_CONSISTENCY_TIMEOUT_SEC = float( + os.getenv('EMBED_CONSISTENCY_TIMEOUT_SEC', '150') +) +EMBED_MODEL_IDENTITY_TIMEOUT_SEC = float( + os.getenv('EMBED_MODEL_IDENTITY_TIMEOUT_SEC', '10') +) EMBED_CONSISTENCY_INCLUDE_111 = os.getenv( 'EMBED_CONSISTENCY_INCLUDE_111', 'false', @@ -147,6 +156,38 @@ def _cosine_distance(vec_a: List[float], vec_b: List[float]) -> float: return max(0.0, 1.0 - dot / (norm_a * norm_b)) +def _fetch_ollama_model_digest( + host: str, + model: str = RAG_EMBED_MODEL, +) -> tuple[str | None, str | None]: + """Read the local Ollama model manifest digest without loading the model.""" + import requests + + try: + response = requests.get( + f"{host.rstrip('/')}/api/tags", + timeout=EMBED_MODEL_IDENTITY_TIMEOUT_SEC, + ) + response.raise_for_status() + models = response.json().get('models') or [] + except Exception as exc: + return None, f"{type(exc).__name__}: {str(exc)[:160]}" + + target = str(model or '').strip() + target_without_latest = target.removesuffix(':latest') + for item in models: + names = { + str(item.get('name') or '').strip(), + str(item.get('model') or '').strip(), + } + if target in names or target_without_latest in names: + digest = str(item.get('digest') or '').strip().lower() + if digest: + return digest, None + return None, 'model_manifest_digest_missing' + return None, f"model_not_found:{target}" + + def verify_embedding_consistency( test_text: str = EMBED_CONSISTENCY_TEST_TEXT, max_diff: float = EMBED_CONSISTENCY_MAX_DIFF, @@ -182,9 +223,35 @@ def verify_embedding_consistency( hosts['ollama_111'] = OLLAMA_HOST_FALLBACK embeddings: Dict[str, List[float]] = {} + model_digests: Dict[str, str] = {} + digest_verified_hosts: List[str] = [] errors: List[str] = [] for label, host in hosts.items(): + digest, digest_error = _fetch_ollama_model_digest(host, RAG_EMBED_MODEL) + if digest: + model_digests[label] = digest + if digest_error: + errors.append(f"{label}: model identity: {digest_error}") + logger.warning("[EmbedVerify] %s model identity failed: %s", label, digest_error) + continue + if not RAG_EMBED_EXPECTED_DIGEST: + errors.append(f"{label}: expected model digest is not configured") + logger.error("[EmbedVerify] expected model digest is not configured") + continue + if digest != RAG_EMBED_EXPECTED_DIGEST: + errors.append( + f"{label}: digest mismatch expected={RAG_EMBED_EXPECTED_DIGEST[:12]} " + f"observed={str(digest or '')[:12]}" + ) + logger.error( + "[EmbedVerify] %s digest mismatch expected=%s observed=%s", + label, + RAG_EMBED_EXPECTED_DIGEST[:12], + str(digest or '')[:12], + ) + continue + digest_verified_hosts.append(label) try: t0 = time.monotonic() vec = ollama_service.generate_embedding( @@ -193,6 +260,7 @@ def verify_embedding_consistency( host=host, # 顯式指定（避免 retry 鏈干擾驗證） timeout=int(EMBED_CONSISTENCY_TIMEOUT_SEC), allow_111_fallback=(label == 'ollama_111'), + timeout_cap=int(EMBED_CONSISTENCY_TIMEOUT_SEC), ) elapsed = time.monotonic() - t0 if vec and len(vec) == RAG_EMBED_DIM: @@ -207,16 +275,24 @@ def verify_embedding_consistency( signature = get_embedding_signature() reachable = list(embeddings.keys()) + digest_verified = bool(reachable) and set(reachable).issubset( + set(digest_verified_hosts) + ) if len(embeddings) < 2: msg = f"only {len(embeddings)} host reachable, cannot cross-verify" logger.warning(f"[EmbedVerify] {msg}") return { - 'ok': True, # fail-safe：1 主機可達不算錯（戰時可能 2 主機暫斷） + 'ok': bool(embeddings) and digest_verified, 'signature': signature, 'reachable': reachable, 'max_diff': 0.0, 'errors': errors + [msg], + 'expected_digest': RAG_EMBED_EXPECTED_DIGEST, + 'model_digests': model_digests, + 'digest_verified_hosts': digest_verified_hosts, + 'digest_verified': digest_verified, + 'redundancy_degraded': len(embeddings) == 1, } # 兩兩比對 cosine 距離 @@ -241,11 +317,16 @@ def verify_embedding_consistency( ) return { - 'ok': consistent, + 'ok': consistent and digest_verified, 'signature': signature, 'reachable': reachable, 'max_diff': max_diff_observed, 'errors': errors, + 'expected_digest': RAG_EMBED_EXPECTED_DIGEST, + 'model_digests': model_digests, + 'digest_verified_hosts': digest_verified_hosts, + 'digest_verified': digest_verified, + 'redundancy_degraded': False, } diff --git a/tests/test_ai_agent_product_integration_service.py b/tests/test_ai_agent_product_integration_service.py index 2df8295..7ed9043 100644 --- a/tests/test_ai_agent_product_integration_service.py +++ b/tests/test_ai_agent_product_integration_service.py @@ -58,6 +58,15 @@ def _canary(*, passed): return {"latest_execution": {"canary_passed": passed}} +def _nemotron_canary(*, passed): + return { + "latest_execution": { + "canary_passed": passed, + "fresh": passed, + } + } + + def test_ai_agent_integration_reports_production_like_partial_truth(monkeypatch): from services import ai_agent_product_integration_service as service @@ -76,6 +85,11 @@ def test_ai_agent_integration_reports_production_like_partial_truth(monkeypatch) "run_internal_rag_candidate_canary", lambda **_kwargs: _canary(passed=False), ) + monkeypatch.setattr( + service, + "run_nemotron_decision_canary", + lambda **_kwargs: _nemotron_canary(passed=False), + ) payload = service.build_ai_agent_product_integration_readback(window_hours=168) assert payload["status"] == "partially_integrated" @@ -105,6 +119,11 @@ def test_ai_agent_integration_requires_full_runtime_closure(monkeypatch): "run_internal_rag_candidate_canary", lambda **_kwargs: _canary(passed=True), ) + monkeypatch.setattr( + service, + "run_nemotron_decision_canary", + lambda **_kwargs: _nemotron_canary(passed=True), + ) payload = service.build_ai_agent_product_integration_readback() assert payload["status"] == "fully_integrated" diff --git a/tests/test_internal_rag_candidate_canary_service.py b/tests/test_internal_rag_candidate_canary_service.py index a3b1417..0531a31 100644 --- a/tests/test_internal_rag_candidate_canary_service.py +++ b/tests/test_internal_rag_candidate_canary_service.py @@ -83,6 +83,7 @@ def test_internal_rag_canary_executes_read_only_pgvector_probe(tmp_path, monkeyp lambda: { "ok": True, "reachable": ["gcp_ollama", "ollama_secondary"], + "digest_verified": True, "max_diff": 0.0, "errors": [], }, @@ -123,7 +124,9 @@ def test_internal_rag_canary_executes_read_only_pgvector_probe(tmp_path, monkeyp assert item["run_identity"] == payload["run_identity"] -def test_internal_rag_canary_requires_both_primary_gcp_hosts(tmp_path, monkeypatch): +def test_internal_rag_canary_allows_one_verified_gcp_host_as_degraded_quorum( + tmp_path, monkeypatch +): from services import internal_rag_candidate_canary_service as service source_root = tmp_path / "candidate" @@ -137,9 +140,11 @@ def test_internal_rag_canary_requires_both_primary_gcp_hosts(tmp_path, monkeypat "_verify_embedding_consistency", lambda: { "ok": True, - "reachable": ["gcp_ollama", "ollama_111"], + "reachable": ["ollama_secondary"], + "digest_verified": True, + "redundancy_degraded": True, "max_diff": 0.0, - "errors": ["ollama_secondary unavailable"], + "errors": ["gcp_ollama unavailable"], }, ) database_calls = [] @@ -162,33 +167,29 @@ def test_internal_rag_canary_requires_both_primary_gcp_hosts(tmp_path, monkeypat write_receipt=True, ) - assert payload["status"] == "canary_failed" + assert payload["status"] == "canary_passed_activation_blocked" item = payload["executed_items"][0] - assert item["canary_checks"]["cross_host_embedding_consistent"] is False - assert item["canary_checks"]["database_call_blocked_by_preflight"] is True + assert item["canary_checks"]["cross_host_embedding_consistent"] is True + assert item["canary_checks"]["embedding_model_digest_verified"] is True assert item["required_consistency_hosts"] == ["gcp_ollama", "ollama_secondary"] - assert item["error"].startswith("embedding_host_preflight_failed") - assert item["database_call_performed"] is False - assert payload["controlled_apply"]["database_call_performed"] is False - assert payload["controlled_apply"]["database_transaction_read_only"] is None + assert item["error"] is None + assert item["database_call_performed"] is True + assert payload["controlled_apply"]["database_call_performed"] is True + assert payload["controlled_apply"]["database_transaction_read_only"] is True assert payload["controlled_apply"]["rollback_terminal"] == ( - "no_database_call_due_embedding_host_preflight" + "transaction_rollback_after_read_only_pgvector_probe" ) assert payload["closure_receipt"]["rollback_or_no_write_terminal"] == ( - "no_database_call_due_embedding_host_preflight" + "transaction_rollback_after_read_only_pgvector_probe" ) - assert payload["next_machine_action"] == ( - "restore_required_embedding_hosts_then_retry_canary" - ) - assert payload["latest_execution"]["database_call_performed"] is False - assert payload["latest_execution"]["transaction_read_only"] is None + assert "rag_embedding_redundancy_degraded" in payload["activation_blockers"] + assert payload["latest_execution"]["database_call_performed"] is True + assert payload["latest_execution"]["transaction_read_only"] is True assert payload["latest_execution"]["rollback_terminal"] == ( - "no_database_call_due_embedding_host_preflight" + "transaction_rollback_after_read_only_pgvector_probe" ) - assert payload["latest_execution"]["error"].startswith( - "embedding_host_preflight_failed" - ) - assert database_calls == [] + assert payload["latest_execution"]["error"] is None + assert len(database_calls) == 1 readback = service.run_internal_rag_candidate_canary( candidate_knowledge_receipt_root=source_root, @@ -207,6 +208,7 @@ def test_internal_rag_canary_normalises_upstream_single_host_fail_safe(monkeypat lambda: { "ok": True, "reachable": [], + "digest_verified": False, "max_diff": 0.0, "errors": ["only 0 host reachable, cannot cross-verify"], }, @@ -217,8 +219,32 @@ def test_internal_rag_canary_normalises_upstream_single_host_fail_safe(monkeypat assert payload["upstream_ok"] is True assert payload["ok"] is False assert payload["required_hosts_ready"] is False + assert payload["minimum_required_host_count"] == 1 assert payload["required_hosts"] == ["gcp_ollama", "ollama_secondary"] - assert payload["errors"][-1].startswith("required embedding hosts unavailable") + assert payload["errors"][-1] == "embedding model digest is not verified" + + +def test_internal_rag_canary_rejects_111_only_quorum(monkeypatch): + from services import internal_rag_candidate_canary_service as service + from services import rag_service + + monkeypatch.setattr( + rag_service, + "verify_embedding_consistency", + lambda: { + "ok": True, + "reachable": ["ollama_111"], + "digest_verified": True, + "max_diff": 0.0, + "errors": [], + }, + ) + + payload = service._verify_embedding_consistency() + + assert payload["ok"] is False + assert payload["approved_reachable_hosts"] == [] + assert payload["minimum_hosts_ready"] is False def test_internal_rag_canary_blocks_empty_embedding_before_database(tmp_path, monkeypatch): @@ -234,6 +260,7 @@ def test_internal_rag_canary_blocks_empty_embedding_before_database(tmp_path, mo lambda: { "ok": True, "reachable": ["gcp_ollama", "ollama_secondary"], + "digest_verified": True, "max_diff": 0.0, "errors": [], }, diff --git a/tests/test_nemotron_decision_canary_service.py b/tests/test_nemotron_decision_canary_service.py new file mode 100644 index 0000000..33ae54a --- /dev/null +++ b/tests/test_nemotron_decision_canary_service.py @@ -0,0 +1,190 @@ +import inspect +import json +import subprocess +import sys + + +class _Response: + def __init__(self, payload, status_code=200): + self._payload = payload + self.status_code = status_code + + def raise_for_status(self): + if self.status_code >= 400: + import requests + + raise requests.HTTPError(f"HTTP {self.status_code}") + + def json(self): + return self._payload + + +def _tags(service, *, digest=None): + return { + "models": [{ + "name": service.NEMOTRON_OLLAMA_MODEL, + "model": service.NEMOTRON_OLLAMA_MODEL, + "digest": digest or service.NEMOTRON_OLLAMA_EXPECTED_DIGEST, + "details": { + "parameter_size": "14.8B", + "quantization_level": "Q4_K_M", + }, + }] + } + + +def test_nemotron_canary_readback_never_calls_network(tmp_path, monkeypatch): + from services import nemotron_decision_canary_service as service + + monkeypatch.setattr( + service.requests, + "get", + lambda *_args, **_kwargs: (_ for _ in ()).throw( + AssertionError("network call forbidden") + ), + ) + payload = service.run_nemotron_decision_canary(output_root=tmp_path) + + assert payload["status"] == "warning" + assert payload["controlled_apply"]["network_call"] is False + assert payload["controlled_apply"]["database_write"] is False + + +def test_nemotron_canary_executes_decision_only_on_verified_secondary( + tmp_path, monkeypatch +): + import requests + from services import nemotron_decision_canary_service as service + + 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") + return _Response(_tags(service)) + + posted = [] + + def fake_post(url, json, timeout): + posted.append((url, json, timeout)) + 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=45, + run_id="nemotron-canary-test-001", + ) + + assert payload["status"] == "canary_passed" + execution = payload["execution"] + assert execution["selected_host"] == service.OLLAMA_HOST_SECONDARY + assert execution["decision_tool"] == "trigger_price_alert" + assert execution["decision_sku"] == "CANARY-NEMO-001" + assert execution["tool_execution_count"] == 0 + assert execution["database_call_performed"] is False + assert execution["writes_price_tables"] is False + assert execution["writes_ai_insights"] is False + assert execution["telegram_sent"] is False + assert execution["rollback_terminal"] == "decision_only_no_tool_or_data_write" + 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 payload["latest_execution"]["canary_passed"] is True + + acknowledged = service.acknowledge_nemotron_decision_canary( + execution["receipt_path"], + telegram_status="acknowledged", + telegram_sent=1, + ) + assert acknowledged["terminal_status"] == "verified_decision_only_no_write" + assert acknowledged["closure_receipt"]["telegram_acknowledgement"] == ( + "acknowledged" + ) + readback = service.run_nemotron_decision_canary(output_root=tmp_path) + assert readback["status"] == "verified" + assert readback["latest_execution"]["terminal_status"] == ( + "verified_decision_only_no_write" + ) + + +def test_nemotron_canary_blocks_model_digest_drift_before_model_call( + tmp_path, monkeypatch +): + from services import nemotron_decision_canary_service as service + + monkeypatch.setattr( + service.requests, + "get", + lambda *_args, **_kwargs: _Response(_tags(service, digest="deadbeef")), + ) + model_calls = [] + monkeypatch.setattr( + service.requests, + "post", + lambda *_args, **_kwargs: model_calls.append(True), + ) + + payload = service.run_nemotron_decision_canary( + output_root=tmp_path, + execute=True, + write_receipt=True, + ) + + 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 model_calls == [] + + +def test_nemotron_canary_route_is_read_only(): + from routes import system_public_routes as routes + + source = inspect.getsource(routes.ai_automation_nemotron_decision_canary_api) + assert "execute=False" in source + assert "write_receipt=False" in source + + +def test_nemotron_canary_cli_outputs_machine_readable_readback(tmp_path): + completed = subprocess.run( + [ + sys.executable, + "scripts/ops/run_nemotron_decision_canary.py", + "--output-root", + str(tmp_path), + ], + capture_output=True, + check=False, + text=True, + ) + + assert completed.returncode == 0 + payload = json.loads(completed.stdout) + assert payload["policy"] == "controlled_nemotron_decision_only_canary_v1" + assert payload["controlled_apply"]["tool_execution_count"] == 0 diff --git a/tests/test_ollama_embedding.py b/tests/test_ollama_embedding.py index 29f1400..b8519d7 100644 --- a/tests/test_ollama_embedding.py +++ b/tests/test_ollama_embedding.py @@ -86,3 +86,25 @@ def test_generate_embedding_caps_timeout_and_clips_input(monkeypatch): 3, ), ] + + +def test_generate_embedding_allows_bounded_canary_timeout_cap(monkeypatch): + calls = [] + + def fake_post(url, json, timeout): + calls.append((url, json, timeout)) + return FakeResponse(200, {"embeddings": [[0.1, 0.2, 0.3]]}) + + monkeypatch.setattr("services.ollama_service.EMBED_MAX_TIMEOUT", 30) + monkeypatch.setattr("services.ollama_service.requests.post", fake_post) + + vec = OllamaService().generate_embedding( + "health", + model="bge-m3:latest", + host="http://ollama", + timeout=150, + timeout_cap=150, + ) + + assert vec == [0.1, 0.2, 0.3] + assert calls[0][2] == 150 diff --git a/tests/test_rag_service.py b/tests/test_rag_service.py index 118aee5..fec969f 100644 --- a/tests/test_rag_service.py +++ b/tests/test_rag_service.py @@ -290,6 +290,11 @@ class TestEmbeddingConsistencyRouting: return _fake_embedding() monkeypatch.setattr(rs, 'EMBED_CONSISTENCY_INCLUDE_111', False) + monkeypatch.setattr( + rs, + '_fetch_ollama_model_digest', + lambda _host, _model: (rs.RAG_EMBED_EXPECTED_DIGEST, None), + ) monkeypatch.setattr(oss.ollama_service, 'generate_embedding', fake_embed) result = rs.verify_embedding_consistency() @@ -299,6 +304,11 @@ class TestEmbeddingConsistencyRouting: assert hosts == [oss.OLLAMA_HOST_PRIMARY, oss.OLLAMA_HOST_SECONDARY] assert oss.OLLAMA_HOST_FALLBACK not in hosts assert all(kwargs.get('allow_111_fallback') is False for _host, kwargs in calls) + assert all( + kwargs.get('timeout_cap') == int(rs.EMBED_CONSISTENCY_TIMEOUT_SEC) + for _host, kwargs in calls + ) + assert result['digest_verified'] is True def test_consistency_check_can_include_111_when_explicitly_enabled(self, monkeypatch): from services import rag_service as rs @@ -311,6 +321,11 @@ class TestEmbeddingConsistencyRouting: return _fake_embedding() monkeypatch.setattr(rs, 'EMBED_CONSISTENCY_INCLUDE_111', True) + monkeypatch.setattr( + rs, + '_fetch_ollama_model_digest', + lambda _host, _model: (rs.RAG_EMBED_EXPECTED_DIGEST, None), + ) monkeypatch.setattr(oss.ollama_service, 'generate_embedding', fake_embed) result = rs.verify_embedding_consistency() @@ -324,6 +339,30 @@ class TestEmbeddingConsistencyRouting: ] assert calls[-1][1].get('allow_111_fallback') is True + def test_consistency_check_blocks_digest_drift_before_embedding(self, monkeypatch): + from services import rag_service as rs + from services import ollama_service as oss + + calls = [] + monkeypatch.setattr( + rs, + '_fetch_ollama_model_digest', + lambda _host, _model: ('deadbeef', None), + ) + monkeypatch.setattr( + oss.ollama_service, + 'generate_embedding', + lambda **kwargs: calls.append(kwargs) or _fake_embedding(), + ) + + result = rs.verify_embedding_consistency() + + assert result['ok'] is False + assert result['digest_verified'] is False + assert result['reachable'] == [] + assert calls == [] + assert any('digest mismatch' in error for error in result['errors']) + # ───────────────────────────────────────────────────────────────────────────── # Test 4: fire-and-forget log 失敗不影響主流程 diff --git a/tests/test_run_scheduler_embed_consistency.py b/tests/test_run_scheduler_embed_consistency.py index 68bf2cb..e703228 100644 --- a/tests/test_run_scheduler_embed_consistency.py +++ b/tests/test_run_scheduler_embed_consistency.py @@ -216,6 +216,7 @@ def test_v2_cron_blind_spot_list_has_failure_notifications(monkeypatch): "run_embed_consistency_check", "run_ollama_111_usage_guard_check", "run_internal_rag_candidate_canary_task", + "run_nemotron_decision_canary_task", ]: source = inspect.getsource(getattr(run_scheduler, fn_name)) assert "_notify_scheduler_failure(" in source @@ -238,6 +239,7 @@ def test_roi_ai_smoke_and_daily_report_schedules_stay_staggered(): assert "schedule.every(6).hours.do(run_action_plan_hygiene_task)" in source assert "schedule.every(15).minutes.do(run_ollama_111_usage_guard_check)" in source assert 'schedule.every().day.at("04:45").do(run_internal_rag_candidate_canary_task)' in source + assert 'schedule.every().day.at("05:00").do(run_nemotron_decision_canary_task)' in source def test_scheduled_internal_rag_canary_is_bounded_and_acknowledged(monkeypatch): @@ -283,6 +285,98 @@ def test_scheduled_internal_rag_canary_is_bounded_and_acknowledged(monkeypatch): assert saved[0][0] == "internal_rag_candidate_canary" +def test_scheduled_nemotron_canary_is_decision_only_and_acknowledged(monkeypatch): + run_scheduler = _load_run_scheduler(monkeypatch) + import services.nemotron_decision_canary_service as canary_service + import services.telegram_templates as telegram_templates + + calls = [] + acknowledgements = [] + saved = [] + monkeypatch.setattr( + canary_service, + "run_nemotron_decision_canary", + lambda **kwargs: calls.append(kwargs) or { + "success": True, + "status": "canary_passed", + "run_identity": {"run_id": "nemotron-run-test"}, + "execution": { + "receipt_path": "/tmp/nemotron-canary.json", + "selected_host": "http://34.21.145.224:11434", + "model": "qwen3:14b", + "decision_tool": "trigger_price_alert", + "model_elapsed_ms": 1234, + }, + }, + ) + monkeypatch.setattr( + canary_service, + "acknowledge_nemotron_decision_canary", + lambda path, **kwargs: acknowledgements.append((path, kwargs)) or { + "canary_passed": True, + "terminal_status": "verified_decision_only_no_write", + }, + ) + monkeypatch.setattr( + telegram_templates, + "send_telegram_with_result", + lambda *_args, **_kwargs: {"ok": True, "sent": 1, "failed": 0}, + ) + monkeypatch.setattr( + run_scheduler, + "_save_stats", + lambda name, data: saved.append((name, data)), + ) + + payload = run_scheduler.run_nemotron_decision_canary_task() + + assert calls[0] == {"execute": True, "write_receipt": True} + assert calls[1] == {"execute": False} + assert acknowledgements[0][0] == "/tmp/nemotron-canary.json" + assert acknowledgements[0][1]["telegram_status"] == "acknowledged" + assert payload["terminal_status"] == "verified_decision_only_no_write" + assert saved[0][0] == "nemotron_decision_canary" + + +def test_scheduled_nemotron_canary_requires_durable_receipt(monkeypatch): + run_scheduler = _load_run_scheduler(monkeypatch) + import services.nemotron_decision_canary_service as canary_service + import services.telegram_templates as telegram_templates + + monkeypatch.setattr( + canary_service, + "run_nemotron_decision_canary", + lambda **_kwargs: { + "success": True, + "status": "canary_passed", + "execution": { + "selected_host": "http://34.21.145.224:11434", + "model": "qwen3:14b", + "decision_tool": "trigger_price_alert", + "model_elapsed_ms": 1234, + }, + }, + ) + monkeypatch.setattr( + telegram_templates, + "send_telegram_with_result", + lambda *_args, **_kwargs: {"ok": True, "sent": 1, "failed": 0}, + ) + notifications = [] + monkeypatch.setattr( + run_scheduler, + "_notify_scheduler_failure", + lambda *args, **kwargs: notifications.append((args, kwargs)), + ) + monkeypatch.setattr(run_scheduler, "_save_stats", lambda *_args: None) + + payload = run_scheduler.run_nemotron_decision_canary_task() + + assert payload["terminal_status"] == "partial" + assert payload["receipt_error"] == "durable_execution_receipt_missing" + assert notifications[0][0][0] == "run_nemotron_decision_canary_task" + + def test_pchome_growth_backfill_catchup_uses_runtime_receipt_cooldown(monkeypatch, tmp_path): run_scheduler = _load_run_scheduler(monkeypatch) stats_path = tmp_path / "scheduler_stats.json"