feat(ai): add PixelRAG platform probe readiness

TODO_NEXT_STEPS.txt

@@ -9,7 +9,7 @@
     python scripts/ops/check_production_version_truth.py
 
 目前最新版本仍以 production `https://mo.wooo.work/health` readback 為準。
-本輪 source target 為 `V10.748`；部署完成前不得宣稱正式環境已是 `V10.748`。
+本輪 source target 為 `V10.758`；部署完成前不得宣稱正式環境已是 `V10.758`。
 舊 iCloud checkout 不是 Gitea dev worktree，不得拿來當最新版本真相。
 
 ================================================================================
@@ -60,8 +60,10 @@ P0-2026-07-09. PixelRAG / MCP / RAG 全自動主線
    - 已完成：多電商 PixelRAG visual evidence lane 與 external MCP/RAG integration readback。
    - 已完成：PixelRAG receipt → internal RAG candidate replay，以及 OCR/VLM replay contract no-write readback。
    - 已完成：PixelRAG application portfolio，把可整合/可運用場景整理成 API/CLI 可讀的 priority lane、status、next machine action 與 forbidden guardrail。
+   - 已完成：PixelRAG Ollama-first VLM replay worker，提供 dry-run/execute、自證 artifact receipt、model_error receipt、blocked page guard、confidence/evidence validation，且 DB write=0、primary human gate=0。
+   - 已完成：PixelRAG VLM route readiness 與 auto-select model，execute 前自動讀 approved Ollama `/api/tags`，避免 configured model 缺失時盲打 generate；完全缺候選時寫 `model_route_not_ready` receipt。
    - 進行中：MCP/RAG runtime health → AI automation smoke。
-   - 未開始：Ollama-first OCR/VLM extraction worker、Ollama-first visual embedding benchmark、pgvector-compatible visual metadata、Coupang platform probe / structured API、跨平台 source contracts。
+   - 未開始：Ollama-first visual embedding benchmark、pgvector-compatible visual metadata、Coupang platform probe / structured API、跨平台 source contracts。
    - 主線文件：`docs/guides/ai_automation_mainline_work_items.md`。
 
 ================================================================================

config.py

@@ -402,7 +402,7 @@ YOUTUBE_API_KEY = os.getenv('YOUTUBE_API_KEY', '')
 # ==========================================
 # 系統版本與路徑
 # ==========================================
-SYSTEM_VERSION = "V10.748"
+SYSTEM_VERSION = "V10.759"
 LOG_FILE_PATH = os.path.join(BASE_DIR, 'logs/system.log')
 public_url = PUBLIC_URL  # 用於模板顯示
 

docs/AI_INTELLIGENCE_MODULE_SOT.md

@@ -116,6 +116,9 @@
 - 2026-07-09 起 PixelRAG visual receipts 必須可轉為內部 RAG candidate replay readback：`/api/ai-automation/pixelrag-rag-candidate-replay` 與 `scripts/ops/report_pixelrag_rag_candidate_replay.py` 只讀 `capture_receipt.json`，拆分 eligible / blocked / invalid receipts，輸出 tile/missing/barrier count、candidate text、promotion boundary 與 next machine action；`/api/ai-automation/smoke` 需包含 `PixelRAG RAG candidate replay`，`/api/ai-automation/scheduled-health-summary` 需輸出 `pixelrag_rag_candidate_replay` family。blocked / 403 / captcha / access denied / verify traffic page 只能進 platform probe 或 structured API 策略，不得當作商品資料、不得寫正式價格表、不得直接寫 `ai_insights`。
 - 2026-07-09 起 PixelRAG visual receipts 進入 OCR/VLM 前必須先輸出 no-write replay contract：`/api/ai-automation/pixelrag-ocr-vlm-replay` 與 `scripts/ops/report_pixelrag_ocr_vlm_replay.py` 只讀 saved tiles，輸出欄位 schema、輸出 schema、confidence/evidence validation rules、Ollama-first route contract 與 next machine action；`/api/ai-automation/smoke` 需包含 `PixelRAG OCR/VLM replay contract`，`/api/ai-automation/scheduled-health-summary` 需輸出 `pixelrag_ocr_vlm_replay` family。此 contract 不執行 OCR/VLM、不呼叫模型、不讀 secret、不連外、不寫 DB、不寫 `ai_insights`、不寫正式價格表；blocked / 403 / captcha / access denied / verify traffic page 只能進 platform probe 或 structured API 策略。
 - 2026-07-09 起 PixelRAG 可整合/可運用範圍必須有機器可讀 application portfolio：`/api/ai-automation/pixelrag-application-portfolio` 與 `scripts/ops/report_pixelrag_application_portfolio.py` 需輸出 commerce、RAG、UX、ops、marketing、governance lanes，每條 lane 必須包含 priority、status、integrates_with、use_cases、current_capability、next_machine_action、no-write 邊界與 forbidden guardrails。此 portfolio 不抓外站、不呼叫模型、不讀 secret、不寫 DB；其用途是把「還可以整合哪些」變成可排程、可驗證、可拒絕違規場景的主線工作項目。
+- 2026-07-09 起 PixelRAG ready receipts 必須有 Ollama-first VLM replay worker：`/api/ai-automation/pixelrag-vlm-replay-worker` 與 `scripts/ops/run_pixelrag_vlm_replay_worker.py` 預設 dry-run，`execute=true&write_receipt=true` 才呼叫 approved Ollama VLM route 並寫 artifact receipt；`/api/ai-automation/smoke` 需包含 `PixelRAG VLM replay worker`，`/api/ai-automation/scheduled-health-summary` 需輸出 `pixelrag_vlm_replay_worker` family。此 worker 不讀 secret、不寫 DB、不寫 `ai_insights`、不寫正式價格表；blocked / 403 / captcha / access denied receipt 自動跳 platform probe 或 structured API，ready receipt 的 VLM 結果仍須 identity matcher replay 與 PromotionGate 才能進候選知識層。
+- 2026-07-09 起 PixelRAG VLM route readiness 必須在 execute 前可讀回：`/api/ai-automation/pixelrag-vlm-route-readiness` 與 `scripts/ops/report_pixelrag_vlm_route_readiness.py` 只讀 approved Ollama `/api/tags`，輸出 configured model、candidate model、reachable host、model_route_ready 與 next machine action；`/api/ai-automation/smoke` 需包含 `PixelRAG VLM route readiness`，`/api/ai-automation/scheduled-health-summary` 需輸出 `pixelrag_vlm_route_readiness` family。此 readback 不呼叫 `/api/generate`、不讀 secret、不寫 DB；worker execute 必須使用它自動避開缺模型路由，完全缺候選時寫 `model_route_not_ready` artifact receipt。
+- 2026-07-10 起 PixelRAG platform probe 必須把 blocked / interstitial receipt 轉成機器可執行下一步：`/api/ai-automation/pixelrag-platform-probe` 與 `scripts/ops/report_pixelrag_platform_probe.py` 需讀取 capture receipts 與 VLM replay receipts，辨識 Shopee language / region / generic marketplace slogan / traffic verification 與 Coupang 403 / access denied，輸出 public browser context probe、structured source fallback、platform backoff、blocked page not product data 與 no-write 邊界；`/api/ai-automation/smoke` 需包含 `PixelRAG platform probe readiness`，`/api/ai-automation/scheduled-health-summary` 需輸出 `pixelrag_platform_probe` family。此 readback 不抓外站、不讀 secret/cookie/session、不寫 DB、不寫 `ai_insights`、不寫正式價格表；capture worker 可消化 `public_browser_context` 的 locale/timezone/Accept-Language，但必須濾掉 Cookie、Authorization、token/secret/key 類 header。
 - 2026-07-02 起 `/ai_intelligence` 商品明細與單品作戰詳情的四格價格證據必須可測：PChome 價格、MOMO 參考價、差距、可信度需以 `data-evidence` 固定，並以 `aria-label="價格證據"` 對應可掃描區塊；候選待確認或缺資料只能顯示「候選待確認 / 待補」，不得捏造價格或讓使用者打開 raw payload 才知道判斷依據。
 - 2026-07-02 起 `/ai_intelligence` 必須是密集 AI 工作台，不得退回大段文字說明頁：首屏與明細可見內容只保留短狀態、數字、四格證據與下一步按鈕；KPI note、benchmark detail、alert 副句、策略說明、decision copy、來源長句與單品 reason list 不得佔用第一層視覺。`tests/test_ai_intelligence_text_density_guardrails.py` 必須鎖住 `data-density-guardrail="compact-ai-workbench"`、短任務文案、detail meta 與 hidden explanatory copy。
 - 2026-07-02 起 `/observability/overview` 也必須採密集 AI 觀測工作台：首屏以 `data-density-guardrail="compact-observability-workbench"`、`AI 觀測 / 風險優先 / 下一步` 與 golden signals 先呈現狀態、數字與操作入口；hero lede、signal note、route desc、host meta 與資料來源長句不得佔用第一層視覺。`tests/test_observability_text_density_guardrails.py` 必須鎖住 compact marker 與 hidden explanatory copy。
@@ -878,6 +881,15 @@ POSTGRES_HOST=momo-db
 | 2026-07-09 | PixelRAG / external MCP/RAG 必須有 runtime monitoring 與 candidate replay | V10.746 起 `/api/ai-automation/smoke`、`/api/ai-automation/scheduled-health-summary` 必須輸出 `External MCP/RAG integration readback` / `external_mcp_rag_integration` 與 `PixelRAG RAG candidate replay` / `pixelrag_rag_candidate_replay`；external MCP/RAG readback 回報 9 個 capability 的 absorbed / unresolved 與 `MCP_ROUTER_ENABLED`、`RAG_ENABLED` runtime flags，PixelRAG replay 只讀 visual receipt，拆分 eligible / blocked / invalid，並明確標記 blocked page 不是商品資料。此路徑不讀 secret、不呼叫外部網路、不寫 DB、不寫 `ai_insights`、不寫正式價格表；eligible receipt 仍需 OCR/VLM replay、identity matcher replay、PromotionGate 與 embedding signature guard。 |
 | 2026-07-09 | PixelRAG OCR/VLM replay contract 必須有 runtime monitoring | V10.747 起 `/api/ai-automation/pixelrag-ocr-vlm-replay`、`scripts/ops/report_pixelrag_ocr_vlm_replay.py`、`/api/ai-automation/smoke` 與 `/api/ai-automation/scheduled-health-summary` 必須輸出 no-write OCR/VLM replay contract / `pixelrag_ocr_vlm_replay` family；readback 只讀 saved tiles 與 RAG candidate replay，輸出 ready / blocked / invalid contracts、field schema、output schema、validation rules、Ollama-first route contract、blocked page guard 與 next machine action。此階段明確標記 `extraction_execution_performed=false`、`ocr_execution_performed=false`、`vlm_execution_performed=false`、`writes_database=false`、`writes_ai_insights=false`、`writes_price_tables=false`、`network_call=false`、`secret_read=false`、`primary_human_gate_count=0`；ready receipt 才能進下一段 `run_ollama_first_vlm_replay_worker`，blocked receipt 只能進 platform probe 或 structured API 策略。 |
 | 2026-07-09 | PixelRAG application portfolio 必須把可整合場景轉成主線工作項目 | V10.748 起 `/api/ai-automation/pixelrag-application-portfolio` 與 `scripts/ops/report_pixelrag_application_portfolio.py` 必須輸出 PixelRAG 在 commerce、RAG、UX、ops、marketing、governance 的可整合/可運用 lanes；每條 lane 需有 priority、status、integrates_with、use_cases、current_capability、next_machine_action、no-write 邊界與 forbidden guardrails。此 readback 依據 PixelRAG visual-RAG pattern、Google Merchant product data、Google Product structured data 與 Baymard product list UX 轉成內部工作項目；它不抓外站、不呼叫模型、不讀 secret、不寫 DB、不把像素結果當正式價格。 |
+| 2026-07-09 | PixelRAG Ollama-first VLM replay worker 必須有 runtime monitoring | V10.751 起 `/api/ai-automation/pixelrag-vlm-replay-worker`、`scripts/ops/run_pixelrag_vlm_replay_worker.py`、`/api/ai-automation/smoke` 與 `/api/ai-automation/scheduled-health-summary` 必須輸出 controlled VLM replay worker / `pixelrag_vlm_replay_worker` family；readback 預設 dry-run，不呼叫模型、不寫 artifact，execute 模式只讀 saved tiles、呼叫 approved Ollama VLM route、驗證 JSON field confidence/evidence refs，並只寫 artifact receipt；model_error 也必須寫 failure artifact receipt，receipt 檔內需自證 `artifact_write_performed=true` 與 `receipt_path`。此 worker 明確標記 `writes_database=false`、`writes_ai_insights=false`、`writes_price_tables=false`、`secret_read=false`、`primary_human_gate_count=0`；blocked page 不得輸出商品欄位，ready VLM 結果仍需 identity matcher replay 與 PromotionGate。 |
+| 2026-07-09 | PixelRAG VLM route readiness 必須有 runtime monitoring 與 execute 前自動選模 | V10.752 起 `/api/ai-automation/pixelrag-vlm-route-readiness`、`scripts/ops/report_pixelrag_vlm_route_readiness.py`、`/api/ai-automation/smoke` 與 `/api/ai-automation/scheduled-health-summary` 必須輸出 read-only approved Ollama route/model readiness / `pixelrag_vlm_route_readiness` family；readback 只打 `/api/tags`，不呼叫 `/api/generate`，輸出 reachable host、configured model available count、candidate model、candidate host、model_route_ready 與 next machine action。`run_pixelrag_vlm_replay_worker.py --execute` 預設必須使用此 readback 自動選擇已安裝候選模型；完全沒有候選時寫 `model_route_not_ready` artifact receipt，不得盲打 missing model。 |
+| 2026-07-10 | PixelRAG VLM execute preflight 必須驗證 generate route | V10.753 起 `/api/ai-automation/pixelrag-vlm-route-readiness?probe_generate=true` 與 `scripts/ops/report_pixelrag_vlm_route_readiness.py --probe-generate` 可對已安裝候選模型執行極小 `/api/generate` preflight；smoke 預設仍不呼叫模型。`run_pixelrag_vlm_replay_worker.py --execute` 預設先執行 generate preflight，若 GCP-A direct / 110 proxy timeout、GCP-B candidate 雖安裝但 generate 不健康，worker 必須在送入 screenshot tiles 前寫 `model_route_not_ready` artifact receipt，輸出 `tag_model_route_ready`、`generate_route_ready`、`route_model_call_performed` 與 `tile_model_call_performed=false`，下一步固定為 `repair_ollama_vlm_generate_runtime_or_proxy_timeout`；不得把 tags 可見誤當 VLM 可執行。 |
+| 2026-07-10 | PixelRAG VLM tile generate 必須綁定 preflight 選出的 host / exact model | V10.754 起 `run_pixelrag_vlm_replay_worker.py --execute` 若 route readiness 已輸出 `candidate_host`，tile VLM generate 必須使用該 approved host 與 exact `candidate_model` 直呼 `/api/generate`，不得重新進入全域 Ollama resolver、不得因 111 fallback 規則把 VLM 模型降級成文字模型、不得再出現 preflight 選 111 但 tile generate tried GCP-B/GCP-A/111 的 route drift。receipt 必須輸出 `route_candidate_host`，模型錯誤時下一步為 `repair_ollama_vlm_generate_runtime_or_proxy_timeout`，仍然不寫 DB、不寫正式價格表。 |
+| 2026-07-10 | PixelRAG VLM no-probe 監控不得把 unknown 顯示成 failed | V10.755 起 smoke / scheduled-health 的 `pixelrag_vlm_route_readiness.details.generate_route_ready` 在 `generate_probe_performed=false` 時必須保留為 `null`，表示尚未做 `/api/generate` preflight；不得把未探測狀態轉成 `false`，避免 UI 或監控把 tags-only readiness 誤讀為 generate failure。 |
+| 2026-07-10 | PixelRAG VLM 空欄位與 Shopee language interstitial 必須自動轉 platform probe | V10.756 起 VLM prompt 明確把 language / region / app-download / landing / loading / logo-only / cookie consent pages 視為非商品卡；若 VLM 回傳所有 offer fields 皆空且 required title / price 缺失，worker 會標記 `non_product_or_interstitial_detected=true`，receipt 下一步改為 `run_platform_probe_or_use_structured_api`，不得反覆要求人工審核或無限 rerun VLM。 |
+| 2026-07-10 | PixelRAG VLM worker preflight 必須容納 111 VLM 冷啟延遲 | V10.757 起 `run_pixelrag_vlm_replay_worker.py --execute` 與對應 API 的 `route_generate_probe_timeout` 預設為 20 秒；111 `minicpm-v` 在正式環境可於約 4 到 20 秒間完成小型 `/api/generate` probe，8 秒容易造成假陰性。若 VLM receipt 已判定 `run_platform_probe_or_use_structured_api`，run-level `next_machine_action` 也必須同樣輸出 platform probe，不得停留在 rerun VLM 摘要。 |
+| 2026-07-10 | PixelRAG VLM language note / generic marketplace slogan 必須視為 interstitial | V10.758 起若 VLM notes 或 title 顯示 `Language selection page`、語言/地區選擇，或只抽到 Shopee `蝦皮購物 | 花得更少買得更好` 這類網站 slogan 且 price/shop 缺失，worker validation 會標記 `interstitial_signal_detected` / `generic_marketplace_title_detected` 與 `non_product_or_interstitial_detected=true`，run-level 與 item-level 下一步都輸出 `run_platform_probe_or_use_structured_api`。 |
+| 2026-07-10 | PixelRAG platform probe 必須自動消化 marketplace interstitial / blocked receipt | V10.759 起 `/api/ai-automation/pixelrag-platform-probe`、`scripts/ops/report_pixelrag_platform_probe.py`、`/api/ai-automation/smoke` 與 `/api/ai-automation/scheduled-health-summary` 必須輸出 platform probe / `pixelrag_platform_probe` family；Shopee traffic / language / generic landing 會轉成 `run_shopee_public_context_probe_then_structured_source_fallback`，Coupang 403 / access denied 會轉成 `use_structured_source_or_platform_backoff_policy`。probe plan 只產生 public empty browser context、structured adapter fallback 與 no-write contract，不讀或注入 cookie/session/login，不寫 DB、不寫 `ai_insights`、不寫正式價格表；capture worker 只允許 locale/timezone/非敏感 header 並濾掉 Cookie / Authorization / token / secret / key 類 header。 |
 | 2026-06-29 | PChome DB apply 授權 lane 必須先通過 no-write guard / decision preflight / decision closeout / issuer gate / signing-decision preflight / signing-decision closeout / signing-issuer guard | V10.725 的 PChome mapping backlog auto-policy 已新增 `/api/ai/pchome-growth/mapping-backlog/auto-policy-db-apply-authorization-lane-guard`、`/api/ai/pchome-growth/mapping-backlog/auto-policy-db-apply-authorization-decision-preflight`、`/api/ai/pchome-growth/mapping-backlog/auto-policy-db-apply-authorization-decision-closeout`、`/api/ai/pchome-growth/mapping-backlog/auto-policy-db-apply-authorization-issuer-gate`、`/api/ai/pchome-growth/mapping-backlog/auto-policy-db-apply-authorization-signing-decision-preflight`、`/api/ai/pchome-growth/mapping-backlog/auto-policy-db-apply-authorization-signing-decision-closeout` 與 `/api/ai/pchome-growth/mapping-backlog/auto-policy-db-apply-authorization-signing-issuer-guard`；這些 endpoint 只驗證 final exact request package、same-run production truth requirement、secret rejection、rollback boundary、lane entry requirements、decision input requirements、rejection policy、post-apply verifier、future authorization decision package、final nonsecret authorization envelope、signing decision preflight inputs、unsigned signing decision package 與 signable request boundary，不讀 secret、不執行 shell/SQL、不寫 DB，也不簽發 database apply authorization。 |
 | 2026-06-29 | PChome DB apply 授權簽署發行者 lane 必須先產出 final signable request package | V10.725 的 PChome mapping backlog auto-policy 新增 `/api/ai/pchome-growth/mapping-backlog/auto-policy-db-apply-authorization-signing-issuer-closeout`；此 endpoint 只把 signing-issuer guard 的 signable request boundary 收斂成 final signable request package 與 closeout contract，確認 fresh production truth、post-apply verifier、migration hash、secret boundary 與 no-side-effect checks，不讀 secret、不簽發 authorization、不執行 shell/SQL、不寫 DB，也不代表正式 DB apply 已授權。 |
 | 2026-06-29 | PChome DB apply 授權簽署執行 lane 必須先通過 operator-held secret boundary preflight | V10.725 的 PChome mapping backlog auto-policy 新增 `/api/ai/pchome-growth/mapping-backlog/auto-policy-db-apply-authorization-signing-execution-preflight`；此 endpoint 只把 final signable request package 轉成 future signing execution preflight package、operator-held secret boundary contract、nonsecret signing inputs、command-shape preview、rollback boundary 與 abort conditions，不讀 secret、不接受 plaintext secret、不簽發 authorization、不執行 shell/SQL、不寫 DB，也不代表正式 DB apply 已授權。 |

docs/guides/ai_automation_mainline_work_items.md

@@ -18,7 +18,9 @@
 | Completed | External MCP/RAG capability inventory absorbed into internal governance readback | `/api/ai-automation/external-mcp-rag-integration` and `scripts/ops/report_external_mcp_rag_integration.py` expose 9 capabilities, absorbed/unresolved counts, and runtime flags. |
 | Completed | PixelRAG receipts to internal RAG candidate replay | `/api/ai-automation/pixelrag-rag-candidate-replay` and `scripts/ops/report_pixelrag_rag_candidate_replay.py` read receipts, split eligible vs blocked, and require OCR/VLM replay plus PromotionGate before knowledge writes. |
 | Completed | PixelRAG application portfolio and integration lanes | `/api/ai-automation/pixelrag-application-portfolio` and `scripts/ops/report_pixelrag_application_portfolio.py` expose commerce, RAG, UX, ops, marketing, and governance uses with priority, status, next machine action, and forbidden guardrails. |
