Add progressive paged screening to run graph

Replace one-shot full-page platform search with progressive screening: fetch
an initial batch, page on demand (rate-limited), and judge as we go. Platform
clients gain search_full_page; downstream builders take a write_runtime flag
so progressive_screening owns persistence and the graph skips redundant steps.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

content_agent/business_modules/content_discovery/content_discovery_builder.py

@@ -20,6 +20,7 @@ def run(
     platform_results: list[dict[str, Any]],
     source_context: dict[str, Any],
     runtime: RuntimeFileStore,
+    write_runtime: bool = True,
 ) -> dict[str, list[dict[str, Any]]]:
     created_at = datetime.now(timezone.utc).isoformat()
     discovered_content_items: list[dict[str, Any]] = []
@@ -41,7 +42,8 @@ def run(
         discovered_content_items.append(discovered_content_item)
         content_media_records.append(content_media_record)
 
-    runtime.append_jsonl(run_id, "content_media_records.jsonl", content_media_records)
+    if write_runtime:
+        runtime.append_jsonl(run_id, "content_media_records.jsonl", content_media_records)
     return {
         "discovered_content_items": discovered_content_items,
         "content_media_records": content_media_records,
@@ -85,6 +87,13 @@ def _build_discovered_content_item(
         "matched_search_query_ids",
         "matched_search_queries",
         "matched_search_query_generation_methods",
+        "progressive_batch_id",
+        "progressive_page_number",
+        "progressive_batch_kind",
+        "progressive_page_item_count",
+        "progressive_item_rank_in_page",
+        "progressive_original_has_more",
+        "progressive_original_next_cursor",
     ]:
         if field in result:
             item[field] = result[field]

content_agent/business_modules/progressive_screening.py

@@ -0,0 +1,413 @@
+from __future__ import annotations
+
+import time
+from copy import deepcopy
+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.errors import ContentAgentError, ErrorCode
+from content_agent.interfaces import GeminiVideoClient, PlatformSearchClient, RuntimeFileStore
+
+
+INITIAL_BATCH_SIZE = 3
+MAX_PAGES = 3
+SEARCH_MIN_INTERVAL_SECONDS = 15.0
+PASS_ACTION = "ADD_TO_CONTENT_POOL"
+
+
+class SearchCallLimiter:
+    def __init__(
+        self,
+        min_interval_seconds: float = SEARCH_MIN_INTERVAL_SECONDS,
+        *,
+        now_fn: Callable[[], float] = time.monotonic,
+        sleep_fn: Callable[[float], None] = time.sleep,
+    ) -> None:
+        self.min_interval_seconds = min_interval_seconds
+        self.now_fn = now_fn
+        self.sleep_fn = sleep_fn
+        self._last_call_at: float | None = None
+
+    def wait(self) -> None:
+        if self._last_call_at is not None:
+            remaining = self.min_interval_seconds - (self.now_fn() - self._last_call_at)
+            if remaining > 0:
+                self.sleep_fn(remaining)
+        self._last_call_at = self.now_fn()
+
+
+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,
+) -> dict[str, list[dict[str, Any]]]:
+    if limiter is None and getattr(platform_client, "requires_progressive_search_rate_limit", False):
+        limiter = SearchCallLimiter()
+    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,
+    )
+    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,
+    }
+
+
+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,
+        limiter: SearchCallLimiter | 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.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()
+
+    def process_query(self, query: dict[str, Any]) -> None:
+        first_page = self._fetch_page(query, page_number=1, cursor=query.get("page_cursor"))
+        if first_page is None:
+            return
+        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,
+        )
+        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,
+        )
+        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:
+            page = self._fetch_page(query, page_number=page_number, cursor=cursor)
+            if page is None or not page:
+                return
+            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,
+            )
+            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,
+    ) -> list[dict[str, Any]] | None:
+        page_query = dict(query)
+        if page_number > 1:
+            page_query["page_cursor"] = str(cursor or "")
+        try:
+            if self.limiter is not None:
+                self.limiter.wait()
+            search = getattr(self.platform_client, "search_full_page", None)
+            if not callable(search):
+                search = self.platform_client.search
+            return list(search(page_query))
+        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,
+    ) -> bool:
+        batch = []
+        for rank, result in enumerate(page_results, start=1):
+            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
+            existing = self._platform_result_by_key.get(key)
+            if existing:
+                platform_access._append_query_source(existing, query)  # noqa: SLF001
+                self._propagate_query_source_merge(key, existing)
+                continue
+            self._platform_result_by_key[key] = prepared
+            batch.append(prepared)
+        if not batch:
+            return False
+
+        discovered = content_discovery.run(
+            self.run_id,
+            self.policy_run_id,
+            batch,
+            self.source_context,
+            self.runtime,
+            write_runtime=False,
+        )
+        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=len(self.discovered_content_items) + 1,
+            write_runtime=False,
+        )
+        decisions = rule_judgment.run(
+            self.run_id,
+            self.policy_run_id,
+            recalled["evidence_bundles"],
+            self.policy_bundle,
+            self.runtime,
+            start_index=len(self.rule_decisions) + 1,
+            write_runtime=False,
+        )
+        self._accumulate(batch, recalled, decisions)
+        return any(decision.get("decision_action") == PASS_ACTION for decision in decisions)
+
+    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]:
+        prepared = platform_access._with_query_source(dict(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 _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:
+        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)
+
+    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 _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 _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])

