Implement V4 direct query sources

content_agent/business_modules/search_intent.py

@@ -45,11 +45,22 @@ def run(
     pattern_seed_pack: dict[str, Any],
     runtime: RuntimeFileStore,
     query_variant_client: QueryVariantClient,
+    *,
+    strategy_version: str | None = None,
 ) -> list[dict[str, Any]]:
     created_at = datetime.now(timezone.utc).isoformat()
     seed_terms = _terms(pattern_seed_pack.get("seed_terms"))
     if not seed_terms:
         raise _query_generation_error("seed_terms_empty")
+    if str(strategy_version or "").upper() == "V4":
+        return _run_v4(
+            run_id,
+            policy_run_id,
+            pattern_seed_pack,
+            runtime,
+            created_at,
+            seed_terms,
+        )
 
     profile = getattr(query_variant_client, "profile", DEFAULT_PROFILE)
     variants_per_seed = int(profile.get("variants_per_seed", 1))
@@ -129,6 +140,214 @@ def run(
     return search_queries
 
 
+def _run_v4(
+    run_id: str,
+    policy_run_id: str,
+    pattern_seed_pack: dict[str, Any],
+    runtime: RuntimeFileStore,
+    created_at: str,
+    seed_terms: list[str],
+) -> list[dict[str, Any]]:
+    candidates = [
+        *_v4_seed_candidates(pattern_seed_pack, seed_terms),
+        *_v4_piaoquan_point_candidates(pattern_seed_pack),
+        *_v4_category_element_candidates(pattern_seed_pack),
+    ]
+    if not candidates:
+        raise _query_generation_error("v4_query_candidates_empty")
+
+    search_queries: list[dict[str, Any]] = []
+    seen_queries: set[str] = set()
+    for candidate in candidates:
+        query = _normalize_query(candidate.get("text"))
+        if not query:
+            continue
+        normalized_key = _normalize_query_key(query)
+        if normalized_key in seen_queries:
+            continue
+        seen_queries.add(normalized_key)
+        query_id = f"q_{len(search_queries) + 1:03d}"
+        row = _base_query_record(
+            run_id=run_id,
+            policy_run_id=policy_run_id,
+            search_query_id=query_id,
+            search_query=query,
+            generation_method=candidate["generation_method"],
+            seed_term=query,
+            pattern_seed_ref=_v4_pattern_seed_ref(pattern_seed_pack, candidate, query),
+            created_at=created_at,
+            query_source_terms=[query],
+            query_source_fields=candidate["query_source_fields"],
+        )
+        row["query_source_refs"] = [_v4_query_source_ref(candidate, query)]
+        search_queries.append(row)
+
+    if not search_queries:
+        raise _query_generation_error("v4_search_queries_empty")
+
+    search_queries = [with_raw_payload(row) for row in search_queries]
+    runtime.append_jsonl(run_id, "search_queries.jsonl", search_queries)
+    return search_queries
+
+
+def _v4_seed_candidates(
+    pattern_seed_pack: dict[str, Any],
+    seed_terms: list[str],
+) -> list[dict[str, Any]]:
+    return [
+        {
+            "text": seed_term,
+            "query_source_type": "seed_term",
+            "generation_method": "seed_term",
+            "query_source_fields": ["seed_terms"],
+            "rank": index + 1,
+            "source_ref": {
+                "source_field": "seed_terms",
+                "source_index": index,
+                "seed_term": seed_term,
+                "query_source_ref_id": f"seed_terms:{index}",
+            },
+        }
+        for index, seed_term in enumerate(seed_terms)
+    ]
+
+
+def _v4_piaoquan_point_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 point_index, sample in enumerate(_sample_elements(binding), start=1):
+            point_type = str(sample.get("point_type") or "").strip()
+            if point_type not in {"目的点", "灵感点"}:
+                continue
+            point_text = _normalize_query(sample.get("point_text"))
+            if not point_text:
+                continue
+            text = _cleanup_purpose_point(point_text) if point_type == "目的点" else point_text
+            if not text:
+                continue
+            rank = len(candidates) + 1
+            source_ref = {
+                "query_source_ref_id": _point_source_ref_id(sample, rank),
+                "binding_rank": binding_index,
+                "point_rank": point_index,
+                "point_type": point_type,
+                "point_text": point_text,
+                "cleaned_text": text,
+                "post_id": sample.get("post_id"),
+                "point_id": sample.get("id") or sample.get("topic_point_id"),
+                "element_id": sample.get("source_element_id") or sample.get("id"),
+                "category_id": sample.get("category_id") or binding.get("category_id"),
+            }
+            candidates.append(
+                {
+                    "text": text,
+                    "query_source_type": "piaoquan_point_text",
+                    "generation_method": "piaoquan_topic_point",
+                    "query_source_fields": ["element_bindings.sample_elements.point_text"],
+                    "rank": rank,
+                    "source_ref": source_ref,
+                }
+            )
+    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:
+                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 _cleanup_purpose_point(value: str) -> str:
+    text = _normalize_query(value)
+    for prefix in ("分享", "讲述", "表达", "传递", "呼吁", "介绍", "展示", "记录", "呈现"):
+        if text.startswith(prefix) and len(text) > len(prefix):
+            text = text[len(prefix):].strip(" ，,。:：")
+            break
+    return _normalize_query(text)
+
+
+def _point_source_ref_id(sample: dict[str, Any], rank: int) -> str:
+    post_id = sample.get("post_id") or "unknown_post"
+    point_id = sample.get("id") or sample.get("topic_point_id") or rank
+    return f"post:{post_id}#point:{point_id}"
+
+
+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}"
+
+
+def _v4_pattern_seed_ref(
+    pattern_seed_pack: dict[str, Any],
+    candidate: dict[str, Any],
+    query: str,
+) -> dict[str, Any]:
+    source_ref = dict(candidate.get("source_ref") or {})
+    return {
+        "query_source_type": candidate["query_source_type"],
+        "query_source_ref_id": source_ref.get("query_source_ref_id"),
+        "query_source_text": query,
+        "query_source_rank": candidate["rank"],
+        "pattern_execution_id": pattern_seed_pack.get("pattern_execution_id"),
+        "mining_config_id": pattern_seed_pack.get("mining_config_id"),
+        "source_post_id": pattern_seed_pack.get("source_post_id"),
+        "matched_post_ids": pattern_seed_pack.get("matched_post_ids") or [],
+        "itemset_ids": _itemset_ids(pattern_seed_pack),
+        "source_ref": source_ref,
+    }
+
+
+def _v4_query_source_ref(candidate: dict[str, Any], query: str) -> dict[str, Any]:
+    source_ref = dict(candidate.get("source_ref") or {})
+    return {
+        "query_source_type": candidate["query_source_type"],
+        "query_source_ref_id": source_ref.get("query_source_ref_id"),
+        "query_source_text": query,
+        "query_source_rank": candidate["rank"],
+        "generation_method": candidate["generation_method"],
+        "source_ref": source_ref,
+    }
+
+
 def _terms(values: Any) -> list[str]:
     if not isinstance(values, list):
         return []
