"""Controlled PixelRAG marketplace candidate knowledge replay worker.

This worker turns marketplace embedding signature guard receipts into
deterministic internal RAG candidate-knowledge receipts. It does not generate
embeddings, call models, call networks, write databases, write AI insights, or
promote candidate prices.
"""

from __future__ import annotations

import hashlib
import json
import os
import re
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Mapping

from services.pixelrag_crawler_integration_service import DEFAULT_ARTIFACT_MAX_AGE_HOURS
from services.pixelrag_marketplace_embedding_signature_guard_replay_service import (
    DEFAULT_OUTPUT_ROOT as DEFAULT_EMBEDDING_SIGNATURE_GUARD_RECEIPT_ROOT,
    EMBEDDING_SIGNATURE_GUARD_REPLAY_VERSION,
)
from services.rag_service import get_embedding_signature


POLICY = "controlled_pixelrag_marketplace_candidate_knowledge_replay_v1"
CANDIDATE_KNOWLEDGE_REPLAY_VERSION = (
    "pixelrag_marketplace_candidate_knowledge_replay_v1"
)
INTERNAL_RAG_TARGET = "rag_service.learning_episode_candidate_preview"
DEFAULT_LIMIT = 25
DEFAULT_OUTPUT_ROOT = os.getenv(
    "PIXELRAG_MARKETPLACE_CANDIDATE_KNOWLEDGE_REPLAY_RECEIPT_ROOT",
    "/app/data/ai_automation/pixelrag_marketplace_candidate_knowledge_replay_receipts"
    if Path("/app/data").exists()
    else "runtime_artifacts/pixelrag_marketplace_candidate_knowledge_replay_receipts",
)

REQUIRED_SIGNATURE_GATES = {
    "public_source_boundary",
    "rate_limit_contract",
    "provenance_contract",
    "identity_matcher_replay",
    "promotion_gate",
    "embedding_signature_guard",
}


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 _parse_iso_datetime(value: Any) -> datetime | None:
    if not value:
        return None
    try:
        return datetime.fromisoformat(str(value).replace("Z", "+00:00"))
    except ValueError:
        return None


def _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(
                    "*/marketplace_embedding_signature_guard_replay_receipt.json"
                )
            )
    else:
        candidates.extend(
            root.glob("*/*/marketplace_embedding_signature_guard_replay_receipt.json")
        )
    return sorted(candidates, key=lambda path: path.stat().st_mtime, reverse=True)[:limit]


def _load_receipt(path: Path) -> tuple[dict[str, Any], list[str]]:
    try:
        return json.loads(path.read_text(encoding="utf-8")), []
    except (OSError, json.JSONDecodeError) as exc:
        return {}, [str(exc)[:300]]


def _as_mapping(value: Any) -> Mapping[str, Any]:
    return value if isinstance(value, Mapping) else {}


def _as_list(value: Any) -> list[Any]:
    return list(value) if isinstance(value, list) else []


def _required_before(signature_payload: Mapping[str, Any]) -> set[str]:
    return set(str(item) for item in _as_list(signature_payload.get("required_before_data_promotion")))


def _knowledge_fingerprint(
    platform: str,
    manifest_id: str,
    candidate_id: str,
    signature_guard_fingerprint: str,
    embedding_signature: str,
) -> str:
    digest = hashlib.sha256(
        (
            f"{platform}:{manifest_id}:{candidate_id}:"
            f"{signature_guard_fingerprint}:{embedding_signature}"
        ).encode("utf-8")
    ).hexdigest()
    return digest[:24]


def _knowledge_text(
    *,
    platform: str,
    manifest_id: str,
    adapter_code: str,
    barrier_type: str,
    candidate: Mapping[str, Any],
    embedding_signature: str,
) -> str:
    parts = [
        f"platform={platform}",
        f"manifest_id={manifest_id}",
        f"adapter_code={adapter_code or 'unknown'}",
        f"barrier_type={barrier_type or 'none'}",
        f"candidate_id={candidate.get('candidate_id') or 'unknown'}",
        f"candidate_index={int(candidate.get('candidate_index') or 0)}",
        f"identity_fingerprint={candidate.get('identity_fingerprint') or 'unknown'}",
        f"promotion_fingerprint={candidate.get('promotion_fingerprint') or 'unknown'}",
        (
            "embedding_signature_guard_fingerprint="
            f"{candidate.get('embedding_signature_guard_fingerprint') or 'unknown'}"
        ),
        f"embedding_signature={embedding_signature}",
        "knowledge_source=pixelrag_marketplace_signature_guard_replay",
        "write_target=internal_rag_candidate_preview",
    ]
    return " | ".join(parts)