content_agent/business_modules/rule_judgment/__init__.py

@@ -13,10 +13,12 @@ def run(
     policy_bundle: dict[str, Any],
     runtime: RuntimeFileStore,
     start_index: int = 1,
+    write_runtime: bool = True,
 ) -> list[dict[str, Any]]:
     decisions = [
         decide(run_id, policy_run_id, start_index + index, bundle, policy_bundle)
         for index, bundle in enumerate(evidence_bundles)
     ]
-    runtime.append_jsonl(run_id, "rule_decisions.jsonl", decisions)
+    if write_runtime:
+        runtime.append_jsonl(run_id, "rule_decisions.jsonl", decisions)
     return decisions

content_agent/graph.py

@@ -12,8 +12,8 @@ from content_agent.errors import ContentAgentError
 from content_agent.business_modules import (
     content_discovery,
     learning_review,
-    platform_access,
     policy_version,
+    progressive_screening,
     result_source_lookup,
     rule_judgment,
     run_record,
@@ -93,10 +93,21 @@ def build_run_graph(deps: RunDependencies):
         return {"search_queries": search_queries, "current_step": "plan_queries"}
 
     def search_platform(state: RunState) -> dict[str, Any]:
-        result = platform_access.run(state["search_queries"], deps.platform_client)
+        result = progressive_screening.run(
+            run_id=state["run_id"],
+            policy_run_id=state["policy_run_id"],
+            search_queries=state["search_queries"],
+            source_context=state["source_context"],
+            policy_bundle=state["policy_bundle"],
+            platform_client=deps.platform_client,
+            runtime=deps.runtime,
+            gemini_video_client=deps.gemini_video_client,
+        )
         return {**result, "current_step": "search_platform"}
 
     def build_discovered_content(state: RunState) -> dict[str, Any]:
+        if state.get("discovered_content_items") is not None:
+            return {"current_step": "build_discovered_content"}
         result = content_discovery.run(
             state["run_id"],
             state["policy_run_id"],
@@ -107,6 +118,8 @@ def build_run_graph(deps: RunDependencies):
         return {**result, "current_step": "build_discovered_content"}
 
     def recall_pattern(state: RunState) -> dict[str, Any]:
+        if state.get("pattern_recall_evidence") is not None:
+            return {"current_step": "recall_pattern"}
         result = pattern_recall.run(
             state["run_id"],
             state["policy_run_id"],
@@ -130,6 +143,8 @@ def build_run_graph(deps: RunDependencies):
         }
 
     def evaluate_rules(state: RunState) -> dict[str, Any]:
+        if state.get("rule_decisions") is not None:
+            return {"current_step": "evaluate_rules"}
         decisions = rule_judgment.run(
             state["run_id"],
             state["policy_run_id"],
@@ -208,11 +223,11 @@ def build_run_graph(deps: RunDependencies):
 
     graph.add_edge(START, "load_source")
     graph.add_edge("load_source", "plan_queries")
-    graph.add_edge("plan_queries", "search_platform")
+    graph.add_edge("plan_queries", "load_policy")
+    graph.add_edge("load_policy", "search_platform")
     graph.add_edge("search_platform", "build_discovered_content")
     graph.add_edge("build_discovered_content", "recall_pattern")
-    graph.add_edge("recall_pattern", "load_policy")
-    graph.add_edge("load_policy", "evaluate_rules")
+    graph.add_edge("recall_pattern", "evaluate_rules")
     graph.add_edge("evaluate_rules", "execute_walk")
     graph.add_edge("execute_walk", "record_run")
     graph.add_edge("record_run", "commit_results")

content_agent/integrations/douyin.py

@@ -32,6 +32,8 @@ BLOGGER_RATE_LIMIT_BUCKET = "douyin_blogger"
 
 
 class CrawapiDouyinClient:
+    requires_progressive_search_rate_limit = True
+
     def __init__(
         self,
         base_url: str,
@@ -92,6 +94,16 @@ class CrawapiDouyinClient:
         )
 
     def search(self, query: dict[str, Any]) -> list[dict[str, Any]]:
+        return self._search(query, self.max_results_per_query)
+
+    def search_full_page(self, query: dict[str, Any]) -> list[dict[str, Any]]:
+        return self._search(query, None)
+
+    def _search(
+        self,
+        query: dict[str, Any],
+        max_results_per_query: int | None,
+    ) -> list[dict[str, Any]]:
         payload = {
             "keyword": query["search_query"],
             "content_type": self.default_content_type,
@@ -110,7 +122,7 @@ class CrawapiDouyinClient:
         next_cursor = str(data_block.get("next_cursor") or "")
 
         results: list[dict[str, Any]] = []
-        selected_items = items[: self.max_results_per_query] if self.max_results_per_query else items
+        selected_items = items[:max_results_per_query] if max_results_per_query else items
         for index, item in enumerate(selected_items, start=1):
             results.append(self._normalize_content_item(query, item, index, has_more, next_cursor))
         return results

content_agent/integrations/kuaishou.py

@@ -110,6 +110,8 @@ def _normalize_kuaishou_item(
 
 
 class CrawapiKuaishouClient:
+    requires_progressive_search_rate_limit = True
+
     def __init__(
         self,
         base_url: str,
@@ -162,6 +164,16 @@ class CrawapiKuaishouClient:
         )
 
     def search(self, query: dict[str, Any]) -> list[dict[str, Any]]:
+        return self._search(query, self.max_results_per_query)
+
+    def search_full_page(self, query: dict[str, Any]) -> list[dict[str, Any]]:
+        return self._search(query, None)
+
+    def _search(
+        self,
+        query: dict[str, Any],
+        max_results_per_query: int | None,
+    ) -> list[dict[str, Any]]:
         data = self._post_json(
             self.keyword_path,
             {"keyword": query["search_query"]},
@@ -172,7 +184,7 @@ class CrawapiKuaishouClient:
         items = block.get("data", []) if isinstance(block.get("data"), list) else []
         has_more = bool(block.get("has_more", False))
         next_cursor = str(block.get("next_cursor") or "")
-        selected = items[: self.max_results_per_query] if self.max_results_per_query else items
+        selected = items[:max_results_per_query] if max_results_per_query else items
         return [
             _normalize_kuaishou_item(query, item, index, has_more, next_cursor)
             for index, item in enumerate(selected, start=1)

content_agent/integrations/shipinhao.py

@@ -101,6 +101,8 @@ def _normalize_shipinhao_item(
 
 
 class CrawapiShipinhaoClient:
+    requires_progressive_search_rate_limit = True
+
     def __init__(
         self,
         base_url: str,
@@ -140,6 +142,16 @@ class CrawapiShipinhaoClient:
         )
 
     def search(self, query: dict[str, Any]) -> list[dict[str, Any]]:
+        return self._search(query, self.max_results_per_query)
+
+    def search_full_page(self, query: dict[str, Any]) -> list[dict[str, Any]]:
+        return self._search(query, None)
+
+    def _search(
+        self,
+        query: dict[str, Any],
+        max_results_per_query: int | None,
+    ) -> list[dict[str, Any]]:
         payload = {
             "keyword": query["search_query"],
             "cursor": str(query.get("page_cursor") or ""),
@@ -177,7 +189,7 @@ class CrawapiShipinhaoClient:
         items = block.get("data", []) if isinstance(block.get("data"), list) else []
         has_more = bool(block.get("has_more", False))
         next_cursor = str(block.get("next_cursor") or "")
-        selected = items[: self.max_results_per_query] if self.max_results_per_query else items
+        selected = items[:max_results_per_query] if max_results_per_query else items
         return [
             _normalize_shipinhao_item(query, item, index, has_more, next_cursor)
             for index, item in enumerate(selected, start=1)

tests/test_douyin_client.py

@@ -253,6 +253,33 @@ def test_douyin_keyword_search_default_limit_is_five():
     assert [result["platform_content_id"] for result in results] == ["0", "1", "2", "3", "4"]
 
 
+def test_douyin_keyword_search_full_page_bypasses_local_limit():
+    client = CrawapiDouyinClient(
+        base_url="http://crawapi.test",
+        keyword_path="/crawler/dou_yin/keyword",
+        max_results_per_query=1,
+        http_client=FakeHttpClient(
+            [
+                _response(
+                    200,
+                    {
+                        "data": {
+                            "data": [
+                                {RAW_CONTENT_ID_KEY: str(index), "author": {}, "statistics": {}}
+                                for index in range(3)
+                            ]
+                        }
+                    },
+                ),
+            ]
+        ),
+    )
+
+    results = client.search_full_page(_search_query("整页"))
+
+    assert [result["platform_content_id"] for result in results] == ["0", "1", "2"]
+
+
 def _author_query(author_id="MS4wLjABAAAA001", **extra):
     return {
         "search_query_id": "author_001",

tests/test_kuaishou_client.py

@@ -108,6 +108,16 @@ def test_kuaishou_search_limits_to_five_by_default():
     ]
 
 
+def test_kuaishou_search_full_page_bypasses_local_limit():
+    items = [_item(f"ks_{index}") for index in range(3)]
+    client = _client([_response(200, {"code": 0, "data": {"data": items}})])
+    client.max_results_per_query = 1
+
+    results = client.search_full_page(_query())
+
+    assert [result["platform_content_id"] for result in results] == ["ks_0", "ks_1", "ks_2"]
+
+
 def test_kuaishou_detail_maps_canonical_fields():
     detail = _item("ks_detail")
     detail["share_count"] = 78

tests/test_progressive_screening.py

@@ -0,0 +1,326 @@
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any
+
+from content_agent.business_modules import (
+    learning_review,
+    progressive_screening,
+    result_source_lookup,
+    run_record,
+    source_seed,
+    walk_engine,
+)
+from content_agent.business_modules.run_record.validation import validate_run
+from content_agent.constants import RUNTIME_RECORD_SCHEMA_VERSION
+from content_agent.integrations.policy_json import JsonPolicyBundleStore
+from content_agent.integrations.runtime_files import LocalRuntimeFileStore
+from content_agent.record_payload import with_raw_payload
+from tests.gemini_helpers import FakeGeminiVideoClient, fake_gemini_pool, fake_gemini_review
+from tests.p1_helpers import real_source_payload
+
+
+class FakeProgressivePlatformClient:
+    def __init__(self, pages: dict[str, list[dict[str, Any]]]) -> None:
+        self.pages = pages
+        self.calls: list[dict[str, Any]] = []
+
+    def search_full_page(self, query: dict[str, Any]) -> list[dict[str, Any]]:
+        self.calls.append(dict(query))
+        cursor = str(query.get("page_cursor") or "")
+        return [dict(item) for item in self.pages.get(cursor, [])]
+
+    def search(self, query: dict[str, Any]) -> list[dict[str, Any]]:
+        return self.search_full_page(query)
+
+
+class FakeClock:
+    def __init__(self) -> None:
+        self.now = 0.0
+        self.sleeps: list[float] = []
+
+    def monotonic(self) -> float:
+        return self.now
+
+    def sleep(self, seconds: float) -> None:
+        self.sleeps.append(seconds)
+        self.now += seconds
+
+
+def test_progressive_screening_two_item_page_has_no_remainder(tmp_path):
+    context = _context(tmp_path)
+    client = FakeProgressivePlatformClient({"": _page(["c1", "c2"], has_more=True, cursor="next")})
+    gemini = FakeGeminiVideoClient(default_result=fake_gemini_pool())
+
+    result = _run_screening(context, client, gemini)
+
+    assert len(client.calls) == 1
+    assert [item["platform_content_id"] for item in result["platform_results"]] == ["c1", "c2"]
+    assert len(gemini.calls) == 2
+
+
+def test_progressive_screening_uses_actual_page_size_without_hardcoded_ten(tmp_path):
+    context = _context(tmp_path)
+    ids = [f"c{i}" for i in range(15)]
+    client = FakeProgressivePlatformClient({"": _page(ids, has_more=True, cursor="next")})
+    gemini = FakeGeminiVideoClient(default_result=fake_gemini_review())
+
+    result = _run_screening(context, client, gemini)
+
+    assert len(client.calls) == 1
+    assert [item["platform_content_id"] for item in result["platform_results"]] == ["c0", "c1", "c2"]
+    assert {item["progressive_page_item_count"] for item in result["discovered_content_items"]} == {15}
+
+
+def test_progressive_screening_pool_in_first_batch_processes_remainder_only(tmp_path):
+    context = _context(tmp_path)
+    ids = [f"c{i}" for i in range(10)]
+    client = FakeProgressivePlatformClient({"": _page(ids, has_more=True, cursor="next")})
+    gemini = FakeGeminiVideoClient(
+        result_by_content_id={"c0": fake_gemini_pool()},
+        default_result=fake_gemini_review(),
+    )
+
+    result = _run_screening(context, client, gemini)
+
+    assert len(client.calls) == 1
+    assert len(result["platform_results"]) == 10
+    assert {decision["decision_action"] for decision in result["rule_decisions"]} == {
+        "ADD_TO_CONTENT_POOL",
+        "KEEP_CONTENT_FOR_REVIEW",
+    }
+
+
+def test_progressive_screening_remainder_and_page_two_gate_page_three(tmp_path):
+    context = _context(tmp_path)
+    pages = {
+        "": _page([f"p1_{i}" for i in range(10)], has_more=True, cursor="cursor_2"),
+        "cursor_2": _page([f"p2_{i}" for i in range(9)], has_more=True, cursor="cursor_3"),
+        "cursor_3": _page([f"p3_{i}" for i in range(11)], has_more=True, cursor="cursor_4"),
+    }
+    client = FakeProgressivePlatformClient(pages)
+    gemini = FakeGeminiVideoClient(
+        result_by_content_id={
+            "p1_0": fake_gemini_pool(),
+            "p1_3": fake_gemini_pool(),
+            "p2_0": fake_gemini_pool(),
+            "p3_0": fake_gemini_pool(),
+        },
+        default_result=fake_gemini_review(),
+    )
+    clock = FakeClock()
+    limiter = progressive_screening.SearchCallLimiter(
+        now_fn=clock.monotonic,
+        sleep_fn=clock.sleep,
+    )
+
+    result = _run_screening(context, client, gemini, limiter=limiter)
+
+    assert [call.get("page_cursor", "") for call in client.calls] == ["", "cursor_2", "cursor_3"]
+    assert len(result["platform_results"]) == 30
+    assert clock.sleeps == [15.0, 15.0]
+
+
+def test_progressive_screening_keep_for_review_does_not_pass(tmp_path):
+    context = _context(tmp_path)
+    client = FakeProgressivePlatformClient({"": _page([f"c{i}" for i in range(10)], has_more=True, cursor="next")})
+    gemini = FakeGeminiVideoClient(default_result=fake_gemini_review())
+
+    result = _run_screening(context, client, gemini)
+
+    assert len(client.calls) == 1
+    assert len(result["platform_results"]) == 3
+    assert {decision["decision_action"] for decision in result["rule_decisions"]} == {
+        "KEEP_CONTENT_FOR_REVIEW"
+    }
+
+
+def test_progressive_screening_deduplicates_across_batches(tmp_path):
+    context = _context(tmp_path)
+    client = FakeProgressivePlatformClient(
+        {
+            "": _page(["dup", "c1", "c2", "dup", "c4"], has_more=True, cursor="next"),
+        }
+    )
+    gemini = FakeGeminiVideoClient(
+        result_by_content_id={"dup": fake_gemini_pool()},
+        default_result=fake_gemini_review(),
+    )
+
+    result = _run_screening(context, client, gemini)
+
+    assert [item["platform_content_id"] for item in result["platform_results"]] == [
+        "dup",
+        "c1",
+        "c2",
+        "c4",
+    ]
+    assert len(result["rule_decisions"]) == 4
+
+
+def test_progressive_screening_runtime_validates_and_does_not_duplicate_old_paging(tmp_path):
+    context = _context(tmp_path)
+    pages = {
+        "": _page([f"p1_{i}" for i in range(10)], has_more=True, cursor="cursor_2"),
+        "cursor_2": _page([f"p2_{i}" for i in range(10)], has_more=True, cursor="cursor_3"),
+    }
+    client = FakeProgressivePlatformClient(pages)
+    gemini = FakeGeminiVideoClient(
+        result_by_content_id={"p1_0": fake_gemini_pool(), "p1_3": fake_gemini_pool()},
+        default_result=fake_gemini_review(),
+    )
+    clock = FakeClock()
+    limiter = progressive_screening.SearchCallLimiter(
+        now_fn=clock.monotonic,
+        sleep_fn=clock.sleep,
+    )
+
+    result = _run_screening(context, client, gemini, limiter=limiter)
+    walk_result = walk_engine.run_bounded_walk(
+        run_id=context["run_id"],
+        policy_run_id=context["policy_run_id"],
+        pattern_seed_pack=context["pattern_seed_pack"],
+        source_context=context["source_context"],
+        search_queries=context["search_queries"],
+        discovered_content_items=result["discovered_content_items"],
+        content_media_records=result["content_media_records"],
+        evidence_bundles=result["evidence_bundles"],
+        rule_decisions=result["rule_decisions"],
+        policy_bundle=context["policy_bundle"],
+        platform_client=client,
+        runtime=context["runtime"],
+        gemini_video_client=gemini,
+    )
+    record = run_record.run(
+        context["run_id"],
+        context["policy_run_id"],
+        context["search_queries"],
+        walk_result["discovered_content_items"],
+        walk_result["rule_decisions"],
+        walk_result["source_path_record_basis"],
+        context["policy_bundle"],
+        context["runtime"],
+        walk_actions=walk_result["walk_actions"],
+        query_failures=result["query_failures"],
+    )
+    result_source_lookup.run(
+        context["run_id"],
+        context["policy_run_id"],
+        context["policy_bundle"],
+        walk_result["discovered_content_items"],
+        walk_result["content_media_records"],
+        walk_result["rule_decisions"],
+        record["source_path_records"],
+        record["search_clues"],
+        context["runtime"],
+    )
+    learning_review.run(context["run_id"], context["policy_run_id"], context["runtime"])
+
+    validation = validate_run(context["run_id"], context["runtime"])
+    page_calls = [call for call in client.calls if call.get("page_cursor") == "cursor_2"]
+    assert len(page_calls) == 1
+    assert validation["status"] == "pass", validation["findings"]
+
+
+def test_progressive_screening_keeps_cursor_when_next_page_was_not_consumed(tmp_path):
+    context = _context(tmp_path)
+    client = FakeProgressivePlatformClient({"": _page([f"c{i}" for i in range(10)], has_more=True, cursor="next")})
+    gemini = FakeGeminiVideoClient(
+        result_by_content_id={"c0": fake_gemini_pool()},
+        default_result=fake_gemini_review(),
+    )
+
+    result = _run_screening(context, client, gemini)
+
+    assert len(client.calls) == 1
+    assert {item["next_cursor"] for item in result["discovered_content_items"]} == {"next"}
+    assert {item["has_more"] for item in result["discovered_content_items"]} == {True}
+
+
+def _context(tmp_path: Path) -> dict[str, Any]:
+    run_id = "run_progressive"
+    policy_run_id = "policy_progressive"
+    runtime = LocalRuntimeFileStore(tmp_path / "runtime")
+    runtime.prepare_run(run_id)
+    seed = source_seed.run(run_id, policy_run_id, real_source_payload(), runtime)
+    query = with_raw_payload(
+        {
+            "record_schema_version": RUNTIME_RECORD_SCHEMA_VERSION,
+            "run_id": run_id,
+            "policy_run_id": policy_run_id,
+            "search_query_id": "q_001",
+            "search_query": "早上好",
+            "search_query_generation_method": "item_single",
+            "discovery_start_source": "pattern_itemset",
+            "previous_discovery_step": "search_query_generated",
+            "pattern_seed_ref": {"seed_term": "早上好"},
+        }
+    )
+    runtime.append_jsonl(run_id, "search_queries.jsonl", [query])
+    return {
+        "run_id": run_id,
+        "policy_run_id": policy_run_id,
+        "search_queries": [query],
+        "source_context": seed["source_context"],
+        "pattern_seed_pack": seed["pattern_seed_pack"],
+        "policy_bundle": JsonPolicyBundleStore(Path(".")).load_policy_bundle("V4"),
+        "runtime": runtime,
+    }
+
+
+def _run_screening(
+    context: dict[str, Any],
+    client: FakeProgressivePlatformClient,
+    gemini: FakeGeminiVideoClient,
+    *,
+    limiter: progressive_screening.SearchCallLimiter | None = None,
+) -> dict[str, Any]:
+    kwargs = {
+        "run_id": context["run_id"],
+        "policy_run_id": context["policy_run_id"],
+        "search_queries": context["search_queries"],
+        "source_context": context["source_context"],
+        "policy_bundle": context["policy_bundle"],
+        "platform_client": client,
+        "runtime": context["runtime"],
+        "gemini_video_client": gemini,
+    }
+    if limiter is not None:
+        kwargs["limiter"] = limiter
+    return progressive_screening.run(**kwargs)
+
+
+def _page(ids: list[str], *, has_more: bool, cursor: str) -> list[dict[str, Any]]:
+    return [_item(content_id, has_more=has_more, cursor=cursor) for content_id in ids]
+
+
+def _item(content_id: str, *, has_more: bool, cursor: str) -> dict[str, Any]:
+    return {
+        "content_discovery_id": f"q_001_content_{content_id}",
+        "search_query_id": "q_001",
+        "platform": "douyin",
+        "platform_content_id": content_id,
+        "platform_content_format": "video",
+        "description": f"内容 {content_id}",
+        "platform_author_id": f"author_{content_id}",
+        "author_display_name": "作者",
+        "statistics": {
+            "digg_count": 1_000_000,
+            "comment_count": 50_000,
+            "share_count": 20_000,
+            "collect_count": 100_000,
+            "play_count": 0,
+        },
+        "tags": [],
+        "play_url": f"https://video.test/{content_id}.mp4",
+        "has_more": has_more,
+        "next_cursor": cursor,
+        "score": 72,
+        "risk_level": "unknown",
+        "discovery_relation": "derived_from_pattern_demand",
+        "discovery_start_source": "pattern_itemset",
+        "previous_discovery_step": "search_query_direct",
+        "content_metadata_source": "fake_progressive_search",
+        "platform_auth_mode": "no_bearer",
+        "platform_raw_payload": {"channel_content_id": content_id},
+    }

tests/test_shipinhao_client.py

@@ -103,6 +103,27 @@ def test_shipinhao_search_default_limit_is_five():
     ]
 
 
+def test_shipinhao_search_full_page_bypasses_local_limit():
+    items = [
+        {
+            "channel_content_id": f"finderobj_{index}",
+            "title": "圆形彩虹",
+            "content_type": "video",
+        }
+        for index in range(3)
+    ]
+    client, _ = _client([_response(200, {"code": 0, "data": {"data": items}})])
+    client.max_results_per_query = 1
+
+    results = client.search_full_page(_query())
+
+    assert [result["platform_content_id"] for result in results] == [
+        "finderobj_0",
+        "finderobj_1",
+        "finderobj_2",
+    ]
+
+
 def test_shipinhao_search_retries_on_25011_then_succeeds():
     client, sleeps = _client([_response(200, _FAIL_25011), _response(200, _SUCCESS)])
     result = client.search(_query())