"""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",
]