def _candidate_knowledge_contracts(
    receipt: Mapping[str, Any],
    *,
    signature_payload: Mapping[str, Any],
) -> list[dict[str, Any]]:
    platform = str(receipt.get("platform") or "unknown")
    manifest_id = str(receipt.get("manifest_id") or "unknown")
    adapter_code = str(signature_payload.get("adapter_code") or receipt.get("adapter_code") or "")
    barrier_type = str(signature_payload.get("barrier_type") or "")
    embedding_contract = _as_mapping(signature_payload.get("embedding_signature_contract"))
    embedding_signature = str(embedding_contract.get("embedding_signature") or "")
    contracts: list[dict[str, Any]] = []
    for candidate in _as_list(signature_payload.get("signature_guard_candidate_contracts")):
        if not isinstance(candidate, Mapping):
            continue
        candidate_id = str(candidate.get("candidate_id") or f"{platform}:{manifest_id}:unknown")
        signature_guard_fingerprint = str(
            candidate.get("embedding_signature_guard_fingerprint") or "missing"
        )
        knowledge_fingerprint = _knowledge_fingerprint(
            platform,
            manifest_id,
            candidate_id,
            signature_guard_fingerprint,
            embedding_signature,
        )
        contracts.append(
            {
                "candidate_id": candidate_id,
                "candidate_index": int(candidate.get("candidate_index") or 0),
                "knowledge_candidate_id": (
                    f"{platform}:{manifest_id}:{knowledge_fingerprint}"
                ),
                "identity_fingerprint": candidate.get("identity_fingerprint"),
                "promotion_fingerprint": candidate.get("promotion_fingerprint"),
                "embedding_signature_guard_fingerprint": signature_guard_fingerprint,
                "candidate_knowledge_fingerprint": knowledge_fingerprint,
                "embedding_signature": embedding_signature,
                "expected_embedding_signature": (
                    candidate.get("expected_embedding_signature") or embedding_signature
                ),
                "candidate_knowledge_text": _knowledge_text(
                    platform=platform,
                    manifest_id=manifest_id,
                    adapter_code=adapter_code,
                    barrier_type=barrier_type,
                    candidate=candidate,
                    embedding_signature=embedding_signature,
                ),
                "internal_rag_target": INTERNAL_RAG_TARGET,
                "knowledge_stage": "pre_internal_rag_candidate_canary",
                "knowledge_strategy": "deterministic_candidate_knowledge_replay_no_db_write",
                "requires_public_source_boundary": True,
                "requires_rate_limit_contract": True,
                "requires_provenance_contract": True,
                "requires_identity_matcher_replay": True,
                "requires_promotion_gate": True,
                "requires_embedding_signature_guard": True,
                "requires_candidate_knowledge_replay": True,
                "requires_internal_rag_candidate_canary": True,
                "embedding_generation_performed": False,
                "model_call_performed": False,
                "ready_for_internal_rag_candidate_replay": True,
                "ready_for_rag_candidate_preview": True,
                "ready_for_ai_insights_write": False,
                "ready_for_price_table_write": False,
                "ai_insights_write_blocked_until_canary": True,
                "price_write_blocked_until_candidate_canary": True,
            }
        )
    return contracts


