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 FakeHydratingProgressivePlatformClient(FakeProgressivePlatformClient): def __init__(self, pages: dict[str, list[dict[str, Any]]]) -> None: super().__init__(pages) self.metadata_calls: list[dict[str, Any]] = [] self.hydrate_calls: list[str] = [] def search_full_page_metadata(self, query: dict[str, Any]) -> list[dict[str, Any]]: self.metadata_calls.append(dict(query)) cursor = str(query.get("page_cursor") or "") return [ {**dict(item), "_platform_search_item": dict(item)} for item in self.pages.get(cursor, []) ] def hydrate_search_result_media( self, result: dict[str, Any], query: dict[str, Any], ) -> dict[str, Any]: content_id = str(result.get("platform_content_id") or "") self.hydrate_calls.append(content_id) hydrated = { key: value for key, value in result.items() if not str(key).startswith("_platform_") } hydrated["play_url"] = f"https://video.example/{content_id}.mp4" return hydrated 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_archives_batch_before_gemini_judgment(tmp_path): context = _context(tmp_path) client = FakeProgressivePlatformClient({"": _page(["c1", "c2", "c3"], has_more=True, cursor="next")}) archive_dispatcher = BatchArchiveDispatcher() gemini = OssRequiredGeminiVideoClient() result = _run_screening(context, client, gemini, archive_dispatcher=archive_dispatcher) assert sorted(archive_dispatcher.archived_content_ids) == ["c1", "c2", "c3"] assert sorted(call["media"]["oss_url"] for call in gemini.calls) == [ "https://res.example/c1.mp4", "https://res.example/c2.mp4", "https://res.example/c3.mp4", ] assert {row["content_media_status"] for row in result["content_media_records"]} == {"oss_uploaded"} def test_progressive_screening_records_batch_timing_event(tmp_path): context = _context(tmp_path) client = FakeProgressivePlatformClient({"": _page(["c1", "c2", "c3"], has_more=True, cursor="next")}) archive_dispatcher = BatchArchiveDispatcher() gemini = FakeGeminiVideoClient(default_result=fake_gemini_pool()) _run_screening(context, client, gemini, archive_dispatcher=archive_dispatcher) events = context["runtime"].read_jsonl(context["run_id"], "run_events.jsonl") timing_events = [row for row in events if row.get("event_type") == "progressive_batch_timing"] assert len(timing_events) == 1 payload = timing_events[0]["raw_payload"] assert payload["batch_kind"] == "initial_top3" assert payload["search_timing_source"] == "page_fetch" for field in ( "search_duration_ms", "oss_batch_duration_ms", "qwen_batch_duration_ms", "portrait_duration_ms", "rule_duration_ms", ): assert isinstance(payload[field], int) assert payload[field] >= 0 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 == [30.0, 30.0] def test_douyin_search_limiter_uses_ten_to_twelve_second_jitter(): limiter = progressive_screening.make_search_limiter("douyin") assert limiter.min_interval_seconds == 10.0 assert limiter.max_interval_seconds == 12.0 clock = FakeClock() limiter = progressive_screening.SearchCallLimiter( min_interval_seconds=10.0, max_interval_seconds=12.0, now_fn=clock.monotonic, sleep_fn=clock.sleep, random_fn=lambda lo, hi: hi, ) limiter.wait() limiter.wait() assert clock.sleeps == [12.0] def test_kuaishou_and_shipinhao_search_limiter_uses_ten_to_twelve_second_jitter(): for platform in ("kuaishou", "shipinhao"): limiter = progressive_screening.make_search_limiter(platform) assert limiter.min_interval_seconds == 10.0 assert limiter.max_interval_seconds == 12.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_hydrates_only_initial_batch_without_pass(tmp_path): context = _context(tmp_path) client = FakeHydratingProgressivePlatformClient( {"": _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 client.metadata_calls == [context["search_queries"][0]] assert client.calls == [] assert client.hydrate_calls == ["c0", "c1", "c2"] assert len(result["platform_results"]) == 3 assert all("_platform_search_item" not in row for row in result["platform_results"]) def test_progressive_screening_hydrates_remainder_only_after_first_batch_pass(tmp_path): context = _context(tmp_path) client = FakeHydratingProgressivePlatformClient( {"": _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 client.hydrate_calls == [f"c{i}" for i in range(10)] assert len(result["platform_results"]) == 10 def test_progressive_screening_does_not_hydrate_duplicate_again(tmp_path): context = _context(tmp_path) client = FakeHydratingProgressivePlatformClient( {"": _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 client.hydrate_calls == ["dup", "c1", "c2", "c4"] assert [item["platform_content_id"] for item in result["platform_results"]] == [ "dup", "c1", "c2", "c4", ] 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, archive_dispatcher: Any | 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 if archive_dispatcher is not None: kwargs["archive_dispatcher"] = archive_dispatcher return progressive_screening.run(**kwargs) class BatchArchiveDispatcher: def __init__(self) -> None: self.archived_content_ids: list[str] = [] self.closed = False def archive_records(self, records: list[dict[str, Any]]) -> list[dict[str, Any]]: archived = [] for row in records: content_id = row["platform_content_id"] self.archived_content_ids.append(content_id) raw_payload = { **(row.get("raw_payload") or {}), "content_media_status": "oss_uploaded", "oss_url": f"https://res.example/{content_id}.mp4", "oss_archive_status": "uploaded", } archived.append( { **row, "content_media_status": "oss_uploaded", "oss_url": f"https://res.example/{content_id}.mp4", "raw_payload": raw_payload, } ) return archived def drain_completed(self) -> list[dict[str, Any]]: return [] def enable_runtime_writes(self) -> None: return None def shutdown(self, *, wait: bool = False) -> None: self.closed = True class OssRequiredGeminiVideoClient(FakeGeminiVideoClient): def analyze( self, content: dict[str, Any], media: dict[str, Any], source_context: dict[str, Any], ) -> dict[str, Any]: assert media["content_media_status"] == "oss_uploaded" assert media["oss_url"] == f"https://res.example/{content['platform_content_id']}.mp4" return super().analyze(content, media, source_context) 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": { # M11 re-baseline:平台分改"量级+收缩比例"。原 digg=1M/低占比新算只得 57.36, # 使 pool 桩(q=80)总分 68.68<70 与 ADD 动作冲突(validate_run 报 threshold_mismatch)。 # 改为中等共鸣占比 → 平台分 ~66:pool 桩总分 ~73≥70 真入池;review 桩(q=60)总分 ~63 仍复看。 "digg_count": 500_000, "comment_count": 24_000, "share_count": 40_000, "collect_count": 120_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}, }