-| In progress | MCP/RAG runtime health in AI automation smoke | `/api/ai-automation/smoke` and `/api/ai-automation/scheduled-health-summary` include external MCP/RAG integration, PixelRAG RAG candidate replay, and PixelRAG OCR/VLM replay contract families. |
+| Completed | PixelRAG Ollama-first VLM replay worker | `/api/ai-automation/pixelrag-vlm-replay-worker` and `scripts/ops/run_pixelrag_vlm_replay_worker.py` dry-run or execute ready visual receipts against approved Ollama VLM routes, emit evidence-bound artifact receipts, and keep blocked pages out of product data. |
+| Completed | PixelRAG VLM route readiness and auto model select | `/api/ai-automation/pixelrag-vlm-route-readiness` and `scripts/ops/report_pixelrag_vlm_route_readiness.py` read approved Ollama `/api/tags`, expose configured/candidate model readiness, and let execute mode avoid missing-model blind generate calls. |
+| In progress | MCP/RAG runtime health in AI automation smoke | `/api/ai-automation/smoke` and `/api/ai-automation/scheduled-health-summary` include external MCP/RAG integration, PixelRAG RAG candidate replay, PixelRAG OCR/VLM replay contract, PixelRAG VLM route readiness, and PixelRAG VLM replay worker families. |
 | In progress | Formal production deploy/readback discipline | Every mainline change must update version, push Gitea main/dev, deploy to 188 without touching `momo-db`, and read back `/health` plus new endpoints. |
 
 ## P1
@@ -26,7 +28,7 @@
 | Status | Work item | Evidence / next machine action |
 |---|---|---|
 | Completed | OCR/VLM replay contract for visual fields | `/api/ai-automation/pixelrag-ocr-vlm-replay` and `scripts/ops/report_pixelrag_ocr_vlm_replay.py` turn saved tiles into no-write field contracts, output schemas, validation rules, and Ollama-first worker actions. |
-| Not started | Ollama-first OCR/VLM extraction worker | Execute the replay contract against approved local multimodal/OCR models and emit confidence/evidence receipts without writing formal price truth. |
+| Completed | Ollama-first OCR/VLM extraction worker | `run_pixelrag_vlm_replay_worker.py --execute --write-receipt` executes the replay contract against approved Ollama multimodal routes and emits confidence/evidence artifact receipts without writing formal price truth. |
 | Not started | Ollama-first multimodal embedding benchmark | Verify local Qwen3-VL or equivalent visual embedding on GCP-A -> GCP-B -> 111 before any visual vector retrieval. |
 | Not started | pgvector-compatible visual evidence metadata | Design metadata-first retrieval without FAISS in production unless ADR approves a different store. |
 | Not started | Coupang platform probe / structured API strategy | Treat 403 as platform barrier; prefer public structured data or approved probe, never count blocked pages as product data. |

docs/guides/ai_automation_session_sop.md

@@ -34,7 +34,11 @@
   或 `python scripts/ops/report_pixelrag_ocr_vlm_replay.py` 可讀回 ready / blocked / invalid replay contracts、field schema、validation rules 與 Ollama-first 下一步，且目前不執行 OCR/VLM、不寫正式價格。
 - PixelRAG 可整合/可運用場景盤點必須確認 `/api/ai-automation/pixelrag-application-portfolio`
   或 `python scripts/ops/report_pixelrag_application_portfolio.py` 可讀回 area、priority、status、use cases、next machine action 與 forbidden guardrails；不得只存在聊天結論。
-- AI automation smoke 必須包含 external MCP/RAG integration、PixelRAG RAG candidate replay 與 PixelRAG OCR/VLM replay contract family，避免 registry 已完成但 runtime flag / receipt replay / VLM contract 未完成時被誤報為全自動閉環。
+- PixelRAG VLM replay worker 必須確認 `/api/ai-automation/pixelrag-vlm-replay-worker`
+  或 `python scripts/ops/run_pixelrag_vlm_replay_worker.py` 可讀回 dry-run/execute、model_call、artifact receipt、blocked/ready 分流、DB write=0 與 primary human gate=0；execute 結果仍只是 candidate evidence。
+- PixelRAG VLM route readiness 必須確認 `/api/ai-automation/pixelrag-vlm-route-readiness`
+  或 `python scripts/ops/report_pixelrag_vlm_route_readiness.py` 可讀回 approved Ollama hosts、configured model 是否安裝、candidate model、自動選模下一步、model_call=false、DB write=0；缺模型不得等 execute 才暴露。
+- AI automation smoke 必須包含 external MCP/RAG integration、PixelRAG RAG candidate replay、PixelRAG OCR/VLM replay contract、PixelRAG VLM route readiness 與 PixelRAG VLM replay worker family，避免 registry 已完成但 runtime flag / receipt replay / VLM route / VLM worker 未完成時被誤報為全自動閉環。
 - AI 自動化 Prometheus 指標變更必須同步檢查 `docker/grafana/provisioning/dashboards/json/ai-automation-overview.json` 是否需要新增 panel 或查詢。
 - 188 線上 active monitoring stack 以 `monitoring/prometheus.yml` 為準；110 gateway 另有 `/home/wooo/monitoring/prometheus.yml`。若 dashboard 無資料，先確認 Prometheus `momo-app` target 與 `momo-network` 連線；所有 Blackbox HTTP target 必須打 `/health`，不可打 Dashboard 首頁 `/`。
 - Smoke dashboard 會保存 JSONL 趨勢；若新增檢查項目，要確保 history compact record 仍保持小而可讀。

docs/guides/browse_sh_crawler_playbook.md

@@ -184,6 +184,18 @@ python scripts/ops/report_pixelrag_ocr_vlm_replay.py --platform shopee_tw --plat
 
 此 contract 只把 saved tiles、欄位 schema、輸出 schema、confidence/evidence 規則與 Ollama-first worker 下一步打包；目前不執行 OCR/VLM、不呼叫模型、不寫 RAG、不寫 `ai_insights`、不寫正式價格表。ready receipt 進 `run_ollama_first_vlm_replay_worker`，blocked / 403 / captcha / access denied receipt 進 platform probe 或 structured API 策略。
 
