Improve V4 search hydration and technical diagnostics

content_agent/business_modules/progressive_screening.py

@@ -12,7 +12,7 @@ from content_agent.interfaces import GeminiVideoClient, PlatformSearchClient, Ru
 
 INITIAL_BATCH_SIZE = 3
 MAX_PAGES = 3
-SEARCH_MIN_INTERVAL_SECONDS = 15.0
+SEARCH_MIN_INTERVAL_SECONDS = 30.0
 PASS_ACTION = "ADD_TO_CONTENT_POOL"
 
 
@@ -191,7 +191,9 @@ class _ProgressiveContext:
         try:
             if self.limiter is not None:
                 self.limiter.wait()
-            search = getattr(self.platform_client, "search_full_page", None)
+            search = getattr(self.platform_client, "search_full_page_metadata", None)
+            if not callable(search):
+                search = getattr(self.platform_client, "search_full_page", None)
             if not callable(search):
                 search = self.platform_client.search
             return list(search(page_query))
@@ -217,6 +219,13 @@ class _ProgressiveContext:
     ) -> bool:
         batch = []
         for rank, result in enumerate(page_results, start=1):
+            key = _content_key(result)
+            if key and key in self._platform_result_by_key:
+                existing = self._platform_result_by_key[key]
+                platform_access._append_query_source(existing, query)  # noqa: SLF001
+                self._propagate_query_source_merge(key, existing)
+                continue
+            result = self._hydrate_result_media(result, query)
             prepared = self._prepare_result(
                 query,
                 result,
@@ -229,11 +238,6 @@ class _ProgressiveContext:
             if not key:
                 batch.append(prepared)
                 continue
-            existing = self._platform_result_by_key.get(key)
-            if existing:
-                platform_access._append_query_source(existing, query)  # noqa: SLF001
-                self._propagate_query_source_merge(key, existing)
-                continue
             self._platform_result_by_key[key] = prepared
             batch.append(prepared)
         if not batch:
@@ -282,7 +286,8 @@ class _ProgressiveContext:
         batch_kind: str,
         rank: int,
     ) -> dict[str, Any]:
-        prepared = platform_access._with_query_source(dict(result), query)  # noqa: SLF001
+        runtime_result = _strip_transient_fields(result)
+        prepared = platform_access._with_query_source(runtime_result, query)  # noqa: SLF001
         original_has_more = bool(prepared.get("has_more"))
         original_next_cursor = str(prepared.get("next_cursor") or "")
         prepared.update(
@@ -302,6 +307,16 @@ class _ProgressiveContext:
         )
         return prepared
 
+    def _hydrate_result_media(
+        self,
+        result: dict[str, Any],
+        query: dict[str, Any],
+    ) -> dict[str, Any]:
+        hydrate = getattr(self.platform_client, "hydrate_search_result_media", None)
+        if not callable(hydrate):
+            return result
+        return dict(hydrate(result, query))
+
     def _accumulate(
         self,
         batch: list[dict[str, Any]],
@@ -418,6 +433,10 @@ def _content_key(result: dict[str, Any]) -> tuple[str, str] | None:
     return str(result.get("platform") or ""), content_id
 
 
+def _strip_transient_fields(result: dict[str, Any]) -> dict[str, Any]:
+    return {key: value for key, value in result.items() if not str(key).startswith("_platform_")}
+
+
 def _content_discovery_id(search_query_id: str, page_number: int, rank: int) -> str:
     if page_number == 1:
         return f"{search_query_id}_content_{rank:03d}"

content_agent/flow_ledger_service.py

@@ -572,10 +572,21 @@ class FlowLedgerService:
         action_counts: Counter[str],
         gemini_counts: Counter[str],
     ) -> dict[str, Any]:
