feat(crawler): add pixelrag visual evidence lane

docs/guides/browse_sh_crawler_playbook.md

@@ -51,3 +51,39 @@ python scripts/tools/browse_sh_probe.py -- screenshot
 - PChome：目前已有搜尋 API 與商品 API，`browse` 只用於確認 API 參數、分頁行為、前端是否切新 endpoint。
 - MOMO：若既有 BeautifulSoup selector 失效，先用 `browse` 找出前端實際 XHR；找到 API 時優先改成 structured API parser。
 - Matcher：`browse` 只提供候選證據；是否為同款仍由 `marketplace_product_matcher.score_marketplace_match()` 決定。
+
+## PixelRAG-style 視覺證據 fallback
+
+2026-07-09 評估 PixelRAG 後，結論是「可導入，但不可直接取代正式爬蟲」。PixelRAG 的核心價值是把渲染後頁面截圖切成 tiles，讓 AI 讀到 HTML parser 可能丟失的視覺結構；本專案第一階段只採用這個視覺證據思路，不直接拉外部 runtime、不用外部 embedding API、不把像素結果寫入正式價格表。
+
+導入順序：
+
+1. 既有 PChome API / MOMO structured parser 先跑；若 parser 結果為空、價格欄位缺失、規格/組合資訊疑似只存在渲染畫面，才啟動視覺證據 fallback。
+2. Phase 1 只輸出 screenshot/tile manifest：URL、platform、crawler、parse failure、viewport、tile coordinates、artifact path、evidence intent。
+3. Phase 2 用 manifest 收集 replay samples，對照 `marketplace_product_matcher` 與 `competitor_match_attempts` 的診斷結果，評估是否提高身份證據覆蓋率。
+4. Phase 3 才評估 Ollama-first multimodal embedding；未完成 GCP-A → GCP-B → 111 視覺 embedding 驗證前，不做自動像素檢索。
+5. Phase 4 若需要索引，優先設計 pgvector-compatible evidence metadata；FAISS 只能先當本地研究/ADR 候選，不直接進 production。
+6. Phase 5 才談 crawler fusion；正式 `competitor_prices` / `competitor_price_history` 寫入仍需 matcher replay/canary 證據。
+
+機器可讀評估：
+
+```bash
+python scripts/ops/report_pixelrag_crawler_integration.py
+python scripts/ops/report_pixelrag_crawler_integration.py --platform momo
+python scripts/ops/report_pixelrag_crawler_integration.py --capability ollama_multimodal_ready --capability pgvector_visual_ready
+python scripts/ops/report_pixelrag_crawler_integration.py \
+  --platform momo \
+  --manifest-url "https://m.momoshop.com.tw/search.momo?searchKeyword=test" \
+  --crawler MomoCrawler.search_products \
+  --trigger-reason parser_empty \
+  --page-size page=1440x1900 \
+  --tile-size tile=512x512
+```
+
+安全邊界：
+
+- read-only；不登入、不下單、不加入購物車、不寫第三方狀態。
+- 不從像素結果直接寫正式價格或同款判斷，只寫 artifact / review diagnostics。
+- 不使用 GitHub runtime 依賴；不讀 secrets / sessions / cookies。
+- 不呼叫 hosted embedding / VLM API；如需 embedding，必須走 Ollama-first 並保留成本/品質 benchmark。
+- 若穩定 XHR/API 可取得同樣資訊，仍優先回補 structured parser，不把視覺 fallback 變成主路徑。

scripts/ops/report_pixelrag_crawler_integration.py