+Ollama-first VLM replay worker：
+
+```bash
+python scripts/ops/report_pixelrag_vlm_route_readiness.py
+python scripts/ops/report_pixelrag_vlm_route_readiness.py --model minicpm-v:latest --include-models
+python scripts/ops/run_pixelrag_vlm_replay_worker.py
+python scripts/ops/run_pixelrag_vlm_replay_worker.py --platform shopee_tw --platform coupang_tw
+python scripts/ops/run_pixelrag_vlm_replay_worker.py --platform shopee_tw --execute --write-receipt --limit 1
+```
+
+API readback: `/api/ai-automation/pixelrag-vlm-route-readiness` 與 `/api/ai-automation/pixelrag-vlm-replay-worker?platform=shopee_tw`。worker 預設為 dry-run，不呼叫模型、不寫 artifact；`execute=true&write_receipt=true` 才呼叫 Ollama VLM 並寫 artifact receipt。execute 前會自動讀 approved Ollama `/api/tags` 選擇已安裝候選模型；若完全沒有候選，會寫 `model_route_not_ready` artifact receipt，不盲打 generate。即使 execute，結果仍只是 candidate evidence；不得直接寫 `ai_insights`、正式價格表或競品價格歷史，且 missing confidence/evidence 會留在 replay / probe lane。
+
 Application portfolio：
 
 ```bash

routes/system_public_routes.py

@@ -753,6 +753,115 @@ def ai_automation_pixelrag_application_portfolio_api():
     ))
 
 
+@system_public_bp.route('/api/ai-automation/pixelrag-vlm-replay-worker')
+@login_required
+def ai_automation_pixelrag_vlm_replay_worker_api():
+    """Dry-run or execute the PixelRAG Ollama-first VLM replay worker."""
+    from services.pixelrag_vlm_replay_worker_service import (
+        run_pixelrag_ollama_vlm_replay_worker,
+    )
+
+    platforms = tuple(
+        str(item or '').strip()
+        for item in request.args.getlist('platform')
+        if str(item or '').strip()
+    )
+    execute = str(request.args.get('execute') or '').strip().lower() in {
+        '1',
+        'true',
+        'yes',
+    }
+    write_receipt = str(request.args.get('write_receipt') or '').strip().lower() in {
+        '1',
+        'true',
+        'yes',
+    }
+    max_age_hours = request.args.get('max_age_hours', 168, type=int)
+    limit = request.args.get('limit', 25, type=int)
+    tile_limit = request.args.get('tile_limit', 4, type=int)
+    timeout = request.args.get('timeout', 90, type=int)
+    auto_select_model = str(request.args.get('auto_select_model', 'true')).strip().lower() not in {
+        '0',
+        'false',
+        'no',
+    }
+    probe_generate_before_execute = (
+        str(request.args.get('probe_generate_before_execute', 'true')).strip().lower()
+        not in {
+            '0',
+            'false',
+            'no',
+        }
+    )
+    route_readiness_timeout = request.args.get('route_readiness_timeout', 3, type=int)
+    route_generate_probe_timeout = request.args.get('route_generate_probe_timeout', 20, type=int)
+    return jsonify(run_pixelrag_ollama_vlm_replay_worker(
+        platform=platforms,
+        max_age_hours=max(1, min(max_age_hours or 168, 720)),
+        limit=max(1, min(limit or 25, 250)),
+        tile_limit=max(1, min(tile_limit or 4, 12)),
+        model=str(request.args.get('model') or '').strip() or None,
+        timeout=max(10, min(timeout or 90, 240)),
+        execute=execute,
+        write_receipt=bool(execute and write_receipt),
+        auto_select_model=auto_select_model,
+        route_readiness_timeout=max(1, min(route_readiness_timeout or 3, 20)),
+        probe_generate_before_execute=probe_generate_before_execute,
+        route_generate_probe_timeout=max(1, min(route_generate_probe_timeout or 20, 30)),
+    ))
+
+
+@system_public_bp.route('/api/ai-automation/pixelrag-vlm-route-readiness')
+@login_required
+def ai_automation_pixelrag_vlm_route_readiness_api():
+    """Read-only PixelRAG VLM approved route/model readiness."""
+    from services.pixelrag_vlm_route_readiness_service import (
+        build_pixelrag_vlm_route_readiness,
+    )
+
+    include_models = str(request.args.get('include_models') or '').strip().lower() in {
+        '1',
+        'true',
+        'yes',
+    }
+    probe_generate = str(request.args.get('probe_generate') or '').strip().lower() in {
+        '1',
+        'true',
+        'yes',
+    }
+    timeout = request.args.get('timeout', 3, type=int)
+    probe_timeout = request.args.get('probe_timeout', 8, type=int)
+    return jsonify(build_pixelrag_vlm_route_readiness(
+        model=str(request.args.get('model') or '').strip() or None,
+        timeout_seconds=max(1, min(timeout or 3, 20)),
+        include_models=include_models,
+        probe_generate=probe_generate,
+        probe_timeout_seconds=max(1, min(probe_timeout or 8, 30)),
+    ))
+
+
+@system_public_bp.route('/api/ai-automation/pixelrag-platform-probe')
+@login_required
+def ai_automation_pixelrag_platform_probe_api():
+    """Read-only PixelRAG platform probe readiness from barrier receipts."""
+    from services.pixelrag_platform_probe_service import (
+        build_pixelrag_platform_probe_readiness,
+    )
+
+    platforms = tuple(
+        str(item or '').strip()
+        for item in request.args.getlist('platform')
+        if str(item or '').strip()
+    )
+    max_age_hours = request.args.get('max_age_hours', 168, type=int)
+    limit = request.args.get('limit', 50, type=int)
+    return jsonify(build_pixelrag_platform_probe_readiness(
+        platform=platforms,
+        max_age_hours=max(1, min(max_age_hours or 168, 720)),
+        limit=max(1, min(limit or 50, 250)),
+    ))
+
+
 @system_public_bp.route('/api/ai-automation/external-mcp-rag-integration')
 @login_required
 def ai_automation_external_mcp_rag_integration_api():

scripts/ops/capture_pixelrag_visual_evidence.js

@@ -17,6 +17,7 @@ const DEFAULT_OUTPUT_DIR = 'runtime_artifacts/pixelrag_visual_evidence';
 const DEFAULT_TIMEOUT_MS = 30000;
 const DEFAULT_SETTLE_MS = 500;
 const DEFAULT_MAX_TILES = 80;
+const FORBIDDEN_CONTEXT_HEADER_TOKENS = ['cookie', 'authorization', 'token', 'secret', 'key'];
 
 const ALLOWED_PLATFORM_HOSTS = {
   momo: new Set(['m.momoshop.com.tw', 'www.momoshop.com.tw']),
@@ -184,6 +185,30 @@ function positiveInt(value, fallback) {
   return Number.isFinite(parsed) && parsed > 0 ? parsed : fallback;
 }
 
+function safePublicBrowserContext(manifest) {
+  const input = manifest.public_browser_context || {};
+  const headers = {};
+  const rawHeaders = input.extra_http_headers || {};
+  for (const [key, value] of Object.entries(rawHeaders)) {
+    const cleanKey = String(key || '').trim();
+    const lowerKey = cleanKey.toLowerCase();
+    if (!cleanKey || FORBIDDEN_CONTEXT_HEADER_TOKENS.some((token) => lowerKey.includes(token))) {
+      continue;
+    }
+    headers[cleanKey] = String(value || '');
+  }
+  return {
+    context_policy: 'public_empty_browser_context_no_login',
+    locale: String(input.locale || 'zh-TW'),
+    timezone_id: String(input.timezone_id || input.timezoneId || 'Asia/Taipei'),
+    extra_http_headers: headers,
+    credentialed_session_allowed: false,
+    storage_state_allowed: false,
+    raw_cookie_or_session_read_allowed: false,
+    login_allowed: false,
+  };
+}
+
 function buildTilePlan({ width, height, tileWidth, tileHeight, maxTiles }) {
   const tiles = [];
   const tilesX = Math.max(1, Math.ceil(width / tileWidth));
@@ -273,6 +298,7 @@ function buildBaseArtifact(manifest, options, errors = []) {
     tile_size: { width: tileWidth, height: tileHeight },
     tile_plan: tilePlan,
     files: [],
+    public_browser_context: safePublicBrowserContext(manifest),
   };
 }
 
@@ -288,8 +314,19 @@ async function capture(manifest, options, artifact) {
     launchOptions.executablePath = chromePath;
   }
 
+  const publicContext = safePublicBrowserContext(manifest);
+  const contextOptions = {
+    ignoreHTTPSErrors: true,
+    viewport,
+    locale: publicContext.locale,
+    timezoneId: publicContext.timezone_id,
+  };
+  if (Object.keys(publicContext.extra_http_headers).length) {
+    contextOptions.extraHTTPHeaders = publicContext.extra_http_headers;
+  }
+
   const browser = await chromium.launch(launchOptions);
-  const context = await browser.newContext({ ignoreHTTPSErrors: true, viewport });
+  const context = await browser.newContext(contextOptions);
   const page = await context.newPage();
   try {
     const response = await page.goto(manifest.capture_target.url, {

scripts/ops/capture_pixelrag_visual_evidence.py

@@ -23,6 +23,7 @@ DEFAULT_OUTPUT_DIR = "runtime_artifacts/pixelrag_visual_evidence"
 DEFAULT_TIMEOUT_MS = 30000
 DEFAULT_SETTLE_MS = 500
 DEFAULT_MAX_TILES = 80
+FORBIDDEN_CONTEXT_HEADER_TOKENS = ("cookie", "authorization", "token", "secret", "key")
 ALLOWED_PLATFORM_HOSTS = {
     "momo": {"m.momoshop.com.tw", "www.momoshop.com.tw"},
     "pchome": {"24h.pchome.com.tw", "ecshweb.pchome.com.tw", "ecapi-cdn.pchome.com.tw"},
@@ -123,6 +124,33 @@ def _build_tile_plan(
     }
 
 
+def _safe_public_browser_context(manifest: dict[str, Any]) -> dict[str, Any]:
+    raw_context = manifest.get("public_browser_context") or {}
+    raw_headers = raw_context.get("extra_http_headers") or {}
+    headers: dict[str, str] = {}
+    if isinstance(raw_headers, dict):
+        for key, value in raw_headers.items():
+            clean_key = str(key or "").strip()
+            lower_key = clean_key.lower()
+            if not clean_key:
+                continue
+            if any(token in lower_key for token in FORBIDDEN_CONTEXT_HEADER_TOKENS):
+                continue
+            headers[clean_key] = str(value or "")
+    return {
+        "context_policy": "public_empty_browser_context_no_login",
+        "locale": str(raw_context.get("locale") or "zh-TW"),
+        "timezone_id": str(
+            raw_context.get("timezone_id") or raw_context.get("timezoneId") or "Asia/Taipei"
+        ),
+        "extra_http_headers": headers,
+        "credentialed_session_allowed": False,
+        "storage_state_allowed": False,
+        "raw_cookie_or_session_read_allowed": False,
+        "login_allowed": False,
+    }
+
+
 def _output_paths(manifest: dict[str, Any], output_dir: str) -> dict[str, Path]:
     target = manifest.get("capture_target") or {}
     manifest_id = manifest.get("manifest_id") or _safe_name(
@@ -187,6 +215,7 @@ def _base_artifact(
             max_tiles=_positive_int(args.max_tiles, DEFAULT_MAX_TILES),
         ),
         "files": [],
+        "public_browser_context": _safe_public_browser_context(manifest),
     }
 
 
@@ -205,10 +234,19 @@ def _capture(
 
     with sync_playwright() as playwright:
         browser = playwright.chromium.launch(headless=True)
-        context = browser.new_context(ignore_https_errors=True, viewport={
-            "width": viewport["width"],
-            "height": viewport["height"],
-        })
+        public_context = _safe_public_browser_context(manifest)
+        context_kwargs = {
+            "ignore_https_errors": True,
+            "viewport": {
+                "width": viewport["width"],
+                "height": viewport["height"],
+            },
+            "locale": public_context["locale"],
+            "timezone_id": public_context["timezone_id"],
+        }
+        if public_context["extra_http_headers"]:
+            context_kwargs["extra_http_headers"] = public_context["extra_http_headers"]
+        context = browser.new_context(**context_kwargs)
         page = context.new_page()
         try:
             response = page.goto(target["url"], wait_until=args.wait_until, timeout=timeout_ms)

scripts/ops/report_pixelrag_platform_probe.py

@@ -0,0 +1,65 @@
+#!/usr/bin/env python3
+"""Report PixelRAG platform probe readiness from barrier receipts."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+from pathlib import Path
+
+
+ROOT = Path(__file__).resolve().parents[2]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+from services.pixelrag_platform_probe_service import (  # noqa: E402
+    build_pixelrag_platform_probe_readiness,
+)
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(
+        description="輸出 PixelRAG platform probe readiness 的機器可讀讀回。"
+    )
+    parser.add_argument(
+        "--artifact-root",
+        help="PixelRAG visual evidence artifact root；預設使用 production/container 設定。",
+    )
+    parser.add_argument(
+        "--vlm-receipt-root",
+        help="PixelRAG VLM replay receipt root；預設使用 production/container 設定。",
+    )
+    parser.add_argument(
+        "--platform",
+        action="append",
+        dest="platforms",
+        help="限制平台，可重複指定，例如 --platform shopee_tw --platform coupang_tw。",
+    )
+    parser.add_argument(
+        "--max-age-hours",
+        type=int,
+        default=168,
+        help="receipt 最大新鮮度小時數。",
+    )
+    parser.add_argument(
+        "--limit",
+        type=int,
+        default=50,
+        help="最多輸出 probe candidate 數。",
+    )
+    args = parser.parse_args()
+
+    payload = build_pixelrag_platform_probe_readiness(
+        artifact_root=args.artifact_root,
+        vlm_receipt_root=args.vlm_receipt_root,
+        platform=tuple(args.platforms or ()),
+        max_age_hours=args.max_age_hours,
+        limit=args.limit,
+    )
+    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())

scripts/ops/report_pixelrag_vlm_route_readiness.py

@@ -0,0 +1,62 @@
+#!/usr/bin/env python3
+"""Report PixelRAG VLM route/model readiness."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+from pathlib import Path
+
+
+ROOT = Path(__file__).resolve().parents[2]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+from services.pixelrag_vlm_route_readiness_service import (  # noqa: E402
+    build_pixelrag_vlm_route_readiness,
+)
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(
+        description="讀回 PixelRAG VLM approved Ollama route / model readiness。"
+    )
+    parser.add_argument("--model", help="要檢查的 configured VLM model。")
+    parser.add_argument(
+        "--timeout",
+        type=int,
+        default=3,
+        help="/api/tags probe timeout 秒數。",
+    )
+    parser.add_argument(
+        "--include-models",
+        action="store_true",
+        help="輸出每個 host 的 model list。",
+    )
+    parser.add_argument(
+        "--probe-generate",
+        action="store_true",
+        help="執行極小 /api/generate preflight；預設不呼叫模型。",
+    )
+    parser.add_argument(
+        "--probe-timeout",
+        type=int,
+        default=8,
+        help="/api/generate preflight timeout 秒數。",
+    )
+    args = parser.parse_args()
+
+    payload = build_pixelrag_vlm_route_readiness(
+        model=args.model,
+        timeout_seconds=args.timeout,
+        include_models=args.include_models,
+        probe_generate=args.probe_generate,
+        probe_timeout_seconds=args.probe_timeout,
+    )
+    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())

scripts/ops/run_pixelrag_vlm_replay_worker.py

@@ -0,0 +1,124 @@
+#!/usr/bin/env python3
+"""Run or dry-run the PixelRAG Ollama VLM replay worker."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+from pathlib import Path
+
+
+ROOT = Path(__file__).resolve().parents[2]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+from services.pixelrag_vlm_replay_worker_service import (  # noqa: E402
+    DEFAULT_MODEL,
+    run_pixelrag_ollama_vlm_replay_worker,
+)
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(
+        description="執行或 dry-run PixelRAG Ollama-first VLM replay worker。"
+    )
+    parser.add_argument(
+        "--artifact-root",
+        help="PixelRAG visual evidence artifact root；預設使用 production/container 設定。",
+    )
+    parser.add_argument(
+        "--output-root",
+        help="VLM replay artifact receipt output root。",
+    )
+    parser.add_argument(
+        "--platform",
+        action="append",
+        dest="platforms",
+        help="限制平台，可重複指定，例如 --platform shopee_tw --platform coupang_tw。",
+    )
+    parser.add_argument(
+        "--max-age-hours",
+        type=int,
+        default=168,
+        help="receipt 最大新鮮度小時數。",
+    )
+    parser.add_argument(
+        "--limit",
+        type=int,
+        default=25,
+        help="最多處理 receipt 數。",
+    )
+    parser.add_argument(
+        "--tile-limit",
+        type=int,
+        default=4,
+        help="每個 receipt 最多送入 VLM 的 tile 數。",
+    )
+    parser.add_argument(
+        "--model",
+        default=DEFAULT_MODEL,
+        help="Ollama VLM model。",
+    )
+    parser.add_argument(
+        "--timeout",
+        type=int,
+        default=90,
+        help="單次 Ollama generate timeout 秒數。",
+    )
+    parser.add_argument(
+        "--execute",
+        action="store_true",
+        help="真的呼叫 Ollama VLM；未指定時只做 no-write dry-run。",
+    )
+    parser.add_argument(
+        "--write-receipt",
+        action="store_true",
+        help="execute 後寫入 VLM replay artifact receipt；不寫 DB。",
+    )
+    parser.add_argument(
+        "--no-auto-select-model",
+        action="store_true",
+        help="停用 execute 前的 approved route / installed model 自動選擇。",
+    )
+    parser.add_argument(
+        "--route-readiness-timeout",
+        type=int,
+        default=3,
+        help="auto-select model 的 /api/tags readiness timeout 秒數。",
+    )
+    parser.add_argument(
+        "--no-probe-generate-before-execute",
+        action="store_true",
+        help="停用 execute 前的極小 /api/generate preflight。",
+    )
+    parser.add_argument(
+        "--route-generate-probe-timeout",
+        type=int,
+        default=20,
+        help="execute 前 /api/generate preflight timeout 秒數。",
+    )
+    args = parser.parse_args()
+
+    payload = run_pixelrag_ollama_vlm_replay_worker(
+        artifact_root=args.artifact_root,
+        output_root=args.output_root,
+        platform=tuple(args.platforms or ()),
+        max_age_hours=args.max_age_hours,
+        limit=args.limit,
+        tile_limit=args.tile_limit,
+        model=args.model,
+        timeout=args.timeout,
+        execute=args.execute,
+        write_receipt=bool(args.write_receipt and args.execute),
+        auto_select_model=not args.no_auto_select_model,
+        route_readiness_timeout=args.route_readiness_timeout,
+        probe_generate_before_execute=not args.no_probe_generate_before_execute,
+        route_generate_probe_timeout=args.route_generate_probe_timeout,
+    )
+    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())

services/ai_automation_smoke_service.py

@@ -542,6 +542,21 @@ def build_scheduled_automation_health_summary(
     if not pixelrag_ocr_vlm_replay or not pixelrag_ocr_vlm_replay_details:
         pixelrag_ocr_vlm_replay = _pixelrag_ocr_vlm_replay_check()
         pixelrag_ocr_vlm_replay_details = pixelrag_ocr_vlm_replay.get("details") or {}
+    pixelrag_vlm_route_readiness = _find_check(source_result, "PixelRAG VLM route readiness")
+    pixelrag_vlm_route_readiness_details = pixelrag_vlm_route_readiness.get("details") or {}
+    if not pixelrag_vlm_route_readiness or not pixelrag_vlm_route_readiness_details:
+        pixelrag_vlm_route_readiness = _pixelrag_vlm_route_readiness_check()
+        pixelrag_vlm_route_readiness_details = pixelrag_vlm_route_readiness.get("details") or {}
+    pixelrag_vlm_replay_worker = _find_check(source_result, "PixelRAG VLM replay worker")
+    pixelrag_vlm_replay_worker_details = pixelrag_vlm_replay_worker.get("details") or {}
+    if not pixelrag_vlm_replay_worker or not pixelrag_vlm_replay_worker_details:
+        pixelrag_vlm_replay_worker = _pixelrag_vlm_replay_worker_check()
+        pixelrag_vlm_replay_worker_details = pixelrag_vlm_replay_worker.get("details") or {}
+    pixelrag_platform_probe = _find_check(source_result, "PixelRAG platform probe readiness")
+    pixelrag_platform_probe_details = pixelrag_platform_probe.get("details") or {}
+    if not pixelrag_platform_probe or not pixelrag_platform_probe_details:
+        pixelrag_platform_probe = _pixelrag_platform_probe_check()
+        pixelrag_platform_probe_details = pixelrag_platform_probe.get("details") or {}
     smoke_status = source_result.get("status") or ("warning" if latest_history else "warning")
     freshness_family = _history_freshness_family(
         latest_history,
@@ -4632,6 +4647,134 @@ def build_scheduled_automation_health_summary(
                 "primary_human_gate_count": 0,
             },
         },
+        {
+            "key": "pixelrag_vlm_route_readiness",
+            "label": "PixelRAG VLM route readiness",
+            "status": pixelrag_vlm_route_readiness.get("status") or "warning",
+            "summary": (
+                pixelrag_vlm_route_readiness.get("summary")
+                or "PixelRAG VLM route readiness has no latest readback."
+            ),
+            "next_machine_action": pixelrag_vlm_route_readiness_details.get("next_machine_action")
+            or "run_pixelrag_vlm_route_readiness_readback",
+            "details": {
+                "policy": pixelrag_vlm_route_readiness_details.get("policy"),
+                "host_count": int(pixelrag_vlm_route_readiness_details.get("host_count") or 0),
+                "reachable_host_count": int(
+                    pixelrag_vlm_route_readiness_details.get("reachable_host_count") or 0
+                ),
+                "configured_model": pixelrag_vlm_route_readiness_details.get("configured_model"),
+                "configured_model_available_count": int(
+                    pixelrag_vlm_route_readiness_details.get("configured_model_available_count") or 0
+                ),
+                "candidate_model": pixelrag_vlm_route_readiness_details.get("candidate_model"),
+                "candidate_host": pixelrag_vlm_route_readiness_details.get("candidate_host"),
+                "candidate_provider": pixelrag_vlm_route_readiness_details.get("candidate_provider"),
+                "tag_model_route_ready": bool(
+                    pixelrag_vlm_route_readiness_details.get("tag_model_route_ready")
+                ),
+                "model_route_ready": bool(
+                    pixelrag_vlm_route_readiness_details.get("model_route_ready")
+                ),
+                "generate_probe_performed": bool(
+                    pixelrag_vlm_route_readiness_details.get("generate_probe_performed")
+                ),
+                "generate_probe_ok_count": int(
+                    pixelrag_vlm_route_readiness_details.get("generate_probe_ok_count") or 0
+                ),
+                "generate_route_ready": (
+                    pixelrag_vlm_route_readiness_details.get("generate_route_ready")
+                    if pixelrag_vlm_route_readiness_details.get("generate_probe_performed")
+                    else None
+                ),
+                "generate_ready_model": pixelrag_vlm_route_readiness_details.get("generate_ready_model"),
+                "generate_ready_host": pixelrag_vlm_route_readiness_details.get("generate_ready_host"),
+                "generate_ready_provider": pixelrag_vlm_route_readiness_details.get("generate_ready_provider"),
+                "model_call_performed": bool(
+                    pixelrag_vlm_route_readiness_details.get("model_call_performed")
+                ),
+                "writes_database": False,
+                "writes_database_count": 0,
+                "primary_human_gate_count": 0,
+            },
+        },
+        {
+            "key": "pixelrag_vlm_replay_worker",
+            "label": "PixelRAG VLM replay worker",
+            "status": pixelrag_vlm_replay_worker.get("status") or "warning",
+            "summary": (
+                pixelrag_vlm_replay_worker.get("summary")
+                or "PixelRAG VLM replay worker has no latest readback."
+            ),
+            "next_machine_action": pixelrag_vlm_replay_worker_details.get("next_machine_action")
+            or "run_pixelrag_vlm_replay_worker_dry_run",
+            "details": {
+                "policy": pixelrag_vlm_replay_worker_details.get("policy"),
+                "receipt_count": int(pixelrag_vlm_replay_worker_details.get("receipt_count") or 0),
+                "ready_count": int(pixelrag_vlm_replay_worker_details.get("ready_count") or 0),
+                "dry_run_count": int(pixelrag_vlm_replay_worker_details.get("dry_run_count") or 0),
+                "executed_count": int(pixelrag_vlm_replay_worker_details.get("executed_count") or 0),
+                "executed_ok_count": int(pixelrag_vlm_replay_worker_details.get("executed_ok_count") or 0),
+                "executed_warning_count": int(
+                    pixelrag_vlm_replay_worker_details.get("executed_warning_count") or 0
+                ),
+                "model_error_count": int(pixelrag_vlm_replay_worker_details.get("model_error_count") or 0),
+                "model_route_not_ready_count": int(
+                    pixelrag_vlm_replay_worker_details.get("model_route_not_ready_count") or 0
+                ),
+                "parse_error_count": int(pixelrag_vlm_replay_worker_details.get("parse_error_count") or 0),
+                "receipt_written_count": int(
+                    pixelrag_vlm_replay_worker_details.get("receipt_written_count") or 0
+                ),
+                "model_call_performed": bool(
+                    pixelrag_vlm_replay_worker_details.get("model_call_performed")
+                ),
+                "artifact_write_performed": bool(
+                    pixelrag_vlm_replay_worker_details.get("artifact_write_performed")
+                ),
+                "writes_database": False,
+                "writes_database_count": 0,
+                "primary_human_gate_count": 0,
+            },
+        },
+        {
+            "key": "pixelrag_platform_probe",
+            "label": "PixelRAG platform probe readiness",
+            "status": pixelrag_platform_probe.get("status") or "warning",
+            "summary": (
+                pixelrag_platform_probe.get("summary")
+                or "PixelRAG platform probe readiness has no latest readback."
+            ),
+            "next_machine_action": pixelrag_platform_probe_details.get("next_machine_action")
+            or "run_pixelrag_platform_probe_readback",
+            "details": {
+                "policy": pixelrag_platform_probe_details.get("policy"),
+                "probe_candidate_count": int(
+                    pixelrag_platform_probe_details.get("probe_candidate_count") or 0
+                ),
+                "ready_for_probe_count": int(
+                    pixelrag_platform_probe_details.get("ready_for_probe_count") or 0
+                ),
+                "shopee_public_context_probe_count": int(
+                    pixelrag_platform_probe_details.get("shopee_public_context_probe_count") or 0
+                ),
+                "language_or_region_interstitial_count": int(
+                    pixelrag_platform_probe_details.get("language_or_region_interstitial_count") or 0
+                ),
+                "traffic_verification_count": int(
+                    pixelrag_platform_probe_details.get("traffic_verification_count") or 0
+                ),
+                "access_denied_count": int(
+                    pixelrag_platform_probe_details.get("access_denied_count") or 0
+                ),
+                "structured_source_fallback_count": int(
+                    pixelrag_platform_probe_details.get("structured_source_fallback_count") or 0
+                ),
+                "writes_database": False,
+                "writes_database_count": 0,
+                "primary_human_gate_count": 0,
+            },
+        },
         freshness_family,
         {
             "key": "daily_summary_delivery",
@@ -13350,6 +13493,187 @@ def _pixelrag_ocr_vlm_replay_check() -> Dict[str, Any]:
         )
 
 
+def _pixelrag_vlm_replay_worker_check() -> Dict[str, Any]:
+    """Dry-run sentinel for the Ollama-first PixelRAG VLM replay worker."""
+    try:
+        from services.pixelrag_vlm_replay_worker_service import (
+            run_pixelrag_ollama_vlm_replay_worker,
+        )
+
+        readback = run_pixelrag_ollama_vlm_replay_worker(execute=False)
+        summary = readback.get("summary") or {}
+        receipt_count = int(summary.get("receipt_count") or 0)
+        ready_count = int(summary.get("ready_count") or 0)
+        dry_run_count = int(summary.get("dry_run_count") or 0)
+        skipped_count = int(summary.get("skipped_count") or 0)
+        executed_count = int(summary.get("executed_count") or 0)
+        status = readback.get("status") or "warning"
+        summary_text = (
+            f"PixelRAG VLM replay worker receipts={receipt_count}, "
+            f"ready={ready_count}, dry_run={dry_run_count}, skipped={skipped_count}, "
+            f"executed={executed_count}"
+        )
+        return _check(
+            "PixelRAG VLM replay worker",
+            status,
+            summary_text,
+            {
+                "policy": readback.get("policy"),
+                "receipt_count": receipt_count,
+                "ready_count": ready_count,
+                "skipped_count": skipped_count,
+                "dry_run_count": dry_run_count,
+                "executed_count": executed_count,
+                "executed_ok_count": int(summary.get("executed_ok_count") or 0),
+                "executed_warning_count": int(summary.get("executed_warning_count") or 0),
+                "model_error_count": int(summary.get("model_error_count") or 0),
+                "model_route_not_ready_count": int(summary.get("model_route_not_ready_count") or 0),
+                "parse_error_count": int(summary.get("parse_error_count") or 0),
+                "receipt_written_count": int(summary.get("receipt_written_count") or 0),
+                "model_call_performed": bool(summary.get("model_call_performed")),
+                "artifact_write_performed": bool(summary.get("artifact_write_performed")),
+                "next_machine_action": readback.get("next_machine_action"),
+                "writes_database": False,
+                "writes_database_count": 0,
+                "primary_human_gate_count": 0,
+            },
+        )
+    except Exception as exc:
+        return _check(
+            "PixelRAG VLM replay worker",
+            "critical",
+            f"PixelRAG VLM replay worker 無法執行 dry-run：{exc}",
+            {
+                "writes_database": False,
+                "writes_database_count": 0,
+                "primary_human_gate_count": 0,
+            },
+        )
+
+
+def _pixelrag_vlm_route_readiness_check() -> Dict[str, Any]:
+    """Read-only sentinel for approved Ollama VLM route/model readiness."""
+    try:
+        from services.pixelrag_vlm_route_readiness_service import (
+            build_pixelrag_vlm_route_readiness,
+        )
+
+        readback = build_pixelrag_vlm_route_readiness(timeout_seconds=3)
+        summary = readback.get("summary") or {}
+        host_count = int(summary.get("host_count") or 0)
+        reachable_host_count = int(summary.get("reachable_host_count") or 0)
+        configured_model = summary.get("configured_model")
+        configured_available = int(summary.get("configured_model_available_count") or 0)
+        candidate_model = summary.get("candidate_model")
+        status = readback.get("status") or "warning"
+        summary_text = (
+            f"PixelRAG VLM route readiness hosts={reachable_host_count}/{host_count}, "
+            f"configured={configured_model}, configured_available={configured_available}, "
+            f"candidate={candidate_model or 'none'}"
+        )
+        return _check(
+            "PixelRAG VLM route readiness",
+            status,
+            summary_text,
+            {
+                "policy": readback.get("policy"),
+                "host_count": host_count,
+                "reachable_host_count": reachable_host_count,
+                "configured_model": configured_model,
+                "configured_model_available_count": configured_available,
+                "candidate_model": candidate_model,
+                "candidate_host": summary.get("candidate_host"),
+                "candidate_provider": summary.get("candidate_provider"),
+                "tag_model_route_ready": bool(summary.get("tag_model_route_ready")),
+                "model_route_ready": bool(summary.get("model_route_ready")),
+                "generate_probe_performed": bool(summary.get("generate_probe_performed")),
+                "generate_probe_ok_count": int(summary.get("generate_probe_ok_count") or 0),
+                "generate_route_ready": (
+                    summary.get("generate_route_ready")
+                    if summary.get("generate_probe_performed")
+                    else None
+                ),
+                "generate_ready_model": summary.get("generate_ready_model"),
+                "generate_ready_host": summary.get("generate_ready_host"),
+                "generate_ready_provider": summary.get("generate_ready_provider"),
+                "model_call_performed": bool(summary.get("model_call_performed")),
+                "next_machine_action": readback.get("next_machine_action"),
+                "writes_database": False,
+                "writes_database_count": 0,
+                "primary_human_gate_count": 0,
+            },
+        )
+    except Exception as exc:
+        return _check(
+            "PixelRAG VLM route readiness",
+            "critical",
+            f"PixelRAG VLM route readiness 無法執行：{exc}",
+            {
+                "writes_database": False,
+                "writes_database_count": 0,
+                "primary_human_gate_count": 0,
+            },
+        )
+
+
+def _pixelrag_platform_probe_check() -> Dict[str, Any]:
+    """Read-only sentinel for PixelRAG platform probe automation readiness."""
+    try:
+        from services.pixelrag_platform_probe_service import (
+            build_pixelrag_platform_probe_readiness,
+        )
+
+        readback = build_pixelrag_platform_probe_readiness()
+        summary = readback.get("summary") or {}
+        candidate_count = int(summary.get("probe_candidate_count") or 0)
+        ready_count = int(summary.get("ready_for_probe_count") or 0)
+        shopee_context_count = int(summary.get("shopee_public_context_probe_count") or 0)
+        structured_count = int(summary.get("structured_source_fallback_count") or 0)
+        status = readback.get("status") or "warning"
+        summary_text = (
+            f"PixelRAG platform probe candidates={candidate_count}, "
+            f"ready={ready_count}, shopee_context={shopee_context_count}, "
+            f"structured_fallback={structured_count}"
+        )
+        return _check(
+            "PixelRAG platform probe readiness",
+            status,
+            summary_text,
+            {
+                "policy": readback.get("policy"),
+                "probe_candidate_count": candidate_count,
+                "ready_for_probe_count": ready_count,
+                "invalid_count": int(summary.get("invalid_count") or 0),
+                "capture_source_count": int(summary.get("capture_source_count") or 0),
+                "vlm_source_count": int(summary.get("vlm_source_count") or 0),
+                "shopee_public_context_probe_count": shopee_context_count,
+                "language_or_region_interstitial_count": int(
+                    summary.get("language_or_region_interstitial_count") or 0
+                ),
+                "traffic_verification_count": int(
+                    summary.get("traffic_verification_count") or 0
+                ),
+                "access_denied_count": int(summary.get("access_denied_count") or 0),
+                "structured_source_fallback_count": structured_count,
+                "next_machine_action": readback.get("next_machine_action"),
+                "writes_database": False,
+                "writes_database_count": 0,
+                "primary_human_gate_count": 0,
+            },
+        )
+    except Exception as exc:
+        return _check(
+            "PixelRAG platform probe readiness",
+            "critical",
+            f"PixelRAG platform probe readiness 無法執行：{exc}",
+            {
+                "writes_database": False,
+                "writes_database_count": 0,
+                "primary_human_gate_count": 0,
+            },
+        )
+
+
 def collect_ai_automation_smoke(*, record_history: bool = True, history_limit: int = 20) -> Dict[str, Any]:
     checks: List[Dict[str, Any]] = [
         _event_router_check(),
@@ -13386,6 +13710,9 @@ def collect_ai_automation_smoke(*, record_history: bool = True, history_limit: i
         _external_mcp_rag_integration_check(),
         _pixelrag_rag_candidate_replay_check(),
         _pixelrag_ocr_vlm_replay_check(),
+        _pixelrag_vlm_route_readiness_check(),
+        _pixelrag_vlm_replay_worker_check(),
+        _pixelrag_platform_probe_check(),
     ]
     worst = max(checks, key=lambda item: STATUS_RANK.get(item["status"], 2))["status"]
     result = {

services/pixelrag_platform_probe_service.py

@@ -0,0 +1,565 @@
+"""Read-only PixelRAG platform probe readiness.
+
+This module turns PixelRAG capture/VLM barrier evidence into concrete machine
+actions for public marketplace probing. It does not read sessions/cookies,
+perform network calls, write DB rows, or promote price truth.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any, Mapping
+from urllib.parse import parse_qs, unquote_plus, urlparse
+
+from services.market_intel.adapters.registry import get_adapter
+from services.pixelrag_crawler_integration_service import (
+    DEFAULT_ARTIFACT_MAX_AGE_HOURS,
+    DEFAULT_ARTIFACT_ROOT,
+    build_pixelrag_marketplace_search_manifest,
+    _parse_iso_datetime,
+)
+from services.pixelrag_rag_candidate_replay_service import (
+    build_pixelrag_rag_candidate_replay_readback,
+)
+
+
+POLICY = "read_only_pixelrag_platform_probe_readiness_v1"
+DEFAULT_LIMIT = 50
+DEFAULT_ACCEPT_LANGUAGE = "zh-TW,zh-Hant;q=0.95,zh;q=0.9,en;q=0.7"
+DEFAULT_VLM_RECEIPT_ROOT = os.getenv(
+    "PIXELRAG_VLM_REPLAY_RECEIPT_ROOT",
+    "/app/data/ai_automation/pixelrag_vlm_replay_receipts"
+    if Path("/app/data").exists()
+    else "runtime_artifacts/pixelrag_vlm_replay_receipts",
+)
+PLATFORM_ADAPTER_ALIASES = {
+    "shopee_tw": "shopee",
+    "coupang_tw": "coupang",
+    "yahoo_shopping_tw": "yahoo",
+}
+
+
+def _normalise_platforms(
+    platform: str | tuple[str, ...] | list[str] | None,
+) -> tuple[str, ...]:
+    if isinstance(platform, str):
+        value = platform.strip().lower()
+        return (value,) if value else ()
+    return tuple(
+        str(item or "").strip().lower()
+        for item in (platform or ())
+        if str(item or "").strip()
+    )
+
+
+def _read_json(path: Path) -> tuple[dict[str, Any], str]:
+    try:
+        parsed = json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, json.JSONDecodeError) as exc:
+        return {}, str(exc)[:300]
+    return parsed if isinstance(parsed, dict) else {}, ""
+
+
+def _safe_int(value: Any, default: int = 0) -> int:
+    try:
+        return int(value or default)
+    except (TypeError, ValueError):
+        return default
+
+
+def _safe_text(value: Any) -> str:
+    if value is None:
+        return ""
+    if isinstance(value, str):
+        return value
+    if isinstance(value, Mapping):
+        return " ".join(_safe_text(item) for item in value.values())
+    if isinstance(value, list):
+        return " ".join(_safe_text(item) for item in value)
+    return str(value)
+
+
+def _extract_keyword(url: str) -> str:
+    parsed = urlparse(str(url or ""))
+    query = parse_qs(parsed.query)
+    for key in ("keyword", "q", "query", "p", "search", "searchKeyword"):
+        values = query.get(key) or []
+        for value in values:
+            text = unquote_plus(str(value or "")).strip()
+            if text:
+                return text
+    return ""
+
+
+def _adapter_code(platform: str) -> str:
+    clean = str(platform or "").strip().lower()
+    return PLATFORM_ADAPTER_ALIASES.get(clean, clean)
+
+
+def _structured_source_plan(platform: str) -> dict[str, Any]:
+    adapter = get_adapter(_adapter_code(platform))
+    if not adapter:
+        return {
+            "available": False,
+            "adapter_code": _adapter_code(platform),
+            "network_request_allowed": False,
+            "database_write_allowed": False,
+            "sources": [],
+        }
+    plan = adapter.build_discovery_plan()
+    return {
+        "available": True,
+        "adapter_code": plan.get("platform_code"),
+        "platform_name": plan.get("platform_name"),
+        "base_url": plan.get("base_url"),
+        "safety_policy": plan.get("safety_policy"),
+        "network_request_allowed": bool(plan.get("network_request_allowed")),
+        "database_write_allowed": bool(plan.get("database_write_allowed")),
+        "sources": list(plan.get("sources") or []),
+    }
+
+
+def _public_browser_context(platform: str) -> dict[str, Any]:
+    return {
+        "context_policy": "public_empty_browser_context_no_login",
+        "locale": "zh-TW",
+        "timezone_id": "Asia/Taipei",
+        "viewport": {"name": "desktop-1440", "width": 1440, "height": 950},
+        "extra_http_headers": {
+            "Accept-Language": DEFAULT_ACCEPT_LANGUAGE,
+        },
+        "credentialed_session_allowed": False,
+        "storage_state_allowed": False,
+        "raw_cookie_or_session_read_allowed": False,
+        "login_allowed": False,
+        "cart_or_checkout_allowed": False,
+        "platform": platform,
+    }
+
+
+def _barrier_type_from_signals(
+    *,
+    platform: str,
+    http_status: int,
+    visual_barrier_reason: str,
+    url: str,
+    title: str,
+    validation: Mapping[str, Any] | None = None,
+    parsed_output: Mapping[str, Any] | None = None,
+) -> str:
+    validation = validation or {}
+    parsed_output = parsed_output or {}
+    haystack = " ".join(
+        [
+            visual_barrier_reason,
+            url,
+            title,
+            _safe_text(parsed_output.get("notes")),
+            _safe_text((parsed_output.get("fields") or {}).get("title")),
+        ]
+    ).lower()
+    if http_status in {401, 403} or "access denied" in haystack:
+        return "access_denied"
+    if "verify/traffic" in haystack or "traffic" in haystack:
+        return "traffic_verification_interstitial"
+    if (
+        validation.get("interstitial_signal_detected")
+        or "language selection" in haystack
+        or "select language" in haystack
+        or "choose language" in haystack
+        or "region selection" in haystack
+        or "select region" in haystack
+        or "language" in haystack
+        or "語言" in haystack
+    ):
+        return "language_or_region_interstitial"
+    if validation.get("generic_marketplace_title_detected") or (
+        platform == "shopee_tw" and "花得更少買得更好" in haystack
+    ):
+        return "generic_marketplace_landing"
+    if visual_barrier_reason:
+        return "platform_visual_barrier"
+    if validation.get("non_product_or_interstitial_detected"):
+        return "non_product_or_interstitial"
+    return "unknown_platform_barrier"
+
+
+def _probe_status(platform: str, barrier_type: str) -> str:
+    if platform == "shopee_tw" and barrier_type in {
+        "language_or_region_interstitial",
+        "generic_marketplace_landing",
+        "traffic_verification_interstitial",
+    }:
+        return "ready_for_public_context_probe"
+    if barrier_type in {"access_denied", "traffic_verification_interstitial"}:
+        return "structured_source_or_backoff_required"
+    return "ready_for_platform_probe"
+
+
+def _next_machine_action(platform: str, barrier_type: str, status: str) -> str:
+    if platform == "shopee_tw" and status == "ready_for_public_context_probe":
+        return "run_shopee_public_context_probe_then_structured_source_fallback"
+    if status == "structured_source_or_backoff_required":
+        return "use_structured_source_or_platform_backoff_policy"
+    return "run_public_platform_context_probe"
+
+
+def _recommended_actions(platform: str, barrier_type: str, status: str) -> list[dict[str, Any]]:
+    actions: list[dict[str, Any]] = []
+    if status == "ready_for_public_context_probe":
+        actions.append({
+            "order": 1,
+            "action": "rerun_visual_capture_with_public_browser_context",
+            "machine_runnable": True,
+            "context_keys": ["locale", "timezone_id", "extra_http_headers", "viewport"],
+        })
+    actions.append({
+        "order": len(actions) + 1,
+        "action": "read_structured_market_intel_adapter_sources",
+        "machine_runnable": True,
+        "adapter_code": _adapter_code(platform),
+    })
+    if barrier_type in {"access_denied", "traffic_verification_interstitial"}:
+        actions.append({
+            "order": len(actions) + 1,
+            "action": "apply_platform_backoff_and_do_not_treat_barrier_as_product_data",
+            "machine_runnable": True,
+        })
+    actions.append({
+        "order": len(actions) + 1,
+        "action": "keep_visual_fields_out_of_formal_price_tables",
+        "machine_runnable": True,
+    })
+    return actions
+
+
+def _manifest_preview(platform: str, url: str) -> dict[str, Any] | None:
+    keyword = _extract_keyword(url)
+    if not keyword:
+        return None
+    manifest = build_pixelrag_marketplace_search_manifest(
+        platform=platform,
+        keyword=keyword,
+        crawler="PixelRAGPlatformProbe.public_context_visual_fallback",
+        trigger_reason="platform_interstitial_or_blocked_page_probe",
+        evidence_intent="collect_public_marketplace_offer_cards_after_platform_probe",
+    )
+    if manifest.get("success"):
+        manifest["public_browser_context"] = _public_browser_context(platform)
+    return manifest
+
+
+def _vlm_receipt_candidates(
+    root: Path,
+    *,
+    platforms: tuple[str, ...],
+    limit: int,
+) -> list[Path]:
+    if not root.exists():
+        return []
+    candidates: list[Path] = []
+    if platforms:
+        for platform in platforms:
+            candidates.extend((root / platform).glob("*/vlm_replay_receipt.json"))
+    else:
+        candidates.extend(root.glob("*/*/vlm_replay_receipt.json"))
+    return sorted(candidates, key=lambda path: path.stat().st_mtime, reverse=True)[:limit]
+
+
+def _candidate_from_capture(candidate: Mapping[str, Any]) -> dict[str, Any] | None:
+    next_action = str(candidate.get("next_machine_action") or "")
+    visual_barrier_reason = str(candidate.get("visual_barrier_reason") or "")
+    http_status = _safe_int(candidate.get("http_status"))
+    if (
+        "platform_probe" not in next_action
+        and not visual_barrier_reason
+        and http_status < 400
+    ):
+        return None
+
+    platform = str(candidate.get("platform") or "unknown").strip().lower()
+    manifest_id = str(candidate.get("manifest_id") or "").strip()
+    url = str(candidate.get("url") or "").strip()
+    title = str(candidate.get("title") or "").strip()
+    barrier_type = _barrier_type_from_signals(
+        platform=platform,
+        http_status=http_status,
+        visual_barrier_reason=visual_barrier_reason,
+        url=url,
+        title=title,
+    )
+    status = _probe_status(platform, barrier_type)
+    return {
+        "platform": platform,
+        "manifest_id": manifest_id,
+        "source_type": "capture_receipt",
+        "source_receipt_path": candidate.get("receipt_path"),
+        "generated_at": candidate.get("generated_at"),
+        "age_hours": candidate.get("age_hours"),
+        "url": url,
+        "title": title,
+        "http_status": http_status,
+        "barrier_type": barrier_type,
+        "visual_barrier_reason": visual_barrier_reason,
+        "probe_status": status,
+        "probe_ready": True,
+        "next_machine_action": _next_machine_action(platform, barrier_type, status),
+        "public_browser_context": _public_browser_context(platform),
+        "capture_manifest_preview": _manifest_preview(platform, url),
+        "structured_source_fallback": _structured_source_plan(platform),
+        "recommended_probe_actions": _recommended_actions(platform, barrier_type, status),
+        "source_next_machine_action": candidate.get("next_machine_action"),
+        "writes_database": False,
+        "primary_human_gate_count": 0,
+    }
+
+
+def _candidate_from_vlm(path: Path, *, now: datetime, max_age_hours: int) -> dict[str, Any] | None:
+    receipt, error = _read_json(path)
+    if error:
+        return {
+            "platform": path.parent.parent.name,
+            "manifest_id": path.parent.name,
+            "source_type": "vlm_replay_receipt",
+            "source_receipt_path": str(path),
+            "probe_status": "invalid_vlm_receipt",
+            "probe_ready": False,
+            "barrier_type": "invalid_receipt",
+            "errors": [error],
+            "next_machine_action": "fix_invalid_pixelrag_vlm_receipt",
+            "writes_database": False,
+            "primary_human_gate_count": 0,
+        }
+
+    validation = receipt.get("validation") if isinstance(receipt.get("validation"), Mapping) else {}
+    parsed_output = receipt.get("parsed_output") if isinstance(receipt.get("parsed_output"), Mapping) else {}
+    next_action = str(receipt.get("next_machine_action") or "")
+    if (
+        "platform_probe" not in next_action
+        and not validation.get("non_product_or_interstitial_detected")
+        and not validation.get("blocked_page_detected")
+    ):
+        return None
+
+    platform = str(receipt.get("platform") or path.parent.parent.name).strip().lower()
+    manifest_id = str(receipt.get("manifest_id") or path.parent.name).strip()
+    generated = _parse_iso_datetime(receipt.get("generated_at"))
+    age_hours = ((now - generated).total_seconds() / 3600) if generated else None
+    source_url = ""
+    source_title = ""
+    source_receipt = str(receipt.get("source_receipt_path") or "")
+    if source_receipt:
+        capture, _ = _read_json(Path(source_receipt))
+        capture_target = capture.get("capture_target") or {}
+        page_metrics = capture.get("page_metrics") or {}
+        source_url = str(capture_target.get("url") or page_metrics.get("final_url") or "")
+        source_title = str(page_metrics.get("title") or "")
+    barrier_type = _barrier_type_from_signals(
+        platform=platform,
+        http_status=0,
+        visual_barrier_reason="",
+        url=source_url,
+        title=source_title,
+        validation=validation,
+        parsed_output=parsed_output,
+    )
+    status = _probe_status(platform, barrier_type)
+    return {
+        "platform": platform,
+        "manifest_id": manifest_id,
+        "source_type": "vlm_replay_receipt",
+        "source_receipt_path": str(path),
+        "source_capture_receipt_path": source_receipt,
+        "generated_at": receipt.get("generated_at"),
+        "age_hours": round(age_hours, 3) if age_hours is not None else None,
+        "stale": age_hours is None or age_hours > max_age_hours,
+        "url": source_url,
+        "title": source_title,
+        "http_status": 0,
+        "barrier_type": barrier_type,
+        "probe_status": status,
+        "probe_ready": True,
+        "validation": {
+            "blocked_page_detected": bool(validation.get("blocked_page_detected")),
+            "non_product_or_interstitial_detected": bool(
+                validation.get("non_product_or_interstitial_detected")
+            ),
+            "interstitial_signal_detected": bool(
+                validation.get("interstitial_signal_detected")
+            ),
+            "generic_marketplace_title_detected": bool(
+                validation.get("generic_marketplace_title_detected")
+            ),
+            "present_field_count": _safe_int(validation.get("present_field_count")),
+            "missing_required_fields": list(validation.get("missing_required_fields") or []),
+        },
+        "next_machine_action": _next_machine_action(platform, barrier_type, status),
+        "public_browser_context": _public_browser_context(platform),
+        "capture_manifest_preview": _manifest_preview(platform, source_url),
+        "structured_source_fallback": _structured_source_plan(platform),
+        "recommended_probe_actions": _recommended_actions(platform, barrier_type, status),
+        "source_next_machine_action": receipt.get("next_machine_action"),
+        "writes_database": False,
+        "primary_human_gate_count": 0,
+    }
+
+
+def _dedupe_items(items: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    by_key: dict[tuple[str, str], dict[str, Any]] = {}
+    source_rank = {"vlm_replay_receipt": 2, "capture_receipt": 1}
+    for item in items:
+        key = (str(item.get("platform") or ""), str(item.get("manifest_id") or ""))
+        current = by_key.get(key)
+        if not current or source_rank.get(str(item.get("source_type")), 0) >= source_rank.get(
+            str(current.get("source_type")), 0
+        ):
+            by_key[key] = item
+    return list(by_key.values())
+
+
+def build_pixelrag_platform_probe_readiness(
+    *,
+    artifact_root: str | Path | None = None,
+    vlm_receipt_root: str | Path | None = None,
+    platform: str | tuple[str, ...] | list[str] | None = None,
+    max_age_hours: int | None = None,
+    limit: int | None = None,
+) -> dict[str, Any]:
+    """Build a no-write platform probe plan from PixelRAG barrier receipts."""
+    capture_root = Path(artifact_root or DEFAULT_ARTIFACT_ROOT)
+    vlm_root = Path(vlm_receipt_root or DEFAULT_VLM_RECEIPT_ROOT)
+    platforms = _normalise_platforms(platform)
+    max_age = max(1, int(max_age_hours or DEFAULT_ARTIFACT_MAX_AGE_HOURS))
+    item_limit = max(1, min(int(limit or DEFAULT_LIMIT), 250))
+    now = datetime.now(timezone.utc)
+
+    capture_readback = build_pixelrag_rag_candidate_replay_readback(
+        artifact_root=capture_root,
+        platform=platforms,
+        max_age_hours=max_age,
+        limit=item_limit,
+    )
+    capture_items = [
+        item
+        for item in (
+            _candidate_from_capture(candidate)
+            for candidate in list(capture_readback.get("candidates") or [])
+        )
+        if item
+    ]
+    vlm_items = [
+        item
+        for item in (
+            _candidate_from_vlm(path, now=now, max_age_hours=max_age)
+            for path in _vlm_receipt_candidates(vlm_root, platforms=platforms, limit=item_limit)
+        )
+        if item
+    ]
+    probe_items = _dedupe_items(capture_items + vlm_items)
+    invalid_count = sum(1 for item in probe_items if not item.get("probe_ready"))
+    ready_count = sum(1 for item in probe_items if item.get("probe_ready"))
+    shopee_public_context_count = sum(
+        1
+        for item in probe_items
+        if item.get("platform") == "shopee_tw"
+        and item.get("probe_status") == "ready_for_public_context_probe"
+    )
+    access_denied_count = sum(1 for item in probe_items if item.get("barrier_type") == "access_denied")
+    traffic_count = sum(
+        1 for item in probe_items if item.get("barrier_type") == "traffic_verification_interstitial"
+    )
+    language_count = sum(
+        1 for item in probe_items if item.get("barrier_type") == "language_or_region_interstitial"
+    )
+    structured_fallback_count = sum(
+        1
+        for item in probe_items
+        if (item.get("structured_source_fallback") or {}).get("available")
+    )
+
+    if invalid_count and invalid_count == len(probe_items):
+        status = "critical"
+    elif probe_items and ready_count:
+        status = "ok"
+    else:
+        status = "warning"
+
+    next_action = (
+        "fix_invalid_pixelrag_platform_probe_receipts"
+        if invalid_count and not ready_count
+        else (
+            "run_platform_probe_or_structured_source_fallback"
+            if ready_count
+            else "run_pixelrag_visual_capture_worker"
+        )
+    )
+
+    return {
+        "success": status != "critical",
+        "policy": POLICY,
+        "status": status,
+        "generated_at": now.isoformat(),
+        "artifact_root": str(capture_root),
+        "vlm_receipt_root": str(vlm_root),
+        "platform_filter": list(platforms),
+        "max_age_hours": max_age,
+        "limit": item_limit,
+        "summary": {
+            "probe_candidate_count": len(probe_items),
+            "ready_for_probe_count": ready_count,
+            "invalid_count": invalid_count,
+            "capture_source_count": len(capture_items),
+            "vlm_source_count": len(vlm_items),
+            "shopee_public_context_probe_count": shopee_public_context_count,
+            "language_or_region_interstitial_count": language_count,
+            "traffic_verification_count": traffic_count,
+            "access_denied_count": access_denied_count,
+            "structured_source_fallback_count": structured_fallback_count,
+            "writes_database_count": 0,
+            "primary_human_gate_count": 0,
+            "platforms": sorted({str(item.get("platform") or "unknown") for item in probe_items}),
+        },
+        "probe_items": probe_items,
+        "source_capture_replay": {
+            "policy": capture_readback.get("policy"),
+            "status": capture_readback.get("status"),
+            "summary": capture_readback.get("summary"),
+            "next_machine_action": capture_readback.get("next_machine_action"),
+        },
+        "probe_contract": {
+            "automation_mode": "platform_probe_plan_no_write",
+            "network_call": False,
+            "db_write": False,
+            "writes_database": False,
+            "writes_ai_insights": False,
+            "writes_price_tables": False,
+            "secret_read": False,
+            "raw_cookie_or_session_read": False,
+            "credentialed_session_allowed": False,
+            "login_allowed": False,
+            "blocked_pages_are_not_product_data": True,
+            "visual_fields_are_candidate_evidence_only": True,
+            "primary_human_gate_count": 0,
+        },
+        "controlled_apply": {
+            "network_call": False,
+            "db_write": False,
+            "writes_database": False,
+            "writes_database_count": 0,
+            "secret_read": False,
+            "raw_cookie_or_session_read": False,
+            "production_price_write": False,
+            "artifact_write": False,
+            "primary_human_gate_count": 0,
+        },
+        "next_machine_action": next_action,
+    }
+
+
+__all__ = [
+    "POLICY",
+    "build_pixelrag_platform_probe_readiness",
+]

services/pixelrag_vlm_replay_worker_service.py

@@ -0,0 +1,843 @@
+"""Ollama-first PixelRAG VLM replay worker.
+
+This worker executes the next machine action emitted by the PixelRAG
+OCR/VLM replay contract. It reads saved screenshot tiles, calls approved
+Ollama hosts, validates evidence-bound JSON fields, and optionally writes an
+artifact receipt. It never writes DB rows, AI insights, or price truth.
+"""
+
+from __future__ import annotations
+
+import base64
+import json
+import os
+import re
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any, Mapping
+
+import requests
+
+from services.ollama_service import (
+    OllamaResponse,
+    OllamaService,
+    get_host_label,
+    get_provider_tag,
+    is_approved_ollama_host,
+)
+from services.pixelrag_crawler_integration_service import (
+    DEFAULT_ARTIFACT_MAX_AGE_HOURS,
+    DEFAULT_ARTIFACT_ROOT,
+)
+from services.pixelrag_ocr_vlm_replay_service import (
+    DEFAULT_CONFIDENCE_THRESHOLD,
+    build_pixelrag_ocr_vlm_replay_contract,
+)
+from services.pixelrag_vlm_route_readiness_service import (
+    build_pixelrag_vlm_route_readiness,
+)
+
+
+POLICY = "controlled_pixelrag_ollama_vlm_replay_worker_v1"
+DEFAULT_LIMIT = 25
+DEFAULT_TILE_LIMIT = 4
+DEFAULT_TIMEOUT_SECONDS = 90
+DEFAULT_ROUTE_GENERATE_PROBE_TIMEOUT_SECONDS = 20
+DEFAULT_OUTPUT_ROOT = os.getenv(
+    "PIXELRAG_VLM_REPLAY_RECEIPT_ROOT",
+    "/app/data/ai_automation/pixelrag_vlm_replay_receipts"
+    if Path("/app/data").exists()
+    else "runtime_artifacts/pixelrag_vlm_replay_receipts",
+)
+DEFAULT_MODEL = (
+    os.getenv("PIXELRAG_VLM_MODEL")
+    or os.getenv("PPT_VISION_MODEL")
+    or "minicpm-v:latest"
+)
+RAW_EXCERPT_LIMIT = 500
+INTERSTITIAL_SIGNAL_TOKENS = (
+    "language selection",
+    "select language",
+    "choose language",
+    "region selection",
+    "select region",
+    "app-download",
+    "app download",
+    "landing page",
+    "loading page",
+    "logo-only",
+    "cookie consent",
+    "選擇語言",
+    "選擇地區",
+    "語言",
+)
+GENERIC_MARKETPLACE_TITLE_TOKENS = (
+    "蝦皮購物 | 花得更少買得更好",
+)
+
+
+def _normalise_platforms(
+    platform: str | tuple[str, ...] | list[str] | None,
+) -> tuple[str, ...]:
+    if isinstance(platform, str):
+        value = platform.strip().lower()
+        return (value,) if value else ()
+    return tuple(
+        str(item or "").strip().lower()
+        for item in (platform or ())
+        if str(item or "").strip()
+    )
+
+
+def _safe_segment(value: Any) -> str:
+    text = str(value or "unknown").strip().lower()
+    text = re.sub(r"[^a-z0-9._-]+", "-", text)
+    return text.strip("-") or "unknown"
+
+
+def _resolve_tile_path(path: str, root: Path) -> Path:
+    tile_path = Path(str(path or "").strip())
+    if tile_path.is_absolute():
+        return tile_path
+    return root / tile_path
+
+
+def _tile_images(item: Mapping[str, Any], *, root: Path, tile_limit: int) -> tuple[list[str], list[dict[str, Any]]]:
+    images: list[str] = []
+    evidence: list[dict[str, Any]] = []
+    for tile in list(item.get("input_tiles") or [])[:tile_limit]:
+        evidence_ref = str(tile.get("evidence_ref") or "")
+        path = _resolve_tile_path(str(tile.get("path") or ""), root)
+        tile_evidence = {
+            "evidence_ref": evidence_ref,
+            "path": str(path),
+            "exists": path.exists(),
+            "loaded": False,
+        }
+        if path.exists():
+            raw = path.read_bytes()
+            images.append(base64.b64encode(raw).decode("ascii"))
+            tile_evidence["loaded"] = True
+            tile_evidence["byte_size"] = len(raw)
+        evidence.append(tile_evidence)
+    return images, evidence
+
+
+def _extract_json_object(content: str) -> dict[str, Any]:
+    text = str(content or "").strip()
+    if not text:
+        raise ValueError("empty_model_output")
+    if text.startswith("```"):
+        text = re.sub(r"^```(?:json)?\s*", "", text, flags=re.IGNORECASE)
+        text = re.sub(r"\s*```$", "", text)
+    try:
+        parsed = json.loads(text)
+    except json.JSONDecodeError:
+        start = text.find("{")
+        end = text.rfind("}")
+        if start < 0 or end <= start:
+            raise
+        parsed = json.loads(text[start:end + 1])
+    if not isinstance(parsed, dict):
+        raise ValueError("model_output_not_json_object")
+    return parsed
+
+
+def _prompt_for_item(item: Mapping[str, Any]) -> str:
+    field_contract = list(item.get("field_contract") or [])
+    compact_contract = [
+        {
+            "field": field.get("field"),
+            "type": field.get("type"),
+            "required": bool(field.get("required")),
+            "min_confidence": field.get("min_confidence"),
+            "evidence_requirement": field.get("evidence_requirement"),
+        }
+        for field in field_contract
+    ]
+    metadata = {
+        "platform": item.get("platform"),
+        "manifest_id": item.get("manifest_id"),
+        "url": item.get("url"),
+        "title_hint": item.get("title_hint"),
+        "http_status": item.get("http_status"),
+        "field_contract": compact_contract,
+        "input_evidence_refs": [
+            tile.get("evidence_ref") for tile in list(item.get("input_tiles") or [])
+        ],
+    }
+    return (
+        "You are a strict public marketplace offer-card VLM extractor.\n"
+        "Return only valid JSON. Do not use markdown. Do not guess.\n"
+        "Use only visible tile evidence and cite evidence_refs like tile:1.\n"
+        "If the tile is access denied, captcha, login, traffic verification, or not a product/search card, "
+        "set blocked_page_detected=true and leave product fields empty.\n"
+        "Language selection, region selection, app-download, landing, loading, logo-only, or cookie consent "
+        "pages are not product/search cards; set blocked_page_detected=true for them.\n"
+        "Required JSON schema:\n"
+        "{\n"
+        '  "blocked_page_detected": false,\n'
+        '  "fields": {"field_name": {"value": null, "confidence": 0.0, "evidence_refs": []}},\n'
+        '  "quality": {"overall_confidence": 0.0, "missing_required_fields": [], '
+        '"requires_identity_matcher_replay": true, "requires_promotion_gate": true},\n'
+        '  "notes": []\n'
+        "}\n"
+        "Metadata and field contract:\n"
+        f"{json.dumps(metadata, ensure_ascii=False, sort_keys=True)}"
+    )
+
+
+def _field_value_present(value: Any) -> bool:
+    if value is None:
+        return False
+    if isinstance(value, str):
+        return bool(value.strip())
+    return True
+
+
+def _stringify_signal(value: Any) -> str:
+    if value is None:
+        return ""
+    if isinstance(value, str):
+        return value
+    if isinstance(value, Mapping):
+        return " ".join(_stringify_signal(item) for item in value.values())
+    if isinstance(value, list):
+        return " ".join(_stringify_signal(item) for item in value)
+    return str(value)
+
+
+def _has_interstitial_signal(*values: Any) -> bool:
+    haystack = " ".join(_stringify_signal(value) for value in values).lower()
+    return any(token.lower() in haystack for token in INTERSTITIAL_SIGNAL_TOKENS)
+
+
+def _validate_model_payload(
+    parsed: Mapping[str, Any],
+    item: Mapping[str, Any],
+) -> dict[str, Any]:
+    fields = parsed.get("fields") if isinstance(parsed.get("fields"), Mapping) else {}
+    quality = parsed.get("quality") if isinstance(parsed.get("quality"), Mapping) else {}
+    missing_required: list[str] = []
+    field_evidence_missing: list[str] = []
+    low_confidence_fields: list[str] = []
+    present_field_count = 0
+    blocked_detected = bool(parsed.get("blocked_page_detected"))
+    title_value = None
+
+    for contract in list(item.get("field_contract") or []):
+        field_name = str(contract.get("field") or "")
+        field_payload = fields.get(field_name) if isinstance(fields, Mapping) else {}
+        if not isinstance(field_payload, Mapping):
+            field_payload = {}
+        value = field_payload.get("value")
+        if field_name == "title":
+            title_value = value
+        evidence_refs = list(field_payload.get("evidence_refs") or [])
+        try:
+            confidence = float(field_payload.get("confidence") or 0)
+        except (TypeError, ValueError):
+            confidence = 0.0
+        min_confidence = float(contract.get("min_confidence") or DEFAULT_CONFIDENCE_THRESHOLD)
+        present = _field_value_present(value)
+        if present:
+            present_field_count += 1
+        if present and not evidence_refs:
+            field_evidence_missing.append(field_name)
+        if present and confidence < min_confidence:
+            low_confidence_fields.append(field_name)
+        if contract.get("required") and (blocked_detected or not present or confidence < min_confidence):
+            missing_required.append(field_name)
+
+    declared_missing = [
+        str(field)
+        for field in list(quality.get("missing_required_fields") or [])
+        if str(field).strip()
+    ]
+    for field in declared_missing:
+        if field not in missing_required:
+            missing_required.append(field)
+    notes_payload = parsed.get("notes")
+    generic_marketplace_title_detected = (
+        isinstance(title_value, str)
+        and any(token in title_value for token in GENERIC_MARKETPLACE_TITLE_TOKENS)
+    )
+    interstitial_signal_detected = _has_interstitial_signal(
+        notes_payload,
+        title_value,
+        item.get("title_hint"),
+    )
+    non_product_or_interstitial_detected = (
+        not blocked_detected
+        and (
+            present_field_count == 0
+            or interstitial_signal_detected
+            or generic_marketplace_title_detected
+        )
+        and bool(missing_required)
+    )
+
+    return {
+        "blocked_page_detected": blocked_detected,
+        "non_product_or_interstitial_detected": non_product_or_interstitial_detected,
+        "interstitial_signal_detected": interstitial_signal_detected,
+        "generic_marketplace_title_detected": generic_marketplace_title_detected,
+        "present_field_count": present_field_count,
+        "missing_required_fields": missing_required,
+        "field_evidence_missing": field_evidence_missing,
+        "low_confidence_fields": low_confidence_fields,
+        "valid_for_identity_matcher_replay": (
+            not blocked_detected
+            and not non_product_or_interstitial_detected
+            and not missing_required
+            and not field_evidence_missing
+        ),
+        "requires_identity_matcher_replay": bool(
+            quality.get("requires_identity_matcher_replay", True)
+        ),
+        "requires_promotion_gate": bool(quality.get("requires_promotion_gate", True)),
+    }
+
+
+def _generate_exact_host(
+    prompt: str,
+    *,
+    host: str,
+    model: str,
+    temperature: float,
+    timeout: int,
+    options: Mapping[str, Any] | None,
+    images: list[str],
+) -> OllamaResponse:
+    """Call the route-readiness selected host without fallback or model downgrade."""
+    clean_host = str(host or "").rstrip("/")
+    if not is_approved_ollama_host(clean_host):
+        return OllamaResponse(
+            success=False,
+            content="",
+            model=model,
+            error=f"unapproved_pixelrag_vlm_candidate_host: {clean_host}",
+            host=clean_host or "unknown",
+        )
+    payload: dict[str, Any] = {
+        "model": model,
+        "prompt": prompt,
+        "stream": False,
+        "options": {"temperature": temperature},
+    }
+    if options:
+        payload["options"].update(dict(options))
+    if images:
+        payload["images"] = images
+    try:
+        response = requests.post(
+            f"{clean_host}/api/generate",
+            json=payload,
+            timeout=max(1, int(timeout or DEFAULT_TIMEOUT_SECONDS)),
+        )
+        if response.status_code != 200:
+            return OllamaResponse(
+                success=False,
+                content="",
+                model=model,
+                error=f"HTTP {response.status_code}: {response.text[:RAW_EXCERPT_LIMIT]}",
+                host=clean_host,
+            )
+        data = response.json()
+        return OllamaResponse(
+            success=True,
+            content=data.get("response", ""),
+            model=model,
+            total_duration=(data.get("total_duration", 0) or 0) / 1e9,
+            host=clean_host,
+            input_tokens=int(data.get("prompt_eval_count", 0) or 0),
+            output_tokens=int(data.get("eval_count", 0) or 0),
+        )
+    except requests.Timeout:
+        return OllamaResponse(
+            success=False,
+            content="",
+            model=model,
+            error=f"timeout ({max(1, int(timeout or DEFAULT_TIMEOUT_SECONDS))}s)",
+            host=clean_host,
+        )
+    except Exception as exc:
+        return OllamaResponse(
+            success=False,
+            content="",
+            model=model,
+            error=f"{type(exc).__name__}: {str(exc)[:RAW_EXCERPT_LIMIT]}",
+            host=clean_host,
+        )
+
+
+def _write_replay_receipt(
+    *,
+    output_root: Path,
+    item: Mapping[str, Any],
+    worker_item: Mapping[str, Any],
+) -> str:
+    target = (
+        output_root
+        / _safe_segment(item.get("platform"))
+        / _safe_segment(item.get("manifest_id"))
+        / "vlm_replay_receipt.json"
+    )
+    target.parent.mkdir(parents=True, exist_ok=True)
+    receipt_payload = dict(worker_item)
+    receipt_payload["artifact_write_performed"] = True
+    receipt_payload["receipt_path"] = str(target)
+    target.write_text(
+        json.dumps(receipt_payload, ensure_ascii=False, indent=2, sort_keys=True),
+        encoding="utf-8",
+    )
+    return str(target)
+
+
+def _skipped_item(item: Mapping[str, Any]) -> dict[str, Any]:
+    return {
+        "platform": item.get("platform"),
+        "manifest_id": item.get("manifest_id"),
+        "source_receipt_path": item.get("source_receipt_path"),
+        "worker_status": "skipped_blocked_or_not_ready",
+        "replay_status": item.get("replay_status"),
+        "blocked_reasons": list(item.get("blocked_reasons") or []),
+        "model_call_performed": False,
+        "artifact_write_performed": False,
+        "writes_database": False,
+        "next_machine_action": item.get("next_machine_action")
+        or "run_platform_probe_or_use_structured_api",
+    }
+
+
+def _dry_run_item(item: Mapping[str, Any]) -> dict[str, Any]:
+    return {
+        "platform": item.get("platform"),
+        "manifest_id": item.get("manifest_id"),
+        "source_receipt_path": item.get("source_receipt_path"),
+        "worker_status": "dry_run_ready",
+        "ready_for_execution": True,
+        "tile_input_count": len(list(item.get("input_tiles") or [])),
+        "field_contract_count": int(item.get("field_contract_count") or 0),
+        "model_call_performed": False,
+        "artifact_write_performed": False,
+        "writes_database": False,
+        "next_machine_action": "run_pixelrag_vlm_replay_worker_execute",
+    }
+
+
+def _model_route_not_ready_item(
+    item: Mapping[str, Any],
+    *,
+    output_root: Path,
+    route_readiness: Mapping[str, Any],
+    write_receipt: bool,
+) -> dict[str, Any]:
+    summary = route_readiness.get("summary") or {}
+    worker_item = {
+        "platform": item.get("platform"),
+        "manifest_id": item.get("manifest_id"),
+        "source_receipt_path": item.get("source_receipt_path"),
+        "worker_status": "model_route_not_ready",
+        "model": summary.get("configured_model"),
+        "candidate_model": summary.get("candidate_model"),
+        "candidate_host": summary.get("candidate_host"),
+        "tag_model_route_ready": bool(summary.get("tag_model_route_ready")),
+        "generate_probe_performed": bool(summary.get("generate_probe_performed")),
+        "generate_probe_ok_count": int(summary.get("generate_probe_ok_count") or 0),
+        "generate_route_ready": bool(summary.get("generate_route_ready")),
+        "generate_ready_model": summary.get("generate_ready_model"),
+        "generate_ready_host": summary.get("generate_ready_host"),
+        "generate_ready_provider": summary.get("generate_ready_provider"),
+        "model_call_performed": False,
+        "artifact_write_performed": False,
+        "writes_database": False,
+        "route_readiness_status": route_readiness.get("status"),
+        "next_machine_action": route_readiness.get("next_machine_action")
+        or "install_or_configure_pixelrag_vlm_model_on_approved_ollama_host",
+    }
+    if write_receipt:
+        worker_item["receipt_path"] = _write_replay_receipt(
+            output_root=output_root,
+            item=item,
+            worker_item=worker_item,
+        )
+        worker_item["artifact_write_performed"] = True
+    return worker_item
+
+
+def _execute_item(
+    item: Mapping[str, Any],
+    *,
+    root: Path,
+    output_root: Path,
+    model: str,
+    route_host: str | None,
+    timeout: int,
+    tile_limit: int,
+    write_receipt: bool,
+) -> dict[str, Any]:
+    images, tile_evidence = _tile_images(item, root=root, tile_limit=tile_limit)
+    base: dict[str, Any] = {
+        "platform": item.get("platform"),
+        "manifest_id": item.get("manifest_id"),
+        "source_receipt_path": item.get("source_receipt_path"),
+        "worker_status": "executing",
+        "model": model,
+        "route_candidate_host": str(route_host or ""),
+        "tile_evidence": tile_evidence,
+        "tile_image_count": len(images),
+        "model_call_performed": bool(images),
+        "artifact_write_performed": False,
+        "writes_database": False,
+    }
+    if not images:
+        base.update({
+            "worker_status": "skipped_no_loadable_tiles",
+            "next_machine_action": "refresh_pixelrag_visual_capture_receipt",
+        })
+        return base
+
+    prompt = _prompt_for_item(item)
+    options = {"num_predict": 700, "num_ctx": 4096}
+    if route_host:
+        response = _generate_exact_host(
+            prompt,
+            host=route_host,
+            model=model,
+            temperature=0.1,
+            timeout=max(10, int(timeout or DEFAULT_TIMEOUT_SECONDS)),
+            options=options,
+            images=images,
+        )
+    else:
+        response = OllamaService(model=model).generate(
+            prompt,
+            model=model,
+            temperature=0.1,
+            timeout=max(10, int(timeout or DEFAULT_TIMEOUT_SECONDS)),
+            options=options,
+            images=images,
+        )
+    base.update({
+        "host": response.host,
+        "host_label": get_host_label(response.host or ""),
+        "provider": get_provider_tag(response.host or ""),
+        "actual_model": response.model,
+        "input_tokens": int(response.input_tokens or 0),
+        "output_tokens": int(response.output_tokens or 0),
+        "total_duration": response.total_duration,
+    })
+    if not response.success:
+        base.update({
+            "worker_status": "model_error",
+            "model_error": str(response.error or "")[:RAW_EXCERPT_LIMIT],
+            "next_machine_action": (
+                "repair_ollama_vlm_generate_runtime_or_proxy_timeout"
+                if route_host
+                else "repair_ollama_vlm_runtime_or_model_route"
+            ),
+        })
+        if write_receipt:
+            base["receipt_path"] = _write_replay_receipt(
+                output_root=output_root,
+                item=item,
+                worker_item=base,
+            )
+            base["artifact_write_performed"] = True
+        return base
+
+    try:
+        parsed = _extract_json_object(response.content)
+    except Exception as exc:
+        base.update({
+            "worker_status": "model_output_parse_error",
+            "parse_error": str(exc)[:RAW_EXCERPT_LIMIT],
+            "raw_model_output_excerpt": str(response.content or "")[:RAW_EXCERPT_LIMIT],
+            "next_machine_action": "tighten_pixelrag_vlm_prompt_or_model",
+        })
+        if write_receipt:
+            base["receipt_path"] = _write_replay_receipt(
+                output_root=output_root,
+                item=item,
+                worker_item=base,
+            )
+            base["artifact_write_performed"] = True
+        return base
+
+    validation = _validate_model_payload(parsed, item)
+    missing_required = list(validation.get("missing_required_fields") or [])
+    evidence_missing = list(validation.get("field_evidence_missing") or [])
+    blocked_detected = bool(validation.get("blocked_page_detected"))
+    non_product_or_interstitial = bool(
+        validation.get("non_product_or_interstitial_detected")
+    )
+    status = "executed_ok"
+    next_action = "run_identity_matcher_replay_then_promotion_gate"
+    if blocked_detected or non_product_or_interstitial:
+        status = "executed_warning"
+        next_action = "run_platform_probe_or_use_structured_api"
+    elif missing_required or evidence_missing:
+        status = "executed_warning"
+        next_action = "rerun_vlm_replay_with_more_tiles_or_ocr"
+
+    base.update({
+        "worker_status": status,
+        "parsed_output": parsed,
+        "validation": validation,
+        "required_field_missing_count": len(missing_required),
+        "field_evidence_missing_count": len(evidence_missing),
+        "next_machine_action": next_action,
+    })
+    if write_receipt:
+        base["receipt_path"] = _write_replay_receipt(
+            output_root=output_root,
+            item=item,
+            worker_item=base,
+        )
+        base["artifact_write_performed"] = True
+    return base
+
+
+def run_pixelrag_ollama_vlm_replay_worker(
+    *,
+    artifact_root: str | Path | None = None,
+    output_root: str | Path | None = None,
+    platform: str | tuple[str, ...] | list[str] | None = None,
+    max_age_hours: int | None = None,
+    limit: int | None = None,
+    tile_limit: int | None = None,
+    model: str | None = None,
+    timeout: int | None = None,
+    execute: bool = False,
+    write_receipt: bool = False,
+    auto_select_model: bool = True,
+    route_readiness_timeout: int | None = None,
+    probe_generate_before_execute: bool = True,
+    route_generate_probe_timeout: int | None = None,
+) -> dict[str, Any]:
+    """Run or dry-run the PixelRAG VLM replay worker."""
+    root = Path(artifact_root or DEFAULT_ARTIFACT_ROOT)
+    output = Path(output_root or DEFAULT_OUTPUT_ROOT)
+    platforms = _normalise_platforms(platform)
+    max_age = max(1, int(max_age_hours or DEFAULT_ARTIFACT_MAX_AGE_HOURS))
+    item_limit = max(1, min(int(limit or DEFAULT_LIMIT), 250))
+    tiles = max(1, min(int(tile_limit or DEFAULT_TILE_LIMIT), 12))
+    selected_model = str(model or DEFAULT_MODEL)
+    selected_route_host = ""
+    selected_timeout = max(10, int(timeout or DEFAULT_TIMEOUT_SECONDS))
+    readiness_timeout = max(1, min(int(route_readiness_timeout or 3), 20))
+    generate_probe_timeout = max(
+        1,
+        min(
+            int(
+                route_generate_probe_timeout
+                or DEFAULT_ROUTE_GENERATE_PROBE_TIMEOUT_SECONDS
+            ),
+            30,
+        ),
+    )
+    generated_at = datetime.now(timezone.utc).isoformat()
+    route_readiness: dict[str, Any] | None = None
+    model_route_ready = True
+
+    contract = build_pixelrag_ocr_vlm_replay_contract(
+        artifact_root=root,
+        platform=platforms,
+        max_age_hours=max_age,
+        limit=item_limit,
+    )
+    replay_items = list(contract.get("replay_items") or [])
+    if execute and auto_select_model:
+        route_readiness = build_pixelrag_vlm_route_readiness(
+            model=selected_model,
+            timeout_seconds=readiness_timeout,
+            probe_generate=bool(probe_generate_before_execute),
+            probe_timeout_seconds=generate_probe_timeout,
+        )
+        route_summary = route_readiness.get("summary") or {}
+        candidate_model = str(route_summary.get("candidate_model") or "").strip()
+        model_route_ready = bool(route_summary.get("model_route_ready"))
+        if candidate_model:
+            selected_model = candidate_model
+            selected_route_host = str(route_summary.get("candidate_host") or "").strip()
+
+    worker_items: list[dict[str, Any]] = []
+    for item in replay_items:
+        if not item.get("ready_for_ollama_vlm_worker"):
+            worker_items.append(_skipped_item(item))
+            continue
+        if not execute:
+            worker_items.append(_dry_run_item(item))
+            continue
+        if not model_route_ready and route_readiness is not None:
+            worker_items.append(_model_route_not_ready_item(
+                item,
+                output_root=output,
+                route_readiness=route_readiness,
+                write_receipt=write_receipt,
+            ))
+            continue
+        worker_items.append(_execute_item(
+            item,
+            root=root,
+            output_root=output,
+            model=selected_model,
+            route_host=selected_route_host,
+            timeout=selected_timeout,
+            tile_limit=tiles,
+            write_receipt=write_receipt,
+        ))
+
+    ready_count = sum(1 for item in replay_items if item.get("ready_for_ollama_vlm_worker"))
+    skipped_count = sum(1 for item in worker_items if item.get("worker_status") == "skipped_blocked_or_not_ready")
+    dry_run_count = sum(1 for item in worker_items if item.get("worker_status") == "dry_run_ready")
+    executed_count = sum(1 for item in worker_items if str(item.get("worker_status") or "").startswith("executed_"))
+    executed_ok_count = sum(1 for item in worker_items if item.get("worker_status") == "executed_ok")
+    executed_warning_count = sum(1 for item in worker_items if item.get("worker_status") == "executed_warning")
+    model_error_count = sum(1 for item in worker_items if item.get("worker_status") == "model_error")
+    route_not_ready_count = sum(1 for item in worker_items if item.get("worker_status") == "model_route_not_ready")
+    parse_error_count = sum(1 for item in worker_items if item.get("worker_status") == "model_output_parse_error")
+    no_tile_count = sum(1 for item in worker_items if item.get("worker_status") == "skipped_no_loadable_tiles")
+    receipt_written_count = sum(1 for item in worker_items if item.get("receipt_path"))
+    required_missing_count = sum(
+        int(item.get("required_field_missing_count") or 0)
+        for item in worker_items
+    )
+    tile_model_call_performed = any(
+        bool(item.get("model_call_performed")) for item in worker_items
+    )
+    route_model_call_performed = bool(
+        route_readiness
+        and (
+            (route_readiness.get("controlled_apply") or {}).get("model_call")
+            or (route_readiness.get("summary") or {}).get("model_call_performed")
+        )
+    )
+    model_call_performed = bool(
+        tile_model_call_performed or route_model_call_performed
+    )
+    artifact_write_performed = any(bool(item.get("artifact_write_performed")) for item in worker_items)
+
+    if parse_error_count or model_error_count or route_not_ready_count or no_tile_count:
+        status = "critical" if ready_count and executed_ok_count == 0 and execute else "warning"
+    elif executed_warning_count or skipped_count or dry_run_count or (not replay_items):
+        status = "warning"
+    else:
+        status = "ok"
+
+    if not replay_items:
+        next_action = "run_pixelrag_visual_capture_worker"
+    elif not execute and ready_count:
+        next_action = "run_pixelrag_vlm_replay_worker_execute"
+    elif route_not_ready_count:
+        next_action = (
+            route_readiness.get("next_machine_action")
+            if route_readiness
+            else None
+        ) or "install_or_configure_pixelrag_vlm_model_on_approved_ollama_host"
+    elif model_error_count or parse_error_count:
+        next_action = "repair_ollama_vlm_runtime_or_model_route"
+    elif executed_warning_count:
+        warning_actions = {
+            str(item.get("next_machine_action") or "")
+            for item in worker_items
+            if item.get("worker_status") == "executed_warning"
+        }
+        if warning_actions == {"run_platform_probe_or_use_structured_api"}:
+            next_action = "run_platform_probe_or_use_structured_api"
+        else:
+            next_action = "rerun_vlm_replay_with_more_tiles_or_platform_probe"
+    elif executed_ok_count:
+        next_action = "run_identity_matcher_replay_then_promotion_gate"
+    else:
+        next_action = "run_platform_probe_or_use_structured_api"
+
+    summary = {
+        "receipt_count": len(replay_items),
+        "ready_count": ready_count,
+        "skipped_count": skipped_count,
+        "dry_run_count": dry_run_count,
+        "executed_count": executed_count,
+        "executed_ok_count": executed_ok_count,
+        "executed_warning_count": executed_warning_count,
+        "model_error_count": model_error_count,
+        "model_route_not_ready_count": route_not_ready_count,
+        "parse_error_count": parse_error_count,
+        "no_tile_count": no_tile_count,
+        "receipt_written_count": receipt_written_count,
+        "required_field_missing_count": required_missing_count,
+        "route_model_call_performed": route_model_call_performed,
+        "tile_model_call_performed": tile_model_call_performed,
+        "model_call_performed": model_call_performed,
+        "artifact_write_performed": artifact_write_performed,
+        "writes_database_count": 0,
+        "primary_human_gate_count": 0,
+        "platforms": sorted({str(item.get("platform") or "unknown") for item in replay_items}),
+    }
+    return {
+        "success": status != "critical",
+        "policy": POLICY,
+        "status": status,
+        "generated_at": generated_at,
+        "artifact_root": str(root),
+        "output_root": str(output),
+        "platform_filter": list(platforms),
+        "max_age_hours": max_age,
+        "limit": item_limit,
+        "tile_limit": tiles,
+        "model": selected_model,
+        "configured_model": str(model or DEFAULT_MODEL),
+        "route_candidate_host": selected_route_host,
+        "timeout_seconds": selected_timeout,
+        "execute": bool(execute),
+        "write_receipt": bool(write_receipt),
+        "auto_select_model": bool(auto_select_model),
+        "route_readiness_timeout_seconds": readiness_timeout,
+        "probe_generate_before_execute": bool(probe_generate_before_execute),
+        "route_generate_probe_timeout_seconds": generate_probe_timeout,
+        "summary": summary,
+        "worker_items": worker_items,
+        "route_readiness": (
+            {
+                "policy": route_readiness.get("policy"),
+                "status": route_readiness.get("status"),
+                "summary": route_readiness.get("summary"),
+                "next_machine_action": route_readiness.get("next_machine_action"),
+            }
+            if route_readiness
+            else None
+        ),
+        "source_contract": {
+            "policy": contract.get("policy"),
+            "status": contract.get("status"),
+            "summary": contract.get("summary"),
+            "next_machine_action": contract.get("next_machine_action"),
+        },
+        "controlled_apply": {
+            "network_call": bool(execute and (route_readiness or model_call_performed)),
+            "model_call": bool(execute and model_call_performed),
+            "artifact_write": artifact_write_performed,
+            "db_write": False,
+            "writes_database": False,
+            "writes_database_count": 0,
+            "secret_read": False,
+            "production_price_write": False,
+            "primary_human_gate_count": 0,
+        },
+        "promotion_boundary": {
+            "writes_ai_insights": False,
+            "writes_price_tables": False,
+            "requires_identity_matcher_replay": True,
+            "requires_promotion_gate": True,
+            "visual_fields_are_candidate_evidence_only": True,
+        },
+        "next_machine_action": next_action,
+    }
+
+
+__all__ = [
+    "DEFAULT_MODEL",
+    "POLICY",
+    "run_pixelrag_ollama_vlm_replay_worker",
+]

