Preserve captured numbers in staged replan reads

loopx/control_plane/quota/cli_projection.py

@@ -1,11 +1,10 @@
 from __future__ import annotations
 
-import json
-from copy import deepcopy
 from typing import Any, cast
 
 from .decision_summary import compact_quota_decision
 from .should_run_packet import reference_duplicate_fallback_guidance
+from .turn_envelope import _apply_context_projection_changes
 
 QUOTA_CLI_TODO_SUMMARY_DETAIL_COMMAND = (
     "quota should-run --include-detail agent-todos"
@@ -464,42 +463,6 @@ def contains_contract(value: Any, *path: str) -> bool:
     return _apply_context_projection_changes(payload, result)
 
 
-def _apply_context_projection_changes(
-    source: dict[str, Any], result: dict[str, Any],
-) -> dict[str, Any]:
-    """Apply typed display edits to the original Python facts, without policy."""
-    if set(result) != {"changes"} or not isinstance(result["changes"], list):
-        raise RuntimeError("TypeScript context projection must return changes")
-    # JSON transport has no aliasing. Keep Python's exact numeric values while
-    # separating repeated caller-owned objects before applying path edits.
-    projected = json.loads(json.dumps(source, allow_nan=False))
-    try:
-        for change in result["changes"]:
-            path = change["path"]
-            if not isinstance(path, list) or not path:
-                raise ValueError("projection path must be nonempty")
-            parent: Any = projected
-            for key in path[:-1]:
-                parent = parent[key]
-            key = path[-1]
-            if "remove" in change:
-                del parent[key]
-                continue
-            if "from" in change:
-                value: Any = source
-                for part in change["from"]:
-                    value = value[part]
-            else:
-                value = change["value"]
-            if isinstance(parent, list) and key == len(parent):
-                parent.append(deepcopy(value))
-            else:
-                parent[key] = deepcopy(value)
-    except (KeyError, IndexError, TypeError, ValueError) as error:
-        raise RuntimeError("Invalid TypeScript context projection changes") from error
-    return cast(dict[str, Any], projected)
-
-
 def compact_refresh_state_cli_payload(payload: dict[str, Any]) -> dict[str, Any]:
     """Share only repeated successful Explore input; full output is the fallback."""
     if (payload.get("ok") is not True or not isinstance(payload.get("explore_result"), dict)

loopx/control_plane/quota/diagnostic_projection.ts

@@ -51,8 +51,28 @@ export function contextSourceIdentity(source: JsonObject, supplied?: unknown):
  * moved into source_projection. Python applies these edits without policy. */
 export type ProjectionSourceValues = WeakMap<object, Map<string, ProjectionPath>>;
 
+/** Retain scalar lineage when a display lens copies or renames source fields. */
+export function contextProjectionRetainFields(source: JsonObject, values: ProjectionSourceValues) {
+  const paths = contextObjectPaths(source);
+  return (projected: JsonObject, original: JsonObject,
+      fields: readonly (string | readonly [string, string])[] = Object.keys(projected)): JsonObject => {
+    const path = paths.get(original), originals = values.get(original);
+    if (!path && !originals) return projected;
+    const retained = values.get(projected) ?? new Map<string, ProjectionPath>();
+    for (const field of fields) {
+      const [from, to] = typeof field === "string" ? [field, field] : field;
+      if (!Object.hasOwn(projected, to) || typeof projected[to] !== "number" ||
+          !Object.hasOwn(original, from) || typeof original[from] !== "number") continue;
+      const origin = originals?.get(from) ?? (path && [...path, from]);
+      if (origin) retained.set(to, origin);
+    }
+    if (retained.size) values.set(projected, retained);
+    return projected;
+  };
+}
+
 export function contextProjectionChanges(source: JsonObject, projected: JsonObject,
-    sourceValues?: ProjectionSourceValues): JsonObject {
+    sourceValues?: ProjectionSourceValues, base: JsonObject = source): JsonObject {
   const paths = contextObjectPaths(source), changes: ProjectionChange[] = [];
   const visit = (before: unknown, after: unknown, path: ProjectionPath, parent?: object): void => {
     const original = parent && sourceValues?.get(parent)?.get(String(path.at(-1)));
@@ -75,7 +95,7 @@ export function contextProjectionChanges(source: JsonObject, projected: JsonObje
         visit(Object.hasOwn(old, key) ? old[key] : undefined, value, [...path, key], after as object);
     } else changes.push({path, value: after});
   };
-  visit(source, projected, []);
+  visit(base, projected, []);
   return {changes};
 }
 

loopx/control_plane/quota/turn_envelope.py

@@ -6,7 +6,7 @@
 from collections.abc import Mapping
 from copy import deepcopy
 from dataclasses import dataclass
-from typing import Any
+from typing import Any, cast
 
 from ..effect_runtime import effect_runtime_projection_result, source_number_tokens
 from ..scheduler.execution_context import (
@@ -50,17 +50,62 @@ def envelope(self) -> dict[str, Any]:
         return envelope
 
 
+def _apply_context_projection_changes(
+    source: dict[str, Any], result: dict[str, Any], *, base: dict[str, Any] | None = None,
+) -> dict[str, Any]:
+    """Apply typed display edits to the original Python facts, without policy."""
+    if set(result) != {"changes"} or not isinstance(result["changes"], list):
+        raise RuntimeError("TypeScript context projection must return changes")
+    # JSON transport has no aliasing. Keep Python's exact numeric values while
+    # separating repeated caller-owned objects before applying path edits.
+    projected = json.loads(json.dumps(source if base is None else base, allow_nan=False))
+    try:
+        for change in result["changes"]:
+            path = change["path"]
+            if not isinstance(path, list) or not path:
+                raise ValueError("projection path must be nonempty")
+            parent: Any = projected
+            for key in path[:-1]:
+                parent = parent[key]
+            key = path[-1]
+            if "remove" in change:
+                del parent[key]
+                continue
+            if "from" in change:
+                value: Any = source
+                for part in change["from"]:
+                    value = value[part]
+            else:
+                value = change["value"]
+            if isinstance(parent, list) and key == len(parent):
+                parent.append(deepcopy(value))
+            else:
+                parent[key] = deepcopy(value)
+    except (KeyError, IndexError, TypeError, ValueError) as error:
+        raise RuntimeError("Invalid TypeScript context projection changes") from error
+    return cast(dict[str, Any], projected)
+
+
 def _turn_envelope_result(
     operation: str,
     **params: object,
 ) -> dict[str, Any]:
     """Call the coarse TS owner and fail closed on transport shape drift."""
 
     request = {"operation": operation, **params}
-    if operation == "reference_context" and (
+    payload = params.get("payload")
+    staged_replan = (
+        operation in {"build", "build_controller_input", "quota_signature", "replan_cli_view", "bind_capture"}
+        and isinstance(payload, Mapping)
+        and (
+            (operation == "replan_cli_view" and params.get("source_edits") is True)
+            or (payload.get("replan_action_packet") and (payload.get("replan_read_path") or params.get("reader_argv")))
+        )
+    )
+    if operation == "replan_view" or staged_replan or operation == "reference_context" and (
         params.get("goal_acceptance") or params.get("vision_audit") or params.get("replan_authoring")
     ):
-        tokens = source_number_tokens(params["payload"])
+        tokens = source_number_tokens(payload)
         if tokens:
             request["source_number_tokens"] = tokens
     elif operation in {"context_delivery", "context_recovery", "progressive_context_delivery", "context_read_sections"}:
@@ -70,6 +115,8 @@ def _turn_envelope_result(
     result = effect_runtime_projection_result("quota.turn_envelope.evaluate", request)
     if not isinstance(result, Mapping):
         raise RuntimeError("TypeScript Turn envelope result must be an object")
+    if operation == "replan_view" and "source_number_tokens" in request:
+        return _apply_context_projection_changes(cast(dict[str, Any], payload), dict(result), base={})
     return dict(result)
 
 
@@ -185,10 +232,7 @@ def project_captured_replan_cli_view(payload: dict[str, Any]) -> dict[str, Any]:
     """Stage replan reads without round-tripping unchanged source values."""
     if not isinstance(payload.get("replan_read_path"), str):
         return payload
-    from .cli_projection import _apply_context_projection_changes
-
-    result = _turn_envelope_result("replan_cli_view", payload=payload, source_edits=True,
-        source_number_tokens=source_number_tokens(payload))
+    result = _turn_envelope_result("replan_cli_view", payload=payload, source_edits=True)
     return _apply_context_projection_changes(payload, result)
 
 
@@ -208,8 +252,6 @@ def prepare_context_delivery(
     Unsupported hosts omit retention and keep complete delivery. A candidate is
     not an ACK. Confirmed input and native delivery assumptions stay distinct.
     """
-    from .cli_projection import _apply_context_projection_changes
-
     source: dict[str, Any] = dict(context=dict(context), scope=dict(scope), capture_id=capture_id, surface=surface,
         baseline=dict(baseline) if baseline is not None else None,
         retained=dict(retained) if retained is not None else None,
@@ -235,8 +277,6 @@ def prepare_context_delivery(
 def prepare_progressive_context(context: Mapping[str, Any], *, source_reference: Mapping[str, Any],
                                 surface: str = "quota_packet") -> dict[str, Any]:
     """Stateless IO adapter to the same typed presentation owner; no Delta identity."""
-    from .cli_projection import _apply_context_projection_changes
-
     source = dict(context=dict(context), native_projection=dict(source_reference), surface=surface)
     return _apply_context_projection_changes(source,
         _turn_envelope_result("progressive_context_delivery", **source, source_edits=True))
@@ -257,8 +297,6 @@ def restore_context_delivery(
     current: Mapping[str, Any],
 ) -> dict[str, Any]:
     """Validate/reconstruct or return the same-capture full body without effects."""
-    from .cli_projection import _apply_context_projection_changes
-
     source = dict(delivery=dict(delivery), baseline=dict(baseline) if baseline is not None else None,
         current=dict(current))
     result = _turn_envelope_result("context_recovery", **source, source_edits=True)