content-find-agent-new/content_agent/business_modules/progressive_screening.py

from __future__ import annotations

import time
import random
from copy import deepcopy
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Any, Callable

from content_agent.business_modules import content_discovery, platform_access, rule_judgment
from content_agent.business_modules.content_discovery import pattern_recall
from content_agent.business_modules.content_discovery.fifty_plus import fifty_plus_score
from content_agent.constants import RUNTIME_RECORD_SCHEMA_VERSION
from content_agent.errors import ContentAgentError, ErrorCode
from content_agent.interfaces import GeminiVideoClient, PlatformSearchClient, RuntimeFileStore
from content_agent.record_payload import with_raw_payload


INITIAL_BATCH_SIZE = 3
MAX_PAGES = 3
SEARCH_MIN_INTERVAL_SECONDS = 30.0
DOUYIN_SEARCH_MIN_INTERVAL_SECONDS = 10.0
DOUYIN_SEARCH_MAX_INTERVAL_SECONDS = 12.0
SHORT_VIDEO_SEARCH_MIN_INTERVAL_SECONDS = 10.0
SHORT_VIDEO_SEARCH_MAX_INTERVAL_SECONDS = 12.0
PASS_ACTION = "ADD_TO_CONTENT_POOL"
# 抖音拉作者画像(适老性)的门:只看 query 相关性,不看平台分,保证适老性始终进三项加权。
# 值=抖音入池 query 下限(score_thresholds.douyin.pool_query=65);低于此靠 query 就被否,不拉省成本。
PORTRAIT_QUERY_GATE = 65
# 画像拉取重试次数(总尝试次数):救热点宝瞬时失败/限速。
PORTRAIT_FETCH_ATTEMPTS = 2


class SearchCallLimiter:
    def __init__(
        self,
        min_interval_seconds: float = SEARCH_MIN_INTERVAL_SECONDS,
        max_interval_seconds: float | None = None,
        *,
        now_fn: Callable[[], float] = time.monotonic,
        sleep_fn: Callable[[float], None] = time.sleep,
        random_fn: Callable[[float, float], float] = random.uniform,
    ) -> None:
        self.min_interval_seconds = min_interval_seconds
        self.max_interval_seconds = max_interval_seconds
        self.now_fn = now_fn
        self.sleep_fn = sleep_fn
        self.random_fn = random_fn
        self._last_call_at: float | None = None

    def wait(self) -> float:
        waited_seconds = 0.0
        if self._last_call_at is not None:
            remaining = self._next_interval_seconds() - (self.now_fn() - self._last_call_at)
            if remaining > 0:
                self.sleep_fn(remaining)
                waited_seconds = remaining
        self._last_call_at = self.now_fn()
        return waited_seconds

    def _next_interval_seconds(self) -> float:
        if self.max_interval_seconds is None or self.max_interval_seconds <= self.min_interval_seconds:
            return self.min_interval_seconds
        return self.random_fn(self.min_interval_seconds, self.max_interval_seconds)


def make_search_limiter(platform: str = "") -> SearchCallLimiter:
    if platform == "douyin":
        return SearchCallLimiter(
            min_interval_seconds=DOUYIN_SEARCH_MIN_INTERVAL_SECONDS,
            max_interval_seconds=DOUYIN_SEARCH_MAX_INTERVAL_SECONDS,
        )
    if platform in {"kuaishou", "shipinhao"}:
        return SearchCallLimiter(
            min_interval_seconds=SHORT_VIDEO_SEARCH_MIN_INTERVAL_SECONDS,
            max_interval_seconds=SHORT_VIDEO_SEARCH_MAX_INTERVAL_SECONDS,
        )
    return SearchCallLimiter()


@dataclass(frozen=True)
class _FetchedPage:
    results: list[dict[str, Any]]
    search_duration_ms: int
    search_wait_ms: int
    search_request_duration_ms: int