+        query_failures = _query_failures(bundle["queries"])
         return {
             "first_round_query_count": len(rows),
             "extension_query_count": len(extension_queries),
             "all_query_count": len(bundle["queries"]),
+            "query_failure_count": len(query_failures),
+            "query_failure_examples": [
+                {
+                    "search_query_id": _text(query.get("search_query_id")),
+                    "search_query": _text(query.get("search_query")),
+                    "message": _text(_query_failure(query).get("message")),
+                    "error_code": _text(_query_failure(query).get("error_code")),
+                }
+                for query in query_failures[:5]
+            ],
             "content_count": len(bundle["content_items"]),
             "oss_uploaded_count": media_counts["oss_uploaded"],
             "oss_pending_count": media_counts["oss_upload_pending"],
@@ -593,6 +604,18 @@ def _query_method(query: dict[str, Any]) -> str:
     return _text(query.get("search_query_generation_method"), "unknown")
 
 
+def _query_failure(query: dict[str, Any]) -> dict[str, Any]:
+    failure = query.get("query_failure")
+    if isinstance(failure, dict):
+        return failure
+    raw_failure = _record(query.get("raw_payload")).get("query_failure")
+    return raw_failure if isinstance(raw_failure, dict) else {}
+
+
+def _query_failures(queries: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    return [query for query in queries if _query_failure(query)]
+
+
 def _content_belongs_to_query(content: dict[str, Any], query_id: str) -> bool:
     if not query_id:
         return False

content_agent/integrations/crawapi_http.py

@@ -30,7 +30,7 @@ class CrawapiTransientError(RuntimeError):
 class RateLimiter:
     def __init__(
         self,
-        min_interval_seconds: float = 12.0,
+        min_interval_seconds: float = 30.0,
         now_fn: Callable[[], float] = time.monotonic,
         sleep_fn: Callable[[float], None] = time.sleep,
     ) -> None:

content_agent/integrations/kuaishou.py

@@ -59,6 +59,10 @@ def _publish_time_seconds(item: dict[str, Any]) -> int | None:
     return int(publish_ms) // 1000
 
 
+def _strip_transient_fields(result: dict[str, Any]) -> dict[str, Any]:
+    return {key: value for key, value in result.items() if not str(key).startswith("_platform_")}
+
+
 def _normalize_kuaishou_item(
     query: dict[str, Any],
     item: dict[str, Any],
@@ -164,7 +168,7 @@ class CrawapiKuaishouClient:
             max_results_per_query=_optional_positive_int(
                 _env("CONTENTFIND_KUAISHOU_MAX_RESULTS_PER_QUERY", env, default="5")
             ),
-            rate_limiter=RateLimiter(min_interval_seconds=15.0),
+            rate_limiter=RateLimiter(min_interval_seconds=30.0),
         )
 
     def search(self, query: dict[str, Any]) -> list[dict[str, Any]]:
@@ -173,10 +177,43 @@ class CrawapiKuaishouClient:
     def search_full_page(self, query: dict[str, Any]) -> list[dict[str, Any]]:
         return self._search(query, None)
 
+    def search_full_page_metadata(self, query: dict[str, Any]) -> list[dict[str, Any]]:
+        return self._search(query, None, hydrate_video=False)
+
+    def hydrate_search_result_media(
+        self,
+        result: dict[str, Any],
+        query: dict[str, Any],
+    ) -> dict[str, Any]:
+        item = result.get("_platform_search_item")
+        if not isinstance(item, dict):
+            return _strip_transient_fields(result)
+        selection = self._select_video_url_with_detail_fallback(item)
+        hydrated = _normalize_kuaishou_item(
+            query,
+            item,
+            1,
+            bool(result.get("has_more")),
+            str(result.get("next_cursor") or ""),
+            selection,
+        )
+        merged = {
+            **_strip_transient_fields(result),
+            "play_url": hydrated.get("play_url"),
+            "platform_raw_payload": hydrated.get("platform_raw_payload", {}),
+        }
+        if selection.get("media_failure_reason"):
+            merged["media_failure_reason"] = selection["media_failure_reason"]
+        else:
+            merged.pop("media_failure_reason", None)
+        return merged
+
     def _search(
         self,
         query: dict[str, Any],
         max_results_per_query: int | None,
+        *,
+        hydrate_video: bool = True,
     ) -> list[dict[str, Any]]:
         data = self._post_json(
             self.keyword_path,
@@ -191,16 +228,20 @@ class CrawapiKuaishouClient:
         selected = items[:max_results_per_query] if max_results_per_query else items
         results = []
         for index, item in enumerate(selected, start=1):
-            results.append(
-                _normalize_kuaishou_item(
-                    query,
-                    item,
-                    index,
-                    has_more,
-                    next_cursor,
-                    self._select_video_url_with_detail_fallback(item),
-                )
+            selection = (
+                self._select_video_url_with_detail_fallback(item) if hydrate_video else None
+            )
+            normalized = _normalize_kuaishou_item(
+                query,
+                item,
+                index,
+                has_more,
+                next_cursor,
+                selection,
             )
+            if not hydrate_video:
+                normalized["_platform_search_item"] = item
+            results.append(normalized)
         return results
 
     def fetch_detail(self, content_id: str) -> dict[str, Any]:
@@ -263,7 +304,9 @@ class CrawapiKuaishouClient:
         payload.update(
             {
                 "kuaishou_detail_fallback_attempted": True,
-                "kuaishou_detail_fallback_status": "used" if detail_selection.get("play_url") else "no_valid_play_url",
+                "kuaishou_detail_fallback_status": (
+                    "used" if detail_selection.get("play_url") else "no_valid_play_url"
+                ),
             }
         )
         return {**detail_selection, "platform_raw_payload": payload}

content_agent/integrations/shipinhao.py

@@ -117,6 +117,10 @@ def _research_keyword(item: dict[str, Any], query: dict[str, Any]) -> str:
     return str(query.get("search_query") or "")[:80]
 
 
+def _strip_transient_fields(result: dict[str, Any]) -> dict[str, Any]:
+    return {key: value for key, value in result.items() if not str(key).startswith("_platform_")}
+
+
 class CrawapiShipinhaoClient:
     requires_progressive_search_rate_limit = True
 
@@ -157,7 +161,7 @@ class CrawapiShipinhaoClient:
             max_results_per_query=_optional_positive_int(
                 _env("CONTENTFIND_SHIPINHAO_MAX_RESULTS_PER_QUERY", env, default="5")
             ),
-            rate_limiter=RateLimiter(min_interval_seconds=15.0),
+            rate_limiter=RateLimiter(min_interval_seconds=30.0),
         )
 
     def search(self, query: dict[str, Any]) -> list[dict[str, Any]]:
@@ -166,10 +170,43 @@ class CrawapiShipinhaoClient:
     def search_full_page(self, query: dict[str, Any]) -> list[dict[str, Any]]:
         return self._search(query, None)
 
+    def search_full_page_metadata(self, query: dict[str, Any]) -> list[dict[str, Any]]:
+        return self._search(query, None, hydrate_video=False)
+
+    def hydrate_search_result_media(
+        self,
+        result: dict[str, Any],
+        query: dict[str, Any],
+    ) -> dict[str, Any]:
+        item = result.get("_platform_search_item")
+        if not isinstance(item, dict):
+            return _strip_transient_fields(result)
+        selection = self._select_video_url_with_title_research(item, query)
+        hydrated = _normalize_shipinhao_item(
+            query,
+            item,
+            1,
+            bool(result.get("has_more")),
+            str(result.get("next_cursor") or ""),
+            selection,
+        )
+        merged = {
+            **_strip_transient_fields(result),
+            "play_url": hydrated.get("play_url"),
+            "platform_raw_payload": hydrated.get("platform_raw_payload", {}),
+        }
+        if selection.get("media_failure_reason"):
+            merged["media_failure_reason"] = selection["media_failure_reason"]
+        else:
+            merged.pop("media_failure_reason", None)
+        return merged
+
     def _search(
         self,
         query: dict[str, Any],
         max_results_per_query: int | None,
+        *,
+        hydrate_video: bool = True,
     ) -> list[dict[str, Any]]:
         data = self._keyword_search(
             {"keyword": query["search_query"], "cursor": str(query.get("page_cursor") or "")}
@@ -181,16 +218,20 @@ class CrawapiShipinhaoClient:
         selected = items[:max_results_per_query] if max_results_per_query else items
         results = []
         for index, item in enumerate(selected, start=1):
-            results.append(
-                _normalize_shipinhao_item(
-                    query,
-                    item,
-                    index,
-                    has_more,
-                    next_cursor,
-                    self._select_video_url_with_title_research(item, query),
-                )
+            selection = (
+                self._select_video_url_with_title_research(item, query) if hydrate_video else None
+            )
+            normalized = _normalize_shipinhao_item(
+                query,
+                item,
+                index,
+                has_more,
+                next_cursor,
+                selection,
             )
+            if not hydrate_video:
+                normalized["_platform_search_item"] = item
+            results.append(normalized)
         return results
 
     def _keyword_search(self, payload: dict[str, Any]) -> dict[str, Any]:
@@ -275,7 +316,9 @@ class CrawapiShipinhaoClient:
         payload.update(
             {
                 "shipinhao_research_source": "title",
-                "shipinhao_research_status": "used" if selection.get("play_url") else "no_valid_play_url",
+                "shipinhao_research_status": (
+                    "used" if selection.get("play_url") else "no_valid_play_url"
+                ),
                 "shipinhao_research_item_count": len(rows),
             }
         )

content_agent/run_service.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import os
+import threading
 from collections import Counter
 from datetime import datetime, timezone
 from pathlib import Path
@@ -18,6 +19,7 @@ from content_agent.integrations.douyin import CrawapiDouyinClient
 from content_agent.integrations.gemini_video import GeminiVideoClient as RealGeminiVideoClient
 from content_agent.integrations.kuaishou import CrawapiKuaishouClient
 from content_agent.integrations.mock_platform import MockPlatformClient
+from content_agent.integrations import oss_archive
 from content_agent.integrations.shipinhao import CrawapiShipinhaoClient
 from content_agent.integrations.policy_json import JsonPolicyBundleStore
 from content_agent.integrations.query_variant import (
@@ -147,6 +149,7 @@ class RunService:
             graph = build_run_graph(deps)
             state = graph.invoke(initial_state)
             self._record_success_metadata(state)
+            self._trigger_post_run_oss_archive(state, request)
             return state
         except Exception as exc:
             error = self._classify_error(exc)
@@ -275,6 +278,70 @@ class RunService:
             },
         )
 
+    def _trigger_post_run_oss_archive(self, state: RunState, request: RunStartRequest) -> None:
+        if request.platform_mode != "real":
+            return
+        run_id = state["run_id"]
+        policy_run_id = state["policy_run_id"]
+        try:
+            records = self.runtime.read_jsonl(run_id, "content_media_records.jsonl")
+            pending_count = sum(
+                1
+                for row in records
+                if row.get("content_media_status") == "oss_upload_pending" and row.get("play_url")
+            )
+            if pending_count <= 0:
+                return
+            self._append_lifecycle_event(
+                run_id,
+                policy_run_id,
+                event_id="oss_archive_post_run_started",
+                event_type="oss_archive_post_run",
+                status="running",
+                message="post-run oss archive started",
+                raw_payload={"pending_due_count": pending_count},
+            )
+        except Exception:
+            return
+
+        def _archive() -> None:
+            try:
+                archived = oss_archive.archive_pending_for_run(self.runtime, run_id)
+                status_counts = Counter(row.get("content_media_status") for row in archived)
+                self._append_lifecycle_event(
+                    run_id,
+                    policy_run_id,
+                    event_id="oss_archive_post_run_completed",
+                    event_type="oss_archive_post_run",
+                    status="success",
+                    message="post-run oss archive completed",
+                    raw_payload={
+                        "pending_due_count": pending_count,
+                        "content_media_status_counts": dict(status_counts),
+                    },
+                )
+            except Exception as exc:  # noqa: BLE001
+                self._append_lifecycle_event(
+                    run_id,
+                    policy_run_id,
+                    event_id="oss_archive_post_run_failed",
+                    event_type="oss_archive_post_run",
+                    status="failed",
+                    message="post-run oss archive failed",
+                    error_code="OSS_ARCHIVE_POST_RUN_FAILED",
+                    raw_payload={
+                        "pending_due_count": pending_count,
+                        "exception_type": type(exc).__name__,
+                        "error": str(exc)[:500],
+                    },
+                )
+
+        self._start_background_thread(_archive, name=f"oss-archive-{run_id}")
+
+    def _start_background_thread(self, target: Any, *, name: str) -> None:
+        thread = threading.Thread(target=target, name=name, daemon=True)
+        thread.start()
+
     def _update_final_output_validation(self, run_id: str, validation: dict[str, Any]) -> None:
         final_output = self.runtime.read_json(run_id, "final_output.json")
         validation_status = validation["status"]

tests/test_crawapi_http.py

@@ -50,13 +50,13 @@ def test_rate_limiter_waits_min_interval_between_same_bucket():
     clock = {"now": 0.0}
     sleeps: list[float] = []
     limiter = RateLimiter(
-        min_interval_seconds=12.0,
+        min_interval_seconds=30.0,
         now_fn=lambda: clock["now"],
         sleep_fn=lambda s: (sleeps.append(s), clock.__setitem__("now", clock["now"] + s)),
     )
     limiter.wait("b")
     limiter.wait("b")
-    assert sleeps == [12.0]
+    assert sleeps == [30.0]
 
 
 def test_http_429_maps_to_platform_rate_limited():

tests/test_douyin_client.py

@@ -398,12 +398,12 @@ def test_rate_limiter_waits_between_keyword_calls():
         sleeps.append(seconds)
         clock["now"] += seconds
 
-    limiter = RateLimiter(min_interval_seconds=12.0, now_fn=lambda: clock["now"], sleep_fn=fake_sleep)
+    limiter = RateLimiter(min_interval_seconds=30.0, now_fn=lambda: clock["now"], sleep_fn=fake_sleep)
 
     limiter.wait("douyin_search")
     limiter.wait("douyin_search")
 
-    assert sleeps == [12.0]
+    assert sleeps == [30.0]
 
 
 def test_search_chain_uses_shared_search_bucket():

tests/test_kuaishou_client.py

@@ -113,6 +113,48 @@ def test_kuaishou_search_falls_back_to_detail_video_url():
     assert "media_failure_reason" not in result
 
 
+def test_kuaishou_metadata_search_does_not_probe_or_fetch_detail():
+    probe_calls: list[str] = []
+    search_item = {**_item("ks_metadata"), "video_url_list": []}
+    client = CrawapiKuaishouClient(
+        base_url="http://crawler.test",
+        http_client=FakeHttpClient(
+            [_response(200, {"code": 0, "data": {"data": [search_item], "has_more": False}})]
+        ),
+        video_url_probe_fn=lambda url, platform: probe_calls.append(url) or {},
+    )
+
+    [result] = client.search_full_page_metadata(_query())
+
+    assert len(client.http_client.requests) == 1
+    assert probe_calls == []
+    assert result["platform_content_id"] == "ks_metadata"
+    assert result["media_failure_reason"] == "no_valid_play_url"
+    assert "_platform_search_item" in result
+
+
+def test_kuaishou_hydrate_search_result_media_uses_detail_fallback():
+    search_item = {**_item("ks_hydrate"), "video_url_list": []}
+    detail_item = {
+        **_item("ks_hydrate"),
+        "video_url_list": [{"video_url": "https://v.kwaicdn.test/detail.mp4"}],
+    }
+    client = _client(
+        [
+            _response(200, {"code": 0, "data": {"data": [search_item], "has_more": False}}),
+            _response(200, {"code": 0, "data": {"data": detail_item}}),
+        ]
+    )
+    [metadata] = client.search_full_page_metadata(_query())
+
+    hydrated = client.hydrate_search_result_media(metadata, _query())
+
+    assert client.http_client.requests[1]["url"].endswith("/crawler/kuai_shou/detail")
+    assert hydrated["play_url"] == "https://v.kwaicdn.test/detail.mp4"
+    assert hydrated["platform_raw_payload"]["kuaishou_detail_fallback_status"] == "used"
+    assert "_platform_search_item" not in hydrated
+
+
 def test_kuaishou_search_keeps_no_valid_after_search_and_detail_fail():
     search_item = {**_item("ks_no_video"), "video_url_list": [], "image_url_list": []}
     detail_item = {**search_item, "content_link": "https://www.gifshow.com/fw/photo/ks_no_video"}

tests/test_progressive_screening.py

@@ -34,6 +34,36 @@ class FakeProgressivePlatformClient:
         return self.search_full_page(query)
 
 
+class FakeHydratingProgressivePlatformClient(FakeProgressivePlatformClient):
+    def __init__(self, pages: dict[str, list[dict[str, Any]]]) -> None:
+        super().__init__(pages)
+        self.metadata_calls: list[dict[str, Any]] = []
+        self.hydrate_calls: list[str] = []
+
+    def search_full_page_metadata(self, query: dict[str, Any]) -> list[dict[str, Any]]:
+        self.metadata_calls.append(dict(query))
+        cursor = str(query.get("page_cursor") or "")
+        return [
+            {**dict(item), "_platform_search_item": dict(item)}
+            for item in self.pages.get(cursor, [])
+        ]
+
+    def hydrate_search_result_media(
+        self,
+        result: dict[str, Any],
+        query: dict[str, Any],
+    ) -> dict[str, Any]:
+        content_id = str(result.get("platform_content_id") or "")
+        self.hydrate_calls.append(content_id)
+        hydrated = {
+            key: value
+            for key, value in result.items()
+            if not str(key).startswith("_platform_")
+        }
+        hydrated["play_url"] = f"https://video.example/{content_id}.mp4"
+        return hydrated
+
+
 class FakeClock:
     def __init__(self) -> None:
         self.now = 0.0
@@ -135,7 +165,7 @@ def test_progressive_screening_remainder_and_page_two_gate_page_three(tmp_path):
 
     assert [call.get("page_cursor", "") for call in client.calls] == ["", "cursor_2", "cursor_3"]
     assert len(result["platform_results"]) == 30
-    assert clock.sleeps == [15.0, 15.0]
+    assert clock.sleeps == [30.0, 30.0]
 
 
 def test_progressive_screening_keep_for_review_does_not_pass(tmp_path):
@@ -152,6 +182,59 @@ def test_progressive_screening_keep_for_review_does_not_pass(tmp_path):
     }
 
 
+def test_progressive_screening_hydrates_only_initial_batch_without_pass(tmp_path):
+    context = _context(tmp_path)
+    client = FakeHydratingProgressivePlatformClient(
+        {"": _page([f"c{i}" for i in range(10)], has_more=True, cursor="next")}
+    )
+    gemini = FakeGeminiVideoClient(default_result=fake_gemini_review())
+
+    result = _run_screening(context, client, gemini)
+
+    assert client.metadata_calls == [context["search_queries"][0]]
+    assert client.calls == []
+    assert client.hydrate_calls == ["c0", "c1", "c2"]
+    assert len(result["platform_results"]) == 3
+    assert all("_platform_search_item" not in row for row in result["platform_results"])
+
+
+def test_progressive_screening_hydrates_remainder_only_after_first_batch_pass(tmp_path):
+    context = _context(tmp_path)
+    client = FakeHydratingProgressivePlatformClient(
+        {"": _page([f"c{i}" for i in range(10)], has_more=True, cursor="next")}
+    )
+    gemini = FakeGeminiVideoClient(
+        result_by_content_id={"c0": fake_gemini_pool()},
+        default_result=fake_gemini_review(),
+    )
+
+    result = _run_screening(context, client, gemini)
+
+    assert client.hydrate_calls == [f"c{i}" for i in range(10)]
+    assert len(result["platform_results"]) == 10
+
+
+def test_progressive_screening_does_not_hydrate_duplicate_again(tmp_path):
+    context = _context(tmp_path)
+    client = FakeHydratingProgressivePlatformClient(
+        {"": _page(["dup", "c1", "c2", "dup", "c4"], has_more=True, cursor="next")}
+    )
+    gemini = FakeGeminiVideoClient(
+        result_by_content_id={"dup": fake_gemini_pool()},
+        default_result=fake_gemini_review(),
+    )
+
+    result = _run_screening(context, client, gemini)
+
+    assert client.hydrate_calls == ["dup", "c1", "c2", "c4"]
+    assert [item["platform_content_id"] for item in result["platform_results"]] == [
+        "dup",
+        "c1",
+        "c2",
+        "c4",
+    ]
+
+
 def test_progressive_screening_deduplicates_across_batches(tmp_path):
     context = _context(tmp_path)
     client = FakeProgressivePlatformClient(

tests/test_run_service_post_archive.py

@@ -0,0 +1,144 @@
+from __future__ import annotations
+
+from content_agent import run_service
+from content_agent.integrations.runtime_files import LocalRuntimeFileStore
+from content_agent.run_service import RunService
+from content_agent.schemas import RunStartRequest
+
+
+class EventRecordingRuntime(LocalRuntimeFileStore):
+    def append_run_event_records(self, run_id, policy_run_id, rows):
+        prepared = [
+            {**row, "run_id": run_id, "policy_run_id": row.get("policy_run_id", policy_run_id)}
+            for row in rows
+        ]
+        self.append_jsonl(run_id, "run_events.jsonl", prepared)
+
+
+def test_post_run_oss_archive_triggers_for_real_pending_records(monkeypatch, tmp_path):
+    runtime = EventRecordingRuntime(tmp_path)
+    runtime.prepare_run("run_001")
+    runtime.append_jsonl(
+        "run_001",
+        "content_media_records.jsonl",
+        [
+            {
+                "run_id": "run_001",
+                "policy_run_id": "policy_001",
+                "platform": "kuaishou",
+                "platform_content_id": "content_001",
+                "content_media_status": "oss_upload_pending",
+                "play_url": "https://source.example/video.mp4",
+                "raw_payload": {},
+            }
+        ],
+    )
+    service = object.__new__(RunService)
+    service.runtime = runtime
+    service._start_background_thread = lambda target, name: target()
+    calls: list[str] = []
+
+    def fake_archive(runtime_arg, run_id):
+        calls.append(run_id)
+        return [
+            {
+                "platform_content_id": "content_001",
+                "content_media_status": "oss_uploaded",
+            }
+        ]
+
+    monkeypatch.setattr(run_service.oss_archive, "archive_pending_for_run", fake_archive)
+
+    service._trigger_post_run_oss_archive(
+        {"run_id": "run_001", "policy_run_id": "policy_001"},
+        RunStartRequest(platform="kuaishou", platform_mode="real"),
+    )
+
+    events = runtime.read_jsonl("run_001", "run_events.jsonl")
+    assert calls == ["run_001"]
+    assert [event["event_id"] for event in events] == [
+        "oss_archive_post_run_started",
+        "oss_archive_post_run_completed",
+    ]
+    assert events[0]["raw_payload"]["pending_due_count"] == 1
+    assert events[1]["raw_payload"]["content_media_status_counts"] == {"oss_uploaded": 1}
+
+
+def test_post_run_oss_archive_failure_is_recorded_without_raising(monkeypatch, tmp_path):
+    runtime = EventRecordingRuntime(tmp_path)
+    runtime.prepare_run("run_001")
+    runtime.append_jsonl(
+        "run_001",
+        "content_media_records.jsonl",
+        [
+            {
+                "run_id": "run_001",
+                "policy_run_id": "policy_001",
+                "platform": "shipinhao",
+                "platform_content_id": "content_001",
+                "content_media_status": "oss_upload_pending",
+                "play_url": "https://source.example/video.mp4",
+                "raw_payload": {},
+            }
+        ],
+    )
+    service = object.__new__(RunService)
+    service.runtime = runtime
+    service._start_background_thread = lambda target, name: target()
+
+    def fail_archive(runtime_arg, run_id):
+        raise TimeoutError("archive stuck")
+
+    monkeypatch.setattr(run_service.oss_archive, "archive_pending_for_run", fail_archive)
+
+    service._trigger_post_run_oss_archive(
+        {"run_id": "run_001", "policy_run_id": "policy_001"},
+        RunStartRequest(platform="shipinhao", platform_mode="real"),
+    )
+
+    events = runtime.read_jsonl("run_001", "run_events.jsonl")
+    assert [event["event_id"] for event in events] == [
+        "oss_archive_post_run_started",
+        "oss_archive_post_run_failed",
+    ]
+    assert events[1]["error_code"] == "OSS_ARCHIVE_POST_RUN_FAILED"
+    assert events[1]["raw_payload"]["exception_type"] == "TimeoutError"
+
+
+def test_post_run_oss_archive_skips_mock_and_non_pending_records(monkeypatch, tmp_path):
+    runtime = EventRecordingRuntime(tmp_path)
+    runtime.prepare_run("run_001")
+    runtime.append_jsonl(
+        "run_001",
+        "content_media_records.jsonl",
+        [
+            {
+                "run_id": "run_001",
+                "policy_run_id": "policy_001",
+                "platform_content_id": "content_uploaded",
+                "content_media_status": "oss_uploaded",
+                "play_url": "https://source.example/video.mp4",
+            }
+        ],
+    )
+    service = object.__new__(RunService)
+    service.runtime = runtime
+    service._start_background_thread = lambda target, name: target()
+    calls: list[str] = []
+    monkeypatch.setattr(
+        run_service.oss_archive,
+        "archive_pending_for_run",
+        lambda runtime_arg, run_id: calls.append(run_id),
+    )
+
+    service._trigger_post_run_oss_archive(
+        {"run_id": "run_001", "policy_run_id": "policy_001"},
+        RunStartRequest(platform="douyin", platform_mode="real"),
+    )
+    service._trigger_post_run_oss_archive(
+        {"run_id": "run_001", "policy_run_id": "policy_001"},
+        RunStartRequest(platform="douyin", platform_mode="mock"),
+    )
+
+    assert calls == []
+    assert runtime.read_jsonl("run_001", "run_events.jsonl") == []

tests/test_shipinhao_client.py

@@ -122,6 +122,75 @@ def test_shipinhao_search_falls_back_to_title_research_for_same_content_id():
     assert "media_failure_reason" not in result
 
 
+def test_shipinhao_metadata_search_does_not_probe_or_title_research():
+    probe_calls: list[str] = []
+    first = {
+        "code": 0,
+        "data": {
+            "data": [
+                {
+                    "channel_content_id": "finderobj_metadata",
+                    "title": "当狗唱dj #心墙",
+                    "content_type": "video",
+                    "image_url_list": [{"image_url": "https://findermp.video.qq.com/cover.jpg"}],
+                }
+            ]
+        },
+    }
+    client = CrawapiShipinhaoClient(
+        base_url="http://crawler.test",
+        http_client=FakeHttpClient([_response(200, first)]),
+        sleep_fn=lambda seconds: None,
+        video_url_probe_fn=lambda url, platform: probe_calls.append(url) or {},
+    )
+
+    [result] = client.search_full_page_metadata(_query())
+
+    assert len(client.http_client.requests) == 1
+    assert probe_calls == []
+    assert result["platform_content_id"] == "finderobj_metadata"
+    assert result["media_failure_reason"] == "no_valid_play_url"
+    assert "_platform_search_item" in result
+
+
+def test_shipinhao_hydrate_search_result_media_uses_title_research():
+    first = {
+        "code": 0,
+        "data": {
+            "data": [
+                {
+                    "channel_content_id": "finderobj_hydrate",
+                    "title": "当狗唱dj #心墙",
+                    "content_type": "video",
+                    "image_url_list": [{"image_url": "https://findermp.video.qq.com/cover.jpg"}],
+                }
+            ]
+        },
+    }
+    second = {
+        "code": 0,
+        "data": {
+            "data": [
+                {
+                    "channel_content_id": "finderobj_hydrate",
+                    "title": "当狗唱dj #心墙",
+                    "content_type": "video",
+                    "video_url_list": [{"video_url": "https://findermp.video.qq.com/video.mp4"}],
+                }
+            ]
+        },
+    }
+    client, _ = _client([_response(200, first), _response(200, second)])
+    [metadata] = client.search_full_page_metadata(_query())
+
+    hydrated = client.hydrate_search_result_media(metadata, _query())
+
+    assert client.http_client.requests[1]["json"] == {"keyword": "当狗唱dj #心墙", "cursor": ""}
+    assert hydrated["play_url"] == "https://findermp.video.qq.com/video.mp4"
+    assert hydrated["platform_raw_payload"]["shipinhao_research_status"] == "used"
+    assert "_platform_search_item" not in hydrated
+
+
 def test_shipinhao_search_keeps_no_valid_when_title_research_does_not_match():
     first = {
         "code": 0,

web2/features/LedgerPage.tsx

@@ -40,6 +40,7 @@ export function LedgerPage({ runId }: { runId: string }) {
       {!loading && !error && !rows.length ? <EmptyState label="没有可展示的搜索流程记录" /> : null}
       {ledger ? (
         <>
+          <QueryFailureNotice summary={ledger.summary} />
           <div className={`ledger-layout ${demandCollapsed ? "demand-collapsed" : ""}`}>
             <DemandPanel
               demand={ledger.demandSummary}
@@ -72,6 +73,24 @@ export function LedgerPage({ runId }: { runId: string }) {
   );
 }
 
+function QueryFailureNotice({ summary }: { summary: Record<string, unknown> }) {
+  const count = Number(summary.query_failure_count || 0);
+  if (!count) return null;
+  const examples = Array.isArray(summary.query_failure_examples)
+    ? summary.query_failure_examples
+      .map((item) => (typeof item === "object" && item ? String((item as { search_query?: unknown }).search_query || "") : ""))
+      .filter(Boolean)
+      .slice(0, 3)
+    : [];
+  return (
+    <section className="source-block open">
+      <div className="source-summary">
+        已完成；有 {count} 条搜索词失败{examples.length ? `：${examples.join("、")}` : ""}。
+      </div>
+    </section>
+  );
+}
+
 function DemandPanel({
   demand,
   collapsed,
@@ -265,9 +284,14 @@ function VideoDecisionCell({
     <td className="rules-cell video-decision-cell">
       <div className="cell-stack">
         <span className={`chip ${decisionChipClass(video.decisionAction)}`}>{video.decisionLabel}</span>
+        {video.decisionAction === "TECHNICAL_RETRY_REQUIRED" ? (
+          <span className="muted">{technicalRetryShortText(video)}</span>
+        ) : null}
         {video.decisionAction === "TECHNICAL_RETRY_REQUIRED" ? <TechnicalRetryDetails detail={video.technicalRetryDetail} compact /> : null}
         <strong>{compactScoreItemsText(video.scoreItems)}</strong>
-        <span className="muted">{video.decisionReasonLabel}</span>
+        {video.decisionAction !== "TECHNICAL_RETRY_REQUIRED" ? (
+          <span className="muted">{video.decisionReasonLabel}</span>
+        ) : null}
         <button className="mini-button score-rule-button" type="button" onClick={() => onOpenScore(scoreRuleDrawer(row, video))}>
           <Info size={13} />
           打分规则
@@ -363,11 +387,37 @@ function videoWalkStatus(video: VideoRef | null, allowWalk: boolean): { title: s
   if (allowWalk) return { title: "通过游走门槛", detail: "可从标签、作者或翻页继续找" };
   if (video.decisionAction === "ADD_TO_CONTENT_POOL") return { title: "入池但未游走", detail: "本条可沉淀，未继续向外扩展" };
   if (video.decisionAction === "KEEP_CONTENT_FOR_REVIEW") return { title: "未游走：待复看停止", detail: "需要人工确认后再决定是否继续" };
-  if (video.decisionAction === "TECHNICAL_RETRY_REQUIRED") return { title: "未游走：技术重试", detail: "视频下载、压缩或模型判断失败，等待系统重试" };
+  if (video.decisionAction === "TECHNICAL_RETRY_REQUIRED") {
+    const detail = video.technicalRetryDetail;
+    return {
+      title: `未游走：${detail?.stageLabel || "技术重试"}`,
+      detail: technicalRetryShortText(video)
+    };
+  }
   if (video.decisionAction === "REJECT_CONTENT") return { title: "未游走：淘汰停止", detail: "不再继续带回新内容" };
   return { title: "未进入游走", detail: "当前没有继续扩展记录" };
 }
 
+function technicalRetryShortText(video: VideoRef): string {
+  const detail = video.technicalRetryDetail;
+  if (!detail) return video.decisionReasonLabel || "系统需要重试后再判断";
+  const stage = detail.stageLabel || "技术重试";
+  const reason = detail.briefReason || detail.failureLabel || detail.failureType || "系统需要重试后再判断";
+  const seconds = technicalRetrySeconds(detail);
+  return `${stage}：${reason}${seconds ? ` · ${seconds}` : ""}`;
+}
+
+function technicalRetrySeconds(detail: NonNullable<VideoRef["technicalRetryDetail"]>): string {
+  const values = [
+    detail.timings.download_seconds,
+    detail.timings.ffmpeg_seconds,
+    detail.timings.gemini_seconds
+  ].filter((value): value is number => typeof value === "number" && !Number.isNaN(value));
+  if (!values.length) return "";
+  const total = values.reduce((sum, value) => sum + value, 0);
+  return `${Number(total.toFixed(1))}s`;
+}
+
 function AssetCell({ row, rowSpan }: { row: FlowLedgerRow; rowSpan?: number }) {
   return (
     <td className="asset-cell" rowSpan={rowSpan}>

web2/features/QueryVideosPage.tsx

@@ -116,6 +116,9 @@ function VideoRow({ index, runId, video }: { index: number; runId: string; video
       </td>
       <td>
         <span className="chip blue">{actionLabel(video.decisionAction)}</span>
+        {video.decisionAction === "TECHNICAL_RETRY_REQUIRED" ? (
+          <div className="muted">{technicalRetryShortText(video)}</div>
+        ) : null}
         {video.decisionAction === "TECHNICAL_RETRY_REQUIRED" ? <TechnicalRetryDetails detail={video.technicalRetryDetail} compact /> : null}
       </td>
       <td>
@@ -134,3 +137,11 @@ function VideoRow({ index, runId, video }: { index: number; runId: string; video
     </tr>
   );
 }
+
+function technicalRetryShortText(video: VideoRef): string {
+  const detail = video.technicalRetryDetail;
+  if (!detail) return "系统需要重试后再判断";
+  const stage = detail.stageLabel || "技术重试";
+  const reason = detail.briefReason || detail.failureLabel || detail.failureType || "系统需要重试后再判断";
+  return `${stage}：${reason}`;
+}

web2/features/RunListPage.tsx

@@ -14,7 +14,8 @@ function statusLabel(value?: string | null): string {
     running: "运行中",
     failed: "运行失败",
     pending: "等待处理",
-    success: "已完成"
+    success: "已完成",
+    partial_success: "已完成"
   };
   return value ? labels[value] || "状态待确认" : "状态待确认";
 }

web2/features/VideoDetailPage.tsx

@@ -8,6 +8,7 @@ import { getFlowLedgerVideo } from "@/lib/api/client";
 import type { FlowLedgerVideoResponse } from "@/lib/api/types";
 import { videoFromApi } from "@/lib/flow-ledger/build";
 import { actionLabel, displayScoreLabel, mediaStatusLabel, metricLabel, platformLabel, scoreItemRows } from "@/lib/flow-ledger/business";
+import type { VideoRef } from "@/lib/flow-ledger/types";
 
 export function VideoDetailPage({ runId, contentId }: { runId: string; contentId: string }) {
   const [data, setData] = useState<FlowLedgerVideoResponse | null>(null);
@@ -51,7 +52,7 @@ export function VideoDetailPage({ runId, contentId }: { runId: string; contentId
             </summary>
             <div className="decl-body">
               <Summary label="互动" value={metricLabel(video)} />
-              <Summary label="判断" value={actionLabel(video.decisionAction)} />
+              <Summary label="判断" value={decisionSummary(video)} />
               <Summary label="视频保存" value={`${video.mediaStatusLabel || mediaStatusLabel(video.mediaStatus)}${video.mediaFailureReason ? ` · ${video.mediaFailureReason}` : ""}`} />
               <Summary label="标签" value={video.tags.length ? video.tags.join(" / ") : "无标签"} />
               {video.decisionAction === "TECHNICAL_RETRY_REQUIRED" ? <TechnicalRetryDetails detail={video.technicalRetryDetail} compact /> : null}
@@ -178,6 +179,15 @@ function platformBreakdownRule(label: string): string {
   return "参与平台表现综合评分。";
 }
 
+function decisionSummary(video: VideoRef): string {
+  if (video.decisionAction !== "TECHNICAL_RETRY_REQUIRED") return actionLabel(video.decisionAction);
+  const detail = video.technicalRetryDetail;
+  if (!detail) return "技术重试";
+  const stage = detail.stageLabel || "技术重试";
+  const reason = detail.briefReason || detail.failureLabel || detail.failureType || "系统需要重试后再判断";
+  return `技术重试 · ${stage}：${reason}`;
+}
+
 function Summary({ label, value }: { label: string; value: string }) {
   return (
     <div className="decl-section">