@@ -155,6 +374,8 @@ def _base_query_record(
     seed_term: str,
     pattern_seed_ref: dict[str, Any],
     created_at: str,
+    query_source_terms: list[str] | None = None,
+    query_source_fields: list[str] | None = None,
 ) -> dict[str, Any]:
     return {
         "record_schema_version": RUNTIME_RECORD_SCHEMA_VERSION,
@@ -166,8 +387,8 @@ def _base_query_record(
         "discovery_start_source": "pattern_itemset",
         "previous_discovery_step": "pattern_search_query",
         "search_query_effect_status": "pending",
-        "query_source_terms": [seed_term],
-        "query_source_fields": ["seed_terms"],
+        "query_source_terms": query_source_terms or [seed_term],
+        "query_source_fields": query_source_fields or ["seed_terms"],
         "pattern_seed_ref": pattern_seed_ref,
         "created_at": created_at,
     }
@@ -338,6 +559,10 @@ def _normalize_query(value: Any) -> str:
     return " ".join(value.split()).strip()
 
 
+def _normalize_query_key(value: Any) -> str:
+    return _normalize_query(value).casefold()
+
+
 def _is_generic_query(query: str, generic_filter: dict[str, Any] | None = None) -> bool:
     generic_queries = set((generic_filter or {}).get("queries") or GENERIC_QUERIES)
     generic_tokens = tuple((generic_filter or {}).get("tokens") or GENERIC_QUERY_TOKENS)

content_agent/business_modules/walk_engine.py

@@ -477,14 +477,11 @@ def _expand_authors(
     decision_by_content_id = _decision_by_content_id(rule_decisions)
     binding, _ = _resolve_edge_binding("author_to_works", walk_strategy)
     unique_author_items = _unique_authors(discovered_content_items)
-    author_items = unique_author_items[: budgets["max_total_actions"]]
-    # R8(2026-06-12 拍板):预算外作者不再静默消失,补 budget_exhausted skip 留痕。
-    overflow_author_items = unique_author_items[budgets["max_total_actions"]:]
     walk_actions: list[dict[str, Any]] = []
 
     # 平台不支持作者边(如视频号 blogger blocked):显式 skip 留痕、不调用平台,游走自然退化。
     if not edge_supported(profile, "author_to_works"):
-        for item in author_items:
+        for item in unique_author_items[: budgets["max_total_actions"]]:
             author_id = item.get("platform_author_id")
             if not author_id:
                 continue
@@ -516,7 +513,8 @@ def _expand_authors(
     }
     platform_results: list[dict[str, Any]] = []
     author_search_queries: list[dict[str, Any]] = []
-    for item in author_items:
+    allowed_author_count = 0
+    for item in unique_author_items:
         author_id = item.get("platform_author_id")
         if not author_id:
             continue
@@ -539,6 +537,19 @@ def _expand_authors(
                 )
             )
             continue
+        if allowed_author_count >= budgets["max_total_actions"]:
+            walk_actions.append(
+                _author_walk_action(
+                    run_id, policy_run_id, author_id, "skipped", created_at,
+                    reason_code="budget_exhausted",
+                    budget_tier="blocked",
+                    binding=binding,
+                    decision=decision,
+                    content_pack=content_pack,
+                )
+            )
+            continue
+        allowed_author_count += 1
         budget_tier = "low_budget" if permission == "allow_low_budget" else "normal"
         try:
             works = fetch_author_works(
@@ -603,20 +614,6 @@ def _expand_authors(
                     "previous_discovery_step": "author_works",
                 }
             )
-    for item in overflow_author_items:
-        author_id = item.get("platform_author_id")
-        if not author_id:
-            continue
-        walk_actions.append(
-            _author_walk_action(
-                run_id, policy_run_id, author_id, "skipped", created_at,
-                reason_code="budget_exhausted",
-                budget_tier="blocked",
-                binding=binding,
-                decision=decision_by_content_id.get(item.get("platform_content_id")),
-                content_pack=content_pack,
-            )
-        )
     if not platform_results:
         return {**_empty_batch(), "walk_actions": walk_actions, "search_queries": author_search_queries}
 
@@ -701,6 +698,7 @@ def _terminal_stage(
     source_path_record_basis: list[dict[str, Any]] = []
 
     for search_query in search_queries:
+        query_source_ref = search_query.get("pattern_seed_ref") or {}
         source_path_record_basis.append(
             {
                 "policy_run_id": search_query["policy_run_id"],
@@ -716,6 +714,13 @@ def _terminal_stage(
                 "previous_discovery_step": search_query["previous_discovery_step"],
                 "origin_path_id": f"pattern_to_search_query:{search_query['search_query_id']}",
                 "source_evidence_ref": "source_context.json#ext_data.evidence_pack",
+                "query_source_type": query_source_ref.get("query_source_type"),
+                "query_source_ref_id": query_source_ref.get("query_source_ref_id"),
+                "query_source_text": query_source_ref.get("query_source_text")
+                or search_query.get("search_query"),
+                "query_source_rank": query_source_ref.get("query_source_rank"),
+                "query_source_refs": search_query.get("query_source_refs")
+                or search_query.get("raw_payload", {}).get("query_source_refs", []),
             }
         )
 

content_agent/graph.py

@@ -88,6 +88,7 @@ def build_run_graph(deps: RunDependencies):
             state["pattern_seed_pack"],
             deps.runtime,
             deps.query_variant_client,
+            strategy_version=state.get("strategy_version"),
         )
         return {"search_queries": search_queries, "current_step": "plan_queries"}
 

tests/test_p0d_p0g.py

@@ -156,7 +156,7 @@ def test_run_service_all_platform_queries_fail_records_failed_query_details(tmp_
     assert state["status"] == "failed"
     assert state["error_code"] == ErrorCode.PLATFORM_REQUEST_FAILED.value
     failed_query_ids = [failure["search_query_id"] for failure in state["error_detail"]["query_failures"]]
-    assert failed_query_ids == ["q_001", "q_002", "q_003", "q_004"]
+    assert failed_query_ids == ["q_001", "q_002", "q_003", "q_004", "q_005"]
     assert runtime.run_updates[-1]["updates"]["status"] == "failed"
     assert runtime.run_updates[-1]["updates"]["error_detail"]["query_failures"]
     assert runtime.lifecycle_events[-1]["event_id"] == "lifecycle_failed"
@@ -190,7 +190,7 @@ def test_run_service_query_generation_failure_records_error_code(tmp_path):
         query_variant_client=FakeQueryVariantClient(error=RuntimeError("model unavailable")),
     )
 
-    state = service.start_run(RunStartRequest(platform_mode="mock"))
+    state = service.start_run(RunStartRequest(platform_mode="mock", strategy_version="V1"))
 
     assert state["status"] == "failed"
     assert state["error_code"] == ErrorCode.QUERY_GENERATION_FAILED.value
@@ -213,7 +213,7 @@ def test_run_service_duplicate_query_variant_records_query_generation_failed(tmp
         query_variant_client=FakeQueryVariantClient({"爱国情感": "爱国情感"}),
     )
 
-    state = service.start_run(RunStartRequest(platform_mode="mock"))
+    state = service.start_run(RunStartRequest(platform_mode="mock", strategy_version="V1"))
 
     assert state["status"] == "failed"
     assert state["error_code"] == ErrorCode.QUERY_GENERATION_FAILED.value
@@ -230,7 +230,7 @@ def test_run_service_generic_query_variant_records_query_generation_failed(tmp_p
         query_variant_client=FakeQueryVariantClient({"爱国情感": "热门视频"}),
     )
 
-    state = service.start_run(RunStartRequest(platform_mode="mock"))
+    state = service.start_run(RunStartRequest(platform_mode="mock", strategy_version="V1"))
 
     assert state["status"] == "failed"
     assert state["error_code"] == ErrorCode.QUERY_GENERATION_FAILED.value
@@ -247,7 +247,7 @@ def test_run_service_malformed_query_variant_result_records_query_generation_fai
         query_variant_client=_MalformedQueryVariantClient(),
     )
 
-    state = service.start_run(RunStartRequest(platform_mode="mock"))
+    state = service.start_run(RunStartRequest(platform_mode="mock", strategy_version="V1"))
 
     assert state["status"] == "failed"
     assert state["error_code"] == ErrorCode.QUERY_GENERATION_FAILED.value
@@ -260,7 +260,7 @@ def test_run_service_missing_query_variant_client_fails_at_p2(tmp_path):
     service = RunService(runtime=runtime)
 
     state = service.start_run(
-        RunStartRequest(platform_mode="mock", source=str(REAL_SOURCE_FIXTURE))
+        RunStartRequest(platform_mode="mock", source=str(REAL_SOURCE_FIXTURE), strategy_version="V1")
     )
 
     assert state["status"] == "failed"

tests/test_search_intent.py

@@ -32,7 +32,35 @@ def _seed_pack():
         "seed_terms": ["中医养生"],
         "itemset_items": ["补气血"],
         "category_bindings": [{"category_id": "c1"}],
-        "element_bindings": [{"element_id": "e1"}],
+        "element_bindings": [
+            {
+                "element_id": "e1",
+                "category_id": "c1",
+                "sample_elements": [
+                    {
+                        "id": "point_1",
+                        "name": "食疗补气血",
+                        "point_type": "目的点",
+                        "point_text": "分享气血食疗方法",
+                        "post_id": "p1",
+                    },
+                    {
+                        "id": "point_2",
+                        "name": "八段锦",
+                        "point_type": "灵感点",
+                        "point_text": "办公室八段锦",
+                        "post_id": "p2",
+                    },
+                    {
+                        "id": "point_3",
+                        "name": "日常保健",
+                        "point_type": "关键点",
+                        "point_text": "不应该进入搜索词",
+                        "post_id": "p3",
+                    },
+                ],
+            }
+        ],
         "pattern_source_system": "pg_pattern_v2",
         "pattern_execution_id": 1987,
         "mining_config_id": 58,
@@ -56,7 +84,7 @@ def test_search_seed_and_queries_do_not_inject_fixed_business_terms(tmp_path):
         ),
     )
     state = service.start_run(
-        RunStartRequest(platform_mode="mock", source=str(REAL_SOURCE_FIXTURE))
+        RunStartRequest(platform_mode="mock", source=str(REAL_SOURCE_FIXTURE), strategy_version="V1")
     )
     run_id = state["run_id"]
 