@@ -0,0 +1,122 @@
+#!/usr/bin/env python3
+"""Report PixelRAG-style crawler integration 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_crawler_integration_service import (  # noqa: E402
+    build_pixelrag_crawler_integration_assessment,
+    build_pixelrag_visual_evidence_manifest,
+)
+
+
+CAPABILITY_FLAGS = {
+    "no_playwright": ("playwright_artifact_pipeline", False),
+    "ollama_multimodal_ready": ("ollama_multimodal_embedding_ready", True),
+    "pgvector_visual_ready": ("pgvector_visual_index_ready", True),
+    "faiss_allowed": ("faiss_allowed_in_production", True),
+}
+
+
+def _size_arg(value: str) -> dict[str, int | str]:
+    try:
+        name, size = value.split("=", 1)
+    except ValueError:
+        name, size = "", value
+    try:
+        width_text, height_text = size.lower().split("x", 1)
+        width = int(width_text)
+        height = int(height_text)
+    except (ValueError, TypeError):
+        raise argparse.ArgumentTypeError("size must be name=WIDTHxHEIGHT or WIDTHxHEIGHT")
+    payload: dict[str, int | str] = {"width": width, "height": height}
+    if name:
+        payload["name"] = name
+    return payload
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(
+        description="輸出 PixelRAG 導入現有爬蟲解決方案的機器可讀評估。"
+    )
+    parser.add_argument(
+        "--platform",
+        action="append",
+        dest="platforms",
+        help="限制目標平台，可重複指定；預設 momo + pchome。",
+    )
+    parser.add_argument(
+        "--capability",
+        action="append",
+        choices=sorted(CAPABILITY_FLAGS),
+        default=[],
+        help="覆寫 readiness 假設，用於模擬下一階段條件。",
+    )
+    parser.add_argument("--manifest-url", help="輸出單一 URL 的視覺證據 manifest。")
+    parser.add_argument("--crawler", default="crawler_diagnostics", help="manifest 來源 crawler 名稱。")
+    parser.add_argument(
+        "--trigger-reason",
+        default="parser_empty_or_low_confidence",
+        help="啟動視覺 fallback 的原因。",
+    )
+    parser.add_argument(
+        "--evidence-intent",
+        default="recover_visual_offer_evidence",
+        help="視覺證據用途。",
+    )
+    parser.add_argument(
+        "--viewport",
+        type=_size_arg,
+        help="viewport，例如 desktop-1440=1440x950。",
+    )
+    parser.add_argument(
+        "--page-size",
+        type=_size_arg,
+        help="預估頁面尺寸，例如 page=1440x1900。",
+    )
+    parser.add_argument(
+        "--tile-size",
+        type=_size_arg,
+        help="tile 尺寸，例如 tile=512x512。",
+    )
+    args = parser.parse_args()
+
+    if args.manifest_url:
+        platform = (args.platforms or ["momo"])[0]
+        payload = build_pixelrag_visual_evidence_manifest(
+            url=args.manifest_url,
+            platform=platform,
+            crawler=args.crawler,
+            trigger_reason=args.trigger_reason,
+            evidence_intent=args.evidence_intent,
+            viewport=args.viewport,
+            page_size=args.page_size,
+            tile_size=args.tile_size,
+        )
+        print(json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True))
+        return 0 if payload.get("success") else 1
+
+    capabilities = {
+        key: value
+        for flag in args.capability
+        for key, value in [CAPABILITY_FLAGS[flag]]
+    }
+    payload = build_pixelrag_crawler_integration_assessment(
+        capabilities=capabilities or None,
+        target_platforms=tuple(args.platforms) if args.platforms else None,
+    )
+    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/pixelrag_crawler_integration_service.py

@@ -0,0 +1,503 @@
+"""PixelRAG-style visual evidence lane for crawler diagnostics.
+
+This module does not call external services, read credentials, or write DB data.
+It turns the PixelRAG research pattern into a controlled, machine-readable
+integration assessment for the existing MOMO/PChome crawler stack.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import math
+from copy import deepcopy
+from datetime import datetime, timezone
+from typing import Any, Mapping
+from urllib.parse import urlparse
+
+
+POLICY = "read_only_pixelrag_crawler_integration_assessment_v1"
+MANIFEST_POLICY = "read_only_pixelrag_visual_evidence_manifest_v1"
+RESULT_PHASE1_READY = "PIXELRAG_VISUAL_EVIDENCE_PHASE1_READY"
+RESULT_BLOCKED_NO_CAPTURE = "PIXELRAG_BLOCKED_NO_VISUAL_CAPTURE"
+RESULT_MANIFEST_READY = "PIXELRAG_VISUAL_EVIDENCE_MANIFEST_READY"
+RESULT_MANIFEST_REJECTED = "PIXELRAG_VISUAL_EVIDENCE_MANIFEST_REJECTED"
+
+ALLOWED_PLATFORMS = ("momo", "pchome", "market_intel", "external_market")
+DEFAULT_VIEWPORT = {"name": "desktop-1440", "width": 1440, "height": 950}
+DEFAULT_TILE_SIZE = {"width": 512, "height": 512}
+VISUAL_FALLBACK_CONFIDENCE_TRIGGERS = {
+    "low",
+    "manual_review",
+    "identity_review",
+    "true_low_confidence",
+    "variant_selection_review",
+}
+
+DEFAULT_CAPABILITIES: dict[str, bool] = {
+    "structured_crawler_api": True,
+    "momo_html_parser_fallback": True,
+    "matcher_guardrails": True,
+    "playwright_artifact_pipeline": True,
+    "browse_sh_optional_probe": True,
+    "ollama_multimodal_embedding_ready": False,
+    "pgvector_visual_index_ready": False,
+    "faiss_allowed_in_production": False,
+    "production_price_auto_write": False,
+}
+
+TARGET_PLATFORMS = ("momo", "pchome")
+
+SOURCE_FINDINGS: tuple[dict[str, str], ...] = (
+    {
+        "code": "pixel_native_retrieval",
+        "finding": (
+            "PixelRAG represents web pages as screenshots, retrieves screenshot "
+            "tiles, and lets a vision-language model read the visual evidence."
+        ),
+        "adoption": "Use as visual evidence fallback after structured parsers fail.",
+    },
+    {
+        "code": "playwright_screenshot_tiles",
+        "finding": (
+            "The research pipeline starts with Playwright rendering, screenshot "
+            "capture, and image tiling."
+        ),
+        "adoption": "Ready for Phase 1 because this repo already has Playwright artifact patterns.",
+    },
+    {
+        "code": "qwen3_vl_embedding_optional",
+        "finding": (
+            "Full pixel-space retrieval needs a multimodal embedding model such as "
+            "Qwen3-VL-Embedding."
+        ),
+        "adoption": "Deferred until Ollama-first multimodal hosting is verified.",
+    },
+    {
+        "code": "faiss_index_policy_gap",
+        "finding": "The paper uses FAISS for visual retrieval indexes.",
+        "adoption": "Do not add FAISS to production before pgvector/ADR review.",
+    },
+)
+
+SAFETY_GUARDS: tuple[dict[str, Any], ...] = (
+    {
+        "code": "read_only_visual_capture",
+        "status": "enforced",
+        "reason": "Screenshots and tile manifests are diagnostic evidence only.",
+    },
+    {
+        "code": "no_price_write_from_pixels",
+        "status": "enforced",
+        "reason": (
+            "Pixel evidence cannot directly write competitor_prices or "
+            "competitor_price_history."
+        ),
+    },
+    {
+        "code": "no_external_embedding_api",
+        "status": "enforced",
+        "reason": "Embedding must stay Ollama-first and cannot use hosted visual APIs by default.",
+    },
+    {
+        "code": "no_github_runtime_dependency",
+        "status": "enforced",
+        "reason": "The integration is derived from public research notes, not GitHub code.",
+    },
+    {
+        "code": "polite_crawler_boundary",
+        "status": "enforced",
+        "reason": "Visual capture must respect delay, target allowlists, and read-only behavior.",
+    },
+)
+
+
+def _merge_capabilities(overrides: Mapping[str, Any] | None) -> dict[str, bool]:
+    capabilities = dict(DEFAULT_CAPABILITIES)
+    for key, value in (overrides or {}).items():
+        if key in capabilities:
+            capabilities[key] = bool(value)
+    return capabilities
+
+
+def _phase_status_counts(phases: list[dict[str, Any]]) -> dict[str, int]:
+    counts: dict[str, int] = {}
+    for phase in phases:
+        status = str(phase.get("status") or "unknown")
+        counts[status] = counts.get(status, 0) + 1
+    return counts
+
+
+def _positive_int(value: Any, fallback: int) -> int:
+    try:
+        parsed = int(value)
+    except (TypeError, ValueError):
+        return fallback
+    return parsed if parsed > 0 else fallback
+
+
+def _dimension_payload(value: Mapping[str, Any] | None, fallback: Mapping[str, Any]) -> dict[str, Any]:
+    payload = dict(fallback)
+    for key, raw_value in (value or {}).items():
+        if key in {"width", "height"}:
+            payload[key] = _positive_int(raw_value, int(fallback[key]))
+        elif key == "name":
+            payload[key] = str(raw_value or fallback.get("name") or "").strip()
+    return payload
+
+
+def _controlled_apply_boundary() -> dict[str, Any]:
+    return {
+        "network_call": False,
+        "db_write": False,
+        "secret_read": False,
+        "github_dependency": False,
+        "production_price_write": False,
+        "rollback": "Disable the visual fallback selector or ignore generated artifacts.",
+    }
+
+
+def _build_phases(capabilities: Mapping[str, bool]) -> list[dict[str, Any]]:
+    capture_ready = bool(capabilities["playwright_artifact_pipeline"])
+    structured_ready = bool(capabilities["structured_crawler_api"])
+    guard_ready = bool(capabilities["matcher_guardrails"])
+    phase1_ready = capture_ready and structured_ready and guard_ready
+    embedding_ready = bool(capabilities["ollama_multimodal_embedding_ready"])
+    visual_index_ready = bool(capabilities["pgvector_visual_index_ready"])
+    faiss_allowed = bool(capabilities["faiss_allowed_in_production"])
+
+    return [
+        {
+            "id": "PXR-1",
+            "name": "Visual capture fallback selector",
+            "status": "ready_to_start" if phase1_ready else "blocked_prerequisite",
+            "integration_point": [
+                "MomoCrawler.search_products parser-empty fallback",
+                "PChomeCrawler search/detail parser anomaly fallback",
+                "market_intel.html_diagnostics low-confidence pages",
+            ],
+            "deliverable": "Emit a screenshot capture request when HTML/API evidence is missing.",
+            "writes": [],
+            "blocking_reason": "" if phase1_ready else "Needs structured crawler and Playwright artifact readiness.",
+        },
+        {
+            "id": "PXR-2",
+            "name": "Tile manifest artifact",
+            "status": "ready_to_start" if phase1_ready else "blocked_prerequisite",
+            "integration_point": [
+                "data/ai_automation visual artifacts",
+                "crawler diagnostics receipt",
+            ],
+            "deliverable": (
+                "Record URL, viewport, tile coordinates, source crawler, parse failure, "
+                "and evidence intent without model inference."
+            ),
+            "writes": ["artifact_file_only"],
+            "blocking_reason": "" if phase1_ready else "Phase 1 selector must be available first.",
+        },
+        {
+            "id": "PXR-3",
+            "name": "Ollama-first multimodal embedding benchmark",
+            "status": "ready_to_benchmark" if embedding_ready else "deferred_model_not_verified",
+            "integration_point": [
+                "Hermes embedding lane",
+                "Ollama GCP-A -> GCP-B -> 111 routing",
+            ],
+            "deliverable": "Benchmark local visual embeddings on saved tiles before enabling retrieval.",
+            "writes": ["benchmark_artifact"],
+            "blocking_reason": "" if embedding_ready else "No verified Ollama-hosted visual embedding model yet.",
+        },
+        {
+            "id": "PXR-4",
+            "name": "Visual retrieval index",
+            "status": (
+                "ready_to_design"
+                if visual_index_ready
+                else "deferred_index_policy"
+            ),
+            "integration_point": [
+                "pgvector-backed evidence index",
+                "AI knowledge retrieval",
+            ],
+            "deliverable": "Use pgvector-compatible visual evidence metadata before any FAISS adoption.",
+            "writes": ["design_artifact"],
+            "blocking_reason": (
+                ""
+                if visual_index_ready
+                else "Production vector policy is pgvector-first; FAISS requires ADR or local-only proof."
+            ),
+            "faiss_allowed_in_production": faiss_allowed,
+        },
+        {
+            "id": "PXR-5",
+            "name": "Crawler fusion and price-write guard",
+            "status": "deferred_replay_required",
+            "integration_point": [
+                "marketplace_product_matcher",
+                "competitor_match_attempts",
+                "manual/AI verification queue",
+            ],
+            "deliverable": (
+                "Fuse text/API evidence and visual evidence into review diagnostics; "
+                "do not auto-write formal price rows until replay proves confidence."
+            ),
+            "writes": ["review_diagnostics_only"],
+            "blocking_reason": "Needs replay/canary evidence before production price decisions.",
+        },
+    ]
+
+
+def should_emit_visual_evidence_fallback(
+    *,
+    parser_success: bool,
+    parsed_item_count: int = 0,
+    confidence_band: str = "",
+    missing_fields: tuple[str, ...] | list[str] | None = None,
+    failure_reason: str = "",
+) -> dict[str, Any]:
+    """Decide whether a crawler result should emit a visual evidence manifest."""
+    triggers: list[str] = []
+    missing = [str(field) for field in (missing_fields or []) if str(field).strip()]
+    normalized_confidence = str(confidence_band or "").strip().lower()
+    normalized_failure = str(failure_reason or "").strip().lower()
+
+    if not parser_success:
+        triggers.append("parser_failed")
+    if int(parsed_item_count or 0) <= 0:
+        triggers.append("parsed_empty")
+    if normalized_confidence in VISUAL_FALLBACK_CONFIDENCE_TRIGGERS:
+        triggers.append(f"confidence:{normalized_confidence}")
+    if any(field in {"price", "product_id", "title", "spec", "image_url"} for field in missing):
+        triggers.append("critical_field_missing")
+    if any(token in normalized_failure for token in ("html", "selector", "render", "visual", "price")):
+        triggers.append("failure_reason_visual_or_parser_related")
+
+    should_emit = bool(triggers)
+    return {
+        "should_emit": should_emit,
+        "triggers": triggers,
+        "fallback_reason": ", ".join(triggers) if should_emit else "structured_evidence_sufficient",
+        "missing_fields": missing,
+        "parser_success": bool(parser_success),
+        "parsed_item_count": int(parsed_item_count or 0),
+        "confidence_band": confidence_band,
+        "policy": MANIFEST_POLICY,
+    }
+
+
+def build_pixelrag_visual_evidence_manifest(
+    *,
+    url: str,
+    platform: str,
+    crawler: str,
+    trigger_reason: str,
+    evidence_intent: str = "recover_visual_offer_evidence",
+    viewport: Mapping[str, Any] | None = None,
+    page_size: Mapping[str, Any] | None = None,
+    tile_size: Mapping[str, Any] | None = None,
+) -> dict[str, Any]:
+    """Build a read-only visual evidence manifest without capturing the page."""
+    errors: list[str] = []
+    raw_url = str(url or "").strip()
+    parsed = urlparse(raw_url)
+    platform_code = str(platform or "").strip().lower()
+    crawler_name = str(crawler or "").strip()
+    reason = str(trigger_reason or "").strip()
+    intent = str(evidence_intent or "recover_visual_offer_evidence").strip()
+
+    if parsed.scheme not in {"http", "https"} or not parsed.netloc:
+        errors.append("URL must be an absolute http(s) URL.")
+    if parsed.username or parsed.password:
+        errors.append("URL credentials are not allowed in visual evidence manifests.")
+    if platform_code not in ALLOWED_PLATFORMS:
+        errors.append(f"platform must be one of: {', '.join(ALLOWED_PLATFORMS)}.")
+    if not crawler_name:
+        errors.append("crawler is required.")
+    if not reason:
+        errors.append("trigger_reason is required.")
+
+    boundary = _controlled_apply_boundary()
+    if errors:
+        return {
+            "success": False,
+            "policy": MANIFEST_POLICY,
+            "result": RESULT_MANIFEST_REJECTED,
+            "errors": errors,
+            "controlled_apply": boundary,
+        }
+
+    viewport_payload = _dimension_payload(viewport, DEFAULT_VIEWPORT)
+    tile_payload = _dimension_payload(tile_size, DEFAULT_TILE_SIZE)
+    page_payload = _dimension_payload(
+        page_size,
+        {
+            "name": "estimated-page",
+            "width": viewport_payload["width"],
+            "height": viewport_payload["height"],
+        },
+    )
+    tiles_x = max(1, math.ceil(page_payload["width"] / tile_payload["width"]))
+    tiles_y = max(1, math.ceil(page_payload["height"] / tile_payload["height"]))
+    tile_count = tiles_x * tiles_y
+    manifest_key = "|".join(
+        [
+            platform_code,
+            crawler_name,
+            raw_url,
+            reason,
+            str(viewport_payload["width"]),
+            str(viewport_payload["height"]),
+            str(tile_payload["width"]),
+            str(tile_payload["height"]),
+        ]
+    )
+    manifest_id = hashlib.sha256(manifest_key.encode("utf-8")).hexdigest()[:20]
+
+    return {
+        "success": True,
+        "policy": MANIFEST_POLICY,
+        "generated_at": datetime.now(timezone.utc).isoformat(),
+        "result": RESULT_MANIFEST_READY,
+        "manifest_id": manifest_id,
+        "status": "capture_requested",
+        "capture_target": {
+            "url": raw_url,
+            "platform": platform_code,
+            "crawler": crawler_name,
+            "trigger_reason": reason,
+            "evidence_intent": intent,
+        },
+        "viewport": viewport_payload,
+        "page_size": page_payload,
+        "tile_size": tile_payload,
+        "tile_plan": {
+            "tiles_x": tiles_x,
+            "tiles_y": tiles_y,
+            "tile_count": tile_count,
+            "overlap_px": 0,
+        },
+        "artifact": {
+            "kind": "pixelrag_visual_evidence_manifest",
+            "suggested_path": (
+                f"data/ai_automation/pixelrag_visual_evidence/"
+                f"{platform_code}/{manifest_id}.json"
+            ),
+            "writes": ["artifact_file_only"],
+        },
+        "safety_guards": deepcopy(list(SAFETY_GUARDS)),
+        "controlled_apply": boundary,
+        "next_action": {
+            "action": "capture_screenshot_and_tiles_from_manifest",
+            "status": "ready_for_capture_worker",
+            "reason": "Manifest is validated; capture worker can run under read-only crawler policy.",
+        },
+    }
+
+
+def build_pixelrag_crawler_integration_assessment(
+    *,
+    capabilities: Mapping[str, Any] | None = None,
+    target_platforms: tuple[str, ...] | list[str] | None = None,
+) -> dict[str, Any]:
+    """Build a safe integration assessment for PixelRAG-style crawler fallback."""
+    merged_capabilities = _merge_capabilities(capabilities)
+    platforms = tuple(target_platforms or TARGET_PLATFORMS)
+    phases = _build_phases(merged_capabilities)
+    phase_counts = _phase_status_counts(phases)
+
+    phase1_ready = all(
+        phase.get("status") == "ready_to_start"
+        for phase in phases[:2]
+    )
+    full_pixelrag_ready = all(
+        merged_capabilities[key]
+        for key in (
+            "playwright_artifact_pipeline",
+            "ollama_multimodal_embedding_ready",
+            "pgvector_visual_index_ready",
+        )
+    ) and not merged_capabilities["faiss_allowed_in_production"]
+
+    result = RESULT_PHASE1_READY if phase1_ready else RESULT_BLOCKED_NO_CAPTURE
+    recommended_mode = (
+        "pixelrag_inspired_visual_evidence_fallback"
+        if phase1_ready
+        else "prepare_visual_capture_prerequisites"
+    )
+
+    payload = {
+        "success": True,
+        "policy": POLICY,
+        "generated_at": datetime.now(timezone.utc).isoformat(),
+        "result": result,
+        "feasible": phase1_ready,
+        "can_start_now": phase1_ready,
+        "full_pixelrag_ready": full_pixelrag_ready,
+        "recommended_mode": recommended_mode,
+        "plain_assessment": (
+            "可導入，但第一階段應定位為爬蟲低信心時的視覺證據 fallback；"
+            "完整像素向量檢索需等 Ollama-first 視覺 embedding 與 pgvector/ADR 驗證。"
+            if phase1_ready
+            else "目前不應啟動，需先補齊 Playwright artifact 或 crawler guardrail 前置條件。"
+        ),
+        "target_platforms": list(platforms),
+        "capabilities": deepcopy(merged_capabilities),
+        "source_findings": deepcopy(list(SOURCE_FINDINGS)),
+        "crawler_fit": {
+            "best_for": [
+                "dynamic pages whose rendered price/spec block is lost by HTML parsing",
+                "visual tables, badges, bundle/spec cards, and image-heavy offer evidence",
+                "MOMO/PChome parser-empty or identity-evidence-missing diagnostics",
+            ],
+            "not_for": [
+                "stable public APIs that already return structured price and product IDs",
+                "bypassing robots, login walls, anti-bot controls, or rate limits",
+                "directly deciding exact product identity without matcher replay",
+            ],
+            "routing_rule": (
+                "Use API/structured parser first; if parser evidence is empty or low-confidence, "
+                "emit visual evidence artifact; matcher remains the authority for identity."
+            ),
+        },
+        "phases": phases,
+        "phase_status_counts": phase_counts,
+        "safety_guards": deepcopy(list(SAFETY_GUARDS)),
+        "controlled_apply": _controlled_apply_boundary(),
+        "next_actions": [
+            {
+                "priority": "P0",
+                "action": "add_visual_evidence_manifest_lane",
+                "status": "ready" if phase1_ready else "blocked_prerequisite",
+                "reason": (
+                    "Starts crawler automation without changing formal price truth."
+                    if phase1_ready
+                    else "Requires Playwright artifact readiness before capture manifests can start."
+                ),
+            },
+            {
+                "priority": "P1",
+                "action": "collect_replay_samples_from_parser_empty_cases",
+                "status": "ready_after_p0" if phase1_ready else "deferred",
+                "reason": "Builds evidence for whether visual capture improves coverage.",
+            },
+            {
+                "priority": "P2",
+                "action": "benchmark_ollama_multimodal_embedding",
+                "status": "deferred",
+                "reason": "Needed before real pixel retrieval; cannot call hosted APIs by default.",
+            },
+        ],
+    }
+    return payload
+
+
+__all__ = [
+    "MANIFEST_POLICY",
+    "POLICY",
+    "RESULT_PHASE1_READY",
+    "RESULT_BLOCKED_NO_CAPTURE",
+    "RESULT_MANIFEST_READY",
+    "RESULT_MANIFEST_REJECTED",
+    "build_pixelrag_crawler_integration_assessment",
+    "build_pixelrag_visual_evidence_manifest",
+    "should_emit_visual_evidence_fallback",
+]

