huangruiteng · Oct 28, 2026 · Oct 27, 2026 · Oct 27, 2026 · Oct 27, 2026
@@ -735,8 +735,19 @@ Implementation PRs must name the replaced caller, remaining gap, actual tests
 and rollback. No full language migration or unrelated manager rewrite is a
 prerequisite. No task is complete merely because a new abstraction compiles.
 
-For C1, the first implementation slice pairs the complete field/clause migration
-with a real single-session producer and consumer replacement. Start at the
+C1 is one acceptance target, not one mandatory PR. Independently useful
+prerequisites may land first when existing callers adopt the shared owner, their
+entrypoint contracts are qualified, and rollback is local. For example, move
+exact context-reference composition and recovery into the existing TypeScript
+owner while the current quota and Turn entries keep their schemas. Name the
+remaining cutover consumer; do not add unused infrastructure or a second ledger.
+Separate shared-owner prerequisites, producer/capture cutover, reader recovery
+and retirement, and adoption/full-cost qualification where each boundary is
+independently reviewable. Record dependencies in the existing tasks and PRs.
+Splitting delivery does not close C1 or waive §9's original quota-baseline test.
+
+The first **cutover** slice pairs the complete field/clause migration with a
+real single-session producer and consumer replacement. Start at the
 shared quota emission boundary and trace both attached heartbeat delivery and
 the Turn adapter; retain their actual host ownership. Do not build an unused
 third serializer while both active writers keep their own semantics. Existing

@@ -559,7 +559,15 @@ C1 盘点是可用替换的一部分，不单独建设架构框架。C1 契约
 缺口、实际测试与回退。完整语言迁移或无关 manager 重写不是前置条件；新抽象能编译
 不能算任务完成。
 
-C1 首个实现片必须把完整字段/条款迁移与真实单会话 producer、consumer 替换一起交付。
+C1 是一个验收目标，不要求放进一个 PR。独立有用的前置改动可以先落地，前提是现有
+调用方实际采用共享 owner、入口契约经过验证，且能局部回退。例如，在现有 quota 和
+Turn 入口保持各自 schema 时，将精确上下文引用组合与恢复收敛到既有 TypeScript
+owner。列明后续切换消费者；不能添加无人使用的基础设施或第二套台账。
+共享 owner 前置、producer/capture 切换、reader 恢复与退役、采用及完整成本验收，
+可按独立可审查的边界拆分，在既有任务和 PR 中记录依赖。拆分交付不代表 C1 完成，
+也不豁免第 9 节相对原 quota 基线的验证。
+
+首个**切换**片必须把完整字段/条款迁移与真实单会话 producer、consumer 替换一起交付。
 从共享 quota emission 边界开始，同时追踪 attached heartbeat 交付与 Turn adapter，
 保留各自真实宿主归属。不能新增无人使用的第三个 serializer，而两个活跃 writer 继续
 各自拥有语义。既有 `test_quota_decision_capture.py` 与

@@ -85,6 +85,28 @@ neither enables native tracing nor changes the default-off transport switch.
 控制字段和原始捕获均保留。`--include-detail agent-todos` 保留内联标题。
 这是普通 CLI 序列化去重，差分开关关闭和开启时均适用，不会开启原生追踪或改变默认关闭。
 
+Exact Task-title and Explore-guidance sharing now use the existing TypeScript
+context owner; the Python quota projection delegates both operations in one
+coarse transport call, preserving each explicit detail selector. With both
+selectors enabled, or no context source carrier, it needs no sharing call. Current
+quota and Turn entrypoints retain their schemas. Explore guidance shares only
+identical inline durable-writeback instructions within the same response, and
+stays inline when the reference would be larger. Before relocating context,
+Turn and staged replan readers restore that guidance from its settlement owner.
+Dangling, chained, conflicting or incorrectly typed references reject rather
+than silently losing instructions. Settlement commands, identities and original
+source content remain unchanged. This shared-owner prerequisite does not qualify
+the proposed complete-contract cutover or claim an end-to-end efficiency gain.
+
+精确任务标题与 Explore 指引共享复用既有 TypeScript 上下文 owner，Python quota
+投影通过一次粗粒度调用委托这两项操作，分别保留显式 detail 选择；两项 detail
+均启用或不存在上下文源载体时无需共享调用。当前 quota 与 Turn 入口保持各自
+schema。Explore 仅共享本响应
+内与 durable-writeback 指引完全相同的正文；引用更大时保留内联。Turn 与分阶段
+replan reader 在移动上下文前，从原结算 owner 恢复指引。悬空、链式、冲突或类型
+错误的引用会被拒绝，不静默丢失指令。结算命令、身份与原始源内容保持不变。
+这一共享 owner 前置不代表完整契约切换已通过验收，也不声称端到端效率提升。
+
 When a replan action carries a complete `vision_authoring` schema, the default
 `quota should-run` packet now retains it, alongside its executable writeback
 summary (`required_fields`, accepted path outcomes, and rule). Only when the

