性能：压缩检索素材并提供增量分页读取

Pattern、Decode、Knowledge 和 Evidence 仅向角色 Bootstrap 暴露摘要与语义 handle，大内容按 8000 字符上限分页读取；Pattern 候选和摘要设硬上限，避免 Build 900049 形态的大 Artifact 反复进入 Planner 与 Worker 上下文。

script_build_host/src/script_build_host/adapters/retrieval.py

@@ -151,9 +151,10 @@ class PatternRetrievalAdapter:
                 task_id=task_id,
                 payload=payload,
             )
+        compact_payload = _compact_pattern_payload(payload)
         return _result(
             "pattern",
-            payload,
+            compact_payload,
             metadata={
                 "task_id": _safe_source_id(task_id, rank=0) if task_id else None,
                 "response_sha256": canonical_sha256(payload).wire,
@@ -429,6 +430,87 @@ def _result(
     )
 
 
+def _compact_pattern_payload(
+    payload: Any,
+    *,
+    max_candidates: int = 3,
+    max_chars: int = 20_000,
+) -> Any:
+    """Keep a bounded Pattern evidence view while preserving the raw digest upstream."""
+
+    if isinstance(payload, list):
+        compact_items = [
+            _compact_pattern_payload(
+                item,
+                max_candidates=max_candidates,
+                max_chars=max_chars,
+            )
+            for item in payload
+        ]
+        while compact_items and _json_chars(compact_items) > max_chars:
+            compact_items.pop()
+        if compact_items:
+            return compact_items
+        return [{"truncated_for_context": True, "items_total": len(payload)}]
+    if not isinstance(payload, dict):
+        if _json_chars(payload) <= max_chars:
+            return payload
+        return {
+            "output": str(payload)[:1_000],
+            "truncated_for_context": True,
+        }
+    compact = dict(payload)
+    data = compact.get("data")
+    if not isinstance(data, dict) or not isinstance(data.get("candidates"), list):
+        if _json_chars(compact) <= max_chars:
+            return compact
+        return {
+            "status": compact.get("status"),
+            "output": str(compact.get("output") or "")[:1_000],
+            "truncated_for_context": True,
+        }
+    compact_data = dict(data)
+    candidates = data["candidates"]
+    compact_candidates: list[Any] = []
+    for candidate in candidates[:max_candidates]:
+        if not isinstance(candidate, dict):
+            compact_candidates.append(candidate)
+            continue
+        compact_candidate = dict(candidate)
+        if compact_candidate.get("target_items") == compact_candidate.get("itemset"):
+            compact_candidate.pop("target_items", None)
+        compact_candidates.append(compact_candidate)
+    compact_data["candidates"] = compact_candidates
+    compact_data["candidates_returned"] = len(compact_candidates)
+    compact_data.setdefault("candidates_total", len(candidates))
+    compact["data"] = compact_data
+    if _json_chars(compact) <= max_chars:
+        return compact
+
+    while compact_candidates:
+        compact_candidates.pop()
+        compact_data["candidates_returned"] = len(compact_candidates)
+        compact_data["truncated_for_context"] = True
+        if _json_chars(compact) <= max_chars:
+            return compact
+
+    return {
+        "status": compact.get("status"),
+        "output": str(compact.get("output") or "")[:1_000],
+        "data": {
+            "interface": compact_data.get("interface"),
+            "matches": compact_data.get("matches"),
+            "candidates_total": compact_data.get("candidates_total", len(candidates)),
+            "candidates_returned": 0,
+            "truncated_for_context": True,
+        },
+    }
+
+
+def _json_chars(value: Any) -> int:
+    return len(json.dumps(value, ensure_ascii=False, default=str))
+
+
 def _safe_source_id(value: Any, *, rank: int) -> str:
     text = str(value if value is not None else rank)
     if re.fullmatch(r"[A-Za-z0-9._:-]{1,128}", text):

script_build_host/src/script_build_host/internal_e2e.py

@@ -205,6 +205,7 @@ async def run_real_e2e(
                 phase_two_root = await _wait_for_root(
                     client,
                     script_build_id,
+                    legacy_state=host.composition.mission_service.legacy_state,
                     expected_status="blocked",
                     expected_reason=PHASE_TWO_CANDIDATE_PORTFOLIO_READY,
                     timeout_seconds=timeout_seconds,
@@ -243,6 +244,7 @@ async def run_real_e2e(
             completed_root = await _wait_for_root(
                 client,
                 script_build_id,
+                legacy_state=host.composition.mission_service.legacy_state,
                 expected_status="completed",
                 expected_reason=None,
                 timeout_seconds=timeout_seconds,
@@ -278,9 +280,7 @@ async def run_real_e2e(
                 )
                 final_payload = _successful_json(finalize, "finalize")
                 if not final_payload.get("committed"):
-                    raise RuntimeError(
-                        "final publication did not report a committed transaction"
-                    )
+                    raise RuntimeError("final publication did not report a committed transaction")
 
             publication_response = await client.get(
                 f"/api/pattern/script_builds/{script_build_id}/publication",
@@ -427,6 +427,7 @@ async def _wait_for_root(
     client: httpx.AsyncClient,
     script_build_id: int,
     *,
+    legacy_state: Any | None = None,
     expected_status: str,
     expected_reason: str | None,
     timeout_seconds: float,
@@ -460,6 +461,12 @@ async def _wait_for_root(
             last_state = state
         if state[0] in {"failed", "cancelled"}:
             raise RuntimeError(f"root task entered terminal failure state: {state}")
+        if legacy_state is not None:
+            build_status = await legacy_state.get_status(script_build_id)
+            if build_status in {BuildStatus.FAILED, BuildStatus.STOPPED}:
+                raise RuntimeError(
+                    f"script build entered terminal status={build_status.value}; root={state}"
+                )
         if state[0] == expected_status and (expected_reason is None or state[1] == expected_reason):
             return root
         await asyncio.sleep(poll_seconds)
@@ -486,9 +493,7 @@ async def _wait_for_build_status(
         if last in expected_statuses:
             return last
         if last in {BuildStatus.FAILED, BuildStatus.STOPPING, BuildStatus.STOPPED}:
-            raise RuntimeError(
-                f"build entered {last.value} while waiting for {expected_values}"
-            )
+            raise RuntimeError(f"build entered {last.value} while waiting for {expected_values}")
         await asyncio.sleep(poll_seconds)
     raise TimeoutError(
         f"timed out waiting for build status={expected_values}; "
@@ -592,12 +597,8 @@ async def _collect_diagnostics(host: Any, script_build_id: int) -> dict[str, Any
     preset_usage: dict[str, dict[str, int | float]] = {}
     domain_errors: list[dict[str, str]] = []
     trace_presets = {binding.root_trace_id: "script_planner"}
-    trace_presets.update(
-        {item.worker_trace_id: item.worker_preset for item in attempts}
-    )
-    trace_presets.update(
-        {item.validator_trace_id: item.validator_preset for item in validations}
-    )
+    trace_presets.update({item.worker_trace_id: item.worker_preset for item in attempts})
+    trace_presets.update({item.validator_trace_id: item.validator_preset for item in validations})
     traces_with_usage: set[str] = set()
     usage_reader = getattr(
         host.composition.mission_service.runner.trace_store, "get_model_usage", None
@@ -607,9 +608,7 @@ async def _collect_diagnostics(host: Any, script_build_id: int) -> dict[str, Any
             model_usage = await usage_reader(trace_id)
             models = model_usage.get("models", []) if isinstance(model_usage, dict) else []
             agent_models = [
-                item
-                for item in models
-                if isinstance(item, dict) and item.get("source") == "agent"
+                item for item in models if isinstance(item, dict) and item.get("source") == "agent"
             ]
             if agent_models:
                 traces_with_usage.add(trace_id)

script_build_host/tests/test_internal_e2e.py

@@ -4,6 +4,7 @@ from pathlib import Path
 import httpx
 import pytest
 
+from script_build_host.domain.records import BuildStatus
 from script_build_host.internal_e2e import _wait_for_root
 
 
@@ -73,3 +74,33 @@ async def test_mission_poll_surfaces_persistent_server_failure() -> None:
                 timeout_seconds=1,
                 poll_seconds=0,
             )
+
+
+@pytest.mark.asyncio
+async def test_mission_poll_stops_when_build_is_terminal() -> None:
+    class FailedState:
+        async def get_status(self, _script_build_id):
+            return BuildStatus.FAILED
+
+    def handler(_request: httpx.Request) -> httpx.Response:
+        return httpx.Response(
+            200,
+            json={
+                "root_task_id": "root",
+                "tasks": [{"task_id": "root", "status": "needs_replan"}],
+            },
+        )
+
+    async with httpx.AsyncClient(
+        transport=httpx.MockTransport(handler), base_url="http://internal"
+    ) as client:
+        with pytest.raises(RuntimeError, match="terminal status=failed"):
+            await _wait_for_root(
+                client,
+                7,
+                legacy_state=FailedState(),
+                expected_status="blocked",
+                expected_reason="BOUNDARY",
+                timeout_seconds=1,
+                poll_seconds=0,
+            )

script_build_host/tests/test_retrieval_adapters.py

@@ -181,6 +181,74 @@ async def test_pattern_timeout_is_frozen_as_a_limitation() -> None:
     assert result.metadata["task_id"] == "task-1"
 
 
+@pytest.mark.asyncio
+async def test_pattern_result_keeps_three_candidates_without_duplicate_items() -> None:
+    candidates = [
+        {
+            "itemset": [{"dimension": f"dimension-{index}"}],
+            "target_items": [{"dimension": f"dimension-{index}"}],
+            "support": 10 - index,
+        }
+        for index in range(6)
+    ]
+
+    async def handler(_request: httpx.Request) -> httpx.Response:
+        return httpx.Response(
+            200,
+            json=[{"status": "success", "data": {"candidates": candidates}}],
+        )
+
+    async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
+        safe = SafeHttpClient(
+            client,
+            OutboundPolicy(frozenset({"api.example.com"}), resolver=_public_resolver),
+        )
+        result = await PatternRetrievalAdapter("https://api.example.com/pattern", safe).retrieve(
+            query={"message": "shape"}, snapshot=SimpleNamespace()
+        )
+
+    payload = json.loads(result.summary)[0]
+    assert payload["data"]["candidates_total"] == 6
+    assert payload["data"]["candidates_returned"] == 3
+    assert len(payload["data"]["candidates"]) == 3
+    assert all("target_items" not in item for item in payload["data"]["candidates"])
+
+
+@pytest.mark.asyncio
+async def test_pattern_result_has_a_hard_context_size_limit() -> None:
+    candidate = {
+        "itemset": [{"dimension": "x" * 10_000}],
+        "target_items": [{"dimension": "x" * 10_000}],
+        "examples": ["post-1"],
+    }
+
+    async def handler(_request: httpx.Request) -> httpx.Response:
+        return httpx.Response(
+            200,
+            json=[
+                {
+                    "status": "success",
+                    "output": "large result",
+                    "data": {"candidates_total": 193, "candidates": [candidate] * 20},
+                }
+            ],
+        )
+
+    async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
+        safe = SafeHttpClient(
+            client,
+            OutboundPolicy(frozenset({"api.example.com"}), resolver=_public_resolver),
+        )
+        result = await PatternRetrievalAdapter("https://api.example.com/pattern", safe).retrieve(
+            query={"message": "shape"}, snapshot=SimpleNamespace()
+        )
+
+    assert len(result.summary) <= 20_100
+    payload = json.loads(result.summary)[0]
+    assert payload["data"]["candidates_total"] == 193
+    assert payload["data"]["truncated_for_context"] is True
+
+
 @pytest.mark.asyncio
 async def test_file_decode_and_knowledge_freeze_file_hash_and_rank(tmp_path) -> None:
     decode_path = tmp_path / "decode.json"