tests/test_pixelrag_crawler_integration_service.py

@@ -0,0 +1,199 @@
+import json
+import subprocess
+import sys
+
+
+def test_pixelrag_assessment_starts_visual_evidence_fallback_not_price_write():
+    from services.pixelrag_crawler_integration_service import (
+        RESULT_PHASE1_READY,
+        build_pixelrag_crawler_integration_assessment,
+    )
+
+    payload = build_pixelrag_crawler_integration_assessment()
+
+    assert payload["success"] is True
+    assert payload["result"] == RESULT_PHASE1_READY
+    assert payload["feasible"] is True
+    assert payload["can_start_now"] is True
+    assert payload["recommended_mode"] == "pixelrag_inspired_visual_evidence_fallback"
+    assert payload["full_pixelrag_ready"] is False
+    assert payload["controlled_apply"] == {
+        "network_call": False,
+        "db_write": False,
+        "secret_read": False,
+        "github_dependency": False,
+        "production_price_write": False,
+        "rollback": "Disable the visual fallback selector or ignore generated artifacts.",
+    }
+
+    phase_ids = {phase["id"]: phase for phase in payload["phases"]}
+    assert phase_ids["PXR-1"]["status"] == "ready_to_start"
+    assert phase_ids["PXR-2"]["writes"] == ["artifact_file_only"]
+    assert phase_ids["PXR-3"]["status"] == "deferred_model_not_verified"
+    assert phase_ids["PXR-5"]["writes"] == ["review_diagnostics_only"]
+    assert any(
+        guard["code"] == "no_price_write_from_pixels"
+        for guard in payload["safety_guards"]
+    )
+
+
+def test_pixelrag_assessment_blocks_when_visual_capture_is_not_ready():
+    from services.pixelrag_crawler_integration_service import (
+        RESULT_BLOCKED_NO_CAPTURE,
+        build_pixelrag_crawler_integration_assessment,
+    )
+
+    payload = build_pixelrag_crawler_integration_assessment(
+        capabilities={"playwright_artifact_pipeline": False}
+    )
+
+    assert payload["result"] == RESULT_BLOCKED_NO_CAPTURE
+    assert payload["feasible"] is False
+    assert payload["can_start_now"] is False
+    assert payload["recommended_mode"] == "prepare_visual_capture_prerequisites"
+    assert payload["phases"][0]["status"] == "blocked_prerequisite"
+
+
+def test_pixelrag_assessment_keeps_github_and_hosted_embedding_out_of_runtime():
+    from services.pixelrag_crawler_integration_service import (
+        build_pixelrag_crawler_integration_assessment,
+    )
+
+    payload = build_pixelrag_crawler_integration_assessment(
+        capabilities={
+            "ollama_multimodal_embedding_ready": True,
+            "pgvector_visual_index_ready": True,
+        }
+    )
+
+    findings = {finding["code"]: finding for finding in payload["source_findings"]}
+    guards = {guard["code"]: guard for guard in payload["safety_guards"]}
+
+    assert payload["full_pixelrag_ready"] is True
+    assert "Ollama" in findings["qwen3_vl_embedding_optional"]["adoption"]
+    assert guards["no_external_embedding_api"]["status"] == "enforced"
+    assert guards["no_github_runtime_dependency"]["status"] == "enforced"
+    assert payload["controlled_apply"]["github_dependency"] is False
+
+
+def test_pixelrag_report_cli_outputs_machine_readable_json():
+    completed = subprocess.run(
+        [
+            sys.executable,
+            "scripts/ops/report_pixelrag_crawler_integration.py",
+            "--platform",
+            "momo",
+        ],
+        capture_output=True,
+        check=False,
+        text=True,
+    )
+
+    assert completed.returncode == 0
+    payload = json.loads(completed.stdout)
+    assert payload["success"] is True
+    assert payload["target_platforms"] == ["momo"]
+    assert payload["next_actions"][0]["action"] == "add_visual_evidence_manifest_lane"
+
+
+def test_visual_fallback_selector_routes_parser_empty_and_low_confidence_cases():
+    from services.pixelrag_crawler_integration_service import (
+        should_emit_visual_evidence_fallback,
+    )
+
+    parser_empty = should_emit_visual_evidence_fallback(
+        parser_success=False,
+        parsed_item_count=0,
+        missing_fields=["price"],
+        failure_reason="HTML selector did not find rendered price block",
+    )
+    clean_parser = should_emit_visual_evidence_fallback(
+        parser_success=True,
+        parsed_item_count=8,
+        confidence_band="high",
+    )
+
+    assert parser_empty["should_emit"] is True
+    assert "parser_failed" in parser_empty["triggers"]
+    assert "critical_field_missing" in parser_empty["triggers"]
+    assert clean_parser["should_emit"] is False
+    assert clean_parser["fallback_reason"] == "structured_evidence_sufficient"
+
+
+def test_visual_evidence_manifest_builds_tile_plan_without_runtime_writes():
+    from services.pixelrag_crawler_integration_service import (
+        RESULT_MANIFEST_READY,
+        build_pixelrag_visual_evidence_manifest,
+    )
+
+    payload = build_pixelrag_visual_evidence_manifest(
+        url="https://m.momoshop.com.tw/search.momo?searchKeyword=test",
+        platform="momo",
+        crawler="MomoCrawler.search_products",
+        trigger_reason="parser_empty",
+        viewport={"name": "desktop-1440", "width": 1440, "height": 950},
+        page_size={"width": 1440, "height": 1900},
+        tile_size={"width": 512, "height": 512},
+    )
+
+    assert payload["success"] is True
+    assert payload["result"] == RESULT_MANIFEST_READY
+    assert payload["status"] == "capture_requested"
+    assert payload["tile_plan"] == {
+        "tiles_x": 3,
+        "tiles_y": 4,
+        "tile_count": 12,
+        "overlap_px": 0,
+    }
+    assert payload["artifact"]["writes"] == ["artifact_file_only"]
+    assert payload["controlled_apply"]["db_write"] is False
+    assert payload["controlled_apply"]["network_call"] is False
+    assert payload["controlled_apply"]["production_price_write"] is False
+
+
+def test_visual_evidence_manifest_rejects_urls_with_credentials():
+    from services.pixelrag_crawler_integration_service import (
+        RESULT_MANIFEST_REJECTED,
+        build_pixelrag_visual_evidence_manifest,
+    )
+
+    payload = build_pixelrag_visual_evidence_manifest(
+        url="https://user:pass@example.test/product",
+        platform="momo",
+        crawler="MomoCrawler",
+        trigger_reason="parser_empty",
+    )
+
+    assert payload["success"] is False
+    assert payload["result"] == RESULT_MANIFEST_REJECTED
+    assert "URL credentials are not allowed" in payload["errors"][0]
+
+
+def test_pixelrag_report_cli_outputs_visual_manifest():
+    completed = subprocess.run(
+        [
+            sys.executable,
+            "scripts/ops/report_pixelrag_crawler_integration.py",
+            "--platform",
+            "pchome",
+            "--manifest-url",
+            "https://24h.pchome.com.tw/prod/TEST-000000001",
+            "--crawler",
+            "PChomeCrawler.search_products",
+            "--trigger-reason",
+            "parser_empty",
+            "--page-size",
+            "page=1024x1024",
+            "--tile-size",
+            "tile=512x512",
+        ],
+        capture_output=True,
+        check=False,
+        text=True,
+    )
+
+    assert completed.returncode == 0
+    payload = json.loads(completed.stdout)
+    assert payload["result"] == "PIXELRAG_VISUAL_EVIDENCE_MANIFEST_READY"
+    assert payload["capture_target"]["platform"] == "pchome"
+    assert payload["tile_plan"]["tile_count"] == 4