services/pixelrag_vlm_route_readiness_service.py

@@ -0,0 +1,330 @@
+"""PixelRAG VLM route readiness.
+
+This module checks approved Ollama routes for installed VLM candidate models.
+By default it only reads /api/tags. Execute preflight can opt into a tiny
+/api/generate probe to avoid blindly sending screenshot tiles to a route that
+is installed but cannot generate.
+"""
+
+from __future__ import annotations
+
+import os
+import time
+from datetime import datetime, timezone
+from typing import Any
+
+import requests
+
+from services.ollama_service import (
+    OLLAMA_HOST_FALLBACK,
+    OLLAMA_HOST_PRIMARY,
+    OLLAMA_HOST_PRIMARY_PROXY,
+    OLLAMA_HOST_SECONDARY,
+    OLLAMA_HOST_SECONDARY_PROXY,
+    get_host_label,
+    get_provider_tag,
+)
+
+
+POLICY = "read_only_pixelrag_vlm_route_readiness_v1"
+DEFAULT_TIMEOUT_SECONDS = 3
+DEFAULT_PROBE_TIMEOUT_SECONDS = 8
+DEFAULT_PROBE_PROMPT = "Return exactly OK."
+DEFAULT_MODEL = (
+    os.getenv("PIXELRAG_VLM_MODEL")
+    or os.getenv("PPT_VISION_MODEL")
+    or "minicpm-v:latest"
+)
+VISION_MODEL_CANDIDATES = (
+    "minicpm-v:latest",
+    "gemma3:4b",
+    "llava:latest",
+)
+
+
+def _unique(values: list[str] | tuple[str, ...]) -> list[str]:
+    seen: set[str] = set()
+    result: list[str] = []
+    for value in values:
+        clean = str(value or "").strip()
+        if not clean or clean in seen:
+            continue
+        seen.add(clean)
+        result.append(clean)
+    return result
+
+
+def _approved_hosts() -> list[str]:
+    return _unique((
+        OLLAMA_HOST_PRIMARY,
+        OLLAMA_HOST_PRIMARY_PROXY,
+        OLLAMA_HOST_SECONDARY,
+        OLLAMA_HOST_SECONDARY_PROXY,
+        OLLAMA_HOST_FALLBACK,
+    ))
+
+
+def _candidate_models(configured_model: str) -> list[str]:
+    env_candidates = [
+        item.strip()
+        for item in os.getenv("PIXELRAG_VLM_MODEL_CANDIDATES", "").split(",")
+        if item.strip()
+    ]
+    return _unique((configured_model, *env_candidates, *VISION_MODEL_CANDIDATES))
+
+
+def _fetch_models(host: str, timeout: int) -> dict[str, Any]:
+    clean_host = str(host or "").rstrip("/")
+    item: dict[str, Any] = {
+        "host": clean_host,
+        "host_label": get_host_label(clean_host),
+        "provider": get_provider_tag(clean_host),
+        "reachable": False,
+        "model_count": 0,
+        "models": [],
+        "error": "",
+    }
+    try:
+        response = requests.get(f"{clean_host}/api/tags", timeout=max(1, timeout))
+        item["http_status"] = response.status_code
+        if response.status_code != 200:
+            item["error"] = f"HTTP {response.status_code}: {response.text[:180]}"
+            return item
+        payload = response.json()
+        models = [
+            str(model.get("name") or "").strip()
+            for model in list(payload.get("models") or [])
+            if str(model.get("name") or "").strip()
+        ]
+        item.update({
+            "reachable": True,
+            "model_count": len(models),
+            "models": models[:80],
+        })
+    except Exception as exc:
+        item["error"] = f"{type(exc).__name__}: {str(exc)[:180]}"
+    return item
+
+
+def _probe_generate(
+    host: str,
+    *,
+    model: str,
+    timeout: int,
+    prompt: str,
+) -> dict[str, Any]:
+    clean_host = str(host or "").rstrip("/")
+    started = time.monotonic()
+    result: dict[str, Any] = {
+        "generate_probe_performed": True,
+        "generate_probe_model": str(model or "").strip(),
+        "generate_probe_ok": False,
+        "generate_probe_error": "",
+        "generate_probe_duration_ms": 0,
+    }
+    payload = {
+        "model": str(model or "").strip(),
+        "prompt": str(prompt or DEFAULT_PROBE_PROMPT),
+        "stream": False,
+        "options": {
+            "temperature": 0,
+            "num_predict": 8,
+            "num_ctx": 512,
+        },
+        "keep_alive": "1m",
+    }
+    try:
+        response = requests.post(
+            f"{clean_host}/api/generate",
+            json=payload,
+            timeout=max(1, timeout),
+        )
+        result["generate_probe_http_status"] = response.status_code
+        if response.status_code != 200:
+            result["generate_probe_error"] = (
+                f"HTTP {response.status_code}: {response.text[:180]}"
+            )
+            return result
+        try:
+            body = response.json()
+        except Exception as exc:
+            result["generate_probe_error"] = f"invalid_json: {type(exc).__name__}"
+            return result
+        result["generate_probe_ok"] = (
+            isinstance(body, dict)
+            and (
+                "response" in body
+                or bool(body.get("done"))
+                or bool(body.get("context"))
+            )
+        )
+        if not result["generate_probe_ok"]:
+            result["generate_probe_error"] = "generate_response_missing_output_fields"
+    except Exception as exc:
+        result["generate_probe_error"] = f"{type(exc).__name__}: {str(exc)[:180]}"
+    finally:
+        result["generate_probe_duration_ms"] = int((time.monotonic() - started) * 1000)
+    return result
+
+
+def build_pixelrag_vlm_route_readiness(
+    *,
+    model: str | None = None,
+    timeout_seconds: int | None = None,
+    include_models: bool = False,
+    probe_generate: bool = False,
+    probe_timeout_seconds: int | None = None,
+    probe_prompt: str | None = None,
+) -> dict[str, Any]:
+    """Read approved Ollama tags and recommend an installed VLM model."""
+    configured_model = str(model or DEFAULT_MODEL).strip() or DEFAULT_MODEL
+    timeout = max(1, min(int(timeout_seconds or DEFAULT_TIMEOUT_SECONDS), 20))
+    probe_timeout = max(
+        1,
+        min(int(probe_timeout_seconds or DEFAULT_PROBE_TIMEOUT_SECONDS), 30),
+    )
+    generated_at = datetime.now(timezone.utc).isoformat()
+    candidates = _candidate_models(configured_model)
+    host_results = [_fetch_models(host, timeout) for host in _approved_hosts()]
+
+    reachable_hosts = [item for item in host_results if item.get("reachable")]
+    configured_hosts = [
+        item for item in reachable_hosts
+        if configured_model in set(item.get("models") or [])
+    ]
+    selected_model = ""
+    selected_host = ""
+    selected_provider = ""
+    selected_reason = ""
+    for candidate in candidates:
+        for host in reachable_hosts:
+            if candidate not in set(host.get("models") or []):
+                continue
+            selected_model = candidate
+            selected_host = str(host.get("host") or "")
+            selected_provider = str(host.get("provider") or "")
+            selected_reason = (
+                "configured_model_available"
+                if candidate == configured_model
+                else "installed_candidate_fallback"
+            )
+            break
+        if selected_model:
+            break
+
+    tag_model_route_ready = bool(selected_model and selected_host)
+    generate_probe_performed = bool(probe_generate and tag_model_route_ready)
+    generate_probe_ok_count = 0
+    generate_ready_model = ""
+    generate_ready_host = ""
+    generate_ready_provider = ""
+    generate_route_ready: bool | None = None
+    if generate_probe_performed:
+        generate_route_ready = False
+        for host in reachable_hosts:
+            if selected_model not in set(host.get("models") or []):
+                continue
+            probe_result = _probe_generate(
+                str(host.get("host") or ""),
+                model=selected_model,
+                timeout=probe_timeout,
+                prompt=probe_prompt or DEFAULT_PROBE_PROMPT,
+            )
+            host.update(probe_result)
+            if probe_result.get("generate_probe_ok"):
+                generate_probe_ok_count += 1
+                if not generate_ready_host:
+                    generate_ready_model = selected_model
+                    generate_ready_host = str(host.get("host") or "")
+                    generate_ready_provider = str(host.get("provider") or "")
+        if generate_ready_host:
+            selected_host = generate_ready_host
+            selected_provider = generate_ready_provider
+            generate_route_ready = True
+
+    model_route_ready = tag_model_route_ready and (
+        bool(generate_route_ready) if generate_probe_performed else True
+    )
+    if not reachable_hosts:
+        status = "critical"
+        next_action = "repair_approved_ollama_host_connectivity"
+    elif not tag_model_route_ready:
+        status = "critical"
+        next_action = "install_or_configure_pixelrag_vlm_model_on_approved_ollama_host"
+    elif generate_probe_performed and not generate_route_ready:
+        status = "critical"
+        next_action = "repair_ollama_vlm_generate_runtime_or_proxy_timeout"
+    elif selected_model != configured_model:
+        status = "warning"
+        next_action = "run_pixelrag_vlm_replay_worker_execute_with_selected_model"
+    else:
+        status = "ok"
+        next_action = "run_pixelrag_vlm_replay_worker_execute"
+
+    public_host_results: list[dict[str, Any]] = []
+    for item in host_results:
+        result = dict(item)
+        result["configured_model_available"] = configured_model in set(item.get("models") or [])
+        result["candidate_models_available"] = [
+            candidate for candidate in candidates
+            if candidate in set(item.get("models") or [])
+        ]
+        result.setdefault("generate_probe_performed", False)
+        result.setdefault("generate_probe_model", "")
+        result.setdefault("generate_probe_ok", False)
+        result.setdefault("generate_probe_error", "")
+        result.setdefault("generate_probe_duration_ms", 0)
+        if not include_models:
+            result.pop("models", None)
+        public_host_results.append(result)
+
+    summary = {
+        "host_count": len(host_results),
+        "reachable_host_count": len(reachable_hosts),
+        "configured_model": configured_model,
+        "configured_model_available_count": len(configured_hosts),
+        "candidate_model": selected_model,
+        "candidate_host": selected_host,
+        "candidate_provider": selected_provider,
+        "candidate_selection_reason": selected_reason,
+        "tag_model_route_ready": tag_model_route_ready,
+        "model_route_ready": model_route_ready,
+        "generate_probe_performed": generate_probe_performed,
+        "generate_probe_ok_count": generate_probe_ok_count,
+        "generate_ready_model": generate_ready_model,
+        "generate_ready_host": generate_ready_host,
+        "generate_ready_provider": generate_ready_provider,
+        "generate_route_ready": generate_route_ready,
+        "model_call_performed": generate_probe_performed,
+        "writes_database_count": 0,
+        "primary_human_gate_count": 0,
+    }
+    return {
+        "success": status != "critical",
+        "policy": POLICY,
+        "status": status,
+        "generated_at": generated_at,
+        "timeout_seconds": timeout,
+        "probe_timeout_seconds": probe_timeout if probe_generate else 0,
+        "configured_model": configured_model,
+        "candidate_models": candidates,
+        "summary": summary,
+        "hosts": public_host_results,
+        "controlled_apply": {
+            "network_call": True,
+            "model_call": generate_probe_performed,
+            "db_write": False,
+            "writes_database": False,
+            "writes_database_count": 0,
+            "secret_read": False,
+            "production_price_write": False,
+            "primary_human_gate_count": 0,
+        },
+        "next_machine_action": next_action,
+    }
+
+
+__all__ = [
+    "POLICY",
+    "build_pixelrag_vlm_route_readiness",
+]