def run(
    *,
    run_id: str,
    policy_run_id: str,
    search_queries: list[dict[str, Any]],
    source_context: dict[str, Any],
    policy_bundle: dict[str, Any],
    platform_client: PlatformSearchClient,
    runtime: RuntimeFileStore,
    gemini_video_client: GeminiVideoClient,
    limiter: SearchCallLimiter | None = None,
    portrait_limiter: SearchCallLimiter | None = None,
    archive_dispatcher: Any | None = None,
    platform: str = "",
) -> dict[str, list[dict[str, Any]]]:
    if limiter is None and getattr(platform_client, "requires_progressive_search_rate_limit", False):
        limiter = make_search_limiter(platform)
    context = _ProgressiveContext(
        run_id=run_id,
        policy_run_id=policy_run_id,
        source_context=source_context,
        policy_bundle=policy_bundle,
        platform_client=platform_client,
        runtime=runtime,
        gemini_video_client=gemini_video_client,
        limiter=limiter,
        portrait_limiter=portrait_limiter,
        archive_dispatcher=archive_dispatcher,
        platform=platform,
    )
    try:
        for query in search_queries:
            context.process_query(query)

        if search_queries and context.query_failures and not context.platform_results:
            raise ContentAgentError(
                ErrorCode.PLATFORM_REQUEST_FAILED,
                "all platform queries failed",
                {"query_failures": context.query_failures},
            )

        context.write_runtime()
        return {
            "platform_results": context.platform_results,
            "query_failures": context.query_failures,
            "discovered_content_items": context.discovered_content_items,
            "content_media_records": context.content_media_records,
            "pattern_recall_evidence": context.pattern_recall_evidence,
            "evidence_bundles": context.evidence_bundles,
            "rule_decisions": context.rule_decisions,
        }
    finally:
        context.close_archive_dispatcher()


