fix(context): keep shared acceptance references resolvable

docs/architecture/rfcs/controller-executor-unified-contract-v0.md

@@ -874,6 +874,27 @@ information and failure consequences.
 | Full/delta/resume delivery | Reuse the existing continuity owner after composition. Establish exact semantic reconstruction and required material availability before action; charge cold fallback and lost-baseline recovery. |
 | Scheduler, admission and settlement identities | One authoritative instance before capture; transport must preserve host ownership, original Turn and once-only effects. Repeated representations do not justify weakening their checks. |
 
+A standalone late move of acceptance sharing is not automatically a useful
+prerequisite. A controlled source-level substitution against main
+`99cfa3ac559bfa237f04f36fb7e926672fea44b2` moved only that rule from the local
+CLI compactor into the existing `reference_context` call. On the same synthetic
+fixture with 100 distinct requirements, serialized projection input grew from
+102,680 to 344,010 bytes; the 102,189-byte caller-visible output stayed equal.
+The ordinary path still made one projection call; all-details delivery changed
+from local sharing to one call. These are projection-parameter measurements,
+not framed traffic, latency, model billing or the integrated C1 comparison.
+Retain this negative result rather than declaring a net efficiency win.
+
+Repair relocated acceptance aliases at the current compatibility owner without
+that extra transport. Existing aliases must follow only inline owners actually
+moved by the projection, including aliases visited before their owner. Preserve
+complete values, original numeric types, carrier order and existing size checks;
+never infer arbitrary reference chains or rewrite saved captures. The complete
+C1 composition must absorb this rule at its source-owning boundary and retire
+the Python compatibility algorithm with its callers, rather than repeatedly
+shipping all acceptance copies across another bridge. Writer/consumer cutover
+and the original-quota outcome/full-cost gate remain open.
+
 A fixed-source synthetic replan audit of candidate `895f38fc5` illustrates why
 the boundary matters. With stride off and no capture staging, the raw source was
 21,461 compact UTF-8 JSON bytes, the original quota CLI projection 20,543, and

docs/architecture/rfcs/controller-executor-unified-contract-v0.zh-CN.md

@@ -662,6 +662,21 @@ compactor：精确共享、省略和延后读取的信息与失败后果各不
 | Full/delta/resume 交付 | 在 composition 后复用既有 continuity owner。行动前验证精确语义重建与必要材料可用；计入冷回退及丢失基线的恢复。 |
 | Scheduler、准入与结算身份 | Capture 前只有一份权威实例；传输保留真实 host owner、原 Turn 与一次性 effects。重复表示不构成削弱检查的理由。 |
 
+单独后移验收共享不自动构成有价值的前置切片。以主线
+`99cfa3ac559bfa237f04f36fb7e926672fea44b2` 为基线的 source-level 受控替换，
+只把该规则从本地 CLI compactor 移到已有 `reference_context` 调用。同一份含 100 条
+不同要求的合成 fixture 中，序列化投影输入从 102,680 增至 344,010 字节，调用者可见
+输出仍为相同的 102,189 字节。普通路径仍调用一次；全部 detail 开启时由本地共享变为
+一次请求。这些是投影参数测量，不是带协议包装的流量、延迟、模型计费或完整 C1 对照。
+保留此负结果，不能宣称净效率收益。
+
+先在当前兼容 owner 修复被迁移的验收引用，不引入上述额外传输。已有别名只跟随本次
+投影实际迁移的内联 owner，包括先访问别名、后迁移 owner 的情况。保留完整值、原始
+数值类型、carrier 顺序和现有大小判断；不推断任意引用链，不重写已存 capture。
+完整 C1 composition 应在持有来源的边界吸收该规则，并随调用方退役 Python 兼容
+算法，不能把所有验收副本反复传过另一条 bridge。Writer／consumer 切换和原 quota
+成果／完整成本门继续保持开放。
+
 候选 `895f38fc5` 的固定 source 合成 replan 审计说明了边界的影响。关闭 stride 且不做
 capture 分阶段时，原始 source 为 21,461 个紧凑 UTF-8 JSON 字节，原 quota CLI
 projection 为 20,543，完整候选为 21,738：**相对 CLI 增加 5.82%**，相对原始 source

docs/reference/required-work-context.md

@@ -34,6 +34,11 @@ relations and authority revision without repeating the body. If no other
 canonical fields need a source record, `work_context.selected_todo_authority`
 carries the source and revision. When the view is bounded, the full canonical
 detail remains in `work_context.sources`.