tests/test_ai_automation_smoke_service.py

@@ -1434,11 +1434,14 @@ def test_collect_ai_automation_smoke_uses_worst_status(monkeypatch):
     monkeypatch.setattr(smoke, "_external_mcp_rag_integration_check", lambda: smoke._check("external mcp rag", "ok", "ok"))
     monkeypatch.setattr(smoke, "_pixelrag_rag_candidate_replay_check", lambda: smoke._check("pixelrag replay", "ok", "ok"))
     monkeypatch.setattr(smoke, "_pixelrag_ocr_vlm_replay_check", lambda: smoke._check("pixelrag ocr vlm", "ok", "ok"))
+    monkeypatch.setattr(smoke, "_pixelrag_vlm_route_readiness_check", lambda: smoke._check("pixelrag vlm route", "ok", "ok"))
+    monkeypatch.setattr(smoke, "_pixelrag_vlm_replay_worker_check", lambda: smoke._check("pixelrag vlm worker", "ok", "ok"))
+    monkeypatch.setattr(smoke, "_pixelrag_platform_probe_check", lambda: smoke._check("pixelrag platform probe", "ok", "ok"))
 
     result = smoke.collect_ai_automation_smoke(record_history=False)
 
     assert result["status"] == "critical"
-    assert result["summary"] == {"ok": 32, "warning": 1, "critical": 1, "total": 34}
+    assert result["summary"] == {"ok": 35, "warning": 1, "critical": 1, "total": 37}
 
 
 def test_pchome_controlled_apply_drift_monitor_reports_verified_zero_drift(monkeypatch):
