fix(context): omit unused Goal body from internal projection transport

docs/reference/required-work-context.md

@@ -21,6 +21,13 @@ not automatically copy that whole document into every response or infer Goal
 intent by trimming headings. Read it before dependent work. Enabled scoped
 acceptance does not replace the original Goal intent.
 
+The Goal availability adapter still reads and UTF-8-validates the complete file
+and hashes all bytes for its revision. Only its path and revision cross the
+internal context-projection bridge; the typed owner does not consume the file
+body there. Missing files and invalid UTF-8 still block dependent delivery.
+Removing that unused internal copy does not shorten the Agent's required Goal
+read or establish model-token, latency or outcome gains.
+
 Already delivered sources disappear from `required_reads`; unresolved provider
 reads retain exact commands and ordering. The typed interaction owner owns this
 selection and fulfillment policy in both ordinary packets and TurnEnvelope.
@@ -105,6 +112,10 @@ transport flag does not restore the old exact Todo schema.
 混合了其他 Todo 和历史证据的 Goal Markdown 文件保留为 `required_reads` 中的全文
 读取命令：quota 核验来源可用性，不每轮自动灌入整个文件，也不按标题猜测哪些文字
 可以替代原始 Goal。依赖交付前仍须读取；局部验收不能替代整体目标。
+Goal 可用性 adapter 仍完整读取文件、校验 UTF-8，并以全部字节计算修订摘要；内部
+context projection 桥接只传路径与修订，typed owner 不在这里消费文件正文。
+文件缺失或非法 UTF-8 仍阻止依赖交付。移除内部未使用副本不会缩短 Agent 的 Goal
+必读范围，也不代表已取得模型 token、延迟或任务效果收益。
 已附带的内容不要求重复读取；未提供的 provider 内容继续保留精确命令与顺序。
 `work_context.complete` 只表示未发现来源失败，未完成的必读义务仍然有效。
 偏好只在对应作用域有记录时出现；Explore 和偏好沿用各自 hook 的真实 adapter，

loopx/control_plane/work_items/context_readback.py

@@ -25,8 +25,10 @@ def _source_content(read, *, registry_path, runtime_root, goal_id, todo_id):
             registry=load_registry(registry_path), goal_id=goal_id,
             project_override=None, state_file_override=None)
         source_bytes = state_file.read_bytes()