+The ordinary CLI shares exact acceptance copies within one response. Existing
+references follow their complete inline owner when that projection relocates it,
+including an owner visited later. Different revisions, types, failures and
+sources stay distinct. This adds no source read or authority, admission or
+settlement change; saved captures retain their original contents.
 Registered Goal state uses the existing writer's strict generated-region
 ownership. When every recognized Todo region is explicitly marked, the source
 adapter delivers **all other text verbatim** in `work_context.sources`: Goal
@@ -579,6 +584,9 @@ all-Goal adoption, actual solver costs or outcome benefit.
 
 普通 Heartbeat/quota 默认在 `agent_channel.work_context` 返回完整选定 Todo、
 启用的验收与当前 Agent 相关的 User Todo，附带来源和修订，不需要开启 TurnEnvelope。
+普通 CLI 在同一响应内共享完全相同的验收副本。已有引用跟随本次投影迁移的完整
+内联 owner，包括后访问的 owner；不同修订、类型、失败和来源分别保留。不增加来源
+读取，不改变权限、准入或结算；已存 capture 保留原始内容。
 Goal Markdown 复用原 writer 对生成区域的严格归属规则：所有已识别 Todo 区域均有
 显式标记时，`work_context.sources` 原样携带区域外的全部文字，保留目标、frontmatter、
 任意章节、注释、历史和末尾停止条件。`excluded_todo_regions` 保留每段的角色、行区间

loopx/control_plane/quota/cli_projection.py

@@ -2,7 +2,7 @@
 
 import json
 from copy import deepcopy
-from typing import Any
+from typing import Any, cast
 
 from .decision_summary import compact_quota_decision
 from .should_run_packet import reference_duplicate_fallback_guidance
@@ -504,7 +504,21 @@ def _reference_inline_goal_acceptance(payload: dict[str, Any]) -> dict[str, Any]
         ("agent_todo_summary", "goal_acceptance_contract"),
     ])
     carriers: dict[str, str] = {}
+    relocated: dict[str, str] = {}
+    references: list[tuple[tuple[str | int, ...], str]] = []
     result = payload
+
+    def writable_parent(path: tuple[str | int, ...]) -> dict[str, Any]:
+        nonlocal result
+        # Copy only the affected spine; never mutate caller-owned facts.
+        result = dict(result)
+        target: Any = result
+        for part in path[:-1]:
+            child = target[part]
+            target[part] = list(child) if isinstance(child, list) else dict(child)
+            target = target[part]
+        return cast(dict[str, Any], target)
+
     for path in paths:
         parent: Any = result
         for part in path[:-1]:
@@ -516,28 +530,33 @@ def _reference_inline_goal_acceptance(payload: dict[str, Any]) -> dict[str, Any]
                 parent = None
                 break
         field = str(path[-1])
+        existing = parent.get(field + "_ref") if isinstance(parent, dict) else None
+        if isinstance(existing, str):
+            references.append((path, existing))
         value = parent.get(field) if isinstance(parent, dict) else None
         if not isinstance(value, dict) or not value or field + "_ref" in parent:
             continue
         # Python equality conflates True/1 and False/0. Compare JSON types and
         # the complete content, including revision, scope, failures and sources.
         encoded = json.dumps(value, sort_keys=True, ensure_ascii=False, allow_nan=False)
         reference = carriers.get(encoded)
+        own_ref = "$" + "".join(
+            f"[{part}]" if isinstance(part, int) else f".{part}" for part in path)
         if reference is None:
-            carriers[encoded] = "$" + "".join(
-                f"[{part}]" if isinstance(part, int) else f".{part}" for part in path)
+            carriers[encoded] = own_ref
             continue
         if len(encoded) <= len(reference) + 8:
             continue
-        # Copy only the affected spine, so callers retain their original facts.
-        result = dict(result)
-        target: Any = result
-        for part in path[:-1]:
-            child = target[part]
-            target[part] = list(child) if isinstance(child, list) else dict(child)
-            target = target[part]
+        target = writable_parent(path)
         del target[field]
         target[field + "_ref"] = reference
+        relocated[own_ref] = reference
+    # Retarget only references to owners actually moved by this projection.
+    # A later owner can move after an earlier alias was visited; a second pass
+    # keeps both directions direct without following arbitrary input chains.
+    for path, reference in references:
+        if reference in relocated:
+            writable_parent(path)[str(path[-1]) + "_ref"] = relocated[reference]
     return result
 
 