@@ -3968,6 +3971,9 @@ def test_collect_ai_automation_smoke_persists_recent_history(tmp_path, monkeypat
     monkeypatch.setattr(smoke, "_external_mcp_rag_integration_check", lambda: smoke._check("external mcp rag", "ok", "ok"))
     monkeypatch.setattr(smoke, "_pixelrag_rag_candidate_replay_check", lambda: smoke._check("pixelrag replay", "ok", "ok"))
     monkeypatch.setattr(smoke, "_pixelrag_ocr_vlm_replay_check", lambda: smoke._check("pixelrag ocr vlm", "ok", "ok"))
+    monkeypatch.setattr(smoke, "_pixelrag_vlm_route_readiness_check", lambda: smoke._check("pixelrag vlm route", "ok", "ok"))
+    monkeypatch.setattr(smoke, "_pixelrag_vlm_replay_worker_check", lambda: smoke._check("pixelrag vlm worker", "ok", "ok"))
+    monkeypatch.setattr(smoke, "_pixelrag_platform_probe_check", lambda: smoke._check("pixelrag platform probe", "ok", "ok"))
 
     first = smoke.collect_ai_automation_smoke(history_limit=5)
     second = smoke.collect_ai_automation_smoke(history_limit=5)
@@ -4023,7 +4029,7 @@ def test_scheduled_automation_health_summary_reads_history_without_side_effects(
         json.dumps({
             "generated_at": datetime.now().isoformat(timespec="seconds"),
             "status": "ok",
-            "summary": {"ok": 34, "warning": 0, "critical": 0, "total": 34},
+            "summary": {"ok": 37, "warning": 0, "critical": 0, "total": 37},
             "checks": [
                 {
                     "name": "PChome 受控落地 drift monitor",
@@ -4148,6 +4154,64 @@ def test_scheduled_automation_health_summary_reads_history_without_side_effects(
                             "blocked_pages_are_not_product_data": True,
                             "next_machine_action": "run_ollama_first_vlm_replay_worker",
                         },
+                    },
+                    {
+                        "name": "PixelRAG VLM route readiness",
+                        "status": "ok",
+                        "summary": "PixelRAG VLM route readiness hosts=1/5, configured=minicpm-v:latest, configured_available=1, candidate=minicpm-v:latest",
+                        "details": {
+                            "policy": "read_only_pixelrag_vlm_route_readiness_v1",
+                            "host_count": 5,
+                            "reachable_host_count": 1,
+                            "configured_model": "minicpm-v:latest",
+                            "configured_model_available_count": 1,
+                            "candidate_model": "minicpm-v:latest",
+                            "candidate_host": "http://34.21.145.224:11434",
+                            "candidate_provider": "ollama_secondary",
+                            "model_route_ready": True,
+                            "model_call_performed": False,
+                            "next_machine_action": "run_pixelrag_vlm_replay_worker_execute",
+                        },
+                    },
+                    {
+                        "name": "PixelRAG VLM replay worker",
+                        "status": "ok",
+                        "summary": "PixelRAG VLM replay worker receipts=1, ready=1, dry_run=1, skipped=0, executed=0",
+                        "details": {
+                            "policy": "controlled_pixelrag_ollama_vlm_replay_worker_v1",
+                            "receipt_count": 1,
+                            "ready_count": 1,
+                            "skipped_count": 0,
+                            "dry_run_count": 1,
+                            "executed_count": 0,
+                            "executed_ok_count": 0,
+                            "executed_warning_count": 0,
+                            "model_error_count": 0,
+                            "model_route_not_ready_count": 0,
+                            "parse_error_count": 0,
+                            "receipt_written_count": 0,
+                            "model_call_performed": False,
+                            "artifact_write_performed": False,
+                            "next_machine_action": "run_pixelrag_vlm_replay_worker_execute",
+                        },
+                    },
+                    {
+                        "name": "PixelRAG platform probe readiness",
+                        "status": "ok",
+                        "summary": "PixelRAG platform probe candidates=1, ready=1, shopee_context=1, structured_fallback=1",
+                        "details": {
+                            "policy": "read_only_pixelrag_platform_probe_readiness_v1",
+                            "probe_candidate_count": 1,
+                            "ready_for_probe_count": 1,
+                            "shopee_public_context_probe_count": 1,
+                            "language_or_region_interstitial_count": 1,
+                            "traffic_verification_count": 0,
+                            "access_denied_count": 0,
+                            "structured_source_fallback_count": 1,
+                            "next_machine_action": "run_platform_probe_or_structured_source_fallback",
+                            "writes_database_count": 0,
+                            "primary_human_gate_count": 0,
+                        },
                     }
                 ],
             }, ensure_ascii=False) + "\n",
@@ -4166,7 +4230,7 @@ def test_scheduled_automation_health_summary_reads_history_without_side_effects(
     )
     assert summary["policy"] == "read_only_ai_automation_scheduled_health_summary"
     assert summary["status"] == "ok"
-    assert summary["summary"]["total"] == 31
+    assert summary["summary"]["total"] == 34
     assert summary["summary"]["primary_human_gate_count"] == 0
     assert summary["summary"]["writes_database_count"] == 0
     assert pchome_family["status"] == "ok"
@@ -5849,6 +5913,30 @@ def test_scheduled_automation_health_summary_reads_history_without_side_effects(
     assert pixelrag_ocr_vlm_replay_family["status"] == "ok"
     assert pixelrag_ocr_vlm_replay_family["details"]["replay_ready_count"] == 1
     assert pixelrag_ocr_vlm_replay_family["details"]["extraction_execution_performed"] is False
+    pixelrag_vlm_route_readiness_family = next(
+        item for item in summary["families"]
+        if item["key"] == "pixelrag_vlm_route_readiness"
+    )
+    assert pixelrag_vlm_route_readiness_family["status"] == "ok"
+    assert pixelrag_vlm_route_readiness_family["details"]["model_route_ready"] is True
+    assert pixelrag_vlm_route_readiness_family["details"]["model_call_performed"] is False
+    assert pixelrag_vlm_route_readiness_family["details"]["writes_database_count"] == 0
+    pixelrag_vlm_replay_worker_family = next(
+        item for item in summary["families"]
+        if item["key"] == "pixelrag_vlm_replay_worker"
+    )
+    assert pixelrag_vlm_replay_worker_family["status"] == "ok"
+    assert pixelrag_vlm_replay_worker_family["details"]["ready_count"] == 1
+    assert pixelrag_vlm_replay_worker_family["details"]["dry_run_count"] == 1
+    assert pixelrag_vlm_replay_worker_family["details"]["model_call_performed"] is False
+    assert pixelrag_vlm_replay_worker_family["details"]["writes_database_count"] == 0
+    pixelrag_platform_probe_family = next(
+        item for item in summary["families"]
+        if item["key"] == "pixelrag_platform_probe"
+    )
+    assert pixelrag_platform_probe_family["status"] == "ok"
+    assert pixelrag_platform_probe_family["details"]["ready_for_probe_count"] == 1
+    assert pixelrag_platform_probe_family["details"]["primary_human_gate_count"] == 0
     assert summary["scheduled_outputs"]["telegram_send_in_preview"] is False
     assert summary["scheduled_outputs"]["writes_database"] is False
     assert summary["automation_policy"]["primary_flow"] == "ai_controlled"
@@ -6470,6 +6558,21 @@ def test_surface_html_readback_check_is_part_of_ai_smoke(monkeypatch):
         "ok",
         "pixelrag ocr vlm ok",
     ))
+    monkeypatch.setattr(smoke, "_pixelrag_vlm_route_readiness_check", lambda: smoke._check(
+        "PixelRAG VLM route readiness",
+        "ok",
+        "pixelrag vlm route ok",
+    ))
+    monkeypatch.setattr(smoke, "_pixelrag_vlm_replay_worker_check", lambda: smoke._check(
+        "PixelRAG VLM replay worker",
+        "ok",
+        "pixelrag vlm worker ok",
+    ))
+    monkeypatch.setattr(smoke, "_pixelrag_platform_probe_check", lambda: smoke._check(
+        "PixelRAG platform probe readiness",
+        "ok",
+        "pixelrag platform probe ok",
+    ))
 
     result = smoke.collect_ai_automation_smoke(record_history=False)
 
@@ -6485,7 +6588,7 @@ def test_surface_html_readback_check_is_part_of_ai_smoke(monkeypatch):
         item for item in result["checks"]
         if item["name"] == "Sitewide visual QA readback"
     )