def _candidate_knowledge_payload(
    receipt: Mapping[str, Any],
    *,
    receipt_path: Path,
) -> dict[str, Any]:
    signature_payload = _as_mapping(receipt.get("embedding_signature_guard_replay"))
    knowledge_contracts = _candidate_knowledge_contracts(
        receipt,
        signature_payload=signature_payload,
    )
    return {
        "candidate_knowledge_replay_version": CANDIDATE_KNOWLEDGE_REPLAY_VERSION,
        "embedding_signature_guard_replay_version": signature_payload.get(
            "embedding_signature_guard_replay_version"
        ),
        "promotion_gate_replay_version": signature_payload.get(
            "promotion_gate_replay_version"
        ),
        "identity_matcher_replay_version": signature_payload.get(
            "identity_matcher_replay_version"
        ),
        "adapter_dry_run_version": signature_payload.get("adapter_dry_run_version"),
        "adapter_preflight_version": signature_payload.get("adapter_preflight_version"),
        "source_contract_version": signature_payload.get("source_contract_version"),
        "adapter_code": signature_payload.get("adapter_code") or receipt.get("adapter_code"),
        "contract_id": signature_payload.get("contract_id"),
        "barrier_type": signature_payload.get("barrier_type"),
        "capture_runtime_unavailable": bool(
            signature_payload.get("capture_runtime_unavailable")
        ),
        "input_embedding_signature_guard_receipt_path": str(receipt_path),
        "input_promotion_gate_receipt_path": signature_payload.get(
            "input_promotion_gate_receipt_path"
        ),
        "input_identity_matcher_receipt_path": signature_payload.get(
            "input_identity_matcher_receipt_path"
        ),
        "input_adapter_dry_run_receipt_path": signature_payload.get(
            "input_adapter_dry_run_receipt_path"
        ),
        "adapter_preflight_receipt_path": signature_payload.get(
            "adapter_preflight_receipt_path"
        ),
        "source_contract_replay_receipt_path": signature_payload.get(
            "source_contract_replay_receipt_path"
        ),
        "source_worker_receipt_path": signature_payload.get("source_worker_receipt_path"),
        "source_receipt_path": signature_payload.get("source_receipt_path"),
        "embedding_signature_contract": signature_payload.get(
            "embedding_signature_contract"
        )
        or {},
        "candidate_knowledge_execution_mode": (
            "deterministic_candidate_knowledge_replay_no_embedding_no_network_no_db"
        ),
        "candidate_knowledge_count": len(knowledge_contracts),
        "candidate_knowledge_contracts": knowledge_contracts,
        "required_before_data_promotion": list(
            signature_payload.get("required_before_data_promotion") or []
        ),
        "allowed_next_step": "run_internal_rag_candidate_canary",
    }


def _candidate_knowledge_checks(
    receipt: Mapping[str, Any],
    *,
    stale: bool,
    errors: list[str],
    knowledge_payload: Mapping[str, Any],
) -> dict[str, bool]:
    signature_payload = _as_mapping(receipt.get("embedding_signature_guard_replay"))
    boundary = _as_mapping(receipt.get("promotion_boundary"))
    signature_checks = _as_mapping(
        receipt.get("embedding_signature_guard_replay_checks")
    )
    signature_contracts = _as_list(
        signature_payload.get("signature_guard_candidate_contracts")
    )
    knowledge_contracts = _as_list(
        knowledge_payload.get("candidate_knowledge_contracts")
    )
    embedding_contract = _as_mapping(
        knowledge_payload.get("embedding_signature_contract")
    )
    expected_signature = str(embedding_contract.get("embedding_signature") or "")
    current_signature = get_embedding_signature()
    required_before = _required_before(signature_payload)
    return {
        "receipt_parse_ok": not errors,
        "receipt_fresh": not stale,
        "embedding_signature_guard_replay_ready": (
            receipt.get("worker_status")
            == "executed_marketplace_embedding_signature_guard_replay_ready"
        ),
        "embedding_signature_guard_version_supported": (
            signature_payload.get("embedding_signature_guard_replay_version")
            == EMBEDDING_SIGNATURE_GUARD_REPLAY_VERSION
        ),
        "embedding_signature_guard_checks_all_passed": (
            int(receipt.get("embedding_signature_guard_replay_check_pass_count") or 0)
            == int(receipt.get("embedding_signature_guard_replay_check_count") or -1)
            and int(receipt.get("embedding_signature_guard_replay_check_count") or 0) > 0
        ),
        "blocked_page_not_product_data": bool(
            signature_checks.get("blocked_page_not_product_data")
        ),
        "signature_guard_candidate_contracts_present": bool(signature_contracts),
        "candidate_knowledge_contracts_generated": bool(knowledge_contracts),
        "candidate_knowledge_count_matches_signature_count": (
            len(knowledge_contracts) == len(signature_contracts)
        ),
        "signature_candidates_ready_for_candidate_knowledge": all(
            bool(item.get("ready_for_candidate_knowledge_replay"))
            for item in signature_contracts
        ),
        "required_signature_gates_present": REQUIRED_SIGNATURE_GATES.issubset(
            required_before
        ),
        "candidate_knowledge_replay_required": all(
            bool(item.get("requires_candidate_knowledge_replay"))
            for item in signature_contracts
        ),
        "embedding_signature_present": (
            len(expected_signature) == 12
            and all(ch in "0123456789abcdef" for ch in expected_signature)
        ),
        "embedding_signature_matches_current_rag_signature": (
            expected_signature == current_signature
        ),
        "embedding_generation_not_performed": (
            not bool(embedding_contract.get("embedding_generation_performed"))
            and not bool(embedding_contract.get("model_call_performed"))
            and not bool(receipt.get("embedding_generation_performed"))
        ),
        "knowledge_text_generated": all(
            bool(item.get("candidate_knowledge_text")) for item in knowledge_contracts
        ),
        "internal_rag_preview_ready": all(
            bool(item.get("ready_for_internal_rag_candidate_replay"))
            and item.get("internal_rag_target") == INTERNAL_RAG_TARGET
            for item in knowledge_contracts
        ),
        "ai_insights_write_disallowed": all(
            not bool(item.get("ready_for_ai_insights_write"))
            and bool(item.get("ai_insights_write_blocked_until_canary"))
            for item in knowledge_contracts
        ),
        "price_write_blocked_until_candidate_canary": all(
            bool(item.get("price_write_blocked_until_candidate_canary"))
            and not bool(item.get("ready_for_price_table_write"))
            for item in knowledge_contracts
        ),
        "direct_price_write_disallowed": not bool(
            boundary.get("direct_price_write_allowed")
        ),
        "direct_ai_insights_write_disallowed": not bool(
            boundary.get("direct_ai_insights_write_allowed")
        ),
        "database_write_disallowed": not bool(receipt.get("writes_database"))
        and int(receipt.get("writes_database_count") or 0) == 0,
        "network_call_disallowed": not bool(receipt.get("network_call_performed")),
        "model_call_disallowed": not bool(receipt.get("model_call_performed")),
        "source_provenance_paths_present": bool(
            knowledge_payload.get("input_embedding_signature_guard_receipt_path")
            and knowledge_payload.get("input_promotion_gate_receipt_path")
            and knowledge_payload.get("input_identity_matcher_receipt_path")
            and knowledge_payload.get("input_adapter_dry_run_receipt_path")
            and knowledge_payload.get("adapter_preflight_receipt_path")
            and knowledge_payload.get("source_contract_replay_receipt_path")
        ),
    }


