feat: implement p3 douyin platform access

content_agent/api.py

@@ -38,7 +38,7 @@ async def validation_exception_handler(request, exc: RequestValidationError):
 @app.post("/runs", response_model=RunStartResponse)
 def start_run(request: RunStartRequest) -> RunStartResponse:
     state = service.start_run(request)
-    if state["status"] != "success":
+    if state["status"] not in {"success", "partial_success"}:
         detail = error_response(
             state.get("error_code", ErrorCode.RUN_START_FAILED),
             state.get("error_message", "run failed"),

content_agent/business_modules/content_discovery/content_discovery_builder.py

@@ -79,6 +79,10 @@ def _build_discovered_content_item(
         "next_cursor",
         "platform_auth_mode",
         "platform_raw_payload",
+        "query_sources",
+        "matched_search_query_ids",
+        "matched_search_queries",
+        "matched_search_query_generation_methods",
     ]:
         if field in result:
             item[field] = result[field]

content_agent/business_modules/content_discovery/source_evidence.py

@@ -19,6 +19,12 @@ def build_source_evidence(
         "search_query_id": discovered_content_item["search_query_id"],
         "search_query": result.get("search_query"),
         "search_query_generation_method": result.get("search_query_generation_method"),
+        "query_sources": copy.deepcopy(result.get("query_sources", [])),
+        "matched_search_query_ids": list(result.get("matched_search_query_ids", [])),
+        "matched_search_queries": list(result.get("matched_search_queries", [])),
+        "matched_search_query_generation_methods": list(
+            result.get("matched_search_query_generation_methods", [])
+        ),
         "discovery_start_source": discovered_content_item["discovery_start_source"],
         "previous_discovery_step": discovered_content_item["previous_discovery_step"],
         "origin_path_id": (

content_agent/business_modules/platform_access.py

@@ -2,28 +2,98 @@ from __future__ import annotations
 
 from typing import Any
 
+from content_agent.errors import ContentAgentError, ErrorCode
 from content_agent.interfaces import PlatformSearchClient
 
 
 def run(
     search_queries: list[dict[str, Any]], platform_client: PlatformSearchClient
-) -> list[dict[str, Any]]:
+) -> dict[str, list[dict[str, Any]]]:
     results: list[dict[str, Any]] = []
-    seen_platform_content_ids: set[str] = set()
+    by_platform_content_id: dict[str, dict[str, Any]] = {}
+    query_failures: list[dict[str, Any]] = []
     for search_query in search_queries:
-        for result in platform_client.search(search_query):
+        try:
+            query_results = platform_client.search(search_query)
+        except Exception as exc:
+            query_failures.append(_query_failure(search_query, exc))
+            continue
+
+        for result in query_results:
             platform_content_id = result.get("platform_content_id")
-            if platform_content_id in seen_platform_content_ids:
+            enriched = _with_query_source(result, search_query)
+            if not platform_content_id:
+                results.append(enriched)
+                continue
+            existing = by_platform_content_id.get(platform_content_id)
+            if existing:
+                _append_query_source(existing, search_query)
                 continue
-            if platform_content_id:
-                seen_platform_content_ids.add(platform_content_id)
-            results.append(
-                {
-                    **result,
-                    "search_query": search_query["search_query"],
-                    "search_query_generation_method": search_query[
-                        "search_query_generation_method"
-                    ],
-                }
-            )
-    return results
+            by_platform_content_id[platform_content_id] = enriched
+            results.append(enriched)
+
+    if search_queries and query_failures and len(query_failures) == len(search_queries) and not results:
+        raise ContentAgentError(
+            ErrorCode.PLATFORM_REQUEST_FAILED,
+            "all platform queries failed",
+            {"query_failures": query_failures},
+        )
+    return {"platform_results": results, "query_failures": query_failures}
+
+
+def _with_query_source(
+    result: dict[str, Any], search_query: dict[str, Any]
+) -> dict[str, Any]:
+    enriched = {
+        **result,
+        "search_query": search_query["search_query"],
+        "search_query_generation_method": search_query["search_query_generation_method"],
+    }
+    _append_query_source(enriched, search_query)
+    return enriched
+
+
+def _append_query_source(result: dict[str, Any], search_query: dict[str, Any]) -> None:
+    source = {
+        "search_query_id": search_query["search_query_id"],
+        "search_query": search_query["search_query"],
+        "search_query_generation_method": search_query["search_query_generation_method"],
+    }
+    if search_query.get("llm_variant_of"):
+        source["llm_variant_of"] = search_query["llm_variant_of"]
+    query_sources = result.setdefault("query_sources", [])
+    if any(
+        item.get("search_query_id") == source["search_query_id"]
+        for item in query_sources
+    ):
+        return
+    query_sources.append(source)
+    result["matched_search_query_ids"] = [
+        item["search_query_id"] for item in query_sources
+    ]
+    result["matched_search_queries"] = [item["search_query"] for item in query_sources]
+    result["matched_search_query_generation_methods"] = [
+        item["search_query_generation_method"] for item in query_sources
+    ]
+
+
+def _query_failure(search_query: dict[str, Any], exc: Exception) -> dict[str, Any]:
+    if isinstance(exc, ContentAgentError):
+        error_code = exc.error_code.value
+        message = exc.message
+        detail = exc.detail
+    else:
+        error_code = ErrorCode.PLATFORM_REQUEST_FAILED.value
+        message = "platform query failed"
+        detail = {"exception_type": type(exc).__name__}
+    return {
+        "search_query_id": search_query["search_query_id"],
+        "search_query": search_query["search_query"],
+        "search_query_generation_method": search_query.get(
+            "search_query_generation_method"
+        ),
+        "status": "failed",
+        "error_code": error_code,
+        "message": message,
+        "error_detail": detail,
+    }

content_agent/business_modules/run_record/recorder.py

@@ -17,6 +17,7 @@ def run(
     decisions: list[dict[str, Any]],
     source_path_record_basis: list[dict[str, Any]],
     runtime: RuntimeFileStore,
+    query_failures: list[dict[str, Any]] | None = None,
 ) -> dict[str, list[dict[str, Any]]]:
     created_at = datetime.now(timezone.utc).isoformat()
     source_path_records = [
@@ -35,9 +36,14 @@ def run(
         for index, basis in enumerate(source_path_record_basis, start=1)
     ]
     search_clues = _build_search_clues(
-        run_id, policy_run_id, search_queries, discovered_content_items, decisions
+        run_id,
+        policy_run_id,
+        search_queries,
+        discovered_content_items,
+        decisions,
+        query_failures or [],
     )
-    run_events = _build_run_events(run_id, policy_run_id)
+    run_events = _build_run_events(run_id, policy_run_id, query_failures or [])
 
     runtime.append_jsonl(run_id, "run_events.jsonl", run_events)
     runtime.append_jsonl(run_id, "source_path_records.jsonl", source_path_records)
@@ -55,14 +61,21 @@ def _build_search_clues(
     search_queries: list[dict[str, Any]],
     discovered_content_items: list[dict[str, Any]],
     decisions: list[dict[str, Any]],
+    query_failures: list[dict[str, Any]],
 ) -> list[dict[str, Any]]:
     decision_by_target_id = {decision["decision_target_id"]: decision for decision in decisions}
+    failure_by_query_id = {failure["search_query_id"]: failure for failure in query_failures}
     items_by_search_query: dict[str, list[dict[str, Any]]] = defaultdict(list)
     for item in discovered_content_items:
-        items_by_search_query[item["search_query_id"]].append(item)
+        query_sources = item.get("query_sources") or [
+            {"search_query_id": item["search_query_id"]}
+        ]
+        for query_source in query_sources:
+            items_by_search_query[query_source["search_query_id"]].append(item)
 
     clues: list[dict[str, Any]] = []
     for index, search_query in enumerate(search_queries, start=1):
+        query_failure = failure_by_query_id.get(search_query["search_query_id"])
         query_items = items_by_search_query[search_query["search_query_id"]]
         action_counts = Counter(
             decision_by_target_id[item["platform_content_id"]]["decision_action"]
@@ -73,7 +86,10 @@ def _build_search_clues(
         review_content_count = action_counts["KEEP_CONTENT_FOR_REVIEW"]
         pending_content_count = action_counts["HOLD_CONTENT_PENDING"]
         rejected_content_count = action_counts["REJECT_CONTENT"]
-        if pooled_content_count or review_content_count:
+        if query_failure:
+            search_query_effect_status = "failed"
+            walk_next_step = "stop_search_query"
+        elif pooled_content_count or review_content_count:
             search_query_effect_status = "success"
             walk_next_step = "keep_search_query"
         elif pending_content_count:
@@ -100,13 +116,16 @@ def _build_search_clues(
                     "rejected_content_count": rejected_content_count,
                     "search_query_effect_status": search_query_effect_status,
                     "walk_next_step": walk_next_step,
+                    **({"query_failure": query_failure} if query_failure else {}),
                 }
             )
         )
     return clues
 
 
-def _build_run_events(run_id: str, policy_run_id: str) -> list[dict[str, Any]]:
+def _build_run_events(
+    run_id: str, policy_run_id: str, query_failures: list[dict[str, Any]]
+) -> list[dict[str, Any]]:
     created_at = datetime.now(timezone.utc).isoformat()
     event_specs = [
         ("source_loaded", "source_context.json", "pattern_seed_pack.json"),
@@ -117,7 +136,7 @@ def _build_run_events(run_id: str, policy_run_id: str) -> list[dict[str, Any]]:
         ("walk_planned", "rule_decisions.jsonl", "source_path_records.jsonl"),
         ("run_recorded", "walk_actions", "run_events.jsonl"),
     ]
-    return [
+    events = [
         with_raw_payload(
             {
                 "record_schema_version": RUNTIME_RECORD_SCHEMA_VERSION,
@@ -133,3 +152,23 @@ def _build_run_events(run_id: str, policy_run_id: str) -> list[dict[str, Any]]:
         )
         for index, (event_type, input_ref, output_ref) in enumerate(event_specs, start=1)
     ]
+    for index, failure in enumerate(query_failures, start=len(events) + 1):
+        events.append(
+            with_raw_payload(
+                {
+                    "record_schema_version": RUNTIME_RECORD_SCHEMA_VERSION,
+                    "run_id": run_id,
+                    "policy_run_id": policy_run_id,
+                    "event_id": f"evt_{index:03d}",
+                    "event_type": "platform_query_failed",
+                    "status": "failed",
+                    "input_ref": f"search_queries.jsonl:{failure['search_query_id']}",
+                    "output_ref": "search_clues.jsonl",
+                    "error_code": failure["error_code"],
+                    "message": failure["message"],
+                    "query_failure": failure,
+                    "created_at": created_at,
+                }
+            )
+        )
+    return events

content_agent/business_modules/run_record/validation.py

@@ -491,11 +491,11 @@ def _check_summary(data: dict[str, Any], findings: list[dict[str, Any]]) -> None
         clue_counts["HOLD_CONTENT_PENDING"] += clue.get("pending_content_count", 0)
         clue_counts["REJECT_CONTENT"] += clue.get("rejected_content_count", 0)
     for action, count in action_counts.items():
-        if clue_counts[action] != count:
+        if clue_counts[action] < count:
             _fail(
                 findings,
                 "search_clue_mismatch",
-                f"search_clues {action} expected {count}, got {clue_counts[action]}",
+                f"search_clues {action} expected at least {count}, got {clue_counts[action]}",
             )
 
 

content_agent/business_modules/walk_strategy.py

@@ -34,23 +34,37 @@ def run(
 
     for item in discovered_content_items:
         decision = decision_by_target_id[item["platform_content_id"]]
-        source_path_record_basis.append(
+        query_sources = item.get("query_sources") or [
             {
-                "policy_run_id": decision["policy_run_id"],
-                "record_schema_version": decision["record_schema_version"],
-                "from_node_type": "SearchQuery",
-                "from_node_id": item["search_query_id"],
-                "to_node_type": "Content",
-                "to_node_id": item["platform_content_id"],
-                "source_path_type": "search_query_to_content",
-                "rule_pack_id": decision["rule_pack_id"],
-                "decision_id": decision["decision_id"],
-                "discovery_start_source": item["discovery_start_source"],
-                "previous_discovery_step": item["previous_discovery_step"],
-                "origin_path_id": decision["source_evidence"]["origin_path_id"],
-                "source_evidence_ref": decision["decision_input_snapshot_id"],
+                "search_query_id": item["search_query_id"],
+                "search_query": item.get("search_query"),
+                "search_query_generation_method": item.get(
+                    "search_query_generation_method"
+                ),
             }
-        )
+        ]
+        for query_source in query_sources:
+            search_query_id = query_source["search_query_id"]
+            source_path_record_basis.append(
+                {
+                    "policy_run_id": decision["policy_run_id"],
+                    "record_schema_version": decision["record_schema_version"],
+                    "from_node_type": "SearchQuery",
+                    "from_node_id": search_query_id,
+                    "to_node_type": "Content",
+                    "to_node_id": item["platform_content_id"],
+                    "source_path_type": "search_query_to_content",
+                    "rule_pack_id": decision["rule_pack_id"],
+                    "decision_id": decision["decision_id"],
+                    "discovery_start_source": item["discovery_start_source"],
+                    "previous_discovery_step": item["previous_discovery_step"],
+                    "origin_path_id": (
+                        f"search_query_to_content:{search_query_id}:"
+                        f"{item['platform_content_id']}"
+                    ),
+                    "source_evidence_ref": decision["decision_input_snapshot_id"],
+                }
+            )
         walk_actions.append(
             {
                 "record_schema_version": decision["record_schema_version"],

content_agent/graph.py

@@ -54,8 +54,8 @@ def build_run_graph(deps: RunDependencies):
         return {"search_queries": search_queries, "current_step": "plan_queries"}
 
     def search_platform(state: RunState) -> dict[str, Any]:
-        results = platform_access.run(state["search_queries"], deps.platform_client)
-        return {"platform_results": results, "current_step": "search_platform"}
+        result = platform_access.run(state["search_queries"], deps.platform_client)
+        return {**result, "current_step": "search_platform"}
 
     def build_discovered_content(state: RunState) -> dict[str, Any]:
         result = content_discovery.run(
@@ -105,6 +105,7 @@ def build_run_graph(deps: RunDependencies):
             state["rule_decisions"],
             state["source_path_record_basis"],
             deps.runtime,
+            state.get("query_failures", []),
         )
         return {**result, "current_step": "record_run"}
 

content_agent/integrations/douyin.py

@@ -135,22 +135,27 @@ class CrawapiDouyinClient:
         }
 
     def _fetch_content_portrait(self, platform_content_id: str) -> dict[str, Any]:
-        try:
-            data = self._post_json(
-                self.content_portrait_path,
-                {
-                    "content_id": platform_content_id,
-                    "need_age": True,
-                    "need_gender": True,
-                    "need_province": True,
-                    "need_city": False,
-                    "need_city_level": False,
-                    "need_phone_brand": False,
-                    "need_phone_price": False,
-                },
-                operation="content_portrait",
-            )
-        except RuntimeError:
+        data = None
+        for _ in range(2):
+            try:
+                data = self._post_json(
+                    self.content_portrait_path,
+                    {
+                        "content_id": platform_content_id,
+                        "need_age": True,
+                        "need_gender": True,
+                        "need_province": True,
+                        "need_city": False,
+                        "need_city_level": False,
+                        "need_phone_brand": False,
+                        "need_phone_price": False,
+                    },
+                    operation="content_portrait",
+                )
+                break
+            except RuntimeError:
+                continue
+        if data is None:
             return {"portrait_available": False, "age_50_plus_level": "missing"}
 
         portrait = _extract_portrait_dimensions(data)
@@ -190,6 +195,9 @@ class CrawapiDouyinClient:
             raise RuntimeError(f"crawapi {operation} failed: bad_json") from exc
         if not isinstance(data, dict):
             raise RuntimeError(f"crawapi {operation} failed: bad_response")
+        code = data.get("code")
+        if code is not None and code not in (0, "0"):
+            raise RuntimeError(f"crawapi {operation} failed: business_error")
         return data
 
 

content_agent/models.py

@@ -3,7 +3,7 @@ from __future__ import annotations
 from typing import Any, Literal, TypedDict
 
 
-RunStatus = Literal["running", "success", "blocked", "failed"]
+RunStatus = Literal["running", "success", "partial_success", "blocked", "failed"]
 DecisionAction = Literal[
     "ADD_TO_CONTENT_POOL",
     "KEEP_CONTENT_FOR_REVIEW",
@@ -30,6 +30,7 @@ class RunState(TypedDict, total=False):
     pattern_seed_pack: dict[str, Any]
     search_queries: list[dict[str, Any]]
     platform_results: list[dict[str, Any]]
+    query_failures: list[dict[str, Any]]
     discovered_content_items: list[dict[str, Any]]
     content_media_records: list[dict[str, Any]]
     evidence_bundles: list[dict[str, Any]]

content_agent/run_service.py

@@ -127,7 +127,7 @@ class RunService:
 
             deps = RunDependencies(
                 runtime=self.runtime,
-                platform_client=self._platform_client(request.platform_mode),
+                platform_client=self._platform_client(request.platform, request.platform_mode),
                 policy_store=self.policy_store,
                 query_variant_client=self.query_variant_client,
             )
@@ -231,12 +231,14 @@ class RunService:
         )
 
     def _record_success_metadata(self, state: RunState) -> None:
+        final_status = "partial_success" if state.get("query_failures") else "success"
+        state["status"] = final_status
         self.runtime.record_policy_run(_policy_run_record_from_state(state))
         validation = self.validate_run(state["run_id"])
         self.runtime.update_run_record(
             state["run_id"],
             {
-                "status": "success",
+                "status": final_status,
                 "current_step": state.get("current_step", "review_strategy"),
                 "validation_status": validation["status"],
                 "completed_at": _utc_now(),
@@ -247,12 +249,15 @@ class RunService:
             state["policy_run_id"],
             event_id="lifecycle_success",
             event_type="run_succeeded",
-            status="success",
-            message="run succeeded",
+            status=final_status,
+            message="run succeeded"
+            if final_status == "success"
+            else "run partially succeeded",
             output_ref="final_output.json",
             raw_payload={
                 "validation_status": validation["status"],
                 "policy_bundle_id": state.get("policy_bundle_id"),
+                "query_failures": state.get("query_failures", []),
             },
         )
 
@@ -365,10 +370,17 @@ class RunService:
             )
         return error_from_exception(exc, detail={"exception_type": type(exc).__name__})
 
-    def _platform_client(self, platform_mode: str) -> PlatformSearchClient:
+    def _platform_client(self, platform: str, platform_mode: str) -> PlatformSearchClient:
         if platform_mode == "mock":
             return MockPlatformClient()
         if platform_mode == "real":
+            if platform != "douyin":
+                raise ContentAgentError(
+                    ErrorCode.INVALID_REQUEST,
+                    "unsupported real platform",
+                    {"platform": platform},
+                    status_code=400,
+                )
             try:
                 return CrawapiDouyinClient.from_env()
             except Exception as exc:
@@ -386,7 +398,7 @@ class RunService:
 
     def get_summary(self, run_id: str) -> dict:
         final_output_exists = (self.runtime.run_dir(run_id) / "final_output.json").exists()
-        status = "success" if final_output_exists else "failed"
+        status = self._summary_status(run_id, final_output_exists)
         validation = self.validate_run(run_id) if final_output_exists else {"status": "fail"}
         policy_run_id = None
         if final_output_exists:
@@ -402,6 +414,25 @@ class RunService:
             "errors": [],
         }
 
+    def _summary_status(self, run_id: str, final_output_exists: bool) -> str:
+        try:
+            run_events = self.runtime.read_jsonl(run_id, "run_events.jsonl")
+            lifecycle_events = [
+                event for event in run_events if str(event.get("event_id", "")).startswith("lifecycle_")
+            ]
+        except Exception:
+            run_events = []
+            lifecycle_events = []
+        if lifecycle_events:
+            return lifecycle_events[-1].get("status") or (
+                "success" if final_output_exists else "failed"
+            )
+        if final_output_exists and any(
+            event.get("event_type") == "platform_query_failed" for event in run_events
+        ):
+            return "partial_success"
+        return "success" if final_output_exists else "failed"
+
     def read_jsonl(self, run_id: str, filename: str) -> list[dict]:
         return self.runtime.read_jsonl(run_id, filename)
 
@@ -438,7 +469,7 @@ def _policy_run_record_from_state(state: RunState) -> dict[str, Any]:
         "rule_pack_source_ref": policy_bundle.get("rule_pack_source_ref"),
         "evidence_bundle_schema_version": policy_bundle.get("evidence_bundle_schema_version"),
         "runtime_record_schema_version": policy_bundle.get("runtime_record_schema_version"),
-        "status": "success",
+        "status": state.get("status", "success"),
         "metrics": (state.get("final_output") or {}).get("summary", {}),
         "decision_summary": _decision_summary(decisions),
         "raw_payload": {

tech_documents/工程落地/implementation_briefs/P3/00_P3_Brief_Index.md

@@ -0,0 +1,210 @@
+# P3 Implementation Brief Index
+
+状态：本目录是 P3 平台接入模块实施前的短实施简报集合。它不是新的架构权威源，也不是产品计划；长期事实仍以 `tech_documents/工程落地/04_V1阶段开发计划.md`、`tech_documents/工程落地/05_阶段验收清单.md`、`product_documents/抖音游走策略/runtime_v1_records_schema.md`、`tech_documents/数据库字段总览/content_agent_schema_registry.json` 和当前代码为准。
+
+## 目标
+
+为 P3 平台接入模块保留实施清单，颗粒度固定到文件级、函数 / 类级、数据合同级、验证级和失败归因级。
+
+P3 只处理平台接入：
+
+1. P3A：平台边界与配置。
+2. P3B：Crawapi 抖音 keyword 搜索合同。
+3. P3C：query 级失败与 run 级 `partial_success`。
+4. P3D：内容点赞画像 retry 与字段归一。
+5. P3E：跨 query 内容去重与来源合并。
+6. P3F：P3 验收与 drift guard。
+
+本目录不处理：
+
+- P2 query 生成。
+- P4 Pattern 回扣。
+- P5 规则包业务拍板。
+- 作者作品。
+- tag 搜索。
+- 账号画像兜底。
+- 小红书、快手、B 站或多平台路由。
+- 队列化限流。
+- 发布任务。
+
+## 现有证据
+
+- `tech_documents/工程落地/04_V1阶段开发计划.md`：P3 已拍板 V1 真实平台只允许抖音；最小闭环为抖音关键词搜索 + 内容点赞画像；每 query 数量走 env；初始只拉第一页并保留 `has_more / next_cursor`；keyword 单 query 失败继续其他 query；部分成功时 run 记 `partial_success`；portrait 失败重试一次；同一 `platform_content_id` 合并全部 query 来源。
+- `content_agent/schemas.py`：当前请求默认 `platform="douyin"`、`platform_mode="real"`。
+- `content_agent/run_service.py`：当前 `platform_mode=real` 会构造 `CrawapiDouyinClient.from_env()`；尚未显式拒绝非 `douyin` platform；平台异常当前会进入整次 run failure。
+- `content_agent/business_modules/platform_access.py`：当前逐 query 调用 `platform_client.search()`；同 `platform_content_id` 只保留第一次命中，后续 query 来源被跳过；尚未做 query 级失败继续。
+- `content_agent/integrations/douyin.py`：当前已接 Crawapi keyword search 和 content portrait；默认每 query 处理 3 条；只拉一页；会保留 `has_more / next_cursor`；portrait 失败直接标记缺失，尚无 retry。
+- `content_agent/integrations/douyin.py`：当前真实结果写 `pattern_recall="pattern_recall_pending"`，不会伪装成 Pattern 通过。
+- `content_agent/models.py`：当前 `RunStatus` 只有 `running / success / blocked / failed`，尚无 `partial_success`。
+- `content_agent/graph.py`：P3 位于 `search_platform` 节点，后续接 `build_discovered_content`、`evaluate_rules`。
+- `content_agent/business_modules/content_discovery/content_discovery_builder.py`：平台归一结果会进入 `discovered_content_items.jsonl`、`content_media_records.jsonl` 和 EvidenceBundle。
+- `content_agent/business_modules/rule_judgment/evaluator.py`：最终入池 / 待复看 / 淘汰由规则包判断，不应由 P3 直接拍板。
+- `tests/test_douyin_client.py`：当前覆盖 keyword 字段映射、HTTP 错误脱敏、portrait 缺失、每 query 限量、41-50 不计 50+。
+- `tests/test_platform_access.py`：当前覆盖跨 query 去重和 `llm_variant` 元数据保留。
+- `product_documents/旧版和上下游理解/旧版ContentFindAgent外部数据源接口审计.md`：旧版 Crawapi keyword endpoint 为 `POST /crawler/dou_yin/keyword`，入参 `keyword / content_type / sort_type / publish_time / cursor / account_id`；返回 `aweme_id / desc / author.sec_uid / statistics / has_more / next_cursor`；内容点赞画像 endpoint 为 `POST /crawler/dou_yin/re_dian_bao/video_like_portrait`。
+- `tech_documents/数据接口与来源/external_data_sources_registry.json`：`PLT_DOUYIN_KEYWORD` 和 `PLT_DOUYIN_CONTENT_PROFILE` 均为 `verified + v1_active`。
+- `product_documents/抖音游走策略/runtime_v1_records_schema.md`：`platform_content_id` 是平台内容 ID，抖音下等于抖音视频 ID；`search_query_effect_status` 支持 `success / weak_effective / pending / blocked / failed`。
+- `product_documents/抖音游走策略/douyin_available_walk_strategy.v1.json`：`next_cursor` 是分页状态节点，不是业务节点；后续分页应由游走策略触发。
+- `product_documents/规则包/douyin_rule_packs.v1.json`：缺 `platform_content_id`、缺内容画像等是规则包 gate / scorecard 输入，P3 不直接做最终业务拍板。
+- `sql/content_agent_schema.sql`：P3 相关承载表已存在，包括 `content_agent_queries`、`content_agent_discovered_content_items`、`content_agent_content_media_records`、`content_agent_search_clues`、`content_agent_run_events`、`content_agent_runs`。
+- `content_agent/integrations/database_runtime.py`：`search_queries.jsonl`、`discovered_content_items.jsonl`、`content_media_records.jsonl`、`search_clues.jsonl`、`run_events.jsonl` 已映射到 DB。
+
+真实 DB 证据：
+
+- DB validator 显示 `content_agent_*` 15 表已存在，`schema_ready=true`。
+- `content_agent_discovered_content_items` 已有唯一键 `run_id + policy_run_id + platform + platform_content_id`，支持跨 query 去重。
+- `content_agent_runs.status` 是字符串字段，`partial_success` 不需要新增 DDL。
+- P3 不新增 DDL；query failure、portrait retry、pagination、source merge 细节先进入 `raw_payload`、`platform_raw_payload`、`run_events.raw_payload`、`source_evidence` 或后续 `source_path_records`。
+
+sub-agent 交叉验证结论：
+
+- 代码逻辑 sub-agent 确认：当前已有 Crawapi keyword / portrait 基础，但 query 级失败继续、run `partial_success`、portrait retry、重复内容合并全部 query 来源尚未完成。
+- 文档 / JSON sub-agent 确认：P3 已拍板无剩余开工前问题；Crawapi 当前联调不可用时按旧文档合同推进，失败归因到登录态、账号态或配置。
+- DB / 数据合同 sub-agent 确认：schema ready，P3 不新增 DDL；相关状态可落现有表和 JSON 字段。
+
+## 修改范围
+
+本目录只新增 P3 brief 文档。后续真正执行 P3 时，预计涉及：
+
+- `content_agent/business_modules/platform_access.py`
+- `content_agent/integrations/douyin.py`
+- `content_agent/run_service.py`
+- `content_agent/models.py`
+- `content_agent/graph.py`
+- `content_agent/business_modules/content_discovery/content_discovery_builder.py`
+- `content_agent/business_modules/rule_judgment/evaluator.py`
+- `.env.example`
+- `tests/test_douyin_client.py`
+- `tests/test_platform_access.py`
+- 后续新增或扩展 RunService / API 相关测试。
+
+## 不修改范围
+
+- 本轮不改代码。
+- 本轮不写 DB。
+- 本轮不执行 DDL。
+- 本轮不更新 `.env` 真实值。
+- 本轮不把任何真实密码、token、API key 写进文档。
+- 本轮不修改旧版归档目录。
+
+## 涉及文件 / 函数 / 类
+
+全局证据文件：
+
+- `content_agent/business_modules/platform_access.py`
+- `content_agent/integrations/douyin.py`
+- `content_agent/run_service.py`
+- `content_agent/models.py`
+- `content_agent/graph.py`
+- `content_agent/business_modules/content_discovery/content_discovery_builder.py`
+- `content_agent/business_modules/rule_judgment/evaluator.py`
+- `tests/test_douyin_client.py`
+- `tests/test_platform_access.py`
+- `product_documents/抖音游走策略/runtime_v1_records_schema.md`
+- `product_documents/抖音游走策略/douyin_available_walk_strategy.v1.json`
+- `product_documents/规则包/douyin_rule_packs.v1.json`
+- `tech_documents/数据接口与来源/external_data_sources_registry.json`
+- `tech_documents/数据库字段总览/content_agent_schema_registry.json`
+- `sql/content_agent_schema.sql`
+
+关键对象：
+
+- `platform_access.run()`
+- `CrawapiDouyinClient.from_env()`
+- `CrawapiDouyinClient.search()`
+- `CrawapiDouyinClient._post_json()`
+- `CrawapiDouyinClient._fetch_content_portrait()`
+- `RunService._platform_client()`
+- `RunService._record_success_metadata()`
+- `RunStatus`
+- `build_run_graph()`
+- `content_discovery.run()`
+- `rule_judgment.run()`
+
+## 数据合同
+
+P3 输入：
+
+- P2 产出的 `search_queries.jsonl` / `content_agent_queries`。
+- 每条 query 至少包含 `search_query_id`、`search_query`、`search_query_generation_method`、`discovery_start_source`、`previous_discovery_step`。
+
+P3 平台输出：
+
+- `platform="douyin"`
+- `platform_content_id`
+- `platform_author_id`
+- `description`
+- `author_display_name`
+- `statistics`
+- `tags`
+- `text_extra`
+- `create_time`
+- `has_more`
+- `next_cursor`
+- `portrait_available`
+- `age_distribution`
+- `age_50_plus_ratio`
+- `age_50_plus_tgi`
+- `pattern_recall="pattern_recall_pending"`
+- `category_or_element_binding="pattern_recall_pending"`
+- `content_metadata_source="douyin_keyword_search"`
+- `platform_raw_payload`
+
+P3 状态合同：
+
+- 单 query keyword 失败：该 query 记录失败，其他 query 继续。
+- 所有 query 都失败且无候选：run 失败或 blocked，按后续 P3C 具体错误合同执行。
+- 部分 query 失败但仍有候选且后续链路跑完：run 记 `partial_success`。
+- portrait 重试后仍失败：内容保留为缺画像输入，后续规则包 reject。
+- 同一 `platform_content_id` 多 query 命中：只保留一个内容对象，但记录全部来源。
+
+DB 承载：
+
+- `content_agent_queries.search_query_effect_status`
+- `content_agent_search_clues.search_query_effect_status`
+- `content_agent_run_events.status / error_code / raw_payload`
+- `content_agent_runs.status`
+- `content_agent_discovered_content_items.raw_payload / platform_raw_payload / source_evidence`
+- `content_agent_content_media_records.raw_payload`
+
+不新增 DB 字段，不修改 DDL。
+
+## 实施步骤
+
+1. 每进入一个 P3 小阶段，先读对应 brief。
+2. 只执行该 brief 的修改范围，不顺手做 P4 Pattern 回扣、P5 规则策略或作者/tag 扩展。
+3. 修改前用 brief 中的证据文件确认当前代码、文档和 DB 状态仍未漂移。
+4. 修改后先跑该 brief 指定的最小验证，再跑 P3 汇总测试。
+5. 如果验证失败，先按 brief 的失败归因判断，不扩大修改面。
+
+## 验证命令
+
+```bash
+find tech_documents/工程落地/implementation_briefs/P3 -maxdepth 1 -type f | sort
+rg -n "Implementation Brief|sub-agent|验证命令|失败归因" tech_documents/工程落地/implementation_briefs/P3
+rg -n "PASS[W]ORD=|TO[K]EN=|API[_]KEY=|SEC[R]ET=" tech_documents/工程落地/implementation_briefs/P3
+rg -n "partial_success|has_more|next_cursor|platform_content_id|portrait|Crawapi" tech_documents/工程落地/implementation_briefs/P3
+uv run python scripts/validate_schema_registry.py
+uv run --with pymysql python scripts/validate_content_agent_db.py --env-file .env
+uv run pytest tests/test_douyin_client.py tests/test_platform_access.py tests/test_api.py tests/test_v1_graph.py tests/test_runtime_files.py tests/test_source_evidence.py tests/test_p0d_p0g.py tests/test_search_intent.py -q
+uv run pytest -q
+```
+
+## 失败归因
+
+- brief 文件缺失：本目录落地问题。
+- 敏感键值命中：brief 内容问题，应改成只写 env key 名，不写赋值。
+- `validate_schema_registry.py` 失败：字段 registry 与 SQL / 文档漂移。
+- DB validator 失败：DB 连接、权限、schema 或 `.env` 目标库问题。
+- P0/P1/P2 测试失败：本次只新增 Markdown，通常不应引发业务测试失败；先检查工作区已有代码改动。
+- P3 后续实现测试失败：按具体 brief 判断是平台配置、Crawapi 合同、query 状态、portrait retry、来源合并还是规则边界问题。
+
+## sub-agent 交叉验证要点
+
+- 代码侧 sub-agent 确认：`platform_access.run()` 当前还没有 query 级失败继续和多 query 来源合并。
+- 代码侧 sub-agent 确认：`CrawapiDouyinClient._fetch_content_portrait()` 当前没有 retry。
+- 代码侧 sub-agent 确认：`RunStatus` 当前还没有 `partial_success`。
+- 文档 / JSON sub-agent 确认：P3 只做抖音 keyword + 内容画像，不能混进作者作品、tag、多平台或 P4 回扣。
+- 文档 / JSON sub-agent 确认：旧 Crawapi 正常合同是 P3 实现依据，真实联调失败归因到登录态、账号态或配置。
+- DB 侧 sub-agent 确认：`content_agent_*` schema ready，P3 不需要新增 DDL。
+- DB 侧 sub-agent 确认：`partial_success`、query failure、portrait retry、source merge 都可用现有字段承载。

tech_documents/工程落地/implementation_briefs/P3/P3A_PlatformBoundary_And_Config.md

@@ -0,0 +1,127 @@
+# P3A Platform Boundary And Config Implementation Brief
+
+状态：本 brief 覆盖 P3 平台边界与配置。它定义 V1 真实平台唯一允许抖音，以及 Crawapi 配置必须来自 CFA 自己的 env。
+
+## 目标
+
+- 真实平台只允许 `douyin`。
+- `platform_mode=mock` 继续稳定用于测试、smoke 和本地开发。
+- `platform_mode=real` 只走 Crawapi 抖音。
+- 非 `douyin` platform 后续实现时必须结构化失败，不静默 fallback。
+- 每 query 拉取数量由 env 控制，不使用 Excel 作为 runtime 配置源。
+
+## 现有证据
+
+- `tech_documents/工程落地/04_V1阶段开发计划.md`：已拍板 V1 真实平台只允许抖音；小红书、快手、B 站和多平台路由不进入 V1 真实平台范围。
+- `content_agent/schemas.py`：当前 `RunStartRequest.platform` 默认是 `douyin`，`platform_mode` 默认是 `real`。
+- `content_agent/run_service.py`：当前 `_platform_client()` 对 `mock` 返回 `MockPlatformClient()`，否则返回 `CrawapiDouyinClient.from_env()`；还没有显式拒绝非 `douyin`。
+- `content_agent/integrations/douyin.py`：`CrawapiDouyinClient.from_env()` 读取 `CONTENTFIND_API_CRAWAPI_BASE_URL`、`CONTENTFIND_DOUYIN_KEYWORD_PATH`、`CONTENTFIND_DOUYIN_VIDEO_LIKE_PORTRAIT_PATH`、`CONTENTFIND_DOUYIN_MAX_RESULTS_PER_QUERY` 等 key。
+- `tech_documents/数据接口与来源/external_data_sources_registry.json`：`PLT_DOUYIN_KEYWORD` 与 `PLT_DOUYIN_CONTENT_PROFILE` 是 V1 active；作者作品、账号画像为 planned read，不属于 P3。
+- `tech_documents/工程落地/05_阶段验收清单.md`：临时每 query 3 条、一页、抖音-only 不能写成长期策略。
+
+sub-agent 交叉验证结论：
+
+- 代码侧 sub-agent 确认：当前代码默认 douyin，但未显式拒绝非 douyin。
+- 文档 / JSON sub-agent 确认：P3 没有剩余平台拍板项。
+- DB 侧 sub-agent 确认：P3 平台边界不需要 DDL。
+
+## 修改范围
+
+后续执行 P3A 时允许：
+
+- 修改 `content_agent/run_service.py` 的平台选择逻辑。
+- 必要时在 `content_agent/errors.py` 增加或复用平台配置相关 error code。
+- 修改 `content_agent/models.py` 或 `content_agent/schemas.py` 中平台字段的注释 / 类型边界。
+- 更新 `.env.example` 的非敏感 key 名。
+- 增加非 douyin 平台拒绝测试。
+
+## 不修改范围
+
+- 不新增小红书、快手、B 站客户端。
+- 不新增平台路由表。
+- 不接作者作品。
+- 不接 tag 搜索。
+- 不接账号画像。
+- 不写 DB。
+- 不修改成功 API 响应字段。
+
+## 涉及文件 / 函数 / 类
+
+文件：
+
+- `content_agent/schemas.py`
+- `content_agent/run_service.py`
+- `content_agent/errors.py`
+- `content_agent/integrations/douyin.py`
+- `content_agent/integrations/mock_platform.py`
+- `.env.example`
+- `tests/test_api.py`
+- `tests/test_v1_graph.py`
+- 后续可新增或扩展 `tests/test_run_service.py`
+
+函数 / 类：
+
+- `RunStartRequest`
+- `RunService._platform_client()`
+- `CrawapiDouyinClient.from_env()`
+- `MockPlatformClient`
+- `ErrorCode`
+
+## 数据合同
+
+请求输入：
+
+- `platform="douyin"` 是 V1 real 模式唯一合法真实平台。
+- `platform_mode="mock"` 可用于测试和 smoke。
+- `platform_mode="real"` 必须使用 Crawapi 抖音。
+
+env key 名：
+
+- `CONTENTFIND_API_CRAWAPI_BASE_URL`
+- `CONTENTFIND_API_CRAWAPI_TIMEOUT_SECONDS`
+- `CONTENTFIND_DOUYIN_KEYWORD_PATH`
+- `CONTENTFIND_DOUYIN_VIDEO_LIKE_PORTRAIT_PATH`
+- `CONTENTFIND_DOUYIN_DEFAULT_ACCOUNT_ID`
+- `CONTENTFIND_DOUYIN_DEFAULT_CONTENT_TYPE`
+- `CONTENTFIND_DOUYIN_DEFAULT_SORT_TYPE`
+- `CONTENTFIND_DOUYIN_DEFAULT_PUBLISH_TIME`
+- `CONTENTFIND_DOUYIN_DEFAULT_CURSOR`
+- `CONTENTFIND_DOUYIN_MAX_RESULTS_PER_QUERY`
+
+错误合同：
+
+- 非 douyin 平台：结构化失败，不能 fallback 到 douyin。
+- 缺 Crawapi 配置：结构化失败，不能进入真实请求。
+- 错误 detail 不得包含 secret、token、真实密码或完整敏感请求头。
+
+## 实施步骤
+
+1. 读取 `RunStartRequest` 当前默认值和 API 测试，确认不破坏成功响应字段。
+2. 在平台 client 选择前显式校验 `request.platform`。
+3. 当 `platform_mode=real` 且 `platform!="douyin"` 时抛结构化错误。
+4. 保持 `platform_mode=mock` 的测试链路稳定。
+5. 确认 `CrawapiDouyinClient.from_env()` 只读 CFA 自己的 env，不借用其他项目 env。
+6. 在 `.env.example` 只补 key 名，不写真实值。
+7. 增加测试：非 douyin 平台失败；mock 模式不受影响；缺 Crawapi 配置返回结构化错误。
+
+## 验证命令
+
+```bash
+uv run pytest tests/test_api.py tests/test_v1_graph.py tests/test_douyin_client.py -q
+rg -n "xiaohongshu|kuaishou|bilibili|B站|小红书|快手" content_agent tests tech_documents/工程落地/implementation_briefs/P3
+rg -n "PASS[W]ORD=|TO[K]EN=|API[_]KEY=|SEC[R]ET=" tech_documents/工程落地/implementation_briefs/P3 .env.example
+```
+
+## 失败归因
+
+- 非 douyin 仍进入 Crawapi：平台边界校验缺失。
+- mock 测试失败：P3A 修改误伤测试客户端。
+- 缺 env 时异常泄漏真实配置：错误清洗不符合 P0G 合同。
+- 新增多平台字样进入代码主链路：越过 P3 边界。
+
+## sub-agent 交叉验证要点
+
+- 代码侧确认：`RunService._platform_client()` 显式拒绝非 douyin。
+- 代码侧确认：mock 模式测试仍走 `MockPlatformClient`。
+- 文档 / JSON 侧确认：作者作品、tag、多平台仍不进入 P3。
+- DB 侧确认：P3A 不涉及 DB schema。

tech_documents/工程落地/implementation_briefs/P3/P3B_CrawapiKeyword_SearchContract.md

@@ -0,0 +1,149 @@
+# P3B Crawapi Keyword Search Contract Implementation Brief
+
+状态：本 brief 覆盖 Crawapi 抖音关键词搜索合同。它定义 P3 如何按旧文档正常接口合同读取第一页 keyword 搜索结果，并保留分页状态。
+
+## 目标
+
+- 按旧文档 Crawapi keyword 合同实现抖音关键词搜索。
+- 每条 query 初始只拉第一页。
+- 保存 `has_more / next_cursor` 给后续游走策略。
+- 每 query 返回数量由 env 控制。
+- 平台接入层只做字段归一，不做 Pattern 回扣或规则拍板。
+
+## 现有证据
+
+- `product_documents/旧版和上下游理解/旧版ContentFindAgent外部数据源接口审计.md`：keyword endpoint 为 `POST /crawler/dou_yin/keyword`；入参为 `keyword / content_type / sort_type / publish_time / cursor / account_id`；返回 `aweme_id / desc / author.nickname / author.sec_uid / statistics.* / has_more / next_cursor`。
+- 同一旧版文档记录：历史可用 payload 包含 `content_type=视频`、`sort_type=综合排序`、`cursor=0`。
+- `tech_documents/数据接口与来源/external_data_sources_registry.json`：`PLT_DOUYIN_KEYWORD` 为 `verified + v1_active`，使用阶段是 `platform_access` 和 `content_discovery`。
+- `product_documents/抖音游走策略/抖音接口真实case样例.json`：keyword case 中 `has_more=true`、`next_cursor` 有值，`platform_content_id` 用于内容去重和后续画像。
+- `product_documents/抖音游走策略/douyin_available_walk_strategy.v1.json`：`next_cursor` 是分页状态节点，不是视频、作者或 tag 业务节点。
+- `content_agent/integrations/douyin.py`：当前 `CrawapiDouyinClient.search()` 构造 keyword payload，只取第一页，限制 `max_results_per_query`，并把 `has_more / next_cursor` 放入归一结果。
+- `tests/test_douyin_client.py`：已有 keyword 字段映射、空结果、HTTP error、每 query 限量测试。
+
+sub-agent 交叉验证结论：
+
+- 文档 / JSON sub-agent 确认：Crawapi 真实联调失败时按旧文档合同推进，失败归因到登录态、账号态或配置。
+- 代码侧 sub-agent 确认：当前 keyword 搜索只取一页，默认每 query 3 条。
+- DB 侧 sub-agent 确认：`has_more / next_cursor` 不新增列，先进入 `raw_payload` 或平台归一结果。
+
+## 修改范围
+
+后续执行 P3B 时允许：
+
+- 修改 `content_agent/integrations/douyin.py` 的 keyword response 校验和字段归一。
+- 增加 business code 校验，避免 HTTP 200 但业务失败被当作成功。
+- 增加 `has_more / next_cursor` 保留测试。
+- 增加 Crawapi 业务失败归因测试。
+- 必要时更新 `.env.example` 的非敏感 key 名。
+
+## 不修改范围
+
+- 不实现第二页请求。
+- 不实现游走策略触发分页。
+- 不接作者作品 `/crawler/dou_yin/blogger`。
+- 不接 TikHub fallback。
+- 不接 AIGC plan。
+- 不接 OSS。
+- 不把完整 Crawapi 原始响应无节制写入运行文件。
+
+## 涉及文件 / 函数 / 类
+
+文件：
+
+- `content_agent/integrations/douyin.py`
+- `content_agent/business_modules/platform_access.py`
+- `tests/test_douyin_client.py`
+- `product_documents/旧版和上下游理解/旧版ContentFindAgent外部数据源接口审计.md`
+- `product_documents/抖音游走策略/抖音接口真实case样例.json`
+- `tech_documents/数据接口与来源/external_data_sources_registry.json`
+
+函数 / 类：
+
+- `CrawapiDouyinClient.search()`
+- `CrawapiDouyinClient._post_json()`
+- `CrawapiDouyinClient._normalize_content_item()`
+- `_extract_tags()`
+- `_content_format()`
+- `_optional_positive_int()`
+
+## 数据合同
+
+请求 payload：
+
+- `keyword`：来自 P2 的 `search_query`。
+- `content_type`：默认来自 env，当前可用默认值是视频。
+- `sort_type`：默认来自 env。
+- `publish_time`：默认来自 env。
+- `cursor`：初始 cursor，当前默认第一页。
+- `account_id`：Crawapi 所需账号态引用。
+
+响应归一字段：
+
+- `content_discovery_id`
+- `search_query_id`
+- `platform="douyin"`
+- `platform_content_id`
+- `platform_content_format`
+- `description`
+- `platform_author_id`
+- `author_display_name`
+- `statistics`
+- `tags`
+- `text_extra`
+- `create_time`
+- `has_more`
+- `next_cursor`
+- `discovery_start_source`
+- `previous_discovery_step`
+- `content_metadata_source="douyin_keyword_search"`
+- `platform_auth_mode`
+- `platform_raw_payload`
+
+分页合同：
+
+- P3 初始只拉第一页。
+- `has_more / next_cursor` 必须保留。
+- P3 不根据 `has_more` 自动翻页。
+- 后续是否拉第二页由游走策略按第一页效果和预算触发。
+
+错误合同：
+
+- HTTP 错误：query 级失败。
+- network error：query 级失败。
+- bad JSON：query 级失败。
+- HTTP 200 但 business code 非成功：query 级失败，归因 Crawapi 登录态、账号态或配置。
+- 错误消息不得包含 secret、token、完整请求头或真实账号敏感信息。
+
+## 实施步骤
+
+1. 读取旧版接口合同，确认当前 payload 与 env 默认值一致。
+2. 加严 `_post_json()` 或 `search()` 的业务响应校验：HTTP 200 但业务失败不能当作空结果成功。
+3. 保持每 query 只调用一次 keyword endpoint。
+4. 保持 `max_results_per_query` 从 env 读取。
+5. 确认每条归一结果都保留 `has_more / next_cursor`。
+6. 确认 `platform_raw_payload` 只保留最小必要字段。
+7. 增加测试：成功、空结果、HTTP 错误、业务错误、分页字段保留、每 query 限量。
+
+## 验证命令
+
+```bash
+uv run pytest tests/test_douyin_client.py -q
+rg -n "has_more|next_cursor|business|keyword_search" content_agent/integrations/douyin.py tests/test_douyin_client.py
+```
+
+真实 Crawapi smoke 只允许显式手动执行，失败归因为登录态、账号态或配置，不阻塞默认测试。
+
+## 失败归因
+
+- 返回数量不符合 env：`max_results_per_query` 解析或 slice 问题。
+- `has_more / next_cursor` 丢失：归一字段遗漏。
+- business code 失败被当作空结果：Crawapi response 校验不足。
+- 真实联调失败：优先归因 Crawapi 登录态、账号态或配置，不改 P3 产品规则。
+- 默认测试访问真实 Crawapi：fake HTTP 注入失败。
+
+## sub-agent 交叉验证要点
+
+- 代码侧确认：keyword 搜索仍只拉第一页，不循环翻页。
+- 代码侧确认：业务失败不被当成成功空结果。
+- 文档 / JSON 侧确认：old Crawapi 合同字段与实现字段对齐。
+- DB 侧确认：分页状态不新增 DB 字段。

tech_documents/工程落地/implementation_briefs/P3/P3C_QueryFailure_And_PartialSuccess.md

@@ -0,0 +1,135 @@
+# P3C Query Failure And PartialSuccess Implementation Brief
+
+状态：本 brief 覆盖 keyword 单 query 失败不中断，以及 run 级 `partial_success`。这是 P3 和现有代码差距最大的状态边界。
+
+## 目标
+
+- 单条 query keyword 拉取失败时，只标记该 query 失败。
+- 其他 query 继续执行。
+- 至少一个 query 成功并产出候选，且后续链路跑完时，run 记 `partial_success`。
+- 所有 query 都失败或无可用候选时，run 才进入失败 / blocked 归因。
+- query 级状态和 run 级状态解耦，避免单点接口波动拖垮整次 run。
+
+## 现有证据
+
+- `tech_documents/工程落地/04_V1阶段开发计划.md`：已拍板 keyword 拉取失败时将该 query 标记为失败并继续其他 query；部分 query 失败但仍有结果跑完时 run 记 `partial_success`。
+- `content_agent/business_modules/platform_access.py`：当前 `platform_client.search()` 异常会直接冒泡，无法继续其他 query。
+- `content_agent/run_service.py`：当前外层捕获异常后整次 run 记 failed；`_record_success_metadata()` 只写 `success`。
+- `content_agent/models.py`：当前 `RunStatus` 没有 `partial_success`。
+- `content_agent/business_modules/run_record/recorder.py`：当前 `search_clues` 根据 discovered items 和 rule decisions 推导 `success / weak_effective / failed`，尚未接收平台 query 级失败信息。
+- `sql/content_agent_schema.sql`：`content_agent_queries.search_query_effect_status`、`content_agent_search_clues.search_query_effect_status`、`content_agent_run_events.status/error_code/raw_payload` 可承载 query 失败；`content_agent_runs.status` 可承载 `partial_success`。
+
+sub-agent 交叉验证结论：
+
+- 代码侧 sub-agent 确认：当前平台异常会导致整次 run failed，不是 query 级失败。
+- DB 侧 sub-agent 确认：`partial_success` 可直接落 `content_agent_runs.status`，不需 DDL。
+- 文档 / JSON sub-agent 确认：query / run 状态解耦是已拍板 P3 事实。
+
+## 修改范围
+
+后续执行 P3C 时允许：
+
+- 修改 `content_agent/business_modules/platform_access.py`，让 `run()` 捕获单 query 平台错误并继续。
+- 修改平台归一输出结构，携带 query 级失败摘要或 status。
+- 修改 `content_agent/models.py`，把 `partial_success` 加入 `RunStatus`。
+- 修改 `content_agent/run_service.py`，根据运行结果写 `success` 或 `partial_success`。
+- 修改 `content_agent/business_modules/run_record/recorder.py`，让 `search_clues` 能体现 query 级失败。
+- 增加平台失败相关 error code 或复用现有结构化错误合同。
+- 增加测试覆盖 partial success。
+
+## 不修改范围
+
+- 不让单 query 失败中断所有 query。
+- 不把 query 失败写成 P2 query generation 失败。
+- 不新增 DB 表或字段。
+- 不把真实 Crawapi 错误原文、账号态或 secret 写入 raw payload。
+- 不改变成功 API 响应字段。
+- 不做重试 / backoff / rate limit 策略；P3C 只处理 query 级失败隔离。
+
+## 涉及文件 / 函数 / 类
+
+文件：
+
+- `content_agent/business_modules/platform_access.py`
+- `content_agent/run_service.py`
+- `content_agent/models.py`
+- `content_agent/business_modules/run_record/recorder.py`
+- `content_agent/errors.py`
+- `tests/test_platform_access.py`
+- 后续新增或扩展 `tests/test_run_service.py`
+- `tests/test_api.py`
+- `tests/test_v1_graph.py`
+
+函数 / 类：
+
+- `platform_access.run()`
+- `RunService.start_run()`
+- `RunService._record_success_metadata()`
+- `RunStatus`
+- `run_record.run()`
+- `_build_search_clues()`
+- `ErrorCode`
+
+## 数据合同
+
+query 级状态：
+
+- 成功并有候选：`success` 或后续由规则结果推导。
+- 接口失败：`failed`。
+- 被平台配置或登录态阻断：`blocked`。
+- 空结果：`failed` 或 `weak_effective`，以后续 P3/P5 具体验收为准。
+
+run 级状态：
+
+- 全部关键链路成功且无 query failure：`success`。
+- 有 query failure，但至少一个 query 成功产出候选且后续链路跑完：`partial_success`。
+- 所有 query 失败或没有候选可继续：`failed` 或 `blocked`。
+
+记录位置：
+
+- query failure 摘要：`content_agent_run_events.raw_payload`。
+- query effect：`content_agent_queries.search_query_effect_status` 或 `content_agent_search_clues.search_query_effect_status`。
+- run 最终状态：`content_agent_runs.status`。
+- 本地兼容导出：`run_events.jsonl`、`search_clues.jsonl`。
+
+建议错误码：
+
+- 平台配置缺失：沿用或新增平台配置错误码。
+- 单 query 平台失败：建议新增或复用 `PLATFORM_QUERY_FAILED`。
+- 所有平台 query 失败：run 级失败可使用平台相关错误码。
+
+## 实施步骤
+
+1. 为 `platform_access.run()` 设计返回结构：既返回成功平台结果，也返回 query 失败摘要。
+2. 对每条 query 包裹 `platform_client.search()`。
+3. 单 query 异常时记录脱敏失败信息，继续下一条 query。
+4. 如果所有 query 都失败且无候选，抛 run 级结构化错误。
+5. 如果存在失败 query 且存在成功候选，把状态信号传到 graph / run_service。
+6. 更新 `RunStatus` 加入 `partial_success`。
+7. 修改 `_record_success_metadata()`，根据 state 中的 query failure 摘要写 `success` 或 `partial_success`。
+8. 确认 lifecycle success event 可标记 `partial_success`，但不写成 failed。
+9. 修改或扩展 `search_clues`，让失败 query 有可追踪记录。
+10. 增加测试：1 成功 1 失败 => run partial_success；所有 query 失败 => run failed；错误信息脱敏。
+
+## 验证命令
+
+```bash
+uv run pytest tests/test_platform_access.py -q
+uv run pytest tests/test_api.py tests/test_v1_graph.py -q
+rg -n "partial_success|PLATFORM_QUERY_FAILED|search_query_effect_status" content_agent tests
+```
+
+## 失败归因
+
+- 单 query 失败仍中断 run：`platform_access.run()` 没有捕获单 query 异常。
+- 部分失败仍写 `success`：RunService 没有读取 query failure 摘要。
+- 部分失败写成 `failed`：query 状态和 run 状态未解耦。
+- DB 写入失败：检查 `content_agent_runs.status`、`content_agent_run_events`、`search_clues` 的字段合同。
+- 错误 detail 泄漏平台账号态或 secret：错误清洗不合格。
+
+## sub-agent 交叉验证要点
+
+- 代码侧确认：单 query 失败会继续其他 query。
+- 代码侧确认：`RunStatus` 包含 `partial_success`，并能落 run metadata。
+- DB 侧确认：`partial_success` 不需要 DDL。
+- 文档 / JSON 侧确认：query 级失败不是 P2 query generation failure。

tech_documents/工程落地/implementation_briefs/P3/P3D_ContentPortrait_Retry_And_Normalization.md

@@ -0,0 +1,137 @@
+# P3D Content Portrait Retry And Normalization Implementation Brief
+
+状态：本 brief 覆盖热点宝内容点赞画像、portrait retry 和画像字段归一。P3 只提供规则包输入，不做最终业务判断。
+
+## 目标
+
+- 对每个 keyword 返回内容拉取内容点赞画像。
+- portrait 拉取失败时重试一次。
+- 重试后仍失败，平台结果标记 `portrait_available=false`。
+- 缺画像内容后续由规则包 reject，不由 P3 直接写最终决策。
+- 50+ 分级和互动量打分由规则包判断；平台接入层只保留归一字段。
+- 错误消息、运行文件和 `run_context` 不泄漏 secret。
+
+## 现有证据
+
+- `tech_documents/工程落地/04_V1阶段开发计划.md`：已拍板 portrait 失败先重试一次；重试后仍失败，则该内容进入 reject；50+ 分级和互动量打分由规则包判断。
+- `product_documents/旧版和上下游理解/旧版ContentFindAgent外部数据源接口审计.md`：内容点赞画像 endpoint 为 `POST /crawler/dou_yin/re_dian_bao/video_like_portrait`，入参 `content_id / need_age / need_gender / need_province`。
+- `tech_documents/数据接口与来源/external_data_sources_registry.json`：`PLT_DOUYIN_CONTENT_PROFILE` 为 `verified + v1_active`，输出年龄、性别、省份画像信号，不直接代表内容质量。
+- `product_documents/抖音游走策略/runtime_v1_records_schema.md`：缺画像样例最终在 `rule_decisions.jsonl` 中体现 `missing_content_portrait`。
+- `product_documents/规则包/douyin_rule_packs.v1.json`：`missing_content_portrait` 是规则包 gate / reason code，P3 不直接决定入池。
+- `content_agent/integrations/douyin.py`：当前 `_fetch_content_portrait()` 捕获 `RuntimeError` 后直接返回 `portrait_available=false`，没有 retry。
+- `content_agent/business_modules/content_discovery/content_discovery_builder.py`：`_build_content_audience_profile()` 在 `portrait_available=false` 时返回空画像，后续规则包可据此判断。
+- `tests/test_douyin_client.py`：已有 portrait HTTP error 标记缺画像、dimensions shape、41-50 不计 50+ 测试。
+
+sub-agent 交叉验证结论：
+
+- 代码侧 sub-agent 确认：当前 portrait 失败无 retry。
+- 文档 / JSON sub-agent 确认：缺画像 reject 属于规则包判断，不是 P3 最终拍板。
+- DB 侧 sub-agent 确认：retry 过程不新增字段，先放 `run_events.raw_payload`；最终缺画像可由 discovered item / rule decision 承载。
+
+## 修改范围
+
+后续执行 P3D 时允许：
+
+- 修改 `content_agent/integrations/douyin.py` 的 `_fetch_content_portrait()`。
+- 增加 retry 一次逻辑。
+- 增加业务响应校验与脱敏错误摘要。
+- 调整平台归一字段，确保画像可作为规则包输入。
+- 修改测试覆盖 retry 成功、retry 后失败。
+- 必要时调整 `content_discovery_builder` 对画像字段的透传，但不做规则判断。
+
+## 不修改范围
+
+- 不在 P3 直接写 `REJECT_CONTENT`。
+- 不在 P3 决定入池、待复看或淘汰。
+- 不把 50+ 分级当成 hard gate。
+- 不把互动量打分作为最终业务分数。
+- 不接账号画像兜底。
+- 不下载视频。
+- 不写 `content_agent_pattern_recall_evidence`。
+
+## 涉及文件 / 函数 / 类
+
+文件：
+
+- `content_agent/integrations/douyin.py`
+- `content_agent/business_modules/content_discovery/content_discovery_builder.py`
+- `content_agent/business_modules/rule_judgment/evaluator.py`
+- `tests/test_douyin_client.py`
+- `product_documents/规则包/douyin_rule_packs.v1.json`
+- `product_documents/抖音游走策略/runtime_v1_records_schema.md`
+
+函数 / 类：
+
+- `CrawapiDouyinClient.search()`
+- `CrawapiDouyinClient._fetch_content_portrait()`
+- `CrawapiDouyinClient._post_json()`
+- `_extract_portrait_dimensions()`
+- `_normalize_age_distribution()`
+- `_build_content_audience_profile()`
+- `rule_judgment.decide()`
+
+## 数据合同
+
+portrait 请求：
+
+- `content_id=platform_content_id`
+- `need_age=true`
+- `need_gender=true`
+- `need_province=true`
+- 其他维度默认 false，除非后续单独拍板。
+
+portrait 归一字段：
+
+- `portrait_available`
+- `age_distribution`
+- `age_50_plus_ratio`
+- `age_50_plus_tgi`
+- `age_50_plus_level`
+
+P3 到规则包输入：
+
+- `content_audience_profile.portrait_available`
+- `content_audience_profile.age_distribution`
+- `content_audience_profile.age_50_plus_ratio`
+- `content_audience_profile.age_50_plus_tgi`
+- `content_engagement_metrics.statistics`
+
+失败状态：
+
+- 第一次 portrait 失败：重试。
+- 第二次仍失败：`portrait_available=false`、`age_50_plus_level="missing"`。
+- 后续 reject：由规则包根据缺画像输入生成 `missing_content_portrait`。
+
+## 实施步骤
+
+1. 为 `_fetch_content_portrait()` 增加最多 2 次调用。
+2. 第一次失败只记录内部 retry 状态，不把内容立即丢弃。
+3. 第二次失败返回缺画像归一结果。
+4. 保护错误摘要，不写真实请求头、secret、token 或账号态敏感值。
+5. 确认成功 portrait 中年龄分布仍支持 dict 和 dimensions list 两种 shape。
+6. 确认 41-50 不计入 50+。
+7. 确认 P3 不把缺画像直接写成 rule decision。
+8. 增加测试：第一次失败第二次成功；两次失败；错误脱敏；画像字段进入 EvidenceBundle。
+
+## 验证命令
+
+```bash
+uv run pytest tests/test_douyin_client.py -q
+uv run pytest tests/test_runtime_files.py tests/test_v1_graph.py -q
+rg -n "missing_content_portrait|portrait_available|age_50_plus" content_agent tests product_documents/规则包
+```
+
+## 失败归因
+
+- portrait 没有 retry：`_fetch_content_portrait()` retry 控制缺失。
+- retry 后仍失败但内容被丢弃：P3 越过规则包边界。
+- 缺画像未触发后续 reject：规则包输入或 gate 连接问题。
+- 50+ / score 在平台接入层直接决定业务动作：越过 P3 边界。
+- 错误日志泄漏 secret：错误清洗不合格。
+
+## sub-agent 交叉验证要点
+
+- 代码侧确认：portrait 第一次失败会重试一次。
+- 代码侧确认：重试后失败只标记缺画像，不在 P3 直接 reject。
+- 文档 / JSON 侧确认：`missing_content_portrait` 属于规则包合同。
+- DB 侧确认：retry 信息不新增 DDL。

tech_documents/工程落地/implementation_briefs/P3/P3E_ContentDedup_And_QuerySourceMerge.md

@@ -0,0 +1,136 @@
+# P3E Content Dedup And Query Source Merge Implementation Brief
+
+状态：本 brief 覆盖同一 `platform_content_id` 被多个 query 命中时的去重和来源合并。P3 必须保留全部命中来源，不能只留下第一条 query。
+
+## 目标
+
+- 同一 `platform_content_id` 在同一 `run_id / policy_run_id / platform` 下只保留一条内容记录。
+- 记录所有命中过该内容的 `search_query_id / search_query / search_query_generation_method`。
+- 不把新发现内容写回上游 `source_post_id` 或 `matched_post_ids`。
+- 不新增 DDL。
+- 为后续 source path 和策略学习保留来源证据。
+
+## 现有证据
+
+- `tech_documents/工程落地/04_V1阶段开发计划.md`：已拍板同一 `platform_content_id` 被多个 query 命中时，合并全部 query 来源；新发现内容不能写入上游 `source_post_id` 或 `matched_post_ids`。
+- `content_agent/business_modules/platform_access.py`：当前使用 `seen_platform_content_ids`，重复内容直接 `continue`，导致后续 query 来源丢失。
+- `tests/test_platform_access.py`：当前测试断言重复内容只保留第一条 query。
+- `sql/content_agent_schema.sql`：`content_agent_discovered_content_items` 有唯一键 `run_id + policy_run_id + platform + platform_content_id`，支持内容去重。
+- `product_documents/抖音游走策略/runtime_v1_records_schema.md`：`source_path_records.jsonl` 可记录 `search_query_to_content` 路径。
+- `tech_documents/数据接口与来源/01_DemandAgent输入合同.md`：新找到的 `platform_content_id` 只能标记为 discovered content，不能声称它 exact 属于上游 Pattern 节点。
+- `content_agent/business_modules/content_discovery/source_evidence.py`：当前 source evidence 负责继承上游证据并追加发现态字段。
+
+sub-agent 交叉验证结论：
+
+- 代码侧 sub-agent 确认：当前重复内容会丢弃后续 query 来源。
+- DB 侧 sub-agent 确认：不新增表；多来源可先放 `raw_payload/source_evidence`，必要时由 `content_agent_source_path_records` 记录 query->content 路径。
+- 文档 / JSON sub-agent 确认：P3 不能污染上游 `source_post_id / matched_post_ids`。
+
+## 修改范围
+
+后续执行 P3E 时允许：
+
+- 修改 `content_agent/business_modules/platform_access.py` 的去重逻辑。
+- 修改平台结果结构，加入多 query 来源数组。
+- 修改 `content_agent/business_modules/content_discovery/content_discovery_builder.py`，透传多来源字段到 `raw_payload` 或 `source_evidence`。
+- 修改 `content_agent/business_modules/content_discovery/source_evidence.py`，确保新发现内容不覆盖上游字段。
+- 修改 `content_agent/business_modules/run_record/recorder.py`，必要时为多个 query 来源生成 source path basis。
+- 修改 `tests/test_platform_access.py` 和 source evidence 相关测试。
+
+## 不修改范围
+
+- 不新增 DB 表。
+- 不新增 `platform_content_id` 来源映射专用列。
+- 不把 `platform_content_id` 追加到 `matched_post_ids`。
+- 不改 DemandAgent 上游证据。
+- 不在 P3 判断该视频是否 Pattern matched。
+- 不实现 P4 Pattern 回扣。
+
+## 涉及文件 / 函数 / 类
+
+文件：
+
+- `content_agent/business_modules/platform_access.py`
+- `content_agent/business_modules/content_discovery/content_discovery_builder.py`
+- `content_agent/business_modules/content_discovery/source_evidence.py`
+- `content_agent/business_modules/run_record/recorder.py`
+- `tests/test_platform_access.py`
+- `tests/test_source_evidence.py`
+- `tests/test_runtime_files.py`
+
+函数 / 类：
+
+- `platform_access.run()`
+- `content_discovery.run()`
+- `_build_discovered_content_item()`
+- `_build_evidence_bundle()`
+- `build_source_evidence()`
+- `run_record.run()`
+- `_build_search_clues()`
+
+## 数据合同
+
+单内容保留字段：
+
+- `platform_content_id`
+- `platform`
+- `content_discovery_id`
+- 首个或稳定主 `search_query_id`
+- 首个或稳定主 `search_query`
+- `statistics`
+- `tags`
+- `platform_raw_payload`
+
+多来源字段建议进入 `raw_payload` 或 `source_evidence`：
+
+- `query_sources`
+- `matched_search_query_ids`
+- `matched_search_queries`
+- `matched_search_query_generation_methods`
+
+上游字段保护：
+
+- `source_post_id` 保持 DemandAgent evidence_pack 原值。
+- `matched_post_ids` 保持 DemandAgent evidence_pack 原值。
+- `decode_case_ids` 保持上游归因字段，不写入新发现内容 ID。
+- 新发现内容使用 `discovered_platform_content_id` 或 `platform_content_id` 表达。
+
+DB 承载：
+
+- `content_agent_discovered_content_items.raw_payload`
+- `content_agent_discovered_content_items.source_evidence`
+- `content_agent_source_path_records` 后续可记录多条 `search_query_to_content`。
+
+## 实施步骤
+
+1. 将 `platform_access.run()` 的 seen set 改成按 `platform_content_id` 聚合。
+2. 首次命中创建内容对象。
+3. 后续 query 命中同一内容时，不丢弃，而是追加 query 来源。
+4. 确定稳定主 query：默认使用首次命中作为主 `search_query_id`，全部来源进入数组。
+5. 确认 `content_discovery_id` 不因来源追加而改变。
+6. 将多 query 来源透传到 discovered content raw payload 或 source evidence。
+7. 确认 source evidence 不覆盖 `source_post_id / matched_post_ids`。
+8. 增加测试：两个 query 命中同一 `platform_content_id` 时只产出一条内容，但来源数组包含两条。
+9. 增加测试：多 query 来源可进入后续 source path 或 search clue 统计。
+
+## 验证命令
+
+```bash
+uv run pytest tests/test_platform_access.py tests/test_source_evidence.py tests/test_runtime_files.py -q
+rg -n "matched_search_query|query_sources|source_post_id|matched_post_ids|platform_content_id" content_agent tests
+```
+
+## 失败归因
+
+- 重复内容仍只保留第一条来源：聚合逻辑未实现。
+- 重复内容产生多条 DB item：去重主键合同未保持。
+- 新发现 ID 写入 `source_post_id`：血缘污染。
+- source path 缺多来源：P3E 与 run_record 交界未传递来源数组。
+- DB unique key 冲突：重复内容未在写入前合并。
+
+## sub-agent 交叉验证要点
+
+- 代码侧确认：重复 `platform_content_id` 不再丢弃后续 query 来源。
+- 代码侧确认：主内容记录仍只有一条。
+- 文档 / JSON 侧确认：新发现内容不能污染上游 Pattern 证据。
+- DB 侧确认：现有唯一键支撑去重，不新增 DDL。

tech_documents/工程落地/implementation_briefs/P3/P3F_P3_Acceptance_And_DriftGuards.md

@@ -0,0 +1,142 @@
+# P3F P3 Acceptance And Drift Guards Implementation Brief
+
+状态：本 brief 覆盖 P3 完整验收、集成测试和漂移防护。P3 完成后应能确认进入 P4 Pattern 回扣阶段。
+
+## 目标
+
+- 验收 P3 抖音平台接入符合已拍板规则。
+- 确认 mock 链路、P0/P1/P2 回归仍稳定。
+- 确认 DB schema ready 且无需 P3 DDL。
+- 确认 P3 不越界到 P4/P5/作者/tag/多平台。
+- 确认真实 Crawapi smoke 失败时有明确归因，不反复卡产品规则。
+
+## 现有证据
+
+- `tech_documents/工程落地/04_V1阶段开发计划.md`：P3 当前无剩余开工前拍板项。
+- `tech_documents/工程落地/05_阶段验收清单.md`：临时每 query 3 条、一页、抖音-only 不能写成长期策略。
+- `tests/test_douyin_client.py`：已有 P3 client 层 fake HTTP 测试基础。
+- `tests/test_platform_access.py`：已有平台接入模块测试基础。
+- `tests/test_api.py`、`tests/test_v1_graph.py`、`tests/test_runtime_files.py`、`tests/test_source_evidence.py`、`tests/test_p0d_p0g.py`、`tests/test_search_intent.py`：P0/P1/P2 回归基础。
+- `scripts/validate_schema_registry.py`：schema registry 校验入口。
+- `scripts/validate_content_agent_db.py`：真实 DB schema validator。
+- `content_agent/integrations/database_runtime.py`：P3 产物可通过现有 runtime 映射写入 DB。
+
+sub-agent 交叉验证结论：
+
+- 代码侧 sub-agent 确认：P3 应补 RunService/API partial success 测试。
+- 文档 / JSON sub-agent 确认：P3 不应混入 P4 Pattern 回扣或 P5 规则拍板。
+- DB 侧 sub-agent 确认：`schema_ready=true`，P3 不新增 DDL。
+
+## 修改范围
+
+后续执行 P3F 时允许：
+
+- 新增或更新 P3 相关测试。
+- 更新 P3 相关 drift guard 命令。
+- 更新 `.env.example` 非敏感 key 名。
+- 如 P3 实现引入新 error code，同步测试错误结构。
+- 更新 implementation brief 的实施后状态说明。
+
+## 不修改范围
+
+- 不新增生产 DB DDL。
+- 不把真实 secret 写进测试、文档或 runtime。
+- 不把真实 Crawapi smoke 设为默认测试。
+- 不接真实抖音之外的平台。
+- 不接作者作品、tag、账号画像。
+- 不让 P3 直接写 `matched` Pattern 回扣状态。
+
+## 涉及文件 / 函数 / 类
+
+文件：
+
+- `tests/test_douyin_client.py`
+- `tests/test_platform_access.py`
+- `tests/test_api.py`
+- `tests/test_v1_graph.py`
+- `tests/test_runtime_files.py`
+- `tests/test_source_evidence.py`
+- `tests/test_p0d_p0g.py`
+- `tests/test_search_intent.py`
+- `scripts/validate_schema_registry.py`
+- `scripts/validate_content_agent_db.py`
+- `content_agent/business_modules/platform_access.py`
+- `content_agent/integrations/douyin.py`
+- `content_agent/run_service.py`
+
+函数 / 类：
+
+- `CrawapiDouyinClient.search()`
+- `CrawapiDouyinClient._fetch_content_portrait()`
+- `platform_access.run()`
+- `RunService.start_run()`
+- `RunService._record_success_metadata()`
+- `run_record.run()`
+- `content_discovery.run()`
+
+## 数据合同
+
+P3 完成后必须满足：
+
+- P2 query 可被 P3 消费。
+- 每 query 拉取数量由 env 控制。
+- 每 query 初始只拉第一页。
+- `has_more / next_cursor` 保留。
+- keyword 单 query 失败不中断其他 query。
+- 部分成功 run 记 `partial_success`。
+- portrait 失败重试一次。
+- portrait 最终缺失由规则包 reject。
+- 同 `platform_content_id` 多 query 命中合并全部来源。
+- 真实待复看继续 `pattern_recall_pending`，不能写 `matched`。
+- P3 不新增 DDL。
+
+## 实施步骤
+
+1. 跑 P3 单元测试：Douyin client 和 platform access。
+2. 跑 RunService/API 级 partial success 测试。
+3. 跑 P0/P1/P2 回归测试。
+4. 跑 schema registry validator。
+5. 跑真实 DB schema validator。
+6. 跑 drift guard：业务模块无 DB 访问、P3 无多平台真实路由、P3 不写 Pattern matched、敏感赋值扫描无结果。
+7. 可选真实 Crawapi smoke：必须显式开启，失败只归因登录态、账号态或配置。
+8. 如果 DB runtime 开启，验证同一 `run_id / policy_run_id` 可在 P0/P2/P3 相关 runtime 表对齐。
+
+## 验证命令
+
+```bash
+find tech_documents/工程落地/implementation_briefs/P3 -maxdepth 1 -type f | sort
+rg -n "Implementation Brief|sub-agent|验证命令|失败归因" tech_documents/工程落地/implementation_briefs/P3
+rg -n "PASS[W]ORD=|TO[K]EN=|API[_]KEY=|SEC[R]ET=" tech_documents/工程落地/implementation_briefs/P3
+rg -n "partial_success|has_more|next_cursor|platform_content_id|portrait|Crawapi" tech_documents/工程落地/implementation_briefs/P3
+uv run python scripts/validate_schema_registry.py
+uv run --with pymysql python scripts/validate_content_agent_db.py --env-file .env
+uv run pytest tests/test_douyin_client.py tests/test_platform_access.py tests/test_api.py tests/test_v1_graph.py tests/test_runtime_files.py tests/test_source_evidence.py tests/test_p0d_p0g.py tests/test_search_intent.py -q
+uv run pytest -q
+```
+
+Drift guard：
+
+```bash
+rg -n "pymysql|sqlalchemy|sqlite|psycopg|SELECT |INSERT |UPDATE |DELETE " content_agent/business_modules
+rg -n "xiaohongshu|kuaishou|bilibili|小红书|快手|B站" content_agent tests tech_documents/工程落地/implementation_briefs/P3
+rg -n "pattern_recall.*matched|category_or_element_binding.*matched" content_agent/business_modules/platform_access.py content_agent/integrations/douyin.py
+```
+
+真实 Crawapi smoke 只允许显式开启，不作为默认 CI / 回归测试。
+
+## 失败归因
+
+- P3 brief 文件缺失：文档落地问题。
+- 敏感赋值扫描命中：文档或测试泄漏配置，应改为只写 key 名。
+- DB validator 失败：DB 目标库、权限或 schema drift。
+- P3 单元测试失败：Crawapi 合同、portrait retry、query failure 或来源合并实现问题。
+- P0/P1/P2 回归失败：P3 改动误伤上游 / 存储 / query 模块。
+- 真实 Crawapi smoke 失败：优先归因登录态、账号态或配置，不改变 P3 产品规则。
+- drift guard 命中多平台或 matched：P3 越界。
+
+## sub-agent 交叉验证要点
+
+- 代码侧 sub-agent 确认：P3 测试覆盖 query 失败继续、partial success、portrait retry、来源合并。
+- 文档 / JSON 侧 sub-agent 确认：P3 仍只做抖音 keyword + 内容画像。
+- DB 侧 sub-agent 确认：P3 产物通过现有 runtime / DB 字段承载，无 DDL。
+- 测试侧确认：默认测试不访问真实 Crawapi。

tests/test_api.py

@@ -53,8 +53,8 @@ def test_api_runs_and_queries_mock_chain(tmp_path, monkeypatch):
 def test_api_defaults_to_real_platform_mode_but_can_select_mock(tmp_path, monkeypatch):
     selected_modes = []
 
-    def fake_platform_client(self, platform_mode):
-        selected_modes.append(platform_mode)
+    def fake_platform_client(self, platform, platform_mode):
+        selected_modes.append((platform, platform_mode))
         return MockPlatformClient()
 
     monkeypatch.setattr(RunService, "_platform_client", fake_platform_client)
@@ -76,7 +76,46 @@ def test_api_defaults_to_real_platform_mode_but_can_select_mock(tmp_path, monkey
     assert default_response.json()["platform_mode"] == "real"
     assert mock_response.status_code == 200
     assert mock_response.json()["platform_mode"] == "mock"
-    assert selected_modes == ["real", "mock"]
+    assert selected_modes == [("douyin", "real"), ("douyin", "mock")]
+
+
+def test_api_rejects_non_douyin_real_platform(tmp_path, monkeypatch):
+    monkeypatch.setattr(
+        api,
+        "service",
+        RunService(
+            runtime_root=tmp_path / "runtime" / "v1",
+            demand_source=FakeDemandSource(),
+            query_variant_client=FakeQueryVariantClient(),
+        ),
+    )
+    client = TestClient(api.app)
+
+    response = client.post(
+        "/runs", json={"platform": "other_platform", "platform_mode": "real"}
+    )
+
+    assert response.status_code == 400
+    assert response.json()["detail"]["error_code"] == "INVALID_REQUEST"
+
+
+def test_api_returns_partial_success_as_successful_response(tmp_path, monkeypatch):
+    service = RunService(
+        runtime_root=tmp_path / "runtime" / "v1",
+        demand_source=FakeDemandSource(),
+        query_variant_client=FakeQueryVariantClient(),
+    )
+    service._platform_client = lambda platform, platform_mode: _PartialFailurePlatformClient()
+    monkeypatch.setattr(api, "service", service)
+    client = TestClient(api.app)
+
+    response = client.post("/runs", json={"platform": "douyin", "platform_mode": "real"})
+
+    assert response.status_code == 200
+    payload = response.json()
+    assert payload["status"] == "partial_success"
+    summary = client.get(f"/runs/{payload['run_id']}").json()
+    assert summary["status"] == "partial_success"
 
 
 def test_api_rejects_legacy_run_identifier_field(tmp_path, monkeypatch):
@@ -90,3 +129,13 @@ def test_api_rejects_legacy_run_identifier_field(tmp_path, monkeypatch):
     )
 
     assert response.status_code == 422
+
+
+class _PartialFailurePlatformClient:
+    def __init__(self) -> None:
+        self.mock = MockPlatformClient()
+
+    def search(self, search_query: dict) -> list[dict]:
+        if search_query["search_query_id"] == "q_002":
+            raise RuntimeError("temporary platform failure")
+        return self.mock.search(search_query)

tests/test_douyin_client.py

@@ -101,6 +101,8 @@ def test_douyin_keyword_search_maps_content_and_portrait_fields():
     assert result["platform_content_id"] == "7615247738577423622"
     assert result["platform_author_id"] == "MS4wLjABAAAAdYc"
     assert result["tags"] == ["#早上好"]
+    assert result["has_more"] is True
+    assert result["next_cursor"] == "10"
     assert result["portrait_available"] is True
     assert result["age_50_plus_level"] == "strong"
     assert result["pattern_recall"] == "pattern_recall_pending"
@@ -108,6 +110,7 @@ def test_douyin_keyword_search_maps_content_and_portrait_fields():
     assert result["platform_auth_mode"] == "no_bearer"
     assert result["platform_raw_payload"][RAW_CONTENT_ID_KEY] == "7615247738577423622"
     assert client.http_client.requests[0]["json"][RAW_AUTHOR_ACCOUNT_KEY] == "771431222"
+    assert len(client.http_client.requests) == 2
 
 
 def test_douyin_keyword_search_returns_empty_list():
@@ -125,7 +128,41 @@ def test_douyin_keyword_search_http_error_is_sanitized():
         client.search(_search_query("接口失败"))
 
 
-def test_douyin_portrait_http_error_marks_content_missing_portrait():
+def test_douyin_keyword_search_business_error_is_failed():
+    client = _client([_response(200, {"code": 22001, "msg": "强制登录", "data": None})])
+
+    with pytest.raises(RuntimeError, match="keyword_search failed: business_error"):
+        client.search(_search_query("账号态失败"))
+
+
+def test_douyin_keyword_search_network_error_is_sanitized():
+    client = _client([httpx.ConnectError("network failed")])
+
+    with pytest.raises(RuntimeError, match="keyword_search failed: network_error"):
+        client.search(_search_query("网络失败"))
+
+
+def test_douyin_keyword_search_bad_json_is_sanitized():
+    client = CrawapiDouyinClient(
+        base_url="http://crawapi.test",
+        keyword_path="/crawler/dou_yin/keyword",
+        content_portrait_path="/crawler/dou_yin/re_dian_bao/video_like_portrait",
+        http_client=FakeHttpClient(
+            [
+                httpx.Response(
+                    200,
+                    content=b"not json",
+                    request=httpx.Request("POST", "http://crawapi.test/endpoint"),
+                )
+            ]
+        ),
+    )
+
+    with pytest.raises(RuntimeError, match="keyword_search failed: bad_json"):
+        client.search(_search_query("坏 JSON"))
+
+
+def test_douyin_portrait_http_error_retries_before_missing():
     client = _client(
         [
             _response(
@@ -147,6 +184,7 @@ def test_douyin_portrait_http_error_marks_content_missing_portrait():
                 },
             ),
             _response(500, {"error": "portrait failed"}),
+            _response(500, {"error": "portrait failed again"}),
         ]
     )
 
@@ -154,6 +192,53 @@ def test_douyin_portrait_http_error_marks_content_missing_portrait():
 
     assert result["portrait_available"] is False
     assert result["age_50_plus_level"] == "missing"
+    assert len(client.http_client.requests) == 3
+
+
+def test_douyin_portrait_retry_can_recover():
+    client = _client(
+        [
+            _response(
+                200,
+                {
+                    "data": {
+                        "data": [
+                            {
+                                RAW_CONTENT_ID_KEY: "7615247738577423622",
+                                "desc": "早上好",
+                                "author": {
+                                    "nickname": "作者",
+                                    RAW_AUTHOR_ID_KEY: "MS4wLjABAAAA001",
+                                },
+                                "statistics": {"digg_count": 1},
+                            }
+                        ],
+                        "has_more": True,
+                        "next_cursor": "10",
+                    }
+                },
+            ),
+            _response(500, {"error": "portrait failed"}),
+            _response(
+                200,
+                {
+                    "data": {
+                        "data": {
+                            "年龄": {
+                                "50+": {"percentage": "18.00%", "preference": "135.0"}
+                            }
+                        }
+                    }
+                },
+            ),
+        ]
+    )
+
+    result = client.search(_search_query("早上好"))[0]
+
+    assert result["portrait_available"] is True
+    assert result["age_50_plus_level"] == "strong"
+    assert len(client.http_client.requests) == 3
 
 
 def test_douyin_keyword_search_can_limit_results_per_query():

tests/test_p0d_p0g.py

@@ -11,6 +11,7 @@ from content_agent.errors import ErrorCode
 from content_agent.integrations.composite_runtime import CompositeRuntimeStore
 from content_agent.integrations.database_runtime import ContentSupplyDbConfig
 from content_agent.integrations.demand_source import DemandSourceService
+from content_agent.integrations.mock_platform import MockPlatformClient
 from content_agent.integrations.runtime_files import LocalRuntimeFileStore
 from content_agent.run_service import RunService
 from content_agent.schemas import RunStartRequest
@@ -83,6 +84,34 @@ def test_run_service_success_records_run_policy_and_lifecycle_events(tmp_path):
     assert runtime.lifecycle_events[0]["raw_payload"]["source_ref"]["demand_content_id"] == 123
 
 
+def test_run_service_partial_platform_failure_records_partial_success(tmp_path):
+    runtime = _SpyRuntimeStore(tmp_path / "runtime")
+    demand_source = FakeDemandSource(real_source_payload(demand_content_id=123))
+    service = RunService(
+        runtime=runtime,
+        demand_source=demand_source,
+        query_variant_client=FakeQueryVariantClient(),
+    )
+    service._platform_client = lambda platform, platform_mode: _PartialFailurePlatformClient()
+
+    state = service.start_run(RunStartRequest(platform_mode="real"))
+
+    assert state["status"] == "partial_success"
+    assert state["query_failures"][0]["search_query_id"] == "q_002"
+    assert runtime.run_updates[-1]["updates"]["status"] == "partial_success"
+    assert runtime.policy_runs[0]["status"] == "partial_success"
+    assert runtime.lifecycle_events[-1]["event_id"] == "lifecycle_success"
+    assert runtime.lifecycle_events[-1]["status"] == "partial_success"
+    assert runtime.lifecycle_events[-1]["raw_payload"]["query_failures"]
+    run_events = service.read_jsonl(state["run_id"], "run_events.jsonl")
+    assert any(event["event_type"] == "platform_query_failed" for event in run_events)
+    search_clues = service.read_jsonl(state["run_id"], "search_clues.jsonl")
+    failed_clue = next(
+        clue for clue in search_clues if clue["search_query_id"] == "q_002"
+    )
+    assert failed_clue["search_query_effect_status"] == "failed"
+
+
 def test_run_service_query_generation_failure_records_error_code(tmp_path):
     runtime = _SpyRuntimeStore(tmp_path / "runtime")
     demand_source = FakeDemandSource(real_source_payload(demand_content_id=123))
@@ -372,6 +401,16 @@ class _MalformedQueryVariantClient:
         return None
 
 
+class _PartialFailurePlatformClient:
+    def __init__(self) -> None:
+        self.mock = MockPlatformClient()
+
+    def search(self, search_query: dict[str, Any]) -> list[dict[str, Any]]:
+        if search_query["search_query_id"] == "q_002":
+            raise RuntimeError("temporary platform failure")
+        return self.mock.search(search_query)
+
+
 class _DemandCursor:
     def __init__(self, connection: "_DemandConnection") -> None:
         self.connection = connection

tests/test_platform_access.py

@@ -1,4 +1,5 @@
 from content_agent.business_modules import platform_access
+from content_agent.errors import ContentAgentError, ErrorCode
 
 
 class DuplicateContentClient:
@@ -27,11 +28,14 @@ def test_platform_access_deduplicates_same_content_across_search_queries():
         },
     ]
 
-    results = platform_access.run(search_queries, DuplicateContentClient())
+    result = platform_access.run(search_queries, DuplicateContentClient())
+    results = result["platform_results"]
 
     assert len(results) == 1
     assert results[0]["search_query_id"] == "q_001"
     assert results[0]["search_query"] == "对比分析"
+    assert results[0]["matched_search_query_ids"] == ["q_001", "q_002"]
+    assert results[0]["matched_search_queries"] == ["对比分析", "对比分析 人物故事"]
 
 
 class UniqueContentClient:
@@ -61,7 +65,7 @@ def test_platform_access_preserves_llm_variant_generation_method():
         },
     ]
 
-    results = platform_access.run(search_queries, UniqueContentClient())
+    results = platform_access.run(search_queries, UniqueContentClient())["platform_results"]
 
     assert [result["search_query_generation_method"] for result in results] == [
         "item_single",
@@ -69,3 +73,63 @@ def test_platform_access_preserves_llm_variant_generation_method():
     ]
     assert results[1]["search_query_id"] == "q_002"
     assert results[1]["search_query"] == "人物叙事素材"
+
+
+class FailingThenSuccessfulClient:
+    def search(self, search_query):
+        if search_query["search_query_id"] == "q_001":
+            raise RuntimeError("temporary platform failure")
+        return [
+            {
+                "content_discovery_id": f"{search_query['search_query_id']}_content_001",
+                "search_query_id": search_query["search_query_id"],
+                "platform_content_id": "7601814454925298995",
+                "description": "成功内容",
+            }
+        ]
+
+
+def test_platform_access_keeps_running_after_single_query_failure():
+    search_queries = [
+        {
+            "search_query_id": "q_001",
+            "search_query": "接口失败",
+            "search_query_generation_method": "item_single",
+        },
+        {
+            "search_query_id": "q_002",
+            "search_query": "正常召回",
+            "search_query_generation_method": "llm_variant",
+        },
+    ]
+
+    result = platform_access.run(search_queries, FailingThenSuccessfulClient())
+
+    assert [item["platform_content_id"] for item in result["platform_results"]] == [
+        "7601814454925298995"
+    ]
+    assert result["query_failures"][0]["search_query_id"] == "q_001"
+    assert result["query_failures"][0]["error_code"] == ErrorCode.PLATFORM_REQUEST_FAILED.value
+
+
+class AlwaysFailingClient:
+    def search(self, search_query):
+        raise RuntimeError("platform unavailable")
+
+
+def test_platform_access_fails_run_when_all_queries_fail():
+    search_queries = [
+        {
+            "search_query_id": "q_001",
+            "search_query": "接口失败",
+            "search_query_generation_method": "item_single",
+        }
+    ]
+
+    try:
+        platform_access.run(search_queries, AlwaysFailingClient())
+    except ContentAgentError as exc:
+        assert exc.error_code == ErrorCode.PLATFORM_REQUEST_FAILED
+        assert exc.detail["query_failures"][0]["search_query_id"] == "q_001"
+    else:
+        raise AssertionError("expected platform request failure")

tests/test_source_evidence.py

@@ -54,3 +54,47 @@ def test_source_evidence_inherits_evidence_pack_without_rewriting_origin(tmp_pat
     assert {
         record["decision_id"] for record in final_output["decision_records"]
     } == {"d_001", "d_002", "d_003"}
+
+
+def test_source_evidence_tracks_multiple_query_sources_without_polluting_origin(tmp_path):
+    service = RunService(
+        runtime_root=tmp_path / "runtime" / "v1",
+        query_variant_client=FakeQueryVariantClient(),
+    )
+    state = service.start_run(
+        RunStartRequest(platform_mode="mock", source=str(REAL_SOURCE_FIXTURE))
+    )
+    run_id = state["run_id"]
+
+    items = service.read_jsonl(run_id, "discovered_content_items.jsonl")
+    multi_source_item = next(
+        item for item in items if len(item.get("query_sources", [])) > 1
+    )
+    decisions = service.read_jsonl(run_id, "rule_decisions.jsonl")
+    decision = next(
+        row
+        for row in decisions
+        if row["decision_target_id"] == multi_source_item["platform_content_id"]
+    )
+    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 source_evidence["discovered_platform_content_id"] != source_evidence["source_post_id"]
+    assert (
+        source_evidence["discovered_platform_content_id"]
+        not in source_evidence["matched_post_ids"]
+    )
+
+    paths = service.read_jsonl(run_id, "source_path_records.jsonl")
+    query_content_paths = [
+        path
+        for path in paths
+        if path["source_path_type"] == "search_query_to_content"
+        and path["to_node_id"] == multi_source_item["platform_content_id"]
+    ]
+    assert {path["from_node_id"] for path in query_content_paths} == {
+        "q_002",
+        "q_003",
+        "q_004",
+    }