-    assert result["summary"]["total"] == 34
+    assert result["summary"]["total"] == 37
     assert surface_check["status"] == "ok"
     assert surface_check["details"]["checked_surface_count"] == 10
     assert sitewide_check["status"] == "ok"

tests/test_pixelrag_platform_probe_service.py

@@ -0,0 +1,240 @@
+import json
+import subprocess
+import sys
+from datetime import datetime, timezone
+
+from tests.test_pixelrag_rag_candidate_replay_service import _write_receipt
+
+
+def _patch_receipt(path, **updates):
+    payload = json.loads(path.read_text(encoding="utf-8"))
+    for key, value in updates.items():
+        if key == "page_metrics":
+            payload.setdefault("page_metrics", {}).update(value)
+        else:
+            payload[key] = value
+    path.write_text(json.dumps(payload, ensure_ascii=False), encoding="utf-8")
+    return payload
+
+
+def test_pixelrag_platform_probe_builds_machine_actions_for_shopee_and_coupang(tmp_path):
+    from services.pixelrag_platform_probe_service import (
+        POLICY,
+        build_pixelrag_platform_probe_readiness,
+    )
+
+    shopee_receipt = _write_receipt(
+        tmp_path,
+        platform="shopee_tw",
+        manifest_id="shopee-traffic",
+        title="蝦皮購物 | 花得更少買得更好",
+        url="https://shopee.tw/search?keyword=sunscreen",
+    )
+    _patch_receipt(
+        shopee_receipt,
+        page_metrics={
+            "final_url": "https://shopee.tw/verify/traffic/error?home_url=https%3A%2F%2Fshopee.tw",
+            "title": "蝦皮購物 | 花得更少買得更好",
+        },
+    )
+    _write_receipt(
+        tmp_path,
+        platform="coupang_tw",
+        manifest_id="coupang-403",
+        title="Access Denied",
+        url="https://www.tw.coupang.com/search?q=iphone",
+        http_status=403,
+    )
+
+    payload = build_pixelrag_platform_probe_readiness(artifact_root=tmp_path)
+    by_platform = {item["platform"]: item for item in payload["probe_items"]}
+
+    assert payload["policy"] == POLICY
+    assert payload["status"] == "ok"
+    assert payload["summary"]["probe_candidate_count"] == 2
+    assert payload["summary"]["ready_for_probe_count"] == 2
+    assert payload["summary"]["traffic_verification_count"] == 1
+    assert payload["summary"]["access_denied_count"] == 1
+    assert payload["summary"]["writes_database_count"] == 0
+    assert payload["controlled_apply"]["primary_human_gate_count"] == 0
+    assert by_platform["shopee_tw"]["probe_status"] == "ready_for_public_context_probe"
+    assert by_platform["shopee_tw"]["next_machine_action"] == (
+        "run_shopee_public_context_probe_then_structured_source_fallback"
+    )
+    assert by_platform["shopee_tw"]["public_browser_context"]["locale"] == "zh-TW"
+    assert by_platform["shopee_tw"]["public_browser_context"]["storage_state_allowed"] is False
+    assert by_platform["shopee_tw"]["capture_manifest_preview"]["success"] is True
+    assert by_platform["coupang_tw"]["probe_status"] == "structured_source_or_backoff_required"
+    assert by_platform["coupang_tw"]["next_machine_action"] == (
+        "use_structured_source_or_platform_backoff_policy"
+    )
+    assert by_platform["coupang_tw"]["structured_source_fallback"]["available"] is True
+    assert payload["probe_contract"]["raw_cookie_or_session_read"] is False
+    assert payload["probe_contract"]["blocked_pages_are_not_product_data"] is True
+
+
+def test_pixelrag_platform_probe_reads_vlm_interstitial_receipts(tmp_path):
+    from services.pixelrag_platform_probe_service import build_pixelrag_platform_probe_readiness
+
+    capture_path = _write_receipt(
+        tmp_path / "capture",
+        platform="shopee_tw",
+        manifest_id="shopee-language",
+        title="Shopee",
+        url="https://shopee.tw/search?keyword=sunscreen",
+    )
+    vlm_dir = tmp_path / "vlm" / "shopee_tw" / "shopee-language"
+    vlm_dir.mkdir(parents=True)
+    (vlm_dir / "vlm_replay_receipt.json").write_text(
+        json.dumps({
+            "generated_at": datetime.now(timezone.utc).isoformat(),
+            "platform": "shopee_tw",
+            "manifest_id": "shopee-language",
+            "source_receipt_path": str(capture_path),
+            "worker_status": "executed_warning",
+            "next_machine_action": "run_platform_probe_or_use_structured_api",
+            "parsed_output": {
+                "blocked_page_detected": False,
+                "fields": {
+                    "title": {
+                        "value": "蝦皮購物 | 花得更少買得更好",
+                        "confidence": 0.95,
+                        "evidence_refs": ["tile:1"],
+                    },
+                    "price": {"value": None, "confidence": 0.0, "evidence_refs": []},
+                },
+                "notes": [{"note": "Language selection page"}],
+            },
+            "validation": {
+                "blocked_page_detected": False,
+                "non_product_or_interstitial_detected": True,
+                "interstitial_signal_detected": True,
+                "generic_marketplace_title_detected": True,
+                "present_field_count": 1,
+                "missing_required_fields": ["price"],
+            },
+        }, ensure_ascii=False),
+        encoding="utf-8",
+    )
+
+    payload = build_pixelrag_platform_probe_readiness(
+        artifact_root=tmp_path / "capture",
+        vlm_receipt_root=tmp_path / "vlm",
+        platform="shopee_tw",
+    )
+
+    assert payload["status"] == "ok"
+    assert payload["summary"]["probe_candidate_count"] == 1
+    assert payload["summary"]["vlm_source_count"] == 1
+    assert payload["summary"]["language_or_region_interstitial_count"] == 1
+    item = payload["probe_items"][0]
+    assert item["source_type"] == "vlm_replay_receipt"
+    assert item["barrier_type"] == "language_or_region_interstitial"
+    assert item["probe_status"] == "ready_for_public_context_probe"
+    assert item["primary_human_gate_count"] == 0
+
+
+def test_pixelrag_platform_probe_cli_outputs_machine_readable_json(tmp_path):
+    _write_receipt(
+        tmp_path,
+        platform="coupang_tw",
+        manifest_id="coupang-403",
+        title="Access Denied",
+        url="https://www.tw.coupang.com/search?q=iphone",
+        http_status=403,
+    )
+
+    completed = subprocess.run(
+        [
+            sys.executable,
+            "scripts/ops/report_pixelrag_platform_probe.py",
+            "--artifact-root",
+            str(tmp_path),
+            "--platform",
+            "coupang_tw",
+        ],
+        capture_output=True,
+        check=False,
+        text=True,
+    )
+
+    assert completed.returncode == 0
+    payload = json.loads(completed.stdout)
+    assert payload["summary"]["probe_candidate_count"] == 1
+    assert payload["probe_items"][0]["platform"] == "coupang_tw"
+    assert payload["probe_contract"]["writes_database"] is False
+
+
+def test_pixelrag_capture_worker_dry_run_uses_safe_public_context(tmp_path):
+    from services.pixelrag_crawler_integration_service import (
+        build_pixelrag_marketplace_search_manifest,
+    )
+
+    manifest = build_pixelrag_marketplace_search_manifest(
+        platform="shopee_tw",
+        keyword="sunscreen",
+        crawler="PixelRAGPlatformProbe.public_context_visual_fallback",
+        trigger_reason="platform_interstitial_or_blocked_page_probe",
+    )
+    manifest["public_browser_context"] = {
+        "locale": "zh-TW",
+        "timezone_id": "Asia/Taipei",
+        "extra_http_headers": {
+            "Accept-Language": "zh-TW,zh-Hant;q=0.95,zh;q=0.9,en;q=0.7",
+            "Cookie": "must_not_survive",
+            "Authorization": "must_not_survive",
+        },
+    }
+    completed = subprocess.run(
+        [
+            sys.executable,
+            "scripts/ops/capture_pixelrag_visual_evidence.py",
+            "--manifest-json",
+            json.dumps(manifest, ensure_ascii=False),
+            "--output-dir",
+            str(tmp_path / "artifacts"),
+            "--dry-run",
+        ],
+        capture_output=True,
+        check=False,
+        text=True,
+    )
+
+    assert completed.returncode == 0
+    payload = json.loads(completed.stdout)
+    context = payload["public_browser_context"]
+    assert context["locale"] == "zh-TW"
+    assert context["timezone_id"] == "Asia/Taipei"
+    assert context["extra_http_headers"]["Accept-Language"].startswith("zh-TW")
+    assert "Cookie" not in context["extra_http_headers"]
+    assert "Authorization" not in context["extra_http_headers"]
+    assert context["raw_cookie_or_session_read_allowed"] is False
+
+
+def test_pixelrag_platform_probe_route_returns_readback(tmp_path, monkeypatch):
+    from flask import Flask
+    from routes import system_public_routes as routes
+    from services import pixelrag_platform_probe_service as service
+
+    _write_receipt(
+        tmp_path,
+        platform="coupang_tw",
+        manifest_id="coupang-403",
+        title="Access Denied",
+        url="https://www.tw.coupang.com/search?q=iphone",
+        http_status=403,
+    )
+    monkeypatch.setattr(service, "DEFAULT_ARTIFACT_ROOT", str(tmp_path))
+
+    app = Flask(__name__)
+    with app.test_request_context(
+        "/api/ai-automation/pixelrag-platform-probe?platform=coupang_tw"
+    ):
+        response = routes.ai_automation_pixelrag_platform_probe_api.__wrapped__()
+    payload = response.get_json()
+
+    assert payload["policy"] == "read_only_pixelrag_platform_probe_readiness_v1"
+    assert payload["summary"]["probe_candidate_count"] == 1
+    assert payload["probe_items"][0]["next_machine_action"] == (
+        "use_structured_source_or_platform_backoff_policy"
+    )

tests/test_pixelrag_vlm_replay_worker_service.py