def _worker_item(
    receipt_path: Path,
    *,
    now: datetime,
    max_age_hours: int,
    execute: bool,
) -> dict[str, Any]:
    receipt, errors = _load_receipt(receipt_path)
    generated_at = _parse_iso_datetime(receipt.get("generated_at"))
    if generated_at is None:
        try:
            generated_at = datetime.fromtimestamp(receipt_path.stat().st_mtime, tz=timezone.utc)
        except OSError:
            generated_at = None
    age_hours = ((now - generated_at).total_seconds() / 3600) if generated_at else None
    stale = age_hours is None or age_hours > max_age_hours
    knowledge_payload = _candidate_knowledge_payload(receipt, receipt_path=receipt_path)
    checks = _candidate_knowledge_checks(
        receipt,
        stale=stale,
        errors=errors,
        knowledge_payload=knowledge_payload,
    )
    check_count = len(checks)
    pass_count = sum(1 for passed in checks.values() if passed)
    ready = pass_count == check_count
    platform = str(receipt.get("platform") or receipt_path.parent.parent.name).strip().lower()
    manifest_id = str(receipt.get("manifest_id") or receipt_path.parent.name).strip()
    status = (
        "executed_marketplace_candidate_knowledge_replay_ready"
        if execute and ready
        else (
            "dry_run_ready_for_marketplace_candidate_knowledge_replay"
            if ready
            else "skipped_marketplace_candidate_knowledge_replay_guard_failed"
        )
    )
    return {
        "worker_status": status,
        "platform": platform,
        "manifest_id": manifest_id,
        "source_type": "marketplace_embedding_signature_guard_replay_receipt",
        "embedding_signature_guard_receipt_path": str(receipt_path),
        "promotion_gate_receipt_path": knowledge_payload.get(
            "input_promotion_gate_receipt_path"
        ),
        "identity_matcher_receipt_path": knowledge_payload.get(
            "input_identity_matcher_receipt_path"
        ),
        "adapter_dry_run_receipt_path": knowledge_payload.get(
            "input_adapter_dry_run_receipt_path"
        ),
        "adapter_preflight_receipt_path": knowledge_payload.get(
            "adapter_preflight_receipt_path"
        ),
        "source_contract_replay_receipt_path": knowledge_payload.get(
            "source_contract_replay_receipt_path"
        ),
        "source_worker_receipt_path": knowledge_payload.get("source_worker_receipt_path"),
        "source_receipt_path": knowledge_payload.get("source_receipt_path"),
        "adapter_code": knowledge_payload.get("adapter_code"),
        "candidate_knowledge_replay_status": "ready" if ready else "blocked",
        "ready_for_execution": ready,
        "execute": bool(execute),
        "generated_at": generated_at.isoformat() if generated_at else None,
        "age_hours": round(age_hours, 3) if age_hours is not None else None,
        "stale": stale,
        "network_call_performed": False,
        "model_call_performed": False,
        "embedding_generation_performed": False,
        "artifact_write_performed": False,
        "writes_database": False,
        "writes_database_count": 0,
        "writes_ai_insights": False,
        "writes_price_tables": False,
        "primary_human_gate_count": 0,
        "candidate_knowledge_replay_checks": checks,
        "candidate_knowledge_replay_check_count": check_count,
        "candidate_knowledge_replay_check_pass_count": pass_count,
        "candidate_knowledge_replay": knowledge_payload,
        "promotion_boundary": {
            "direct_ai_insights_write_allowed": False,
            "direct_price_write_allowed": False,
            "candidate_knowledge_replay_only": True,
            "requires_public_source_boundary": True,
            "requires_rate_limit_contract": True,
            "requires_provenance_contract": True,
            "requires_identity_matcher_replay": True,
            "requires_promotion_gate": True,
            "requires_embedding_signature_guard": True,
            "requires_internal_rag_candidate_canary": True,
        },
        "next_machine_action": (
            "run_internal_rag_candidate_canary"
            if execute and ready
            else (
                "run_marketplace_candidate_knowledge_replay_execute"
                if ready
                else "repair_marketplace_candidate_knowledge_replay_inputs"
            )
        ),
    }