-        text = source_bytes.decode("utf-8")
-        return {"state_file": str(state_file), "text": text,
+        # Validate the entire source, including its tail. The typed owner keeps
+        # Goal consumption as a pending read and does not consume this body.
+        source_bytes.decode("utf-8")
+        return {"state_file": str(state_file),
             "revision": "sha256:" + sha256(source_bytes).hexdigest()}
     if source == "goal_acceptance":
         # The typed acceptance owner scopes task-binding observations while

tests/control_plane/test_goal_state_context_transport.py

@@ -0,0 +1,115 @@
+"""Full Goal availability checks do not transport an unconsumed source body."""
+from __future__ import annotations
+
+import contextlib
+from copy import deepcopy
+from hashlib import sha256
+import io
+import json
+
+import pytest
+
+from canonical_authority_fixture import initialize_canonical_authority, isolate_sqlite_runtime
+from loopx.cli import main
+from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection
+from loopx.control_plane.effect_runtime import MAX_REQUEST_BYTES
+from loopx.control_plane.testing.canary_harness import write_fixture_registry
+from loopx.control_plane.work_items import context_readback
+
+
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+def test_goal_availability_transport_preserves_full_read_and_recovery(tmp_path, monkeypatch, provider):
+    isolate_sqlite_runtime(tmp_path, monkeypatch)
+    runtime, registry, state = tmp_path / "runtime", tmp_path / "registry.json", tmp_path / "state.md"
+    goal, todo_id = "goal-context", "todo_export"
+    initial = b"---\nstatus: active\n---\n# Goal\n## Objective\nPreserve the export contract.\n"
+    state.write_bytes(initial)
+    write_fixture_registry(project=tmp_path, runtime_root=runtime, registry_path=registry,
+        goal_id=goal, domain="software", adapter_kind="generic_project_goal_v0",
+        state_file=str(state), registered_agents=["agent-a"], quota_allowed_slots=None)
+    todo = {"schema_version": "todo_item_v0", "todo_id": todo_id, "role": "agent",
+        "text": "Preserve every export row and cancellation. Stop before publication.",
+        "status": "open", "done": False, "task_class": "advancement_task", "action_kind": "implement",
+        "index": 1, "source_section": "Agent Todo", "archive_state": "active", "claimed_by": "agent-a"}
+    initialize_canonical_authority(runtime, goal,
+        build_todo_runtime_shadow_projection(goal_id=goal, todos=[todo], handoff_mode="soft_claim"),
+        state_path=state, provider=provider)
+    source_args = dict(registry_path=registry, runtime_root=runtime, goal_id=goal, todo_id=todo_id)
+    original_effect = context_readback.effect_runtime_result
+    requests, plans = [], []
+
+    def observe(operation, request, **kwargs):
+        if operation == "work_item.context.plan":
+            plans.append(deepcopy(request["packet"]))
+        if operation == "work_item.context.project":
+            requests.append(deepcopy(request))
+        return original_effect(operation, request, **kwargs)
+
+    monkeypatch.setattr(context_readback, "effect_runtime_result", observe)
+
+    def guard():
+        output = io.StringIO()
+        with contextlib.redirect_stdout(output):
+            code = main(["--registry", str(registry), "--runtime-root", str(runtime), "--format", "json",
+                "quota", "should-run", "--goal-id", goal, "--agent-id", "agent-a"])
+        packet = json.loads(output.getvalue())
+        assert code == 0 and packet["should_run"] is True, packet.get("reason")
+        return packet["interaction_contract"]["agent_channel"]
+
+    small_channel = guard()
+    small_request = requests[-1]
+    large = initial + b"\n## Evidence\n" + b"Prior evidence remains available. " * 100_000
+    large += "\n## Stop Conditions\n未经授权不得发布。\n".encode()
+    assert len(large) > MAX_REQUEST_BYTES
+    state.write_bytes(large)
+    large_channel = guard()
+    large_request = requests[-1]
+    captured_plan = plans[-1]
+    assert state.read_bytes() == large
+    # Source growth changes neither the complete Agent contract nor transport
+    # size. The full Goal read remains pending; availability is not consumption.
+    assert small_channel == large_channel
+    assert large_channel["delivery_allowed"] is True
+    assert large_channel["work_context"]["complete"] is True
+    assert large_channel["work_context"]["selected_todo_text_verified"] is True
+    assert any(row.get("source") == "goal_state" for row in large_channel["required_reads"])
+    assert not any(row.get("source") == "goal_state" for row in large_channel["work_context"]["sources"])
+    assert len(json.dumps(small_request).encode()) == len(json.dumps(large_request).encode())
+    content = context_readback._source_content({"source": "goal_state"}, **source_args)
+    assert content == {"state_file": str(state), "revision": "sha256:" + sha256(large).hexdigest()}
+    # Independently compare the current typed projector with the former full
+    # source input. Removing the transport must not weaken its obligations.
+    legacy_request = deepcopy(large_request)
+    source = next(row for row in legacy_request["source_results"]
+        if row.get("content", {}).get("state_file") == str(state))
+    source["content"]["text"] = large.decode()
+    assert original_effect("work_item.context.project", legacy_request, large_local_snapshot=True) == \
+        original_effect("work_item.context.project", large_request, large_local_snapshot=True)
+
+    changed = large + b"\nAdditional stop at the final byte.\n"
+    state.write_bytes(changed)
+    revised = context_readback._source_content({"source": "goal_state"}, **source_args)
+    assert revised["revision"] == "sha256:" + sha256(changed).hexdigest()
+    assert revised["revision"] != content["revision"]
+
+    # Fault the disposable source after admission: exercise actual file reads
+    # and the real TS context projection, not a mocked failure result. Then
+    # recover through the complete native CLI guard.
+    for fault in ("invalid_utf8_tail", "missing"):
+        if fault == "missing":
+            state.unlink()
+            error = FileNotFoundError
+        else:
+            state.write_bytes(large + b"\xff")
+            error = UnicodeDecodeError
+        with pytest.raises(error):
+            context_readback._source_content({"source": "goal_state"}, **source_args)
+        failed = deepcopy(captured_plan)
+        context_readback.attach_work_context(failed, registry_path=registry,
+            runtime_root=runtime, hook_dispatch=None)
+        channel = failed["interaction_contract"]["agent_channel"]
+        assert channel["delivery_allowed"] is False
+        assert channel["work_context"]["complete"] is False
+        assert any(row.get("source") == "goal_state" for row in channel["required_reads"])
+        state.write_bytes(large)
+        assert guard() == large_channel