class _ProgressiveContext:
    def __init__(
        self,
        *,
        run_id: str,
        policy_run_id: str,
        source_context: dict[str, Any],
        policy_bundle: dict[str, Any],
        platform_client: PlatformSearchClient,
        runtime: RuntimeFileStore,
        gemini_video_client: GeminiVideoClient,
        archive_dispatcher: Any | None,
        limiter: SearchCallLimiter | None,
        portrait_limiter: SearchCallLimiter | None = None,
        external_seen_content_ids: set[str] | None = None,
        recall_index_base: int = 0,
        decision_index_base: int = 0,
        platform: str = "",
        portrait_client: Any | None = None,
    ) -> None:
        self.run_id = run_id
        self.policy_run_id = policy_run_id
        self.source_context = source_context
        self.policy_bundle = policy_bundle
        self.platform_client = platform_client
        self.runtime = runtime
        self.gemini_video_client = gemini_video_client
        self.limiter = limiter
        self.portrait_limiter = portrait_limiter
        self.archive_dispatcher = archive_dispatcher
        self._archive_dispatcher_closed = False
        self.platform_results: list[dict[str, Any]] = []
        self.query_failures: list[dict[str, Any]] = []
        self.discovered_content_items: list[dict[str, Any]] = []
        self.content_media_records: list[dict[str, Any]] = []
        self.pattern_recall_evidence: list[dict[str, Any]] = []
        self.evidence_bundles: list[dict[str, Any]] = []
        self.rule_decisions: list[dict[str, Any]] = []
        self._platform_result_by_key: dict[tuple[str, str], dict[str, Any]] = {}
        self._record_indexes_by_key: dict[tuple[str, str], dict[str, int]] = {}
        self._queries_with_consumed_next_pages: set[str] = set()
        # M8C:游走复用统一搜索单元时注入的全 run 级状态。
        # external_seen_content_ids:与 _FrontierContext 共享的同一 set,承载首轮 + 跨层
        #   已发现 content_id;为 None 时(首轮)行为完全等价。
        # *_index_base:游走批次的 id 起算基线(=首轮计数),避免 recall/decision id 撞首轮。
        self._external_seen_content_ids = external_seen_content_ids
        self._recall_index_base = recall_index_base
        self._decision_index_base = decision_index_base
        # M9A:抖音 50+ 受众子分。platform!="douyin" 时不注入 content_audience_50plus 块,
        # evaluator 即走旧 50/50(非抖音零改动)。portrait_client 默认复用 platform_client。
        self.platform = platform
        self.portrait_client = portrait_client or platform_client
        # M9 修复:作者画像按作者缓存(画像是作者级,同作者结果一致)。避免对首轮大量视频
        # 逐条串行拉画像 → 同作者只拉一次,把调用量从"视频数"降到"作者数"。None=该作者拉取失败。
        self._portrait_by_author: dict[str, dict[str, Any] | None] = {}
        self._portrait_error_by_author: dict[str, str | None] = {}
        self._batch_event_index = 0

    def process_query(self, query: dict[str, Any]) -> dict[str, list[dict[str, Any]]]:
        """统一搜索单元(前3命中→剩余→翻页≤3页)。

        M8C:返回本次搜索新增的 discovered_content_items / rule_decisions 切片,
        供游走 frontier 取回再判 allow_walk。首轮 run() 忽略返回值,行为等价。
        """
        start_discovered = len(self.discovered_content_items)
        start_decisions = len(self.rule_decisions)
        self._process_query_pages(query)
        return {
            "discovered_content_items": self.discovered_content_items[start_discovered:],
            "rule_decisions": self.rule_decisions[start_decisions:],
        }

    def process_prefetched_batch(
        self,
        query: dict[str, Any],
        results: list[dict[str, Any]],
        *,
        search_duration_ms: int = 0,
        search_wait_ms: int = 0,
        search_request_duration_ms: int = 0,
        search_timing_source: str = "prefetched_author_works",
    ) -> dict[str, list[dict[str, Any]]]:
        """处理已抓取的一批结果(作者作品:列表抓取+截断,不分页、不限速)。

        复用 _process_batch 的 discovery→recall→decision+去重+index 路径,
        返回本批新增切片。去重/index 与统一搜索单元共享同一 context 状态。
        """
        start_discovered = len(self.discovered_content_items)
        start_decisions = len(self.rule_decisions)
        if results:
            self._process_batch(
                query,
                results,
                page_number=1,
                page_item_count=len(results),
                batch_kind="author_works",
                rank_offset=0,
                search_duration_ms=search_duration_ms,
                search_wait_ms=search_wait_ms,
                search_request_duration_ms=search_request_duration_ms,
                search_timing_source=search_timing_source,
            )
        return {
            "discovered_content_items": self.discovered_content_items[start_discovered:],
            "rule_decisions": self.rule_decisions[start_decisions:],
        }

    def _process_query_pages(self, query: dict[str, Any]) -> None:
        first_fetch = self._fetch_page(query, page_number=1, cursor=query.get("page_cursor"))
        if first_fetch is None:
            return
        first_page = first_fetch.results
        page_count = len(first_page)
        if page_count == 0:
            return

        first_batch = first_page[:INITIAL_BATCH_SIZE]
        remainder_batch = first_page[INITIAL_BATCH_SIZE:]
        passed_first = self._process_batch(
            query,
            first_batch,
            page_number=1,
            page_item_count=page_count,
            batch_kind="initial_top3",
            rank_offset=0,
            search_duration_ms=first_fetch.search_duration_ms,
            search_wait_ms=first_fetch.search_wait_ms,
            search_request_duration_ms=first_fetch.search_request_duration_ms,
            search_timing_source="page_fetch",
        )
        if not passed_first or not remainder_batch:
            return

        passed_remainder = self._process_batch(
            query,
            remainder_batch,
            page_number=1,
            page_item_count=page_count,
            batch_kind="page_remainder",
            rank_offset=INITIAL_BATCH_SIZE,
            search_duration_ms=0,
            search_wait_ms=0,
            search_request_duration_ms=0,
            search_timing_source="same_page_remainder",
        )
        if not passed_remainder:
            return

        cursor = _page_next_cursor(first_page)
        has_more = _page_has_more(first_page)
        page_number = 2
        while page_number <= MAX_PAGES and has_more and cursor:
            fetched_page = self._fetch_page(query, page_number=page_number, cursor=cursor)
            if fetched_page is None or not fetched_page.results:
                return
            page = fetched_page.results
            self._queries_with_consumed_next_pages.add(str(query["search_query_id"]))
            page_count = len(page)
            passed_page = self._process_batch(
                query,
                page,
                page_number=page_number,
                page_item_count=page_count,
                batch_kind=f"page_{page_number:02d}",
                rank_offset=0,
                search_duration_ms=fetched_page.search_duration_ms,
                search_wait_ms=fetched_page.search_wait_ms,
                search_request_duration_ms=fetched_page.search_request_duration_ms,
                search_timing_source="page_fetch",
            )
            if page_number >= MAX_PAGES or not passed_page:
                return
            cursor = _page_next_cursor(page)
            has_more = _page_has_more(page)
            page_number += 1

    def _fetch_page(
        self,
        query: dict[str, Any],
        *,
        page_number: int,
        cursor: Any,
    ) -> _FetchedPage | None:
        page_query = dict(query)
        if page_number > 1:
            page_query["page_cursor"] = str(cursor or "")
        try:
            search_started_at = time.monotonic()
            waited_seconds = self.limiter.wait() if self.limiter is not None else 0.0
            request_started_at = time.monotonic()
            search = getattr(self.platform_client, "search_full_page_metadata", None)
            if not callable(search):
                search = getattr(self.platform_client, "search_full_page", None)
            if not callable(search):
                search = self.platform_client.search
            results = list(search(page_query))
            request_ended_at = time.monotonic()
            return _FetchedPage(
                results=results,
                search_duration_ms=_elapsed_ms(search_started_at, request_ended_at),
                search_wait_ms=int(waited_seconds * 1000),
                search_request_duration_ms=_elapsed_ms(request_started_at, request_ended_at),
            )
        except Exception as exc:
            failure_query = page_query if page_number == 1 else {
                **page_query,
                "search_query_id": f"{query['search_query_id']}_progressive_page_{page_number:03d}",
            }
            failure = platform_access._query_failure(failure_query, exc)  # noqa: SLF001
            failure["progressive_page_number"] = page_number
            self.query_failures.append(failure)
            return None

    def _process_batch(
        self,
        query: dict[str, Any],
        page_results: list[dict[str, Any]],
        *,
        page_number: int,
        page_item_count: int,
        batch_kind: str,
        rank_offset: int,
        search_duration_ms: int = 0,
        search_wait_ms: int = 0,
        search_request_duration_ms: int = 0,
        search_timing_source: str = "unknown",
    ) -> bool:
        batch_started_at = time.monotonic()
        batch = []
        for rank, result in enumerate(page_results, start=1):
            key = _content_key(result)
            if key and key in self._platform_result_by_key:
                existing = self._platform_result_by_key[key]
                platform_access._append_query_source(existing, query)  # noqa: SLF001
                self._propagate_query_source_merge(key, existing)
                continue
            # M8C:游走跨层 / 首轮去重——已发现的 content_id 静默跳过(不二次判定、不撞 DB 唯一索引)。
            if (
                key
                and self._external_seen_content_ids is not None
                and key[1] in self._external_seen_content_ids
            ):
                continue
            result = self._hydrate_result_media(result, query)
            prepared = self._prepare_result(
                query,
                result,
                page_number=page_number,
                page_item_count=page_item_count,
                batch_kind=batch_kind,
                rank=rank_offset + rank,
            )
            key = _content_key(prepared)
            if not key:
                batch.append(prepared)
                continue
            self._platform_result_by_key[key] = prepared
            if self._external_seen_content_ids is not None:
                self._external_seen_content_ids.add(key[1])
            batch.append(prepared)
        if not batch:
            return False

        discovery_started_at = time.monotonic()
        discovered = content_discovery.run(
            self.run_id,
            self.policy_run_id,
            batch,
            self.source_context,
            self.runtime,
            write_runtime=False,
        )
        discovery_duration_ms = _elapsed_ms(discovery_started_at, time.monotonic())
        recalled = pattern_recall.run(
            self.run_id,
            self.policy_run_id,
            discovered["discovered_content_items"],
            discovered["content_media_records"],
            discovered["evidence_bundles"],
            self.source_context,
            self.runtime,
            self.gemini_video_client,
            start_index=self._recall_index_base + len(self.discovered_content_items) + 1,
            write_runtime=False,
            archive_dispatcher=self.archive_dispatcher,
        )
        recall_timing = recalled.get("batch_timing") or {}
        portrait_started_at = time.monotonic()
        self._inject_fifty_plus(recalled["evidence_bundles"])
        portrait_duration_ms = _elapsed_ms(portrait_started_at, time.monotonic())
        rule_started_at = time.monotonic()
        decisions = rule_judgment.run(
            self.run_id,
            self.policy_run_id,
            recalled["evidence_bundles"],
            self.policy_bundle,
            self.runtime,
            start_index=self._decision_index_base + len(self.rule_decisions) + 1,
            write_runtime=False,
        )
        rule_duration_ms = _elapsed_ms(rule_started_at, time.monotonic())
        self._accumulate(batch, recalled, decisions)
        has_pass = any(decision.get("decision_action") == PASS_ACTION for decision in decisions)
        self._record_batch_timing_event(
            query,
            batch,
            decisions,
            page_number=page_number,
            page_item_count=page_item_count,
            batch_kind=batch_kind,
            search_duration_ms=search_duration_ms,
            search_wait_ms=search_wait_ms,
            search_request_duration_ms=search_request_duration_ms,
            search_timing_source=search_timing_source,
            oss_batch_duration_ms=int(recall_timing.get("oss_batch_duration_ms") or 0),
            qwen_batch_duration_ms=int(recall_timing.get("qwen_batch_duration_ms") or 0),
            portrait_duration_ms=portrait_duration_ms,
            rule_duration_ms=rule_duration_ms,
            discovery_duration_ms=discovery_duration_ms,
            batch_duration_ms=_elapsed_ms(batch_started_at, time.monotonic()),
            has_pass=has_pass,
        )
        return has_pass

    def _inject_fifty_plus(self, bundles: list[dict[str, Any]]) -> None:
        """M9A:仅抖音,对相关性达标(query≥PORTRAIT_QUERY_GATE)的视频拉作者画像补 50+。

        门只看 query 相关性、不看平台分:适老性要始终进三项加权(35/35/30),
        否则"平台互动一般但极度适老"的视频会被错杀(拍板 2026-06-22)。阈值=抖音入池 query 下限 65,
        低于此的视频靠 query 这关就被否、适老性也救不回来,不拉(省成本)。
        单点覆盖首轮 + 游走。注入 bundle["content_audience_50plus"]:
        not_attempted(相关性不够,没拉)/ ok(画像算出)/ unavailable(拉取失败)/ incomplete(字段缺)。
        非抖音不注入该块 → evaluator 走旧 50/50。
        """
        if self.platform != "douyin":
            return
        fetch_portrait = getattr(self.portrait_client, "fetch_account_fans_portrait", None)
        if not callable(fetch_portrait):
            # 平台 client 不提供画像(mock/replay)→ 不注入,evaluator 走旧 50/50。
            return
        for bundle in bundles:
            query = _get_path(bundle, "pattern_match_result.query_relevance_score")
            if not (_is_number(query) and query >= PORTRAIT_QUERY_GATE):
                bundle["content_audience_50plus"] = {"status": "not_attempted"}
                continue
            author_id = _get_path(bundle, "content.author.platform_author_id")
            if not author_id:
                bundle["content_audience_50plus"] = {"status": "incomplete"}
                continue
            author_id = str(author_id)
            if author_id in self._portrait_by_author:
                portrait = self._portrait_by_author[author_id]  # 命中缓存,不再 30s 限速
                fetch_error = self._portrait_error_by_author.get(author_id)
            else:
                portrait, fetch_error = None, None
                for attempt in range(PORTRAIT_FETCH_ATTEMPTS):  # 重试,救瞬时失败/限速
                    try:
                        portrait = fetch_portrait(author_id)
                        fetch_error = None
                        break
                    except Exception as exc:  # 不再吞掉原因:记下来供技术详情展示
                        fetch_error = f"{type(exc).__name__}: {exc}".strip()[:200]
                        portrait = None
                self._portrait_by_author[author_id] = portrait
                self._portrait_error_by_author[author_id] = fetch_error
            if portrait is None:
                bundle["content_audience_50plus"] = {
                    "status": "unavailable",
                    "failure_reason": fetch_error or "热点宝画像接口未返回数据",
                }
                continue
            fans_age = _get_path(portrait, "fans.age.data")
            result = fifty_plus_score(fans_age)
            block: dict[str, Any] = {"status": result["status"]}
            if result["status"] == "ok":
                block["score"] = result["score"]
                block["components"] = {
                    key: result[key]
                    for key in ("band", "band_tgi", "band_percent", "tgi_score", "percent_score")
                }
            bundle["content_audience_50plus"] = block

    def _record_batch_timing_event(
        self,
        query: dict[str, Any],
        batch: list[dict[str, Any]],
        decisions: list[dict[str, Any]],
        *,
        page_number: int,
        page_item_count: int,
        batch_kind: str,
        search_duration_ms: int,
        search_wait_ms: int,
        search_request_duration_ms: int,
        search_timing_source: str,
        oss_batch_duration_ms: int,
        qwen_batch_duration_ms: int,
        portrait_duration_ms: int,
        rule_duration_ms: int,
        discovery_duration_ms: int,
        batch_duration_ms: int,
        has_pass: bool,
    ) -> None:
        self._batch_event_index += 1
        action_counts: dict[str, int] = {}
        for decision in decisions:
            action = str(decision.get("decision_action") or "unknown")
            action_counts[action] = action_counts.get(action, 0) + 1
        raw_payload = {
            "batch_timing_schema_version": "progressive_batch_timing.v1",
            "progressive_batch_id": (
                f"{query['search_query_id']}_p{page_number:02d}_{batch_kind}"
            ),
            "search_query_id": query.get("search_query_id"),
            "search_query": query.get("search_query"),
            "search_query_generation_method": query.get("search_query_generation_method"),
            "page_number": page_number,
            "page_item_count": page_item_count,
            "batch_kind": batch_kind,
            "batch_video_count": len(batch),
            "decision_action_counts": action_counts,
            "has_add_to_content_pool": has_pass,
            "search_duration_ms": search_duration_ms,
            "search_wait_ms": search_wait_ms,
            "search_request_duration_ms": search_request_duration_ms,
            "search_timing_source": search_timing_source,
            "oss_batch_duration_ms": oss_batch_duration_ms,
            "qwen_batch_duration_ms": qwen_batch_duration_ms,
            "portrait_duration_ms": portrait_duration_ms,
            "rule_duration_ms": rule_duration_ms,
            "discovery_duration_ms": discovery_duration_ms,
            "batch_duration_ms": batch_duration_ms,
            "content_ids": [str(item.get("platform_content_id") or "") for item in batch],
        }
        row = with_raw_payload(
            {
                "record_schema_version": RUNTIME_RECORD_SCHEMA_VERSION,
                "run_id": self.run_id,
                "policy_run_id": self.policy_run_id,
                "event_id": (
                    f"evt_progressive_batch_{self._batch_event_index:04d}_"
                    f"{_safe_token(query.get('search_query_id'))}_p{page_number:02d}_{batch_kind}"
                ),
                "event_type": "progressive_batch_timing",
                "status": "success",
                "input_ref": "platform_results",
                "output_ref": "rule_decisions",
                "error_code": None,
                "message": (
                    f"{batch_kind} videos={len(batch)} add={action_counts.get(PASS_ACTION, 0)}"
                ),
                "created_at": datetime.now(timezone.utc).isoformat(),
            }
        )
        row["raw_payload"].update(raw_payload)
        self.runtime.append_jsonl(self.run_id, "run_events.jsonl", [row])

    def _prepare_result(
        self,
        query: dict[str, Any],
        result: dict[str, Any],
        *,
        page_number: int,
        page_item_count: int,
        batch_kind: str,
        rank: int,
    ) -> dict[str, Any]:
        runtime_result = _strip_transient_fields(result)
        prepared = platform_access._with_query_source(runtime_result, query)  # noqa: SLF001
        original_has_more = bool(prepared.get("has_more"))
        original_next_cursor = str(prepared.get("next_cursor") or "")
        prepared.update(
            {
                "search_query_id": query["search_query_id"],
                "content_discovery_id": _content_discovery_id(query["search_query_id"], page_number, rank),
                "progressive_batch_id": (
                    f"{query['search_query_id']}_p{page_number:02d}_{batch_kind}"
                ),
                "progressive_page_number": page_number,
                "progressive_batch_kind": batch_kind,
                "progressive_page_item_count": page_item_count,
                "progressive_item_rank_in_page": rank,
                "progressive_original_has_more": original_has_more,
                "progressive_original_next_cursor": original_next_cursor,
            }
        )
        return prepared

    def _hydrate_result_media(
        self,
        result: dict[str, Any],
        query: dict[str, Any],
    ) -> dict[str, Any]:
        hydrate = getattr(self.platform_client, "hydrate_search_result_media", None)
        if not callable(hydrate):
            return result
        return dict(hydrate(result, query))

    def _accumulate(
        self,
        batch: list[dict[str, Any]],
        recalled: dict[str, Any],
        decisions: list[dict[str, Any]],
    ) -> None:
        for offset, result in enumerate(batch):
            key = _content_key(result)
            if key:
                self._record_indexes_by_key[key] = {
                    "platform_result": len(self.platform_results) + offset,
                    "discovered": len(self.discovered_content_items) + offset,
                    "evidence_bundle": len(self.evidence_bundles) + offset,
                    "decision": len(self.rule_decisions) + offset,
                }
        self.platform_results.extend(batch)
        self.discovered_content_items.extend(recalled["discovered_content_items"])
        self.content_media_records.extend(recalled["content_media_records"])
        self.pattern_recall_evidence.extend(recalled["pattern_recall_evidence"])
        self.evidence_bundles.extend(recalled["evidence_bundles"])
        self.rule_decisions.extend(decisions)

    def _propagate_query_source_merge(
        self,
        key: tuple[str, str],
        merged_result: dict[str, Any],
    ) -> None:
        indexes = self._record_indexes_by_key.get(key)
        if not indexes:
            return
        fields = [
            "query_sources",
            "matched_search_query_ids",
            "matched_search_queries",
            "matched_search_query_generation_methods",
        ]
        for collection_name in ["discovered_content_items"]:
            row = getattr(self, collection_name)[indexes["discovered"]]
            _copy_fields(row, merged_result, fields)
        bundle = self.evidence_bundles[indexes["evidence_bundle"]]
        _copy_fields(bundle.setdefault("source_evidence", {}), merged_result, fields)
        decision = self.rule_decisions[indexes["decision"]]
        _copy_fields(decision.setdefault("source_evidence", {}), merged_result, fields)
        raw_payload = decision.setdefault("raw_payload", {})
        if isinstance(raw_payload.get("source_evidence"), dict):
            _copy_fields(raw_payload["source_evidence"], merged_result, fields)

    def write_runtime(self) -> None:
        if self.archive_dispatcher is not None:
            self._merge_archive_updates(self.archive_dispatcher.drain_completed())
        self.runtime.append_jsonl(self.run_id, "content_media_records.jsonl", self.content_media_records)
        self.runtime.append_jsonl(self.run_id, "pattern_recall_evidence.jsonl", self.pattern_recall_evidence)
        self._hide_consumed_query_cursors()
        self.runtime.append_jsonl(self.run_id, "discovered_content_items.jsonl", self.discovered_content_items)
        self.runtime.append_jsonl(self.run_id, "rule_decisions.jsonl", self.rule_decisions)
        if self.archive_dispatcher is not None:
            self.archive_dispatcher.enable_runtime_writes()

    def close_archive_dispatcher(self) -> None:
        if self.archive_dispatcher is None or self._archive_dispatcher_closed:
            return
        self._archive_dispatcher_closed = True
        self.archive_dispatcher.shutdown(wait=False)

    def _merge_archive_updates(self, updates: list[dict[str, Any]]) -> None:
        if not updates:
            return
        by_key = {
            (row.get("platform"), row.get("platform_content_id")): row
            for row in updates
            if row.get("platform_content_id")
        }
        if not by_key:
            return
        self.content_media_records = [
            by_key.get((row.get("platform"), row.get("platform_content_id")), row)
            for row in self.content_media_records
        ]

    def _hide_consumed_query_cursors(self) -> None:
        if not self._queries_with_consumed_next_pages:
            return
        for row in self.platform_results:
            self._hide_cursor_if_consumed(row)
        for row in self.discovered_content_items:
            self._hide_cursor_if_consumed(row)
        for bundle in self.evidence_bundles:
            source_evidence = bundle.get("source_evidence")
            if isinstance(source_evidence, dict):
                self._hide_cursor_if_consumed(source_evidence)
        for decision in self.rule_decisions:
            source_evidence = decision.get("source_evidence")
            if isinstance(source_evidence, dict):
                self._hide_cursor_if_consumed(source_evidence)
            raw_payload = decision.get("raw_payload")
            if isinstance(raw_payload, dict) and isinstance(raw_payload.get("source_evidence"), dict):
                self._hide_cursor_if_consumed(raw_payload["source_evidence"])

    def _hide_cursor_if_consumed(self, row: dict[str, Any]) -> None:
        if str(row.get("search_query_id") or "") not in self._queries_with_consumed_next_pages:
            return
        row["has_more"] = False
        row["next_cursor"] = ""
        raw_payload = row.get("raw_payload")
        if isinstance(raw_payload, dict):
            raw_payload["has_more"] = False
            raw_payload["next_cursor"] = ""