@@ -102,7 +130,7 @@ def test_search_queries_preserve_source_terms_for_replay(tmp_path):
         ),
     )
     state = service.start_run(
-        RunStartRequest(platform_mode="mock", source=str(REAL_SOURCE_FIXTURE))
+        RunStartRequest(platform_mode="mock", source=str(REAL_SOURCE_FIXTURE), strategy_version="V1")
     )
     queries = service.read_jsonl(state["run_id"], "search_queries.jsonl")
     p2_queries = [
@@ -147,6 +175,73 @@ def test_search_intent_custom_evidence_fields_whitelist():
     assert llm_query["query_source_fields"] == ["seed_terms"]
 
 
+def test_v4_search_intent_uses_direct_sources_without_llm_variant():
+    client = FakeQueryVariantClient({"中医养生": "气血食疗"})
+    runtime = _Runtime()
+
+    queries = search_intent.run(
+        "run_1",
+        "policy_1",
+        _seed_pack(),
+        runtime,
+        client,
+        strategy_version="V4",
+    )
+
+    assert client.calls == []
+    assert [row["search_query"] for row in queries] == [
+        "中医养生",
+        "气血食疗方法",
+        "办公室八段锦",
+        "食疗补气血",
+        "八段锦",
+        "日常保健",
+    ]
+    assert [row["search_query_generation_method"] for row in queries] == [
+        "seed_term",
+        "piaoquan_topic_point",
+        "piaoquan_topic_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_point = queries[1]
+    assert first_point["pattern_seed_ref"]["query_source_type"] == "piaoquan_point_text"
+    assert first_point["pattern_seed_ref"]["query_source_text"] == "气血食疗方法"
+    assert first_point["pattern_seed_ref"]["query_source_rank"] == 1
+    assert first_point["query_source_refs"][0]["source_ref"]["point_type"] == "目的点"
+    assert first_point["raw_payload"]["query_source_refs"][0]["query_source_text"] == "气血食疗方法"
+    assert queries[3]["pattern_seed_ref"]["query_source_type"] == "category_terminal_element"
+    assert queries[3]["raw_payload"]["query_source_refs"][0]["generation_method"] == "category_leaf_element"
+
+
+def test_v4_search_intent_dedupes_by_direct_source_priority():
+    client = FakeQueryVariantClient({"中医养生": "气血食疗"})
+    runtime = _Runtime()
+    seed_pack = _seed_pack()
+    seed_pack["seed_terms"] = ["中医养生", "气血食疗方法"]
+    seed_pack["element_bindings"][0]["sample_elements"][0]["point_text"] = "分享气血食疗方法"
+    seed_pack["element_bindings"][0]["sample_elements"][0]["name"] = "气血食疗方法"
+
+    queries = search_intent.run(
+        "run_1",
+        "policy_1",
+        seed_pack,
+        runtime,
+        client,
+        strategy_version="V4",
+    )
+
+    assert client.calls == []
+    matching = [row for row in queries if row["search_query"] == "气血食疗方法"]
+    assert len(matching) == 1
+    assert matching[0]["search_query_generation_method"] == "seed_term"
+    assert matching[0]["pattern_seed_ref"]["query_source_type"] == "seed_term"
+
+
 def test_search_intent_custom_generic_filter_blocks_query():
     client = FakeQueryVariantClient({"中医养生": "禁用泛词"})
     client.profile = copy.deepcopy(DEFAULT_PROFILE)

tests/test_source_evidence.py

@@ -74,8 +74,8 @@ def test_source_evidence_tracks_multiple_query_sources_without_polluting_origin(
     )
     source_evidence = decision["source_evidence"]
 
-    assert multi_source_item["matched_search_query_ids"] == ["q_002", "q_003", "q_004"]
-    assert source_evidence["matched_search_query_ids"] == ["q_002", "q_003", "q_004"]
+    assert multi_source_item["matched_search_query_ids"] == ["q_002", "q_003", "q_004", "q_005"]
+    assert source_evidence["matched_search_query_ids"] == ["q_002", "q_003", "q_004", "q_005"]
     assert source_evidence["discovered_platform_content_id"] != source_evidence["source_post_id"]
     assert (
         source_evidence["discovered_platform_content_id"]
@@ -93,4 +93,5 @@ def test_source_evidence_tracks_multiple_query_sources_without_polluting_origin(
         "q_002",
         "q_003",
         "q_004",
+        "q_005",
     }

tests/test_v4_m1_query_sources_replay.py

@@ -0,0 +1,57 @@
+from content_agent.business_modules import search_intent, source_seed
+from content_agent.integrations.runtime_files import LocalRuntimeFileStore
+from content_agent.run_service import RunService
+from content_agent.schemas import RunStartRequest
+from tests.p1_helpers import FakeQueryVariantClient, REAL_SOURCE_FIXTURE, real_source_payload
+
+
+def test_v4_m1_direct_query_sources_replay_from_seed_pack(tmp_path):
+    runtime = LocalRuntimeFileStore(tmp_path / "runtime")
+    runtime.prepare_run("run_1")
+    seed = source_seed.run("run_1", "policy_1", real_source_payload(), runtime)
+    client = FakeQueryVariantClient({"爱国情感": "不应该调用"})
+
+    queries = search_intent.run(
+        "run_1",
+        "policy_1",
+        seed["pattern_seed_pack"],
+        runtime,
+        client,
+        strategy_version="V4",
+    )
+
+    assert client.calls == []
+    methods = {row["search_query_generation_method"] for row in queries}
+    assert {"seed_term", "piaoquan_topic_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)
+
+    by_method = {row["search_query_generation_method"]: row for row in queries}
+    assert by_method["seed_term"]["pattern_seed_ref"]["query_source_type"] == "seed_term"
+    assert by_method["piaoquan_topic_point"]["pattern_seed_ref"]["query_source_type"] == (
+        "piaoquan_point_text"
+    )
+    assert by_method["category_leaf_element"]["pattern_seed_ref"]["query_source_type"] == (
+        "category_terminal_element"
+    )
+
+
+def test_v4_m1_run_service_does_not_call_query_variant_client(tmp_path):
+    client = FakeQueryVariantClient({"爱国情感": "不应该调用"})
+    service = RunService(
+        runtime_root=tmp_path / "runtime" / "v1",
+        query_variant_client=client,
+    )
+
+    state = service.start_run(
+        RunStartRequest(platform_mode="mock", source=str(REAL_SOURCE_FIXTURE), strategy_version="V4")
+    )
+
+    queries = service.read_jsonl(state["run_id"], "search_queries.jsonl")
+    assert state["status"] == "success"
+    assert client.calls == []
+    assert queries
+    assert "llm_variant" not in {row["search_query_generation_method"] for row in queries}

tests/test_v4_m6_e2e_acceptance.py

@@ -0,0 +1,54 @@
+from content_agent.run_service import RunService
+from content_agent.schemas import RunStartRequest
+from tests.gemini_helpers import FakeGeminiVideoClient
+from tests.p1_helpers import FakeQueryVariantClient, REAL_SOURCE_FIXTURE
+
+
+def test_v4_m6_controlled_e2e_builds_complete_runtime_without_external_clients(tmp_path):
+    query_client = FakeQueryVariantClient({"爱国情感": "不应该调用"})
+    gemini_client = FakeGeminiVideoClient()
+    service = RunService(
+        runtime_root=tmp_path / "runtime" / "v1",
+        query_variant_client=query_client,
+        gemini_video_client=gemini_client,
+    )
+
+    state = service.start_run(
+        RunStartRequest(platform_mode="mock", source=str(REAL_SOURCE_FIXTURE), strategy_version="V4")
+    )
+    run_id = state["run_id"]
+
+    assert state["status"] == "success"
+    assert query_client.calls == []
+    assert gemini_client.calls
+
+    search_queries = service.read_jsonl(run_id, "search_queries.jsonl")
+    content_items = service.read_jsonl(run_id, "discovered_content_items.jsonl")
+    media_records = service.read_jsonl(run_id, "content_media_records.jsonl")
+    recall_evidence = service.read_jsonl(run_id, "pattern_recall_evidence.jsonl")
+    rule_decisions = service.read_jsonl(run_id, "rule_decisions.jsonl")
+    walk_actions = service.read_jsonl(run_id, "walk_actions.jsonl")
+    source_paths = service.read_jsonl(run_id, "source_path_records.jsonl")
+    search_clues = service.read_jsonl(run_id, "search_clues.jsonl")
+    final_output = service.read_json(run_id, "final_output.json")
+    strategy_review = service.read_json(run_id, "strategy_review.json")
+
+    assert {"seed_term", "piaoquan_topic_point", "category_leaf_element"} <= {
+        row["search_query_generation_method"] for row in search_queries
+    }
+    assert "llm_variant" not in {row["search_query_generation_method"] for row in search_queries}
+    assert content_items
+    assert media_records
+    assert recall_evidence
+    assert rule_decisions
+    assert walk_actions
+    assert source_paths
+    assert search_clues
+    assert final_output["policy"]["strategy_version"] == "V4"
+    assert strategy_review["schema_version"] == "runtime_record.v1"
+    assert all(row["scorecard"]["schema_version"] == "v4_scorecard.v1" for row in rule_decisions)
+    assert all("allow_walk" in row["decision_replay_data"] for row in rule_decisions)
+    assert any(path["source_path_type"] == "pattern_to_search_query" for path in source_paths)
+
+    validation = service.validate_run(run_id)
+    assert validation["status"] == "pass", validation["findings"]

tests/test_walk_engine_author.py

@@ -1,3 +1,5 @@
+import copy
+
 from content_agent.business_modules.walk_engine import run_bounded_walk
 from tests.p6_walk_helpers import FakeWalkPlatformClient, build_initial_walk_context, set_v4_allow_walk
 
@@ -89,6 +91,46 @@ def test_v4_author_edge_denies_allow_walk_false(tmp_path):
     assert skipped[0]["raw_payload"]["walk_gate_snapshot"]
 
 
+def test_v4_author_edge_filters_denied_before_consuming_author_budget(tmp_path):
+    context = build_initial_walk_context(tmp_path)
+    base_item = context["discovered_content_items"][0]
+    base_decision = context["rule_decisions"][0]
+    items = []
+    decisions = []
+    for index in range(4):
+        item = copy.deepcopy(base_item)
+        item["platform_content_id"] = f"73900000000000004{index:02d}"
+        item["platform_author_id"] = f"author_{index}"
+        item["has_more"] = False
+        item["tags"] = []
+        decision = copy.deepcopy(base_decision)
+        decision["decision_id"] = f"decision_{index}"
+        decision["decision_target_id"] = item["platform_content_id"]
+        set_v4_allow_walk(decision, index == 3)
+        items.append(item)
+        decisions.append(decision)
+    context["discovered_content_items"] = items
+    context["rule_decisions"] = decisions
+    client = FakeWalkPlatformClient()
+
+    result = run_bounded_walk(platform_client=client, **context)
+
+    assert [call["platform_author_id"] for call in client.author_calls] == ["author_3"]
+    skipped = [
+        row for row in result["walk_actions"]
+        if row["edge_id"] == "author_to_works" and row["walk_status"] == "skipped"
+    ]
+    assert [row["reason_code"] for row in skipped[:3]] == [
+        "v4_allow_walk_denied",
+        "v4_allow_walk_denied",
+        "v4_allow_walk_denied",
+    ]
+    assert not any(
+        row["reason_code"] == "budget_exhausted" and row["raw_payload"].get("allow_walk") is True
+        for row in skipped
+    )
+
+
 def test_non_v4_author_edge_keeps_legacy_permission(tmp_path):
     context = build_initial_walk_context(tmp_path)
     decision = context["rule_decisions"][0]