工具面回归：锁定 Broker权限、Prompt组合与身份脱敏

更新 Agent preset 精确 allowlist，断言旧快照和低层写工具不可见、模型数据不暴露物理ID/URI/digest。验证每个真实模型角色恰好冻结一次公共 Context Access Protocol，Retrieval分页语义、Root闭包说明和工具schema描述完整。

script_build_host/tests/test_agent_surface.py

@@ -1,6 +1,5 @@
 from __future__ import annotations
 
-import json
 from dataclasses import replace
 from datetime import UTC, datetime
 from hashlib import sha256
@@ -16,14 +15,13 @@ from script_build_host.agents.model_resolver import (
 from script_build_host.agents.presets import register_script_presets
 from script_build_host.agents.prompt_resolver import SnapshotRoleSystemPromptResolver
 from script_build_host.agents.prompts import (
+    CONTEXT_ACCESS_PROTOCOL,
     phase_three_prompt_manifest,
     script_build_prompt_manifest,
 )
-from script_build_host.domain.canonical_json import canonical_sha256
 from script_build_host.domain.errors import ProtocolViolation
 from script_build_host.domain.input_snapshot import ScriptBuildInputSnapshotV1
-from script_build_host.domain.workbench import strategy_handle
-from script_build_host.tools.gateway import LegacyScriptToolGateway, _project_snapshot
+from script_build_host.domain.workbench import protect_model_value
 from script_build_host.tools.registry import register_script_tools
 
 
@@ -63,9 +61,11 @@ def test_script_presets_are_exact_and_never_expose_legacy_control_tools() -> Non
 
     planner = get_preset("script_planner")
     assert set(planner.allowed_tools or []) == {
+        "get_current_context",
         "plan_script_tasks",
         "decide_script_task",
-        "read_workbench_detail",
+        "search_mission_context",
+        "read_mission_context",
         "dispatch_script_tasks",
         "validate_attempt",
     }
@@ -86,12 +86,15 @@ def test_script_presets_are_exact_and_never_expose_legacy_control_tools() -> Non
     assert "batch_link_paragraph_elements" not in element_tools
 
     assert set(get_preset("script_root_worker").allowed_tools or []) == {
-        "read_workbench_detail",
+        "search_mission_context",
+        "read_mission_context",
         "legacy_projection_dry_run",
         "save_root_delivery_manifest",
         "submit_attempt",
     }
     assert set(get_preset("script_root_validator").allowed_tools or []) == {
+        "search_mission_context",
+        "read_mission_context",
         "query_validation_evidence",
         "deterministic_precheck",
         "submit_validation",
@@ -100,10 +103,11 @@ def test_script_presets_are_exact_and_never_expose_legacy_control_tools() -> Non
 
 def test_new_build_prompt_manifest_freezes_every_phase_one_and_phase_two_role() -> None:
     manifest = script_build_prompt_manifest()
-    assert len(manifest) == 14
+    assert len(manifest) == 15
     assert len({str(item["preset"]) for item in manifest}) == len(manifest)
     assert {str(item["preset"]) for item in manifest} >= {
         "script_planner",
+        "script_context_broker",
         "script_structure_worker",
         "script_paragraph_worker",
         "script_element_set_worker",
@@ -126,6 +130,38 @@ def test_phase_three_prompt_manifest_is_versioned_separately() -> None:
     assert all(str(item["content_sha256"]).startswith("sha256:") for item in manifest)
 
 
+def test_every_model_role_freezes_one_shared_context_access_protocol() -> None:
+    manifest = {
+        str(item["preset"]): str(item["content"])
+        for item in (*script_build_prompt_manifest(), *phase_three_prompt_manifest())
+    }
+    deterministic_or_utility = {
+        "script_context_broker",
+        "script_compose_worker",
+        "script_candidate_portfolio_worker",
+    }
+
+    for preset, content in manifest.items():
+        if preset in deterministic_or_utility:
+            assert not content.startswith(CONTEXT_ACCESS_PROTOCOL)
+        else:
+            assert content.startswith(CONTEXT_ACCESS_PROTOCOL), preset
+            assert content.count("## Context Access Protocol") == 1
+
+    assert "active_has_more" in manifest["script_planner"]
+    for preset in (
+        "script_pattern_retrieval_worker",
+        "script_decode_retrieval_worker",
+        "script_external_retrieval_worker",
+        "script_knowledge_retrieval_worker",
+    ):
+        assert "read_mission_context` 的 cursor" in manifest[preset]
+        assert "不计作第二次" in manifest[preset]
+    assert "Direction、CandidatePortfolio" in manifest["script_root_worker"]
+    assert "required_handles" in manifest["script_candidate_validator"]
+    assert "required_handles" in manifest["script_root_validator"]
+
+
 class _Gateway:
     coordinator = SimpleNamespace(task_store=None)
 
@@ -136,6 +172,9 @@ def test_legacy_retrieval_tool_schemas_keep_old_argument_shapes() -> None:
     schemas = {
         item["function"]["name"]: item["function"]["parameters"] for item in registry.get_schemas()
     }
+    descriptions = {
+        item["function"]["name"]: item["function"]["description"] for item in registry.get_schemas()
+    }
     decode = schemas["search_script_decode_case"]
     assert decode["required"] == ["return_field"]
     assert decode["properties"]["match_fields"]["type"] == "object"
@@ -147,8 +186,13 @@ def test_legacy_retrieval_tool_schemas_keep_old_argument_shapes() -> None:
     assert knowledge["required"] == ["keyword"]
     assert knowledge["properties"]["max_count"]["default"] == 3
     assert schemas["query_pattern_qa"]["required"] == ["message"]
-    assert schemas["load_images"]["required"] == ["image_urls"]
-    assert schemas["load_frozen_strategy"]["required"] == ["strategy_handle"]
+    assert schemas["search_mission_context"]["required"] == ["collection"]
+    assert schemas["read_mission_context"]["required"] == ["handle"]
+    assert "sufficiency" in descriptions["search_mission_context"]
+    assert "exhausted=true" in descriptions["read_mission_context"]
+    assert "load_images" not in schemas
+    assert "load_frozen_strategy" not in schemas
+    assert "read_workbench_detail" not in schemas
     assert schemas["save_script_elements"]["required"] == [
         "elements",
         "links",
@@ -223,6 +267,9 @@ def test_agent_visible_write_schemas_never_expose_mechanical_identity_fields() -
         "read_active_frontier",
         "read_pinned_candidates",
         "read_accepted_artifact",
+        "read_workbench_detail",
+        "load_frozen_strategy",
+        "load_images",
         "create_script_paragraph",
         "create_script_element",
         "update_script_element",
@@ -243,56 +290,37 @@ def test_agent_visible_write_schemas_never_expose_mechanical_identity_fields() -
         assert forbidden_tools.isdisjoint(get_preset(preset_name).allowed_tools or []), preset_name
 
 
-def test_input_snapshot_projection_is_bounded_and_keeps_business_inputs() -> None:
-    persona = [
+def test_model_visible_values_replace_storage_identity_and_uri_fields() -> None:
+    protected = protect_model_value(
         {
-            "列": ("实质", "形式", "作用", "感受")[index % 4],
-            "点名称": f"point-{index}",
-            "人设权重分": str(1 - index / 100),
-            "帖子覆盖率": "0.8",
-            "分类路径": ["/account/style"],
-        }
-        for index in range(87)
-    ]
-    payload = {
-        "snapshot_id": "50",
-        "script_build_id": 900049,
-        "canonical_sha256": "sha256:" + "a" * 64,
-        "account": {"account_name": "account"},
-        "topic": {
-            "topic": {"result": "gym story", "topic_direction": "satirical comic"},
-            "composition_items": [
-                {
-                    "id": index,
-                    "point_type": "关键点",
-                    "dimension": "形式",
-                    "element_name": f"item-{index}",
-                    "note": "useful",
-                    "created_at": "unused metadata",
-                    "is_active": True,
-                }
-                for index in range(30)
-            ],
+            "uri": "script-build://artifact-versions/9",
+            "url": "https://private.example/source",
+            "raw_artifact_ref": "script-build://raw-artifacts/sha256/secret",
+            "artifact_version_id": 9,
+            "branch_id": 7,
+            "strategy_id": 11,
+            "topic_id": 12,
+            "topic_build_id": 13,
+            "execution_id": 14,
+            "content_sha256": "sha256:secret",
+            "goal_id": "goal-allowed",
+            "decision_id": "decision-allowed",
         },
-        "persona_points": persona,
-        "section_patterns": [{"模式名称": "parallel atlas"}],
-        "strategies": [],
-        "prompt_manifest": [{"content": "x" * 100_000}],
-        "model_manifest": {"secret": "x" * 100_000},
-        "datasource_manifest": {"endpoint": "x" * 100_000},
-    }
-    projected = _project_snapshot(payload, view="element-set")
-    encoded = json.dumps(projected, ensure_ascii=False)
+        namespace="surface-test",
+    )
+    encoded = str(protected)
 
-    assert projected["view"] == "element-set"
-    assert projected["persona_points_total"] == 87
-    assert len(projected["persona_points"]) == 12
-    assert projected["section_patterns"] == [{"模式名称": "parallel atlas"}]
-    assert "prompt_manifest" not in projected
-    assert "model_manifest" not in projected
-    assert "datasource_manifest" not in projected
-    assert "created_at" not in projected["topic"]["composition_items"][0]
-    assert len(encoded) < 20_000
+    assert "script-build://" not in encoded
+    assert "https://" not in encoded
+    assert "sha256:" not in encoded
+    assert "artifact_version_id" not in protected
+    assert "branch_id" not in protected
+    assert "strategy_id" not in protected
+    assert "topic_id" not in protected
+    assert "topic_build_id" not in protected
+    assert "execution_id" not in protected
+    assert protected["goal_id"] == "goal-allowed"
+    assert protected["decision_id"] == "decision-allowed"
 
 
 class _Bindings:
@@ -417,89 +445,5 @@ async def test_role_prompt_resolver_uses_task_pinned_snapshot_and_rechecks_diges
         )
 
 
-@pytest.mark.asyncio
-async def test_snapshot_tool_hides_on_demand_body_until_explicit_digest_verified_load() -> None:
-    on_demand_body = "strategy detail " * 1_200
-    snapshot = ScriptBuildInputSnapshotV1(
-        snapshot_id="3",
-        script_build_id=9,
-        execution_id=1,
-        topic_build_id=2,
-        topic_id=3,
-        topic={},
-        account={},
-        persona_points=(),
-        section_patterns=(),
-        strategies=(
-            {
-                "strategy_id": 7,
-                "version": 1,
-                "mode": "always_on",
-                "content": "always body",
-                "content_sha256": "sha256:" + "1" * 64,
-            },
-            {
-                "strategy_id": 8,
-                "version": 2,
-                "mode": "on_demand",
-                "description": "optional",
-                "content": on_demand_body,
-                "content_sha256": canonical_sha256(on_demand_body).wire,
-            },
-        ),
-        prompt_manifest=(),
-        datasource_manifest={},
-        model_manifest={},
-        canonical_sha256="sha256:" + "a" * 64,
-        created_at=datetime.now(UTC),
-    )
-
-    class Snapshots:
-        async def get(self, _snapshot_id: str, *, script_build_id: int):
-            assert script_build_id == 9
-            return snapshot
-
-    task = SimpleNamespace(current_spec=SimpleNamespace(context_refs=["script-build://inputs/3"]))
-    coordinator = SimpleNamespace(
-        task_store=SimpleNamespace(load=lambda _root: _async(SimpleNamespace(tasks={"task": task})))
-    )
-    gateway = LegacyScriptToolGateway(
-        bindings=_Bindings(),
-        snapshots=Snapshots(),
-        artifacts=SimpleNamespace(),
-        coordinator=coordinator,
-    )
-    context = {"root_trace_id": "root-a", "task_id": "task"}
-    summary = await gateway.read_input_snapshot(context)
-    assert summary["strategies"][0]["content"] == "always body"
-    assert "content" not in summary["strategies"][1]
-    selected_handle = strategy_handle(snapshot.snapshot_id, 1, snapshot.strategies[1])
-    loaded = await gateway.load_frozen_strategy(selected_handle, context)
-    assert loaded["content_fragment"] == on_demand_body[:7_000]
-    assert loaded["next_cursor"] == 7_000
-    second = await gateway.load_frozen_strategy(
-        selected_handle, context, cursor=loaded["next_cursor"]
-    )
-    assert second["content_fragment"] == on_demand_body[7_000:14_000]
-    assert "content_sha256" not in loaded["metadata"]
-
-    tampered = replace(
-        snapshot,
-        strategies=(
-            snapshot.strategies[0],
-            {**snapshot.strategies[1], "content": "changed after freeze"},
-        ),
-    )
-
-    class TamperedSnapshots:
-        async def get(self, _snapshot_id: str, *, script_build_id: int):
-            assert script_build_id == 9
-            return tampered
-
-    gateway.snapshots = TamperedSnapshots()
-    with pytest.raises(ProtocolViolation, match="strategy digest"):
-        await gateway.load_frozen_strategy(selected_handle, context)
-
-
 async def _async(value):
     return value