fix(web): 游走带回计数对旧 run 鲁棒 — 按 query 方法判定首轮/游走

labeling 修复前跑的 run,标签带回视频被误标 search_query_direct,
导致 _is_first_round_content 误判为首轮 → 游走带回=0(但内容其实带回了 11-12 条)。
改:内容所属 query 是扩展(tag_query/author_works)query 即判为游走带回,
这条 query→内容关联在旧 run 里也正确。新增 _walk_query_ids,首轮筛选/run总览/预算总览同步。

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

content_agent/flow_ledger_service.py

@@ -391,7 +391,8 @@ class FlowLedgerService:
     def _query_content(self, query_id: str, bundle: dict[str, Any], *, first_round_only: bool) -> list[dict[str, Any]]:
         items = [item for item in bundle["content_items"] if _content_belongs_to_query(item, query_id)]
         if first_round_only:
-            items = [item for item in items if _is_first_round_content(item)]
+            walk_query_ids = _walk_query_ids(bundle)
+            items = [item for item in items if _is_first_round_content(item, walk_query_ids)]
         return items
 
     def _query_related_content_ids(self, query_id: str, bundle: dict[str, Any]) -> set[str]:
@@ -679,7 +680,8 @@ class FlowLedgerService:
     ) -> dict[str, Any]:
         query_failures = _query_failures(bundle["queries"])
         # 游走指标(run 级):游走带回视频、其中入池、向外游走次数、最深层数。
-        walk_items = [c for c in bundle["content_items"] if not _is_first_round_content(c)]
+        walk_query_ids = _walk_query_ids(bundle)
+        walk_items = [c for c in bundle["content_items"] if not _is_first_round_content(c, walk_query_ids)]
         pooled_targets = {
             _text(d.get("decision_target_id"))
             for d in bundle["decisions"]
@@ -752,13 +754,25 @@ def _content_belongs_to_query(content: dict[str, Any], query_id: str) -> bool:
     return False
 
 
-def _is_first_round_content(content: dict[str, Any]) -> bool:
+def _is_first_round_content(content: dict[str, Any], walk_query_ids: set[str] | None = None) -> bool:
+    # 优先用「内容所属 query 是不是扩展(标签/作者)query」判定——这条 query→内容 的关联即使在
+    # labeling 修复前的旧 run 里也正确(旧 run 把 previous_discovery_step 误标成 search_query_direct)。
+    if walk_query_ids and _text(content.get("search_query_id")) in walk_query_ids:
+        return False
     previous = _text(content.get("previous_discovery_step") or _record(content.get("raw_payload")).get("previous_discovery_step"))
     if not previous:
         return True
     return not any(token in previous for token in ["tag", "hashtag", "author", "next_page", "walk", "pagination"])
 
 
+def _walk_query_ids(bundle: dict[str, Any]) -> set[str]:
+    return {
+        _text(query.get("search_query_id"))
+        for query in bundle["queries"]
+        if _query_method(query) in EXTENSION_QUERY_METHODS and _text(query.get("search_query_id"))
+    }
+
+
 def _content_ids(content: dict[str, Any]) -> set[str]:
     return {
         _text(content.get("content_discovery_id")),
@@ -840,7 +854,8 @@ def _walk_budget_overview(
     statuses = Counter(_text(action.get("status"), "unknown") for action in actions)
     tag_used = sum(1 for a in actions if a.get("edge_id") == "hashtag_to_query" and a.get("status") == "success")
     author_used = sum(1 for a in actions if a.get("edge_id") == "author_to_works" and a.get("status") == "success")
-    walk_items = [c for c in bundle["content_items"] if not _is_first_round_content(c)]
+    walk_query_ids = _walk_query_ids(bundle)
+    walk_items = [c for c in bundle["content_items"] if not _is_first_round_content(c, walk_query_ids)]
     return {
         "total_actions": len(actions),
         "success_count": statuses["success"],

tests/test_flow_ledger_api.py

@@ -5,6 +5,18 @@ from content_agent.flow_ledger_service import _FLOW_LEDGER_CACHE
 from content_agent.run_service import RunService
 
 
+def test_first_round_detection_robust_to_old_labeling():
+    from content_agent.flow_ledger_service import _is_first_round_content
+
+    # 旧 run:标签带回视频被误标 search_query_direct;只看 previous_discovery_step 会误判成首轮。
+    old = {"search_query_id": "tag_q", "previous_discovery_step": "search_query_direct"}
+    assert _is_first_round_content(old) is True
+    # 传入扩展 query 集合后(query→内容关联在旧 run 里仍正确),正确识别为游走带回。
+    assert _is_first_round_content(old, {"tag_q"}) is False
+    # 真首轮内容不受影响。
+    assert _is_first_round_content({"search_query_id": "q1", "previous_discovery_step": "seed_term"}, {"tag_q"}) is True
+
+
 def test_flow_ledger_api_returns_business_v4_ledger(tmp_path, monkeypatch):
     _FLOW_LEDGER_CACHE.clear()
     service = RunService(runtime_root=tmp_path / "runtime" / "v1")