def _write_candidate_knowledge_receipt(
    *,
    output_root: Path,
    item: Mapping[str, Any],
) -> str:
    target = (
        output_root
        / _safe_segment(item.get("platform"))
        / _safe_segment(item.get("manifest_id"))
        / "marketplace_candidate_knowledge_replay_receipt.json"
    )
    target.parent.mkdir(parents=True, exist_ok=True)
    receipt = dict(item)
    receipt["artifact_write_performed"] = True
    receipt["receipt_path"] = str(target)
    receipt["generated_at"] = datetime.now(timezone.utc).isoformat()
    receipt["policy"] = POLICY
    target.write_text(
        json.dumps(receipt, ensure_ascii=False, indent=2, sort_keys=True),
        encoding="utf-8",
    )
    return str(target)


def run_pixelrag_marketplace_candidate_knowledge_replay(
    *,
    embedding_signature_guard_receipt_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,
    execute: bool = False,
    write_receipt: bool = False,
) -> dict[str, Any]:
    """Run or dry-run marketplace candidate knowledge replay."""
    source_root = Path(
        embedding_signature_guard_receipt_root
        or DEFAULT_EMBEDDING_SIGNATURE_GUARD_RECEIPT_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))
    now = datetime.now(timezone.utc)
    receipt_paths = _receipt_candidates(source_root, platforms=platforms, limit=item_limit)
    worker_items: list[dict[str, Any]] = []
    for receipt_path in receipt_paths:
        item = _worker_item(
            receipt_path,
            now=now,
            max_age_hours=max_age,
            execute=execute,
        )
        if execute and write_receipt and item.get("ready_for_execution"):
            item["receipt_path"] = _write_candidate_knowledge_receipt(
                output_root=output,
                item=item,
            )
            item["artifact_write_performed"] = True
        worker_items.append(item)

    ready_count = sum(1 for item in worker_items if item.get("ready_for_execution"))
    blocked_count = len(worker_items) - ready_count
    dry_run_count = sum(
        1 for item in worker_items if str(item.get("worker_status") or "").startswith("dry_run_")
    )
    executed_count = sum(
        1 for item in worker_items if str(item.get("worker_status") or "").startswith("executed_")
    )
    receipt_written_count = sum(1 for item in worker_items if item.get("receipt_path"))
    guard_failed_count = sum(
        1
        for item in worker_items
        if item.get("candidate_knowledge_replay_status") != "ready"
    )
    capture_runtime_gap_count = sum(
        1
        for item in worker_items
        if (item.get("candidate_knowledge_replay") or {}).get(
            "capture_runtime_unavailable"
        )
    )
    knowledge_candidate_count = sum(
        int(
            (item.get("candidate_knowledge_replay") or {}).get(
                "candidate_knowledge_count"
            )
            or 0
        )
        for item in worker_items
    )
    internal_rag_ready_count = sum(
        sum(
            1
            for candidate in _as_list(
                (item.get("candidate_knowledge_replay") or {}).get(
                    "candidate_knowledge_contracts"
                )
            )
            if isinstance(candidate, Mapping)
            and candidate.get("ready_for_internal_rag_candidate_replay")
        )
        for item in worker_items
    )
    expected_signatures = sorted(
        {
            str(
                (
                    (item.get("candidate_knowledge_replay") or {})
                    .get("embedding_signature_contract")
                    or {}
                ).get("embedding_signature")
                or ""
            )
            for item in worker_items
        }
        - {""}
    )

    if not worker_items:
        status = "warning"
    elif guard_failed_count:
        status = "warning"
    else:
        status = "ok"

    if not worker_items:
        next_action = "run_marketplace_embedding_signature_guard_replay"
    elif not execute and ready_count:
        next_action = "run_marketplace_candidate_knowledge_replay_execute"
    elif execute and receipt_written_count:
        next_action = "run_internal_rag_candidate_canary"
    elif guard_failed_count:
        next_action = "repair_marketplace_candidate_knowledge_replay_inputs"
    else:
        next_action = "refresh_marketplace_candidate_knowledge_replay_candidates"

    summary = {
        "candidate_count": len(worker_items),
        "ready_count": ready_count,
        "blocked_count": blocked_count,
        "dry_run_count": dry_run_count,
        "executed_count": executed_count,
        "receipt_written_count": receipt_written_count,
        "guard_failed_count": guard_failed_count,
        "capture_runtime_gap_count": capture_runtime_gap_count,
        "knowledge_candidate_count": knowledge_candidate_count,
        "internal_rag_ready_count": internal_rag_ready_count,
        "expected_embedding_signatures": expected_signatures,
        "platforms": sorted({str(item.get("platform") or "unknown") for item in worker_items}),
        "network_call_performed": False,
        "model_call_performed": False,
        "embedding_generation_performed": False,
        "artifact_write_performed": bool(receipt_written_count),
        "writes_database_count": 0,
        "writes_ai_insights": False,
        "writes_price_tables": False,
        "primary_human_gate_count": 0,
    }
    return {
        "success": status != "critical",
        "policy": POLICY,
        "status": status,
        "generated_at": now.isoformat(),
        "candidate_knowledge_replay_version": CANDIDATE_KNOWLEDGE_REPLAY_VERSION,
        "embedding_signature_guard_receipt_root": str(source_root),
        "output_root": str(output),
        "platform_filter": list(platforms),
        "max_age_hours": max_age,
        "limit": item_limit,
        "execute": bool(execute),
        "write_receipt": bool(write_receipt and execute),
        "summary": summary,
        "worker_items": worker_items,
        "controlled_apply": {
            "network_call": False,
            "model_call": False,
            "embedding_generation": False,
            "artifact_write": bool(receipt_written_count),
            "db_write": False,
            "writes_database": False,
            "writes_database_count": 0,
            "writes_ai_insights": False,
            "writes_price_tables": False,
            "secret_read": False,
            "raw_cookie_or_session_read": False,
            "credentialed_session_allowed": False,
            "login_allowed": False,
            "production_price_write": False,
            "primary_human_gate_count": 0,
        },
        "promotion_boundary": {
            "writes_ai_insights": False,
            "writes_price_tables": False,
            "candidate_knowledge_replay_only": True,
            "requires_public_source_boundary": True,
            "requires_rate_limit_contract": True,
            "requires_provenance_contract": True,
            "requires_identity_matcher_replay": True,
            "requires_promotion_gate": True,
            "requires_embedding_signature_guard": True,
            "requires_internal_rag_candidate_canary": True,
        },
        "next_machine_action": next_action,
    }


__all__ = [
    "CANDIDATE_KNOWLEDGE_REPLAY_VERSION",
    "DEFAULT_OUTPUT_ROOT",
    "INTERNAL_RAG_TARGET",
    "POLICY",
    "run_pixelrag_marketplace_candidate_knowledge_replay",
]
