Refine V4 query sources and walk budgets

content_agent/business_modules/content_discovery/pattern_recall/recall_decision.py

@@ -9,6 +9,7 @@ id 编号、三个 list 的组装与落盘全部留主线程按 offset 串行
 from __future__ import annotations
 
 import os
+import time
 from datetime import datetime, timezone
 from typing import Any
 
@@ -40,6 +41,7 @@ def run(
     archive_dispatcher: Any | None = None,
 ) -> dict[str, Any]:
     created_at = datetime.now(timezone.utc).isoformat()
+    archive_started_at = time.monotonic()
     content_media_records = oss_archive.mark_archive_pending(content_media_records)
     if archive_dispatcher is not None:
         archive_records = getattr(archive_dispatcher, "archive_records", None)
@@ -47,12 +49,15 @@ def run(
             content_media_records = archive_records(content_media_records)
         else:
             archive_dispatcher.submit_records(content_media_records)
+    archive_duration_ms = _elapsed_ms(archive_started_at, time.monotonic())
     media_by_content_id = {
         row["platform_content_id"]: row for row in content_media_records
     }
+    judgment_started_at = time.monotonic()
     judgments = _collect_judgments(
         discovered_content_items, media_by_content_id, source_context, gemini_video_client
     )
+    judgment_duration_ms = _elapsed_ms(judgment_started_at, time.monotonic())
     evidence_rows: list[dict[str, Any]] = []
     updated_items: list[dict[str, Any]] = []
     updated_bundles: list[dict[str, Any]] = []
@@ -82,6 +87,10 @@ def run(
         "content_media_records": updated_media_records,
         "discovered_content_items": updated_items,
         "evidence_bundles": updated_bundles,
+        "batch_timing": {
+            "oss_batch_duration_ms": archive_duration_ms,
+            "qwen_batch_duration_ms": judgment_duration_ms,
+        },
     }
 
 
@@ -199,6 +208,10 @@ def _resolve_max_workers() -> int:
         return 4
 
 
+def _elapsed_ms(start: float, end: float) -> int:
+    return max(0, int((end - start) * 1000))
+
+
 def _build_pattern_match_result(judgment: dict[str, Any], recall_evidence_id: str) -> dict[str, Any]:
     final_status = str(judgment.get("final_status") or judgment.get("status") or "ok")
     result = {

content_agent/business_modules/progressive_screening.py

@@ -2,13 +2,17 @@ from __future__ import annotations
 
 import time
 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
@@ -35,12 +39,23 @@ class SearchCallLimiter:
         self.sleep_fn = sleep_fn
         self._last_call_at: float | None = None
 
-    def wait(self) -> None:
+    def wait(self) -> float:
+        waited_seconds = 0.0
         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)
+                waited_seconds = remaining
         self._last_call_at = self.now_fn()
+        return waited_seconds
+
+
+@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(
@@ -54,11 +69,14 @@ def run(
     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 = SearchCallLimiter()
+    if portrait_limiter is None and getattr(platform_client, "requires_progressive_search_rate_limit", False):
+        portrait_limiter = SearchCallLimiter()
     context = _ProgressiveContext(
         run_id=run_id,
         policy_run_id=policy_run_id,
@@ -68,6 +86,7 @@ def run(
         runtime=runtime,
         gemini_video_client=gemini_video_client,
         limiter=limiter,
+        portrait_limiter=portrait_limiter,
         archive_dispatcher=archive_dispatcher,
         platform=platform,
     )
@@ -107,8 +126,9 @@ class _ProgressiveContext:
         platform_client: PlatformSearchClient,
         runtime: RuntimeFileStore,
         gemini_video_client: GeminiVideoClient,
-        limiter: SearchCallLimiter | None,
         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,
@@ -123,6 +143,7 @@ class _ProgressiveContext:
         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]] = []
@@ -150,6 +171,7 @@ class _ProgressiveContext:
         # 逐条 30s 限速拉画像 → 同作者只拉一次,把调用量从"视频数"降到"作者数"。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页)。
@@ -166,7 +188,14 @@ class _ProgressiveContext:
         }
 
     def process_prefetched_batch(
-        self, query: dict[str, Any], results: list[dict[str, Any]]
+        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]]]:
         """处理已抓取的一批结果(作者作品:列表抓取+截断,不分页、不限速)。
 
@@ -183,6 +212,10 @@ class _ProgressiveContext:
                 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:],
@@ -190,9 +223,10 @@ class _ProgressiveContext:
         }
 
     def _process_query_pages(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:
+        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
@@ -206,6 +240,10 @@ class _ProgressiveContext:
             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
@@ -217,6 +255,10 @@ class _ProgressiveContext:
             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
@@ -225,9 +267,10 @@ class _ProgressiveContext:
         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:
+            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(
@@ -237,6 +280,10 @@ class _ProgressiveContext:
                 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
@@ -250,19 +297,27 @@ class _ProgressiveContext:
         *,
         page_number: int,
         cursor: Any,
-    ) -> list[dict[str, Any]] | None:
+    ) -> _FetchedPage | 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_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
-            return list(search(page_query))
+            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,
@@ -282,7 +337,12 @@ class _ProgressiveContext:
         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)
@@ -318,6 +378,7 @@ class _ProgressiveContext:
         if not batch:
             return False
 
+        discovery_started_at = time.monotonic()
         discovered = content_discovery.run(
             self.run_id,
             self.policy_run_id,
@@ -326,6 +387,7 @@ class _ProgressiveContext:
             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,
@@ -339,7 +401,11 @@ class _ProgressiveContext:
             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,
@@ -349,8 +415,29 @@ class _ProgressiveContext:
             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)
-        return any(decision.get("decision_action") == PASS_ACTION for decision in 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+。
@@ -385,8 +472,8 @@ class _ProgressiveContext:
                 portrait, fetch_error = None, None
                 for attempt in range(PORTRAIT_FETCH_ATTEMPTS):  # 重试,救瞬时失败/限速
                     try:
-                        if self.limiter is not None:
-                            self.limiter.wait()
+                        if self.portrait_limiter is not None:
+                            self.portrait_limiter.wait()
                         portrait = fetch_portrait(author_id)
                         fetch_error = None
                         break
@@ -412,6 +499,81 @@ class _ProgressiveContext:
                 }
             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],
@@ -571,6 +733,15 @@ def _get_path(data: Any, path: str) -> Any:
     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)
 

content_agent/business_modules/search_intent.py

@@ -156,10 +156,11 @@ def _run_v4(
     query_variant_client: QueryVariantClient,
     platform: str,
 ) -> list[dict[str, Any]]:
-    # 首轮搜索池主用上游 query_seed_points(Top30 真实点位);为空才回退 seed_terms。
+    # 首轮搜索池三路并行:pattern seed_terms + 上游 query_seed_points + 分类叶子元素。
     seed_candidates = _v4_seed_candidates(pattern_seed_pack, seed_terms)
-    candidates = _v4_query_seed_point_candidates(pattern_seed_pack) or seed_candidates
-    if not candidates:
+    query_seed_point_candidates = _v4_query_seed_point_candidates(pattern_seed_pack)
+    category_candidates = _v4_category_element_candidates(pattern_seed_pack)
+    if not seed_candidates and not query_seed_point_candidates and not category_candidates:
         raise _query_generation_error("v4_query_candidates_empty")
 
     # M9D Gate 2:仅非抖音,AI 判 query 是否易搜 50+;拿不准/异常从宽放过。
@@ -196,12 +197,10 @@ def _run_v4(
             row["query_source_refs"] = [_v4_query_source_ref(candidate, query)]
             search_queries.append(row)
 
-    _admit(candidates, use_gate2=apply_gate2)
-
-    # M13:Gate2 把候选清空时,用 seed_terms 兜底(宽词最稳,绕过 Gate2),而非整条 run 失败。
-    # seed_terms 契约保证非空,故非抖音抽象需求永远有词、不空跑;抖音不过 Gate2,不会进这里。
-    if not search_queries:
-        _admit(seed_candidates, use_gate2=False, fallback=True)
+    # seed_terms 和分类叶子元素是需求侧必搜来源;query_seed_points 仍保留非抖音 Gate2。
+    _admit(seed_candidates, use_gate2=False)
+    _admit(query_seed_point_candidates, use_gate2=apply_gate2)
+    _admit(category_candidates, use_gate2=False)
 
     if not search_queries:
         raise _query_generation_error("v4_search_queries_empty")
@@ -280,6 +279,56 @@ def _v4_query_seed_point_candidates(pattern_seed_pack: dict[str, Any]) -> list[d
     return candidates
 
 
+def _v4_category_element_candidates(pattern_seed_pack: dict[str, Any]) -> list[dict[str, Any]]:
+    candidates: list[dict[str, Any]] = []
+    for binding_index, binding in enumerate(_bindings(pattern_seed_pack), start=1):
+        for element_index, sample in enumerate(_sample_elements(binding), start=1):
+            name = _normalize_query(sample.get("name"))
+            if not name or _is_generic_query(name):
+                continue
+            rank = len(candidates) + 1
+            category_id = sample.get("category_id") or binding.get("category_id")
+            element_id = sample.get("source_element_id") or sample.get("id")
+            source_ref = {
+                "query_source_ref_id": _element_source_ref_id(category_id, element_id, name, rank),
+                "binding_rank": binding_index,
+                "element_rank": element_index,
+                "element_name": name,
+                "element_id": element_id,
+                "category_id": category_id,
+                "itemset_item_id": binding.get("itemset_item_id"),
+                "post_id": sample.get("post_id"),
+                "point_type": sample.get("point_type"),
+            }
+            candidates.append(
+                {
+                    "text": name,
+                    "query_source_type": "category_terminal_element",
+                    "generation_method": "category_leaf_element",
+                    "query_source_fields": ["element_bindings.sample_elements.name"],
+                    "rank": rank,
+                    "source_ref": source_ref,
+                }
+            )
+    return candidates
+
+
+def _bindings(pattern_seed_pack: dict[str, Any]) -> list[dict[str, Any]]:
+    values = pattern_seed_pack.get("element_bindings")
+    return [item for item in values if isinstance(item, dict)] if isinstance(values, list) else []
+
+
+def _sample_elements(binding: dict[str, Any]) -> list[dict[str, Any]]:
+    values = binding.get("sample_elements")
+    return [item for item in values if isinstance(item, dict)] if isinstance(values, list) else []
+
+
+def _element_source_ref_id(category_id: Any, element_id: Any, name: str, rank: int) -> str:
+    category = category_id or "unknown_category"
+    element = element_id or name or rank
+    return f"category:{category}#element:{element}"
+
+
 _LOG = logging.getLogger(__name__)
 _PURPOSE_PREFIX_PATH = Path("product_documents/配置/purpose_point_prefixes.v1.json")
 

content_agent/business_modules/walk_engine.py

@@ -9,6 +9,7 @@
 from __future__ import annotations
 
 import hashlib
+import time
 from datetime import datetime, timezone
 from typing import Any, Callable
 
@@ -102,6 +103,12 @@ def run_bounded_walk(
         if getattr(platform_client, "requires_progressive_search_rate_limit", False)
         else None
     )
+    # 作者/标签 search 和抖音热点宝画像是不同接口桶:各自 30s,互不占用。
+    portrait_limiter = (
+        SearchCallLimiter()
+        if getattr(platform_client, "requires_progressive_search_rate_limit", False)
+        else None
+    )
     # M8C 统一搜索单元:游走自有 _ProgressiveContext(独立累积 + 注入首轮 seen + index 基线)。
     walk_ctx = _ProgressiveContext(
         run_id=run_id,
@@ -112,6 +119,7 @@ def run_bounded_walk(
         runtime=runtime,
         gemini_video_client=gemini_video_client,
         limiter=limiter,
+        portrait_limiter=portrait_limiter,
         archive_dispatcher=archive_dispatcher,
         external_seen_content_ids=seen_content_ids,
         recall_index_base=len(discovered_content_items),
@@ -496,8 +504,9 @@ def _expand_authors(
         ctx.seen_author_ids.add(author_id)
         budget_tier = "low_budget" if permission == "allow_low_budget" else "normal"
         author_query_id = f"author_{_short_hash(author_id)}"
-        if limiter is not None:
-            limiter.wait()
+        author_fetch_started_at = time.monotonic()
+        waited_seconds = limiter.wait() if limiter is not None else 0.0
+        request_started_at = time.monotonic()
         try:
             works = fetch_author_works(
                 {
@@ -506,6 +515,7 @@ def _expand_authors(
                     "discovery_start_source": item.get("discovery_start_source", "pattern_itemset"),
                 }
             )
+            request_ended_at = time.monotonic()
         except Exception as exc:
             # 失败的作者游走不扣预算(作者已 seen,不会重试)
             walk_actions.append(
@@ -546,7 +556,14 @@ def _expand_authors(
             discovery_start_source, "author_to_works", created_at,
             raw_extra={"platform_author_id": author_id},
         )
-        batch = walk_ctx.process_prefetched_batch(author_query, fresh_works)
+        batch = walk_ctx.process_prefetched_batch(
+            author_query,
+            fresh_works,
+            search_duration_ms=_elapsed_ms(author_fetch_started_at, request_ended_at),
+            search_wait_ms=int(waited_seconds * 1000),
+            search_request_duration_ms=_elapsed_ms(request_started_at, request_ended_at),
+            search_timing_source="author_works_fetch",
+        )
         if batch["discovered_content_items"]:
             # 血缘补全:作者作品内容引用的合成 query id 必须真实存在于 search_queries,
             # 否则 validate_run 断链(真实 E2E v1_run_3a3bc9f0d72d 实证:missing_search_query_ref 等 8 条 fail)。
@@ -934,6 +951,10 @@ def _blocked_tag(tag: str) -> bool:
     return any(term in lowered for term in BLOCKED_TAG_TERMS)
 
 
+def _elapsed_ms(start: float, end: float) -> int:
+    return max(0, int((end - start) * 1000))
+
+
 def _walk_action_id(
     run_id: str,
     policy_run_id: str,

content_agent/flow_ledger_service.py

@@ -19,6 +19,7 @@ from content_agent.interfaces import RuntimeStore
 FIRST_ROUND_QUERY_METHODS = {
     "seed_term",
     "piaoquan_topic_point",
+    "query_seed_point",
     "category_leaf_element",
 }
 EXTENSION_QUERY_METHODS = {
@@ -29,11 +30,13 @@ EXTENSION_QUERY_METHODS = {
 SOURCE_LABELS = {
     "seed_term": "从需求单的 pattern 词来",
     "piaoquan_topic_point": "票圈帖子具体的点",
+    "query_seed_point": "票圈帖子具体的点",
     "category_leaf_element": "分类叶子元素",
 }
 QUERY_METHOD_LABELS = {
     "seed_term": "从需求单的 pattern 词来",
     "piaoquan_topic_point": "票圈帖子具体的点",
+    "query_seed_point": "票圈帖子具体的点",
     "category_leaf_element": "分类叶子元素",
     "tag_query": "从视频标签继续游走",
     "author_works": "从作者继续游走",
@@ -1755,7 +1758,7 @@ def _source_details(
             _detail("种子词", source_text),
             _detail("来源帖子", _source_post_id(seed_ref, source_ref, bundle)),
         ])
-    if method == "piaoquan_topic_point":
+    if method in {"piaoquan_topic_point", "query_seed_point"}:
         point_text = _text(source_ref.get("point_text") or source_ref.get("cleaned_text") or source_text)
         return _dedupe_texts([
             _detail("帖子 ID", source_ref.get("post_id") or _source_post_id(seed_ref, source_ref, bundle)),
@@ -1796,7 +1799,7 @@ def _source_ref_label(
 ) -> str:
     if method == "seed_term":
         return _text(_demand_id(bundle) or _source_post_id(seed_ref, source_ref, bundle), "来源编号待确认")
-    if method == "piaoquan_topic_point":
+    if method in {"piaoquan_topic_point", "query_seed_point"}:
         return _text(source_ref.get("post_id") or _source_post_id(seed_ref, source_ref, bundle), "帖子编号待确认")
     if method == "category_leaf_element":
         binding = _category_binding(bundle, source_ref)

tech_documents/数据接口与来源/walk_policy.json

@@ -2,16 +2,16 @@
   "schema_version": "walk_policy.v1",
   "title": "游走全局护栏 + 判定→边通行证(配 walk_graph.v2 使用)",
   "generated_at": "2026-06-11",
-  "value_provenance": "基线=walk_engine v1 实际硬限(行为等价,M4 拍板 2026-06-11);2026-06-12 R7 拍板放宽 tag 1→3、作者 2→3;2026-06-17 M8 拍板:深度游走重做——max_depth 3→5、总闸 60→10(口径=向外游走次数)、删 max_reseed_rounds、tag/作者预算 3→5、每作者作品 3→5、删 query_next_page 边;max_depth/max_total_actions_per_run 由 walk_engine frontier 真消费",
+  "value_provenance": "基线=walk_engine v1 实际硬限(行为等价,M4 拍板 2026-06-11);2026-06-12 R7 拍板放宽 tag 1→3、作者 2→3;2026-06-17 M8 拍板:深度游走重做——max_depth 3→5、总闸 60→10(口径=向外游走次数)、删 max_reseed_rounds、tag/作者预算 3→5、每作者作品 3→5、删 query_next_page 边;2026-06-24 收敛真实跑测成本:总闸10、tag/作者各5;max_depth/max_total_actions_per_run 由 walk_engine frontier 真消费",
   "global": {
-    "max_total_actions_per_run": { "value": 30, "provenance": "2026-06-19 调整:全 run 向外游走(标签跳+作者跳合计)硬闸 10→30(标签20+作者20=40,总闸30<40 仍为绑定上限);不含结局标记;由 walk_engine frontier 真生效", "tbd": false },
+    "max_total_actions_per_run": { "value": 10, "provenance": "2026-06-24 调整:全 run 向外游走(标签跳+作者跳合计)硬闸 30→10;不含结局标记;由 walk_engine frontier 真生效", "tbd": false },
     "max_depth": { "value": 5, "provenance": "M8 拍板 2026-06-17:游走树深度天花板=5,run 级、挂视频继承父+1、不重置;由 walk_engine frontier 真生效", "tbd": false },
     "gemini_max_workers": { "value": 2, "provenance": "拍板 2026-06-16:4核8G服务器资源收敛;判定总并发2,ffmpeg随单条链路自然不超过2", "tbd": false }
   },
   "edge_budgets": [
     { "edge_id": "video_to_author",  "max_total_actions": null, "max_per_content": 1, "max_pages": null, "provenance": "作者标识随 video 携带,loop 不单独执行此边、无预算消费;作者数上限见 author_to_works" },
-    { "edge_id": "author_to_works",  "max_total_actions": 20,  "max_works_per_author": 5, "max_pages": null, "provenance": "2026-06-19 调整:全 run 作者游走 5→20、每作者仍 ≤5 新作品;由 walk_engine frontier 跨层运行扣减" },
-    { "edge_id": "hashtag_to_query", "max_total_actions": 20,  "max_per_content": null, "max_tag_hops": null, "provenance": "2026-06-19 调整:全 run 标签游走 5→20;由 walk_engine frontier 跨层运行扣减" }
+    { "edge_id": "author_to_works",  "max_total_actions": 5,  "max_works_per_author": 5, "max_pages": null, "provenance": "2026-06-24 调整:全 run 作者游走 20→5、每作者仍 ≤5 新作品;由 walk_engine frontier 跨层运行扣减" },
+    { "edge_id": "hashtag_to_query", "max_total_actions": 5,  "max_per_content": null, "max_tag_hops": null, "provenance": "2026-06-24 调整:全 run 标签游走 20→5;由 walk_engine frontier 跨层运行扣减" }
   ],
   "dedup": {
     "query_key": "归一化 keyword(去空白/全半角/大小写统一)后全 run 去重;tag query 与 P2 初始 query 共用一个去重集",

tests/test_flow_ledger_api.py

@@ -123,7 +123,7 @@ def test_flow_ledger_api_returns_business_v4_ledger(tmp_path, monkeypatch):
                 "policy_run_id": policy_run_id,
                 "search_query_id": "q_003",
                 "search_query": "情绪放松练习",
-                "search_query_generation_method": "piaoquan_topic_point",
+                "search_query_generation_method": "query_seed_point",
                 "pattern_seed_ref": {"query_source_text": "情绪放松练习", "source_post_id": "post_001"},
                 "raw_payload": {
                     "query_source_refs": [
@@ -590,8 +590,9 @@ def test_flow_ledger_api_returns_business_v4_ledger(tmp_path, monkeypatch):
     run_walk = client.get(f"/runs/{run_id}/flow-ledger/walk").json()
     assert run_walk["summary"]["total_actions"] >= 1
     assert run_walk["summary"]["tag_used"] == 1  # walk_001 hashtag success
-    assert run_walk["summary"]["tag_cap"] == 20
+    assert run_walk["summary"]["tag_cap"] == 5
     assert run_walk["summary"]["author_used"] == 0
+    assert run_walk["summary"]["author_cap"] == 5
     assert run_walk["summary"]["max_depth_reached"] == 1
     assert run_walk["summary"]["max_depth_cap"] == 5
     assert run_walk["summary"]["walk_video_total"] == 1  # douyin_002 是游走带回

tests/test_gate2_query_50plus.py

@@ -91,15 +91,15 @@ def _qsp_seed_pack():
     }
 
 
-def test_gate2_all_rejected_falls_back_to_seed_terms_for_non_douyin():
-    # M13:快手/视频号 Gate2 把 query_seed_points 全判否 → 不 raise,用 seed_terms 兜底(绕过 Gate2)。
+def test_gate2_rejects_query_seed_points_but_keeps_mandatory_seed_terms_for_non_douyin():
+    # seed_terms 是首轮必搜;非抖音 Gate2 只过滤 query_seed_points。
     queries = search_intent.run(
         "r", "p", _qsp_seed_pack(), _Runtime(), _Judge(False),
         strategy_version="V4", platform="kuaishou",
     )
-    assert [q["search_query"] for q in queries] == ["养生"]  # seed_terms 兜底,非被否的窄词
+    assert [q["search_query"] for q in queries] == ["养生"]
     assert queries[0]["search_query_generation_method"] == "seed_term"
-    assert queries[0]["query_source_refs"][0]["source_ref"].get("gate2_fallback") is True
+    assert "gate2_fallback" not in queries[0]["query_source_refs"][0]["source_ref"]
 
 
 def test_douyin_skips_gate2_and_fallback():
@@ -108,7 +108,7 @@ def test_douyin_skips_gate2_and_fallback():
         "r", "p", _qsp_seed_pack(), _Runtime(), _Judge(False),
         strategy_version="V4", platform="douyin",
     )
-    assert len(queries) == 4
+    assert len(queries) == 5
     assert all(
         "gate2_fallback" not in q["query_source_refs"][0]["source_ref"] for q in queries
     )

tests/test_portrait_cache.py

@@ -3,6 +3,15 @@
 from content_agent.business_modules.progressive_screening import _ProgressiveContext
 
 
+class CountingLimiter:
+    def __init__(self):
+        self.calls = 0
+
+    def wait(self):
+        self.calls += 1
+        return 0.0
+
+
 class CountingPortraitClient:
     def __init__(self):
         self.calls = []
@@ -21,11 +30,12 @@ class CountingPortraitClient:
         }
 
 
-def _ctx(client):
+def _ctx(client, *, limiter=None, portrait_limiter=None):
     return _ProgressiveContext(
         run_id="r", policy_run_id="p", source_context={}, policy_bundle={},
         platform_client=client, runtime=None, gemini_video_client=None,
-        limiter=None, archive_dispatcher=None, platform="douyin", portrait_client=client,
+        limiter=limiter, portrait_limiter=portrait_limiter,
+        archive_dispatcher=None, platform="douyin", portrait_client=client,
     )
 
 
@@ -46,6 +56,16 @@ def test_portrait_cached_by_author():
     assert all(b["content_audience_50plus"]["status"] == "ok" for b in bundles)
 
 
+def test_portrait_uses_separate_limiter_bucket():
+    client = CountingPortraitClient()
+    search_limiter = CountingLimiter()
+    portrait_limiter = CountingLimiter()
+    ctx = _ctx(client, limiter=search_limiter, portrait_limiter=portrait_limiter)
+    ctx._inject_fifty_plus([_bundle("a1"), _bundle("a2")])
+    assert search_limiter.calls == 0
+    assert portrait_limiter.calls == 2
+
+
 def test_portrait_fetched_when_query_passes_regardless_of_platform():
     # 修复:门只看 query 相关性,平台分再低也拉画像、算适老性(原来 platform<65 会跳过)。
     client = CountingPortraitClient()

tests/test_progressive_screening.py

@@ -106,6 +106,31 @@ def test_progressive_screening_archives_batch_before_gemini_judgment(tmp_path):
     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)]

tests/test_qwen_video.py

@@ -5,6 +5,7 @@ import json
 from content_agent.integrations.qwen_video import QwenVideoClient
 from content_agent.run_service import _gemini_video_client_from_env
 from content_agent.integrations.gemini_video import GeminiVideoClient
+from content_agent.integrations import timeout_config
 
 
 class _FakeResponse:
@@ -34,6 +35,7 @@ def test_qwen_analyze_sends_video_url_oss_and_parses():
     def fake_post(url, headers=None, json=None, timeout=None):
         captured["url"] = url
         captured["json"] = json
+        captured["timeout"] = timeout
         return _FakeResponse(
             {"choices": [{"message": {"content": '{"query_relevance_score": 88, "query_relevance_reason": "贴合"}'}}]}
         )
@@ -51,6 +53,7 @@ def test_qwen_analyze_sends_video_url_oss_and_parses():
     content = captured["json"]["messages"][1]["content"]
     video_part = next(p for p in content if p["type"] == "video_url")
     assert video_part["video_url"]["url"] == "https://res.example.com/v.mp4"
+    assert captured["timeout"].read == timeout_config.read_timeout("video_llm") == 120.0
 
 
 def test_content_inspection_split_detection():

tests/test_search_intent.py

@@ -195,7 +195,7 @@ def test_search_intent_custom_evidence_fields_whitelist():
     assert llm_query["query_source_fields"] == ["seed_terms"]
 
 
-def test_v4_search_intent_uses_query_seed_points_without_llm_variant():
+def test_v4_search_intent_uses_three_direct_sources_without_llm_variant():
     client = FakeQueryVariantClient({"中医养生": "气血食疗"})
     runtime = _Runtime()
 
@@ -209,16 +209,27 @@ def test_v4_search_intent_uses_query_seed_points_without_llm_variant():
     )
 
     assert client.calls == []
-    # query_seed_points 主用:目的点剥"分享"前缀、灵感点不剥、按 rank 序;不再有 element_bindings 两路。
-    assert [row["search_query"] for row in queries] == ["气血食疗", "办公室八段锦"]
+    # 三路首轮搜索池:seed_terms、query_seed_points、分类叶子元素;不再调 LLM 变体。
+    assert [row["search_query"] for row in queries] == [
+        "中医养生",
+        "气血食疗",
+        "办公室八段锦",
+        "食疗补气血",
+        "八段锦",
+        "日常保健",
+    ]
     assert [row["search_query_generation_method"] for row in queries] == [
+        "seed_term",
         "query_seed_point",
         "query_seed_point",
+        "category_leaf_element",
+        "category_leaf_element",
+        "category_leaf_element",
     ]
     assert all("llm_variant_of" not in row for row in queries)
     assert runtime.rows["search_queries.jsonl"] == queries
 
-    first = queries[0]
+    first = next(row for row in queries if row["search_query"] == "气血食疗")
     assert first["pattern_seed_ref"]["query_source_type"] == "query_seed_point"
     assert first["pattern_seed_ref"]["query_source_text"] == "气血食疗"
     assert first["pattern_seed_ref"]["query_source_rank"] == 1
@@ -234,6 +245,7 @@ def test_v4_falls_back_to_seed_terms_when_query_seed_points_empty():
     runtime = _Runtime()
     seed_pack = _seed_pack()
     seed_pack["query_seed_points"] = []
+    seed_pack["element_bindings"] = []
 
     queries = search_intent.run(
         "run_1",

tests/test_search_previous_step.py

@@ -8,6 +8,6 @@ def test_tag_query_labeled_as_walk():
 
 
 def test_first_round_methods_labeled_search_query_direct():
-    for method in ["seed_term", "item_single", "piaoquan_topic_point", "category_leaf_element", None]:
+    for method in ["seed_term", "item_single", "piaoquan_topic_point", "query_seed_point", "category_leaf_element", None]:
         assert search_previous_discovery_step({"search_query_generation_method": method}) == "search_query_direct"
     assert search_previous_discovery_step({}) == "search_query_direct"

tests/test_timeout_hardening.py

@@ -47,6 +47,8 @@ def test_httpx_timeout_is_segmented_with_short_read():
     assert t.read == 120.0          # read 短,停吐字节即抛
     assert t.write == 600.0         # write 承载总时长
     assert t.connect == timeout_config.CONNECT_TIMEOUT_SECONDS
+    video_llm = timeout_config.httpx_timeout("video_llm", env={})
+    assert video_llm.read == 120.0  # Qwen/Gemini 停吐字节 120s 即抛
 
 
 def test_as_httpx_timeout_read_capped_by_total():

tests/test_v4_m1_query_sources_replay.py

@@ -21,17 +21,16 @@ def test_v4_m1_direct_query_sources_replay_from_seed_pack(tmp_path):
     )
 
     assert client.calls == []
-    # M12B:首轮搜索池主用上游 query_seed_points,不再有 element_bindings 两路、不调 LLM 变体。
+    # 首轮搜索池三路直连:seed_terms + query_seed_points + 分类叶子元素,不调 LLM 变体。
     methods = {row["search_query_generation_method"] for row in queries}
-    assert methods == {"query_seed_point"}
+    assert {"seed_term", "query_seed_point", "category_leaf_element"} <= methods
     assert "llm_variant" not in methods
     assert all("llm_variant_of" not in row for row in queries)
     assert all(row["pattern_seed_ref"]["query_source_text"] == row["search_query"] for row in queries)
     assert all(row["pattern_seed_ref"]["query_source_rank"] >= 1 for row in queries)
     assert all(row["raw_payload"]["query_source_refs"] for row in queries)
-    assert all(
-        row["pattern_seed_ref"]["query_source_type"] == "query_seed_point" for row in queries
-    )
+    source_types = {row["pattern_seed_ref"]["query_source_type"] for row in queries}
+    assert {"seed_term", "query_seed_point", "category_terminal_element"} <= source_types
 
 
 def test_v4_m1_run_service_does_not_call_query_variant_client(tmp_path):

tests/test_walk_graph_config.py

@@ -24,8 +24,8 @@ def test_graph_has_8_nodes_and_9_edges():
 
 def test_policy_unwraps_pinned_values():
     policy = WalkGraphStore().load_policy()
-    # M8 拍板:总闸 60→10、删 max_reseed_rounds;deny / halve_min_1(max(1, budget//2))。
-    assert policy["global"]["max_total_actions_per_run"] == 30
+    # 2026-06-24 拍板:总闸收敛到 10、删 max_reseed_rounds;deny / halve_min_1(max(1, budget//2))。
+    assert policy["global"]["max_total_actions_per_run"] == 10
     assert policy["global"]["max_depth"] == 5
     assert "max_reseed_rounds" not in policy["global"]
     assert policy["edge_permissions"]["KEEP_CONTENT_FOR_REVIEW"]["video_to_hashtag"] == "deny"
@@ -63,12 +63,12 @@ def test_policy_allows_smoke_env_overrides(monkeypatch):
 
 
 def test_policy_edge_budgets_match_decided_values():
-    # M8 拍板 2026-06-17:标签 5、作者 5、每作者作品 5;query_next_page 已删。
+    # 2026-06-24 拍板:标签 5、作者 5、每作者作品 5;query_next_page 已删。
     budgets = WalkGraphStore().load_policy()["edge_budgets_by_id"]
     assert "query_next_page" not in budgets
-    assert budgets["author_to_works"]["max_total_actions"] == 20
+    assert budgets["author_to_works"]["max_total_actions"] == 5
     assert budgets["author_to_works"]["max_works_per_author"] == 5
-    assert budgets["hashtag_to_query"]["max_total_actions"] == 20
+    assert budgets["hashtag_to_query"]["max_total_actions"] == 5
 
 
 def test_shipinhao_author_edges_blocked_and_terminal_edges_supported():

tests/test_walk_policy_real_knobs.py

@@ -11,7 +11,7 @@ def test_global_knobs_unwrap_to_plain_int():
     glob = _policy()["global"]
     assert glob["max_depth"] == 5
     assert isinstance(glob["max_depth"], int)
-    assert glob["max_total_actions_per_run"] == 30
+    assert glob["max_total_actions_per_run"] == 10
     assert isinstance(glob["max_total_actions_per_run"], int)
 
 
@@ -23,8 +23,8 @@ def test_max_reseed_rounds_removed_and_loader_ok():
 
 def test_edge_budgets_are_m8_values():
     budgets = _policy()["edge_budgets_by_id"]
-    assert budgets["hashtag_to_query"]["max_total_actions"] == 20
-    assert budgets["author_to_works"]["max_total_actions"] == 20
+    assert budgets["hashtag_to_query"]["max_total_actions"] == 5
+    assert budgets["author_to_works"]["max_total_actions"] == 5
     assert budgets["author_to_works"]["max_works_per_author"] == 5
 
 

web2/features/LedgerPage.tsx

@@ -271,7 +271,7 @@ function VideoDetailCell({ runId, video }: { runId: string; video: VideoRef | nu
 
 function sourceKindClass(kind: string): string {
   if (kind === "seed_term") return "seed";
-  if (kind === "piaoquan_topic_point") return "piaoquan";
+  if (kind === "piaoquan_topic_point" || kind === "query_seed_point") return "piaoquan";
   if (kind === "category_leaf_element") return "category";
   return "unknown";
 }

web2/lib/flow-ledger/business.ts

@@ -29,6 +29,7 @@ const REASON_LABELS: Record<string, string> = {
 const SOURCE_LABELS: Record<string, string> = {
   seed_term: "从需求单的 pattern 词来",
   piaoquan_topic_point: "票圈帖子具体的点",
+  query_seed_point: "票圈帖子具体的点",
   category_leaf_element: "分类叶子元素",
   pattern_itemset: "来自需求 Pattern",
   pattern_search_query: "来自需求 Pattern",
@@ -42,6 +43,7 @@ const SOURCE_LABELS: Record<string, string> = {
 const METHOD_LABELS: Record<string, string> = {
   seed_term: "从需求单的 pattern 词来",
   piaoquan_topic_point: "票圈帖子具体的点",
+  query_seed_point: "票圈帖子具体的点",
   category_leaf_element: "分类叶子元素",
   query_next_page: "翻页继续找",
   tag_query: "从视频标签继续游走",

web2/lib/flow-ledger/format.ts

@@ -44,6 +44,7 @@ export function label(value: string): string {
   const labels: Record<string, string> = {
     seed_term: "从需求单的 pattern 词来",
     piaoquan_topic_point: "票圈帖子具体的点",
+    query_seed_point: "票圈帖子具体的点",
     category_leaf_element: "分类叶子元素",
     item_single: "使用原始需求词",
     tag_query: "从视频标签继续游走",