def _get_path(data: Any, path: str) -> Any:
    cur = data
    for part in path.split("."):
        if not isinstance(cur, dict):
            return None
        cur = cur.get(part)
    return cur


def _elapsed_ms(start: float, end: float) -> int:
    return max(0, int((end - start) * 1000))


def _safe_token(value: Any) -> str:
    token = "".join(ch if ch.isalnum() or ch in {"_", "-"} else "_" for ch in str(value or "query"))
    return token[:80] or "query"


def _is_number(value: Any) -> bool:
    return isinstance(value, (int, float)) and not isinstance(value, bool)


def _content_key(result: dict[str, Any]) -> tuple[str, str] | None:
    content_id = str(result.get("platform_content_id") or "")
    if not content_id:
        return None
    return str(result.get("platform") or ""), content_id


def _strip_transient_fields(result: dict[str, Any]) -> dict[str, Any]:
    return {key: value for key, value in result.items() if not str(key).startswith("_platform_")}


def _content_discovery_id(search_query_id: str, page_number: int, rank: int) -> str:
    if page_number == 1:
        return f"{search_query_id}_content_{rank:03d}"
    return f"{search_query_id}_page_{page_number:03d}_content_{rank:03d}"


def _page_has_more(results: list[dict[str, Any]]) -> bool:
    return any(bool(result.get("has_more")) for result in results)


def _page_next_cursor(results: list[dict[str, Any]]) -> str:
    for result in results:
        cursor = str(result.get("next_cursor") or "")
        if cursor:
            return cursor
    return ""


def _copy_fields(target: dict[str, Any], source: dict[str, Any], fields: list[str]) -> None:
    for field in fields:
        if field in source:
            target[field] = deepcopy(source[field])
    raw_payload = target.get("raw_payload")
    if isinstance(raw_payload, dict):
        for field in fields:
            if field in source:
                raw_payload[field] = deepcopy(source[field])