@@ -0,0 +1,523 @@
+import json
+import subprocess
+import sys
+from types import SimpleNamespace
+
+from tests.test_pixelrag_ocr_vlm_replay_service import _write_receipt
+
+
+def test_pixelrag_vlm_replay_worker_dry_run_keeps_blocked_pages_out(tmp_path):
+    from services.pixelrag_vlm_replay_worker_service import (
+        POLICY,
+        run_pixelrag_ollama_vlm_replay_worker,
+    )
+
+    _write_receipt(
+        tmp_path,
+        platform="shopee_tw",
+        manifest_id="shopee-ok",
+        title="Shopee 防曬乳",
+        url="https://shopee.tw/search?keyword=sunscreen",
+    )
+    _write_receipt(
+        tmp_path,
+        platform="coupang_tw",
+        manifest_id="coupang-403",
+        title="Access Denied",
+        url="https://www.tw.coupang.com/search?q=iphone",
+        http_status=403,
+    )
+
+    payload = run_pixelrag_ollama_vlm_replay_worker(
+        artifact_root=tmp_path,
+        platform=("shopee_tw", "coupang_tw"),
+    )
+
+    assert payload["policy"] == POLICY
+    assert payload["status"] == "warning"
+    assert payload["execute"] is False
+    assert payload["summary"]["receipt_count"] == 2
+    assert payload["summary"]["ready_count"] == 1
+    assert payload["summary"]["dry_run_count"] == 1
+    assert payload["summary"]["skipped_count"] == 1
+    assert payload["summary"]["model_call_performed"] is False
+    assert payload["summary"]["artifact_write_performed"] is False
+    assert payload["summary"]["writes_database_count"] == 0
+    assert payload["controlled_apply"]["primary_human_gate_count"] == 0
+    assert payload["next_machine_action"] == "run_pixelrag_vlm_replay_worker_execute"
+    by_platform = {item["platform"]: item for item in payload["worker_items"]}
+    assert by_platform["shopee_tw"]["worker_status"] == "dry_run_ready"
+    assert by_platform["coupang_tw"]["worker_status"] == "skipped_blocked_or_not_ready"
+
+
+def test_pixelrag_vlm_replay_worker_execute_writes_artifact_receipt(tmp_path, monkeypatch):
+    from services import pixelrag_vlm_replay_worker_service as service
+
+    _write_receipt(
+        tmp_path,
+        platform="shopee_tw",
+        manifest_id="shopee-ok",
+        title="Shopee 防曬乳",
+        url="https://shopee.tw/search?keyword=sunscreen",
+    )
+
+    class FakeOllama:
+        def __init__(self, model):
+            self.model = model
+
+        def generate(self, *args, **kwargs):
+            return SimpleNamespace(
+                success=True,
+                content=json.dumps({
+                    "blocked_page_detected": False,
+                    "fields": {
+                        "title": {
+                            "value": "防曬乳 SPF50",
+                            "confidence": 0.92,
+                            "evidence_refs": ["tile:1"],
+                        },
+                        "price": {
+                            "value": "399",
+                            "confidence": 0.88,
+                            "evidence_refs": ["tile:2"],
+                        },
+                    },
+                    "quality": {
+                        "overall_confidence": 0.89,
+                        "missing_required_fields": [],
+                        "requires_identity_matcher_replay": True,
+                        "requires_promotion_gate": True,
+                    },
+                    "notes": [],
+                }),
+                model="minicpm-v:latest",
+                error=None,
+                total_duration=1.5,
+                host="http://34.87.90.216:11434",
+                input_tokens=12,
+                output_tokens=80,
+            )
+
+    monkeypatch.setattr(service, "OllamaService", FakeOllama)
+    payload = service.run_pixelrag_ollama_vlm_replay_worker(
+        artifact_root=tmp_path,
+        output_root=tmp_path / "receipts",
+        platform="shopee_tw",
+        execute=True,
+        write_receipt=True,
+        tile_limit=1,
+        auto_select_model=False,
+    )
+
+    assert payload["status"] == "ok"
+    assert payload["summary"]["executed_count"] == 1
+    assert payload["summary"]["executed_ok_count"] == 1
+    assert payload["summary"]["receipt_written_count"] == 1
+    assert payload["summary"]["writes_database_count"] == 0
+    item = payload["worker_items"][0]
+    assert item["worker_status"] == "executed_ok"
+    assert item["provider"] == "gcp_ollama"
+    assert item["validation"]["valid_for_identity_matcher_replay"] is True
+    receipt_path = tmp_path / "receipts" / "shopee_tw" / "shopee-ok" / "vlm_replay_receipt.json"
+    assert receipt_path.exists()
+    receipt = json.loads(receipt_path.read_text(encoding="utf-8"))
+    assert receipt["parsed_output"]["fields"]["title"]["value"] == "防曬乳 SPF50"
+    assert receipt["artifact_write_performed"] is True
+    assert receipt["receipt_path"] == str(receipt_path)
+
+
+def test_pixelrag_vlm_replay_worker_writes_model_error_receipt(tmp_path, monkeypatch):
+    from services import pixelrag_vlm_replay_worker_service as service
+
+    _write_receipt(
+        tmp_path,
+        platform="shopee_tw",
+        manifest_id="shopee-ok",
+        title="Shopee 防曬乳",
+        url="https://shopee.tw/search?keyword=sunscreen",
+    )
+
+    class FakeOllama:
+        def __init__(self, model):
+            self.model = model
+
+        def generate(self, *args, **kwargs):
+            return SimpleNamespace(
+                success=False,
+                content="",
+                model="minicpm-v:latest",
+                error="all hosts failed",
+                total_duration=None,
+                host="http://34.21.145.224:11434",
+                input_tokens=0,
+                output_tokens=0,
+            )
+
+    monkeypatch.setattr(service, "OllamaService", FakeOllama)
+    payload = service.run_pixelrag_ollama_vlm_replay_worker(
+        artifact_root=tmp_path,
+        output_root=tmp_path / "receipts",
+        platform="shopee_tw",
+        execute=True,
+        write_receipt=True,
+        tile_limit=1,
+        auto_select_model=False,
+    )
+
+    assert payload["status"] == "critical"
+    assert payload["summary"]["model_error_count"] == 1
+    assert payload["summary"]["receipt_written_count"] == 1
+    assert payload["summary"]["artifact_write_performed"] is True
+    assert payload["controlled_apply"]["writes_database"] is False
+    receipt_path = tmp_path / "receipts" / "shopee_tw" / "shopee-ok" / "vlm_replay_receipt.json"
+    receipt = json.loads(receipt_path.read_text(encoding="utf-8"))
+    assert receipt["worker_status"] == "model_error"
+    assert receipt["artifact_write_performed"] is True
+    assert receipt["receipt_path"] == str(receipt_path)
+    assert receipt["next_machine_action"] == "repair_ollama_vlm_runtime_or_model_route"
+
+
+def test_pixelrag_vlm_replay_worker_empty_fields_routes_to_platform_probe(tmp_path, monkeypatch):
+    from services import pixelrag_vlm_replay_worker_service as service
+
+    _write_receipt(
+        tmp_path,
+        platform="shopee_tw",
+        manifest_id="shopee-language",
+        title="蝦皮購物 | 花得更少買得更好",
+        url="https://shopee.tw/search?keyword=sunscreen",
+    )
+
+    class FakeOllama:
+        def __init__(self, model):
+            self.model = model
+
+        def generate(self, *args, **kwargs):
+            return SimpleNamespace(
+                success=True,
+                content=json.dumps({
+                    "blocked_page_detected": False,
+                    "fields": {
+                        "title": {"value": None, "confidence": 0.0, "evidence_refs": []},
+                        "price": {"value": None, "confidence": 0.0, "evidence_refs": []},
+                        "currency": {"value": None, "confidence": 0.0, "evidence_refs": []},
+                    },
+                    "quality": {
+                        "overall_confidence": 0.0,
+                        "missing_required_fields": [],
+                        "requires_identity_matcher_replay": True,
+                        "requires_promotion_gate": True,
+                    },
+                    "notes": [],
+                }),
+                model="minicpm-v:latest",
+                error=None,
+                total_duration=2.0,
+                host="http://192.168.0.111:11434",
+                input_tokens=10,
+                output_tokens=40,
+            )
+
+    monkeypatch.setattr(service, "OllamaService", FakeOllama)
+    payload = service.run_pixelrag_ollama_vlm_replay_worker(
+        artifact_root=tmp_path,
+        output_root=tmp_path / "receipts",
+        platform="shopee_tw",
+        execute=True,
+        write_receipt=True,
+        tile_limit=4,
+        auto_select_model=False,
+    )
+
+    assert payload["status"] == "warning"
+    assert payload["summary"]["executed_warning_count"] == 1
+    assert payload["summary"]["required_field_missing_count"] == 2
+    assert payload["next_machine_action"] == "run_platform_probe_or_use_structured_api"
+    item = payload["worker_items"][0]
+    assert item["worker_status"] == "executed_warning"
+    assert item["next_machine_action"] == "run_platform_probe_or_use_structured_api"
+    assert item["validation"]["present_field_count"] == 0
+    assert item["validation"]["non_product_or_interstitial_detected"] is True
+    assert item["validation"]["valid_for_identity_matcher_replay"] is False
+    receipt_path = tmp_path / "receipts" / "shopee_tw" / "shopee-language" / "vlm_replay_receipt.json"
+    receipt = json.loads(receipt_path.read_text(encoding="utf-8"))
+    assert receipt["next_machine_action"] == "run_platform_probe_or_use_structured_api"
+
+
+def test_pixelrag_vlm_replay_worker_language_note_routes_to_platform_probe(tmp_path, monkeypatch):
+    from services import pixelrag_vlm_replay_worker_service as service
+
+    _write_receipt(
+        tmp_path,
+        platform="shopee_tw",
+        manifest_id="shopee-language-note",
+        title="蝦皮購物 | 花得更少買得更好",
+        url="https://shopee.tw/search?keyword=sunscreen",
+    )
+
+    class FakeOllama:
+        def __init__(self, model):
+            self.model = model
+
+        def generate(self, *args, **kwargs):
+            return SimpleNamespace(
+                success=True,
+                content=json.dumps({
+                    "blocked_page_detected": False,
+                    "fields": {
+                        "title": {
+                            "value": "蝦皮購物 | 花得更少買得更好",
+                            "confidence": 0.95,
+                            "evidence_refs": ["tile:1"],
+                        },
+                        "price": {"value": None, "confidence": 0.0, "evidence_refs": []},
+                        "shop": {"value": None, "confidence": 0.0, "evidence_refs": []},
+                    },
+                    "quality": {
+                        "overall_confidence": 0.5,
+                        "missing_required_fields": ["title", "price", "shop"],
+                        "requires_identity_matcher_replay": True,
+                        "requires_promotion_gate": False,
+                    },
+                    "notes": [{"note": "Language selection page"}],
+                }),
+                model="minicpm-v:latest",
+                error=None,
+                total_duration=2.0,
+                host="http://192.168.0.111:11434",
+                input_tokens=10,
+                output_tokens=40,
+            )
+
+    monkeypatch.setattr(service, "OllamaService", FakeOllama)
+    payload = service.run_pixelrag_ollama_vlm_replay_worker(
+        artifact_root=tmp_path,
+        output_root=tmp_path / "receipts",
+        platform="shopee_tw",
+        execute=True,
+        write_receipt=True,
+        tile_limit=4,
+        auto_select_model=False,
+    )
+
+    assert payload["next_machine_action"] == "run_platform_probe_or_use_structured_api"
+    item = payload["worker_items"][0]
+    assert item["next_machine_action"] == "run_platform_probe_or_use_structured_api"
+    assert item["validation"]["present_field_count"] == 1
+    assert item["validation"]["interstitial_signal_detected"] is True
+    assert item["validation"]["generic_marketplace_title_detected"] is True
+    assert item["validation"]["non_product_or_interstitial_detected"] is True
+
+
+def test_pixelrag_vlm_replay_worker_auto_selects_installed_candidate(tmp_path, monkeypatch):
+    from services import pixelrag_vlm_replay_worker_service as service
+
+    _write_receipt(
+        tmp_path,
+        platform="shopee_tw",
+        manifest_id="shopee-ok",
+        title="Shopee 防曬乳",
+        url="https://shopee.tw/search?keyword=sunscreen",
+    )
+
+    monkeypatch.setattr(
+        service,
+        "build_pixelrag_vlm_route_readiness",
+        lambda **kwargs: {
+            "policy": "read_only_pixelrag_vlm_route_readiness_v1",
+            "status": "warning",
+            "summary": {
+                "configured_model": "minicpm-v:latest",
+                "candidate_model": "gemma3:4b",
+                "candidate_host": "http://34.21.145.224:11434",
+                "model_route_ready": True,
+            },
+            "next_machine_action": "run_pixelrag_vlm_replay_worker_execute_with_selected_model",
+        },
+    )
+
+    def fake_generate_exact_host(prompt, **kwargs):
+        assert kwargs["host"] == "http://34.21.145.224:11434"
+        assert kwargs["model"] == "gemma3:4b"
+        return SimpleNamespace(
+            success=True,
+            content=json.dumps({
+                "blocked_page_detected": False,
+                "fields": {
+                    "title": {
+                        "value": "防曬乳 SPF50",
+                        "confidence": 0.92,
+                        "evidence_refs": ["tile:1"],
+                    },
+                    "price": {
+                        "value": "399",
+                        "confidence": 0.88,
+                        "evidence_refs": ["tile:1"],
+                    },
+                },
+                "quality": {
+                    "overall_confidence": 0.90,
+                    "missing_required_fields": [],
+                    "requires_identity_matcher_replay": True,
+                    "requires_promotion_gate": True,
+                },
+            }),
+            model="gemma3:4b",
+            error=None,
+            total_duration=1.0,
+            host="http://34.21.145.224:11434",
+            input_tokens=10,
+            output_tokens=50,
+        )
+
+    monkeypatch.setattr(service, "_generate_exact_host", fake_generate_exact_host)
+    payload = service.run_pixelrag_ollama_vlm_replay_worker(
+        artifact_root=tmp_path,
+        platform="shopee_tw",
+        model="minicpm-v:latest",
+        execute=True,
+        tile_limit=1,
+        auto_select_model=True,
+    )
+
+    assert payload["model"] == "gemma3:4b"
+    assert payload["configured_model"] == "minicpm-v:latest"
+    assert payload["route_candidate_host"] == "http://34.21.145.224:11434"
+    assert payload["route_readiness"]["summary"]["candidate_model"] == "gemma3:4b"
+    assert payload["worker_items"][0]["route_candidate_host"] == "http://34.21.145.224:11434"
+    assert payload["summary"]["executed_ok_count"] == 1
+
+
+def test_pixelrag_vlm_replay_worker_generate_probe_blocks_tile_generate(tmp_path, monkeypatch):
+    from services import pixelrag_vlm_replay_worker_service as service
+
+    _write_receipt(
+        tmp_path,
+        platform="shopee_tw",
+        manifest_id="shopee-ok",
+        title="Shopee 防曬乳",
+        url="https://shopee.tw/search?keyword=sunscreen",
+    )
+
+    monkeypatch.setattr(
+        service,
+        "build_pixelrag_vlm_route_readiness",
+        lambda **kwargs: {
+            "policy": "read_only_pixelrag_vlm_route_readiness_v1",
+            "status": "critical",
+            "summary": {
+                "configured_model": "minicpm-v:latest",
+                "candidate_model": "gemma3:4b",
+                "candidate_host": "http://34.21.145.224:11434",
+                "tag_model_route_ready": True,
+                "model_route_ready": False,
+                "generate_probe_performed": True,
+                "generate_probe_ok_count": 0,
+                "generate_route_ready": False,
+                "generate_ready_model": "",
+                "generate_ready_host": "",
+                "generate_ready_provider": "",
+                "model_call_performed": True,
+            },
+            "controlled_apply": {"model_call": True},
+            "next_machine_action": "repair_ollama_vlm_generate_runtime_or_proxy_timeout",
+        },
+    )
+
+    class FakeOllama:
+        def __init__(self, model):
+            self.model = model
+
+        def generate(self, *args, **kwargs):
+            raise AssertionError("tile VLM generate should not be called")
+
+    monkeypatch.setattr(service, "OllamaService", FakeOllama)
+    payload = service.run_pixelrag_ollama_vlm_replay_worker(
+        artifact_root=tmp_path,
+        output_root=tmp_path / "receipts",
+        platform="shopee_tw",
+        model="minicpm-v:latest",
+        execute=True,
+        write_receipt=True,
+        tile_limit=1,
+        auto_select_model=True,
+        probe_generate_before_execute=True,
+    )
+
+    assert payload["status"] == "critical"
+    assert payload["model"] == "gemma3:4b"
+    assert payload["summary"]["model_route_not_ready_count"] == 1
+    assert payload["summary"]["route_model_call_performed"] is True
+    assert payload["summary"]["tile_model_call_performed"] is False
+    assert payload["summary"]["model_call_performed"] is True
+    assert payload["controlled_apply"]["model_call"] is True
+    assert payload["next_machine_action"] == "repair_ollama_vlm_generate_runtime_or_proxy_timeout"
+    item = payload["worker_items"][0]
+    assert item["worker_status"] == "model_route_not_ready"
+    assert item["generate_probe_performed"] is True
+    assert item["generate_route_ready"] is False
+    assert item["model_call_performed"] is False
+    receipt_path = tmp_path / "receipts" / "shopee_tw" / "shopee-ok" / "vlm_replay_receipt.json"
+    assert receipt_path.exists()
+    receipt = json.loads(receipt_path.read_text(encoding="utf-8"))
+    assert receipt["worker_status"] == "model_route_not_ready"
+    assert receipt["artifact_write_performed"] is True
+    assert receipt["generate_route_ready"] is False
+
+
+def test_pixelrag_vlm_replay_worker_cli_outputs_machine_readable_json(tmp_path):
+    _write_receipt(
+        tmp_path,
+        platform="shopee_tw",
+        manifest_id="shopee-ok",
+        title="Shopee 防曬乳",
+        url="https://shopee.tw/search?keyword=sunscreen",
+    )
+
+    completed = subprocess.run(
+        [
+            sys.executable,
+            "scripts/ops/run_pixelrag_vlm_replay_worker.py",
+            "--artifact-root",
+            str(tmp_path),
+            "--platform",
+            "shopee_tw",
+        ],
+        capture_output=True,
+        check=False,
+        text=True,
+    )
+
+    assert completed.returncode == 0
+    payload = json.loads(completed.stdout)
+    assert payload["summary"]["receipt_count"] == 1
+    assert payload["summary"]["dry_run_count"] == 1
+    assert payload["controlled_apply"]["model_call"] is False
+    assert payload["controlled_apply"]["writes_database"] is False
+
+
+def test_pixelrag_vlm_replay_worker_route_returns_readback(tmp_path, monkeypatch):
+    from flask import Flask
+    from routes import system_public_routes as routes
+    from services import pixelrag_vlm_replay_worker_service as service
+
+    _write_receipt(
+        tmp_path,
+        platform="shopee_tw",
+        manifest_id="shopee-ok",
+        title="Shopee 防曬乳",
+        url="https://shopee.tw/search?keyword=sunscreen",
+    )
+    monkeypatch.setattr(service, "DEFAULT_ARTIFACT_ROOT", str(tmp_path))
+
+    app = Flask(__name__)
+    with app.test_request_context(
+        "/api/ai-automation/pixelrag-vlm-replay-worker?platform=shopee_tw"
+    ):
+        response = routes.ai_automation_pixelrag_vlm_replay_worker_api.__wrapped__()
+    payload = response.get_json()
+
+    assert payload["policy"] == "controlled_pixelrag_ollama_vlm_replay_worker_v1"
+    assert payload["summary"]["receipt_count"] == 1
+    assert payload["summary"]["dry_run_count"] == 1
+    assert payload["controlled_apply"]["model_call"] is False

tests/test_pixelrag_vlm_route_readiness_service.py

@@ -0,0 +1,175 @@
+import json
+
+
+class FakeResponse:
+    def __init__(self, status_code=200, payload=None, text=""):
+        self.status_code = status_code
+        self._payload = payload or {}
+        self.text = text
+
+    def json(self):
+        return self._payload
+
+
+def test_pixelrag_vlm_route_readiness_selects_installed_candidate(monkeypatch):
+    from services import pixelrag_vlm_route_readiness_service as service
+
+    hosts = [
+        "http://primary:11434",
+        "http://secondary:11434",
+    ]
+    monkeypatch.setattr(service, "_approved_hosts", lambda: hosts)
+
+    def fake_get(url, timeout):
+        if url.startswith("http://primary"):
+            raise TimeoutError("primary timeout")
+        return FakeResponse(
+            payload={
+                "models": [
+                    {"name": "gemma3:4b"},
+                    {"name": "bge-m3:latest"},
+                ]
+            }
+        )
+
+    monkeypatch.setattr(service.requests, "get", fake_get)
+    payload = service.build_pixelrag_vlm_route_readiness(
+        model="minicpm-v:latest",
+        timeout_seconds=1,
+    )
+
+    assert payload["policy"] == "read_only_pixelrag_vlm_route_readiness_v1"
+    assert payload["status"] == "warning"
+    assert payload["summary"]["reachable_host_count"] == 1
+    assert payload["summary"]["configured_model_available_count"] == 0
+    assert payload["summary"]["candidate_model"] == "gemma3:4b"
+    assert payload["summary"]["model_route_ready"] is True
+    assert payload["controlled_apply"]["model_call"] is False
+    assert payload["controlled_apply"]["writes_database"] is False
+    assert payload["next_machine_action"] == "run_pixelrag_vlm_replay_worker_execute_with_selected_model"
+
+
+def test_pixelrag_vlm_route_readiness_critical_when_no_candidate(monkeypatch):
+    from services import pixelrag_vlm_route_readiness_service as service
+
+    monkeypatch.setattr(service, "_approved_hosts", lambda: ["http://secondary:11434"])
+    monkeypatch.setattr(
+        service.requests,
+        "get",
+        lambda url, timeout: FakeResponse(payload={"models": [{"name": "bge-m3:latest"}]}),
+    )
+
+    payload = service.build_pixelrag_vlm_route_readiness(model="minicpm-v:latest")
+
+    assert payload["status"] == "critical"
+    assert payload["success"] is False
+    assert payload["summary"]["model_route_ready"] is False
+    assert payload["next_machine_action"] == "install_or_configure_pixelrag_vlm_model_on_approved_ollama_host"
+
+
+def test_pixelrag_vlm_route_readiness_generate_probe_success(monkeypatch):
+    from services import pixelrag_vlm_route_readiness_service as service
+
+    monkeypatch.setattr(service, "_approved_hosts", lambda: ["http://secondary:11434"])
+    monkeypatch.setattr(
+        service.requests,
+        "get",
+        lambda url, timeout: FakeResponse(payload={"models": [{"name": "gemma3:4b"}]}),
+    )
+    monkeypatch.setattr(
+        service.requests,
+        "post",
+        lambda url, json, timeout: FakeResponse(payload={"response": "OK", "done": True}),
+    )
+
+    payload = service.build_pixelrag_vlm_route_readiness(
+        model="minicpm-v:latest",
+        probe_generate=True,
+        probe_timeout_seconds=1,
+    )
+
+    assert payload["status"] == "warning"
+    assert payload["summary"]["candidate_model"] == "gemma3:4b"
+    assert payload["summary"]["tag_model_route_ready"] is True
+    assert payload["summary"]["model_route_ready"] is True
+    assert payload["summary"]["generate_probe_performed"] is True
+    assert payload["summary"]["generate_probe_ok_count"] == 1
+    assert payload["summary"]["generate_route_ready"] is True
+    assert payload["summary"]["generate_ready_host"] == "http://secondary:11434"
+    assert payload["controlled_apply"]["model_call"] is True
+
+
+def test_pixelrag_vlm_route_readiness_generate_probe_failure_blocks_route(monkeypatch):
+    from services import pixelrag_vlm_route_readiness_service as service
+
+    monkeypatch.setattr(service, "_approved_hosts", lambda: ["http://secondary:11434"])
+    monkeypatch.setattr(
+        service.requests,
+        "get",
+        lambda url, timeout: FakeResponse(payload={"models": [{"name": "gemma3:4b"}]}),
+    )
+
+    def fake_post(url, json, timeout):
+        raise TimeoutError("generate timeout")
+
+    monkeypatch.setattr(service.requests, "post", fake_post)
+
+    payload = service.build_pixelrag_vlm_route_readiness(
+        model="minicpm-v:latest",
+        probe_generate=True,
+        probe_timeout_seconds=1,
+    )
+
+    assert payload["status"] == "critical"
+    assert payload["success"] is False
+    assert payload["summary"]["candidate_model"] == "gemma3:4b"
+    assert payload["summary"]["tag_model_route_ready"] is True
+    assert payload["summary"]["model_route_ready"] is False
+    assert payload["summary"]["generate_probe_performed"] is True
+    assert payload["summary"]["generate_probe_ok_count"] == 0
+    assert payload["summary"]["generate_route_ready"] is False
+    assert payload["next_machine_action"] == "repair_ollama_vlm_generate_runtime_or_proxy_timeout"
+
+
+def test_pixelrag_vlm_route_readiness_route_returns_readback(monkeypatch):
+    from flask import Flask
+    from routes import system_public_routes as routes
+    from services import pixelrag_vlm_route_readiness_service as service
+
+    monkeypatch.setattr(
+        service,
+        "build_pixelrag_vlm_route_readiness",
+        lambda **kwargs: {
+            "success": True,
+            "policy": "read_only_pixelrag_vlm_route_readiness_v1",
+            "status": "ok",
+            "summary": {"model_route_ready": True, "candidate_model": "gemma3:4b"},
+            "next_machine_action": "run_pixelrag_vlm_replay_worker_execute",
+        },
+    )
+
+    app = Flask(__name__)
+    with app.test_request_context(
+        "/api/ai-automation/pixelrag-vlm-route-readiness?model=minicpm-v:latest"
+        "&probe_generate=true&probe_timeout=4"
+    ):
+        response = routes.ai_automation_pixelrag_vlm_route_readiness_api.__wrapped__()
+    payload = response.get_json()
+
+    assert payload["policy"] == "read_only_pixelrag_vlm_route_readiness_v1"
+    assert payload["summary"]["candidate_model"] == "gemma3:4b"
+
+
+def test_pixelrag_vlm_route_readiness_cli_outputs_json(monkeypatch, tmp_path):
+    from services import pixelrag_vlm_route_readiness_service as service
+
+    monkeypatch.setattr(service, "_approved_hosts", lambda: ["http://secondary:11434"])
+    monkeypatch.setattr(
+        service.requests,
+        "get",
+        lambda url, timeout: FakeResponse(payload={"models": [{"name": "gemma3:4b"}]}),
+    )
+
+    payload = service.build_pixelrag_vlm_route_readiness(model="minicpm-v:latest")
+    assert json.dumps(payload, ensure_ascii=False)
+    assert payload["summary"]["candidate_model"] == "gemma3:4b"