@@ -582,58 +582,14 @@ def _deduplicate_next_action_warning(
     return compact
 
 
-def _deduplicate_agent_lane_next_action(
-    item: dict[str, Any], *, payload: dict[str, Any],
-) -> dict[str, Any]:
+def _deduplicate_agent_lane_next_action(item: dict[str, Any]) -> dict[str, Any]:
+    # Display-only omission stays local; exact source sharing runs once below.
     text = item.get("text")
     title = item.get("title")
-    if not isinstance(title, str):
-        return item
-    if isinstance(text, str) and (text == title or text.endswith(f" {title}")):
+    if isinstance(title, str) and isinstance(text, str) and (text == title or text.endswith(f" {title}")):
         compact = dict(item)
         compact.pop("title", None)
         return compact
-
-    # The typed work-context owner has already verified the current full Task.
-    # A bounded lane label may differ from its full title. Reference only an
-    # exact duplicate in this same packet, never a summary or a prior capture.
-    contract = payload.get("interaction_contract")
-    channel = contract.get("agent_channel") if isinstance(contract, dict) else None
-    context = channel.get("work_context") if isinstance(channel, dict) else None
-    selected = payload.get("selected_todo")
-    if not isinstance(context, dict) or not isinstance(selected, dict):
-        return item
-    if (context.get("complete") is not True or context.get("selected_todo_text_verified") is not True
-            or not isinstance(item.get("todo_id"), str) or not item["todo_id"] or "title_ref" in item
-            or selected.get("todo_id") != item["todo_id"]):
-        return item
-    sources = context.get("sources", [])
-    if not isinstance(sources, list):
-        return item
-    matches = [(index, source.get("content", {})) for index, source in enumerate(sources)
-        if isinstance(source, dict) and source.get("source") == "selected_todo"]
-    if len(matches) != 1:
-        return item
-    index, content = matches[0]
-    if not isinstance(content, dict):
-        return item
-    todo = content.get("todo", {})
-    if (not isinstance(todo, dict) or not isinstance(todo.get("text"), str)
-            or content.get("ok") is not True or content.get("matched") is not True
-            or content.get("ambiguous") is True or content.get("read_complete") is False
-            or todo.get("archive_state") == "archive" or todo.get("status") == "done"
-            or todo.get("todo_id") != item["todo_id"]):
-        return item
-    for field in ("text", "title"):
-        if todo.get(field) != title:
-            continue
-        ref = f"$.interaction_contract.agent_channel.work_context.sources[{index}].content.todo.{field}"
-        if len(json.dumps(title, ensure_ascii=False)) <= len(ref) + 8:
-            return item
-        compact = dict(item)
-        compact.pop("title")
-        compact["title_ref"] = ref
-        return compact
     return item
 
 
@@ -810,40 +766,21 @@ def _reference_inline_goal_acceptance(payload: dict[str, Any]) -> dict[str, Any]
     return result
 
 
-def _reference_explore_writeback_guidance(payload: dict[str, Any]) -> dict[str, Any]:
-    """Keep identical optional instructions once in this same CLI packet.
+def _reference_context(
+    payload: dict[str, Any], *, task_title: bool = True, explore_guidance: bool = True,
+) -> dict[str, Any]:
+    """One coarse transport call for the existing typed context owner."""
+    contract = payload.get("interaction_contract")
+    agent = contract.get("agent_channel") if isinstance(contract, dict) else None
+    context = agent.get("work_context") if isinstance(agent, dict) else None
+    # No source carrier means neither operation can act. Leave semantic matching
+    # and reference validation in TypeScript, rather than replicating its rules.
+    if not (task_title or explore_guidance) or not isinstance(context, dict) or not context.get("sources"):
+        return payload
+    from .turn_envelope import _turn_envelope_result
 
-    This is transport-only reference projection, after all decisions. Preserve
-    standalone Explore reads, differing text and the complete settlement owner.
-    """
-    contract = payload.get("interaction_contract", {})
-    agent = contract.get("agent_channel", {})
-    context = agent.get("work_context", {})
-    sources = context.get("sources", [])
-    steps = contract.get("cli_channel", {}).get("settlement_plan", {}).get("ordered_steps", [])
-    for step_index, step in enumerate(steps):
-        if step.get("kind") != "durable_writeback":
-            continue
-        for attachment_index, attachment in enumerate(step.get("optional_attachments", [])):
-            guidance = attachment.get("guidance")
-            if attachment.get("capability_id") != "explore" or not isinstance(guidance, str) or not guidance:
-                continue
-            ref = (f"$.interaction_contract.cli_channel.settlement_plan.ordered_steps[{step_index}]"
-                   f".optional_attachments[{attachment_index}].guidance")
-            updated = list(sources)
-            for index, source in enumerate(sources):
-                content = source.get("content", {})
-                graph = content.get("graph", {})
-                if source.get("kind") != "explore_turn_context" or graph.get("result_writeback_guidance") != guidance:
-                    continue
-                graph = dict(graph)
-                graph.pop("result_writeback_guidance")
-                graph["result_writeback_guidance_ref"] = ref
-                updated[index] = {**source, "content": {**content, "graph": graph}}
-            if updated != sources:
-                return {**payload, "interaction_contract": {**contract, "agent_channel": {
-                    **agent, "work_context": {**context, "sources": updated}}}}
-    return payload
+    return _turn_envelope_result("reference_context", payload=payload,
+        task_title=task_title, explore_guidance=explore_guidance)
 
 
 def compact_quota_should_run_cli_payload(
@@ -926,7 +863,7 @@ def compact_quota_should_run_cli_payload(
         if isinstance(agent_lane_next_action, dict):
             compact = dict(compact)
             compact["agent_lane_next_action"] = _deduplicate_agent_lane_next_action(
-                agent_lane_next_action, payload=compact,
+                agent_lane_next_action,
             )
         goal_route_hint = payload.get("goal_route_hint")
         if isinstance(goal_route_hint, dict):
@@ -941,8 +878,8 @@ def compact_quota_should_run_cli_payload(
         agent_lane_next_action.pop("content_revision", None)
         compact["agent_lane_next_action"] = agent_lane_next_action
     compact = _reference_inline_goal_acceptance(compact)
-    if not include_vision_detail:
-        compact = _reference_explore_writeback_guidance(compact)
+    compact = _reference_context(compact, task_title=not include_todo_summary_detail,
+        explore_guidance=not include_vision_detail)
     return _promote_interaction_contract(
         _promote_runtime_capability_reentry(compact)
     )

@@ -219,6 +219,92 @@ function canonicalHash(value: unknown): string {
   return `sha256:${createHash("sha256").update(JSON.stringify(canonicalValue(value)), "utf8").digest("hex")}`;
 }
 
+function resolveExploreGuidance(payload: JsonObject, context: JsonObject): JsonObject {
+  if (!Array.isArray(context.sources)) return context;
+  const steps = object(object(object(payload.interaction_contract).cli_channel).settlement_plan).ordered_steps;
+  return {...context, sources: context.sources.map(value => {
+    const source = object(value), content = object(source.content), graph = object(content.graph);
+    if (source.kind !== "explore_turn_context" || !Object.hasOwn(graph, "result_writeback_guidance_ref")) return value;
+    const reference = graph.result_writeback_guidance_ref;
+    const match = typeof reference === "string"
+      ? /^\$\.interaction_contract\.cli_channel\.settlement_plan\.ordered_steps\[(0|[1-9]\d*)\]\.optional_attachments\[(0|[1-9]\d*)\]\.guidance$/.exec(reference) : null;
+    const step = object(match && Array.isArray(steps) ? steps[Number(match[1])] : null);
+    const attachment = object(match && Array.isArray(step.optional_attachments)
+      ? step.optional_attachments[Number(match[2])] : null);
+    if (Object.hasOwn(graph, "result_writeback_guidance") || step.kind !== "durable_writeback" ||
+        attachment.capability_id !== "explore" || Object.hasOwn(attachment, "guidance_ref") ||
+        typeof attachment.guidance !== "string" || !attachment.guidance) {
+      throw new EffectRuntimeRequestError("Explore guidance reference must resolve to one inline durable-writeback Explore attachment");
+    }
+    delete graph.result_writeback_guidance_ref;
+    return {...source, content: {...content, graph: {...graph, result_writeback_guidance: attachment.guidance}}};
+  })};
+}
+
+function referenceExploreGuidance(payload: JsonObject): JsonObject {
+  const interaction = object(payload.interaction_contract), agent = object(interaction.agent_channel);
+  const context = object(agent.work_context);
+  resolveExploreGuidance(payload, context); // Validate existing references, including idempotent reads.
+  if (!Array.isArray(context.sources)) return payload;
+  const steps = object(object(interaction.cli_channel).settlement_plan).ordered_steps;
+  if (!Array.isArray(steps)) return payload;
+  const owners = new Map<string, string>();
+  steps.forEach((value, stepIndex) => {
+    const step = object(value);
+    if (step.kind !== "durable_writeback" || !Array.isArray(step.optional_attachments)) return;
+    step.optional_attachments.forEach((value, attachmentIndex) => {
+      const attachment = object(value), guidance = attachment.guidance;
+      if (attachment.capability_id !== "explore" || Object.hasOwn(attachment, "guidance_ref") ||
+          typeof guidance !== "string" || !guidance || owners.has(guidance)) return;
+      owners.set(guidance, `$.interaction_contract.cli_channel.settlement_plan.ordered_steps[${stepIndex}].optional_attachments[${attachmentIndex}].guidance`);
+    });
+  });
+  const sources = context.sources.map(value => {
+    const source = object(value), content = object(source.content), graph = object(content.graph);
+    const guidance = graph.result_writeback_guidance;
+    const reference = typeof guidance === "string" ? owners.get(guidance) : undefined;
+    if (source.kind !== "explore_turn_context" || typeof guidance !== "string" || !reference ||
+        Buffer.byteLength(JSON.stringify(guidance), "utf8") <= Buffer.byteLength(reference, "utf8") + 8) return value;
+    delete graph.result_writeback_guidance;
+    return {...source, content: {...content, graph: {...graph, result_writeback_guidance_ref: reference}}};
+  });
+  return {...payload, interaction_contract: {...interaction, agent_channel: {...agent,
+    work_context: {...context, sources}}}};
+}
+
+function currentTaskTitleOwner(payload: JsonObject): {index: number; todo: JsonObject} | null {
+  const lane = object(payload.agent_lane_next_action), selected = object(payload.selected_todo);
+  const context = object(object(object(payload.interaction_contract).agent_channel).work_context);
+  if (context.complete !== true || context.selected_todo_text_verified !== true ||
+      typeof lane.todo_id !== "string" || !lane.todo_id || selected.todo_id !== lane.todo_id ||
+      !Array.isArray(context.sources)) return null;
+  const matches = context.sources.map((value, index) => ({source: object(value), index}))
+    .filter(({source}) => source.source === "selected_todo");
+  if (matches.length !== 1) return null;
+  const {source, index} = matches[0], content = object(source.content), todo = object(content.todo);
+  if (content.ok !== true || content.matched !== true || content.ambiguous === true ||
+      content.read_complete === false || typeof todo.text !== "string" ||
+      todo.archive_state === "archive" || todo.status === "done" || todo.todo_id !== lane.todo_id) return null;
+  return {index, todo};
+}
+
+function referenceCurrentTaskTitle(payload: JsonObject): JsonObject {
+  const lane = object(payload.agent_lane_next_action), title = lane.title;
+  // Preserve historical caller-owned references; complete readers validate them
+  // separately. Display-only omission of a title already in a label stays local.
+  if (typeof title !== "string" || Object.hasOwn(lane, "title_ref")) return payload;
+  const owner = currentTaskTitleOwner(payload);
+  if (!owner) return payload;
+  for (const field of ["text", "title"]) {
+    if (owner.todo[field] !== title || Object.hasOwn(owner.todo, `${field}_ref`)) continue;
+    const reference = `$.interaction_contract.agent_channel.work_context.sources[${owner.index}].content.todo.${field}`;
+    if (Array.from(JSON.stringify(title)).length <= reference.length + 8) return payload;
+    delete lane.title;
+    return {...payload, agent_lane_next_action: {...lane, title_ref: reference}};
+  }
+  return payload;
+}
+
 function compactFields(
   value: unknown,
   fields: readonly string[],
@@ -355,6 +441,7 @@ function replanGoalContext(payload: JsonObject, reads: JsonObject | null, coreRe
  * Different revisions, other source fields and unavailable context stay intact.
  */
 function replanWorkContext(payload: JsonObject, context: JsonObject): JsonObject {
+  context = resolveExploreGuidance(payload, context);
   const reads = replanReadCommands(payload);
   const core = object(object(object(payload.autonomous_replan_obligation).replan_context).core_goal);
   const acceptance = object(core.acceptance_contract);
@@ -1272,6 +1359,12 @@ export function evaluateTurnEnvelope(value: unknown): JsonObject {
     assertTurnEnvelopeConsumerVersion(requireJsonObject(request.envelope, "turn_envelope"));
     return {ok: true};
   }
+  if (operation === "reference_context") {
+    let payload = requireJsonObject(request.payload, "quota decision");
+    if (request.task_title === true) payload = referenceCurrentTaskTitle(payload);
+    if (request.explore_guidance === true) payload = referenceExploreGuidance(payload);
+    return payload;
+  }
   if (operation === "build") return buildTurnEnvelope(request);
   if (operation === "context_delivery") return prepareContextDelivery(request, LEGACY_QUOTA_CONTEXT_FIELDS);
   if (operation === "native_context_lifecycle") return updateNativeContextLifecycle(request);