loopx/semantics/inventory.py

@@ -116,7 +116,7 @@ def _assigned_name(node: ast.AST) -> tuple[str, ast.AST] | None:
     return None
 
 
-def parse_python(source: SourceFile) -> ast.AST:
+def parse_python(source: SourceFile) -> ast.Module:
     """Parse one tracked Python source, rejecting an unparseable one by path.
 
     Deliberately not cached. Caching the trees would hold roughly two million

loopx/semantics/python_production.py

@@ -9,8 +9,9 @@
 import ast
 from collections import Counter
 from dataclasses import dataclass
+from functools import lru_cache
 from types import SimpleNamespace
-from typing import Mapping, TypeVar
+from typing import Mapping, TypeVar, cast
 
 from .inventory import SourceFile
 
@@ -137,20 +138,29 @@ def enum_members(source: SourceFile, symbol: str, *, strict: bool = False) -> di
     return result
 
 
+@lru_cache(maxsize=4096)
+def _field_mentions(source: SourceFile) -> frozenset[str]:
+    """Reuse parsed names, not another retained AST or a source-text filter."""
+    names: set[str] = set()
+    for node in ast.walk(_parsed(source)):
+        if isinstance(node, ast.Name):
+            names.add(node.id)
+        elif isinstance(node, ast.Attribute):
+            names.add(node.attr)
+        elif isinstance(node, ast.Constant) and isinstance(node.value, str):
+            names.add(node.value)
+        elif isinstance(node, ast.keyword) and node.arg is not None:
+            names.add(node.arg)
+    return frozenset(names)
+
+
 def python_literal_uses(source: SourceFile, field: str) -> set[str]:
     """Literal writes and direct dispatch operands; not alias/data-flow proof."""
-    tree = _parsed(source)
-    nodes = list(ast.walk(tree))
-    # Inspect parsed names/keys, including escaped strings and normalized
-    # identifiers. Parse errors must still fail even when the field is absent.
-    if not any(
-        isinstance(node, ast.Name) and node.id == field
-        or isinstance(node, ast.Attribute) and node.attr == field
-        or isinstance(node, ast.Constant) and node.value == field
-        or isinstance(node, ast.keyword) and node.arg == field
-        for node in nodes
-    ):
+    # Parsing precedes absence: escaped keys and normalized identifiers count,
+    # and invalid syntax must fail even without a textual field mention.
+    if field not in _field_mentions(source):
         return set()
+    tree = _parsed(source)
 
     def reads(node: ast.AST) -> bool:
         if isinstance(node, ast.Name):
@@ -188,7 +198,7 @@ def literals(node: ast.AST) -> set[str]:
     # conditions and does not inspect strings containing sample source text.
     rows = scan_python_production(source, field=field, enums={})
     result = set().union(*(row.values for row in rows))
-    for node in nodes:
+    for node in ast.walk(tree):
         if isinstance(node, ast.Compare):
             for left, operator, right in zip(
                 [node.left, *node.comparators[:-1]], node.ops, node.comparators, strict=True,
@@ -222,9 +232,9 @@ def _qualified_bindings(source: SourceFile, tree: ast.Module, owners: Mapping[st
                 if value is None and modules is not None and alias.name in symbols:
                     # One unrenamed re-export hop through a tracked module binds the
                     # same owner; only names that are owner symbols are followed.
-                    target = _tracked_module(module, modules)
-                    if target is not None and target.path != source.path:
-                        value = _reexported(target, alias.name, imports)
+                    reexport_source = _tracked_module(module, modules)
+                    if reexport_source is not None and reexport_source.path != source.path:
+                        value = _reexported(reexport_source, alias.name, imports)
                 if value is not None:
                     bindings[name] = value
                 else:
@@ -323,11 +333,11 @@ def _module_functions(source: SourceFile, tree: ast.Module) -> dict[str, ast.Fun
     # name some scope declares ``global`` and then stores is no longer reliably
     # the ``def`` above; binding a call to that ``def`` would report the original
     # body's returns for a function the module can swap at runtime.
-    for node in ast.walk(tree):
-        if not isinstance(node, ast.Global):
+    for global_node in ast.walk(tree):
+        if not isinstance(global_node, ast.Global):
             continue
-        scope = _enclosing_scope(tree, node)
-        for name in node.names:
+        scope = _enclosing_scope(tree, global_node)
+        for name in global_node.names:
             if name in defined and _stores_name(scope, name):
                 bound[name] += 1
     functions = {name: node for name, node in defined.items()
@@ -338,14 +348,14 @@ def _module_functions(source: SourceFile, tree: ast.Module) -> dict[str, ast.Fun
 
 def _index_value(node: ast.AST) -> str | int | None:
     if isinstance(node, ast.Constant) and type(node.value) in (str, int):
-        return node.value
+        return cast(str | int, node.value)
     if (isinstance(node, ast.UnaryOp) and isinstance(node.op, ast.USub)
             and isinstance(node.operand, ast.Constant) and type(node.operand.value) is int):
         return -node.operand.value
     return None
 
 
-def _union(values: list[ast.AST]) -> ast.AST:
+def _union(values: list[ast.expr]) -> ast.expr:
     """Fold several definitions of one local into a finite selection node.
 
     The scan reports syntactic result possibilities, so a name written more than
