运行诊断：聚合 Broker预算、分页与语义调用指标

E2E报告增加 Context Broker 请求数、p50/p95/max token、遗漏后读取、重复详情页、语义调用、缓存和 fallback 统计。工具计数改为当前 save_script_paragraphs 与 save_script_elements，并保持成功和异常诊断结构一致。

script_build_host/src/script_build_host/internal_e2e.py

@@ -668,6 +668,13 @@ async def _collect_diagnostics(host: Any, script_build_id: int) -> dict[str, Any
     trace_ids.update(item.worker_trace_id for item in attempts)
     trace_ids.update(item.validator_trace_id for item in validations)
     tool_calls: dict[str, int] = {}
+    broker_events = [
+        item
+        for item in await host.composition.mission_service.runner.trace_store.get_events(
+            binding.root_trace_id
+        )
+        if item.get("event") == "context_broker_receipt"
+    ]
     for trace_id in trace_ids:
         for message in await host.composition.mission_service.runner.trace_store.get_trace_messages(
             trace_id
@@ -722,15 +729,45 @@ async def _collect_diagnostics(host: Any, script_build_id: int) -> dict[str, Any
         "preset_usage": preset_usage,
         "tool_calls": tool_calls,
         "paragraph_tool_calls": {
-            "single": tool_calls.get("create_script_paragraph", 0),
-            "batch": tool_calls.get("create_script_paragraphs", 0),
+            "paragraph_upsert": tool_calls.get("save_script_paragraphs", 0),
+            "element_upsert": tool_calls.get("save_script_elements", 0),
         },
+        "context_broker": _context_broker_diagnostics(broker_events),
         "deterministic_preflight_rejection": any(
             "Deterministic preflight rejected" in item.summary for item in validations
         ),
     }
 
 
+def _context_broker_diagnostics(events: list[dict[str, Any]]) -> dict[str, Any]:
+    sizes = sorted(int(item.get("estimated_tokens") or 0) for item in events)
+    detail_pages = [
+        str(item.get("detail_page_key"))
+        for item in events
+        if int(item.get("detail_reads") or 0) > 0 and item.get("detail_page_key")
+    ]
+    return {
+        "requests": len(events),
+        "bundle_tokens": {
+            "p50": _percentile(sizes, 0.50),
+            "p95": _percentile(sizes, 0.95),
+            "max": max(sizes, default=0),
+        },
+        "omitted_count": sum(int(item.get("omitted_count") or 0) for item in events),
+        "detail_reads": sum(int(item.get("detail_reads") or 0) for item in events),
+        "duplicate_detail_reads": len(detail_pages) - len(set(detail_pages)),
+        "semantic_calls": sum(int(item.get("semantic_calls") or 0) for item in events),
+        "semantic_fallbacks": sum(bool(item.get("fallback")) for item in events),
+        "semantic_cache_hits": sum(bool(item.get("cache_hit")) for item in events),
+    }
+
+
+def _percentile(values: list[int], ratio: float) -> int:
+    if not values:
+        return 0
+    return values[min(len(values) - 1, round((len(values) - 1) * ratio))]
+
+
 def _empty_preset_usage() -> dict[str, int | float]:
     return {
         "agent_runs": 0,

script_build_host/tests/test_internal_e2e.py

@@ -5,7 +5,7 @@ import httpx
 import pytest
 
 from script_build_host.domain.records import BuildStatus
-from script_build_host.internal_e2e import _wait_for_root
+from script_build_host.internal_e2e import _context_broker_diagnostics, _wait_for_root
 
 
 def test_recovery_checkpoint_fixtures_cover_all_segment_boundaries() -> None:
@@ -18,6 +18,40 @@ def test_recovery_checkpoint_fixtures_cover_all_segment_boundaries() -> None:
     )
 
 
+def test_context_broker_diagnostics_count_repeated_pages_not_shared_revisions() -> None:
+    diagnostics = _context_broker_diagnostics(
+        [
+            {
+                "revision": "same-artifact",
+                "estimated_tokens": 100,
+                "detail_reads": 1,
+                "detail_page_key": "page-1",
+            },
+            {
+                "revision": "same-artifact",
+                "estimated_tokens": 300,
+                "detail_reads": 1,
+                "detail_page_key": "page-2",
+            },
+            {
+                "revision": "same-artifact",
+                "estimated_tokens": 100,
+                "detail_reads": 1,
+                "detail_page_key": "page-1",
+                "semantic_calls": 1,
+                "fallback": True,
+            },
+        ]
+    )
+
+    assert diagnostics["requests"] == 3
+    assert diagnostics["bundle_tokens"] == {"p50": 100, "p95": 300, "max": 300}
+    assert diagnostics["detail_reads"] == 3
+    assert diagnostics["duplicate_detail_reads"] == 1
+    assert diagnostics["semantic_calls"] == 1
+    assert diagnostics["semantic_fallbacks"] == 1
+
+
 @pytest.mark.asyncio
 async def test_mission_poll_allows_only_bounded_startup_failures() -> None:
     calls = 0