@@ -407,6 +417,12 @@ def scan_python_production(
     call_arguments = call_arguments or {}
     calls = _qualified_bindings(source, tree, call_arguments, modules)
     return_paths = return_paths or {}
+    # Every output form needs a field, registered return or qualified owner.
+    # Keep enum results and builder arguments even when the field is absent.
+    if not bindings and not calls and not return_functions and (
+        not field or field not in _field_mentions(source)
+    ):
+        return []
     module_functions = _module_functions(source, tree)
     call_memo: dict[tuple[str, str], tuple[frozenset[str], tuple[str, ...]]] = {}
     environments: dict[tuple[int, str], SimpleNamespace] = {}
@@ -481,7 +497,7 @@ def collect(node: ast.AST, loops: frozenset[int] = frozenset()) -> None:
         if scope == '<module>':
             nested_names -= local_owner_names | {owner.split('::')[1] for owner in call_arguments
                                                    if owner.split('::')[0] == source.path}
-        import_bound = set()
+        import_bound: set[str] = set()
         for node in nodes:
             if isinstance(node, (ast.Import, ast.ImportFrom)):
                 import_bound.update(alias.asname or alias.name.split('.')[0] for alias in node.names)
@@ -510,7 +526,7 @@ def collect(node: ast.AST, loops: frozenset[int] = frozenset()) -> None:
         local_bindings = {k: v for k, v in bindings.items() if k not in shadows}
         local_calls = {k: v for k, v in calls.items() if k not in shadows}
 
-        plain: dict[str, list[ast.AST]] = {}
+        plain: dict[str, list[ast.expr]] = {}
         for node in nodes:
             target = value = None
             if (isinstance(node, ast.Assign) and len(node.targets) == 1
@@ -520,7 +536,7 @@ def collect(node: ast.AST, loops: frozenset[int] = frozenset()) -> None:
                     and node.value is not None):
                 target, value = node.target.id, node.value
             if target is not None:
-                plain.setdefault(target, []).append(value)
+                plain.setdefault(target, []).append(cast(ast.expr, value))
         declared = {name for node in nodes if isinstance(node, (ast.Global, ast.Nonlocal))
                     for name in node.names}
         # Every store of the name must be one of those plain writes, so a value
@@ -772,7 +788,7 @@ def same_module_call(node: ast.Call, mode: str) -> tuple[set[str], bool] | None:
                 blocked(reason)
             return set(values), bool(reasons)
 
-        def enum_object_value(node: ast.AST, seen: frozenset[str]) -> tuple[set[str], bool]:
+        def enum_object_value(node: ast.AST | None, seen: frozenset[str]) -> tuple[set[str], bool]:
             node, seen = bound(node, seen)
             if isinstance(node, ast.IfExp):
                 parts = [enum_object_value(value, seen) for value in (node.body, node.orelse)]
@@ -855,7 +871,7 @@ def scan_scope(body: list[ast.stmt], scope: str, parameters: set[str],
         env = env if env is not None else environment(body, scope, parameters, call_shadows, own)
         blockers = env.blockers
 
-        def record(node: ast.AST | None, form: str, location: ast.AST) -> None:
+        def record(node: ast.AST | None, form: str, location: ast.expr | ast.stmt) -> None:
             # Reasons are read back by position: one scope's environment is
             # shared with same-module call resolution, which may nest inside.
             start = len(blockers)