Validate quota audit references before a staged replan

loopx/control_plane/quota/cli_projection.py

@@ -291,57 +291,6 @@ def _reference_inline_replan_authoring(
     return compact
 
 
-def _vision_continuation_ref(audit: dict[str, Any]) -> dict[str, Any]:
-    compact = {
-        key: audit[key]
-        for key in ("schema_version", "required", "decision")
-        if key in audit
-    }
-    judge = audit.get("vision_gap_judge")
-    if isinstance(judge, dict):
-        compact["vision_gap_judge"] = {
-            key: judge[key] for key in ("done", "decision") if key in judge
-        }
-        compact["vision_gap_judge"]["projection_ref"] = (
-            "$.vision_continuation_audit.vision_gap_judge"
-        )
-    compact["projection_ref"] = "$.vision_continuation_audit"
-    return compact
-
-
-def _replace_nested_vision_continuation_audits(
-    payload: dict[str, Any],
-    *,
-    source_audit: dict[str, Any],
-    audit_ref: dict[str, Any],
-) -> None:
-    source_key = _json_content_key(source_audit)
-    frontier = payload.get("goal_frontier_projection")
-    if (
-        isinstance(frontier, dict)
-        and _json_content_key(frontier.get("vision_continuation_audit")) == source_key
-    ):
-        compact_frontier = dict(frontier)
-        compact_frontier["vision_continuation_audit"] = dict(audit_ref)
-        payload["goal_frontier_projection"] = compact_frontier
-
-    interaction = payload.get("interaction_contract")
-    if not isinstance(interaction, dict):
-        return
-    compact_interaction = dict(interaction)
-    for channel_name in ("agent_channel", "cli_channel"):
-        channel = compact_interaction.get(channel_name)
-        if (
-            not isinstance(channel, dict)
-            or _json_content_key(channel.get("vision_continuation_audit")) != source_key
-        ):
-            continue
-        compact_channel = dict(channel)
-        compact_channel["vision_continuation_audit"] = dict(audit_ref)
-        compact_interaction[channel_name] = compact_channel
-    payload["interaction_contract"] = compact_interaction
-
-
 def _deduplicate_next_action_warning(
     warning: dict[str, Any],
     *,
@@ -573,6 +522,7 @@ def writable_parent(path: tuple[str | int, ...]) -> dict[str, Any]:
 
 def _reference_context(
     payload: dict[str, Any], *, task_title: bool = True, explore_guidance: bool = True,
+    vision_audit: bool = False,
     diagnostics: dict[str, Any] | None = None,
 ) -> dict[str, Any]:
     """One coarse transport call for the existing typed context owner."""
@@ -581,12 +531,34 @@ def _reference_context(
     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 diagnostics is None and (not (task_title or explore_guidance) or not isinstance(context, dict) or not context.get("sources")):
+    if not vision_audit and diagnostics is None and (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
 
-    result = _turn_envelope_result("reference_context", payload=payload,
+    # The typed projection reads and edits these roots. Untouched capture
+    # facts stay local. Apply the returned paths against the original payload.
+    source = {key: payload[key] for key in (
+        "mode", "groups", "next_automatic_turn", "agent_todo_summary", "user_todo_summary",
+        "todo_summary_projection",
+        "interaction_contract", "agent_lane_next_action", "selected_todo",
+    ) if key in payload}
+    if isinstance(context, dict) and isinstance(context.get("sources"), list):
+        # Edits address rows by their original index. An unrelated source has
+        # no fields read by this projection; its complete body stays local.
+        rows = [row if isinstance(row, dict) and (
+            (task_title and row.get("source") == "selected_todo")
+            or (explore_guidance and row.get("kind") == "explore_turn_context")
+        ) else None for row in context["sources"]]
+        source["interaction_contract"] = {**contract, "agent_channel": {
+            **agent, "work_context": {**context, "sources": rows}}}
+    if vision_audit:
+        source["vision_continuation_audit"] = payload["vision_continuation_audit"]
+        frontier = payload.get("goal_frontier_projection")
+        if isinstance(frontier, dict) and "vision_continuation_audit" in frontier:
+            source["goal_frontier_projection"] = {"vision_continuation_audit": frontier["vision_continuation_audit"]}
+    result = _turn_envelope_result("reference_context", payload=source,
         task_title=task_title, explore_guidance=explore_guidance,
+        **({"vision_audit": True} if vision_audit else {}),
         **({"diagnostics": diagnostics} if diagnostics is not None else {}))
     return _apply_context_projection_changes(payload, result)
 
@@ -664,15 +636,7 @@ def compact_quota_should_run_cli_payload(
             compact = dict(compact)
             compact["goal_boundary"] = compact_goal_boundary
     vision_audit = payload.get("vision_continuation_audit")
-    if not include_vision_detail and isinstance(vision_audit, dict):
-        compact = dict(compact)
-        compact_vision_audit = _compact_vision_continuation_audit(vision_audit)
-        compact["vision_continuation_audit"] = compact_vision_audit
-        _replace_nested_vision_continuation_audits(
-            compact,
-            source_audit=vision_audit,
-            audit_ref=_vision_continuation_ref(compact_vision_audit),
-        )
+    share_audit = not include_vision_detail and isinstance(vision_audit, dict)
     replan_action = payload.get("replan_action_packet")
     if not include_vision_detail and isinstance(replan_action, dict):
         compact_replan_action = _reference_inline_replan_authoring(
@@ -713,7 +677,9 @@ def compact_quota_should_run_cli_payload(
         diagnostics = {"agent_detail": include_todo_summary_detail,
             "user_detail": include_user_todo_summary_detail}
     compact = _reference_context(compact, task_title=not include_todo_summary_detail,
-        explore_guidance=not include_vision_detail, diagnostics=diagnostics)
+        vision_audit=share_audit, explore_guidance=not include_vision_detail, diagnostics=diagnostics)
+    if share_audit:
+        compact = {**compact, "vision_continuation_audit": _compact_vision_continuation_audit(vision_audit)}
     return _promote_interaction_contract(
         _promote_runtime_capability_reentry(compact)
     )

loopx/control_plane/quota/turn_envelope.py

@@ -56,7 +56,11 @@ def _turn_envelope_result(
     """Call the coarse TS owner and fail closed on transport shape drift."""
 
     request = {"operation": operation, **params}
-    if operation in {"context_delivery", "context_recovery", "progressive_context_delivery", "context_read_sections"}:
+    if operation == "reference_context" and params.get("vision_audit"):
+        tokens = source_number_tokens(params["payload"])
+        if tokens:
+            request["source_number_tokens"] = tokens
+    elif operation in {"context_delivery", "context_recovery", "progressive_context_delivery", "context_read_sections"}:
         tokens = source_number_tokens(params)
         if tokens:
             request["source_number_tokens"] = tokens

loopx/control_plane/quota/turn_envelope.ts

@@ -1,6 +1,6 @@
 import {capturedContextSelection} from "./context_selection.ts";
 import {contextReadSectionDigests, initialContextReadPointers, prepareProgressiveContext, prepareProgressiveContextDelivery} from "./progressive_context.ts";
-import {contextProjectionChanges, projectQuotaDiagnostics} from "./diagnostic_projection.ts";
+import {contextProjectionChanges, contextSourceIdentity, projectQuotaDiagnostics} from "./diagnostic_projection.ts";
 import type { LoopXTurnRouteValue } from "../turn_driver/turn_contract_generated.ts";
 import { EffectiveAction, type QuotaEffectiveActionValue } from "./effective_action.generated.ts";
 import { createHash } from "node:crypto";
@@ -281,6 +281,70 @@ function referenceExploreGuidance(payload: JsonObject): JsonObject {
     work_context: {...context, sources}}}};
 }
 
+/** Keep the complete vision audit once. These existing projection references
+ * share identical facts only; a channel-specific obligation is never a summary
+ * of the root. The old display adapter delegates here before adding metadata. */
+function visionAuditReference(audit: JsonObject): JsonObject {
+  const reference: JsonObject = {projection_ref: "$.vision_continuation_audit"};
+  for (const key of ["schema_version", "required", "decision"]) {
+    if (Object.hasOwn(audit, key)) reference[key] = audit[key];
+  }
+  if (typeof audit.vision_gap_judge === "object" && audit.vision_gap_judge !== null &&
+      !Array.isArray(audit.vision_gap_judge)) {
+    const judge = object(audit.vision_gap_judge);
+    reference.vision_gap_judge = {projection_ref: "$.vision_continuation_audit.vision_gap_judge"};
+    for (const key of ["done", "decision"]) {
+      if (Object.hasOwn(judge, key)) (reference.vision_gap_judge as JsonObject)[key] = judge[key];
+    }
+  }
+  return reference;
+}
+
+function mapNestedVisionAudits(payload: JsonObject,
+    visit: (audit: JsonObject, path: (string | number)[]) => JsonObject): JsonObject {
+  const map = (value: unknown, path: (string | number)[]) => {
+    const parent = object(value), audit = parent.vision_continuation_audit;
+    if (typeof audit !== "object" || audit === null || Array.isArray(audit)) return value;
+    return {...parent, vision_continuation_audit: visit(audit as JsonObject, [...path, "vision_continuation_audit"])};
+  };
+  const rawInteraction = payload.interaction_contract, interaction = object(rawInteraction);
+  const hasInteraction = typeof rawInteraction === "object" && rawInteraction !== null && !Array.isArray(rawInteraction);
+  return {...payload,
+    ...(Object.hasOwn(payload, "goal_frontier_projection") ? {
+      goal_frontier_projection: map(payload.goal_frontier_projection, ["goal_frontier_projection"])} : {}),
+    ...(hasInteraction ? {interaction_contract: {...interaction,
+      ...Object.fromEntries(["agent_channel", "cli_channel"].filter(key => Object.hasOwn(interaction, key))
+        .map(key => [key, map(interaction[key], ["interaction_contract", key])]))}} : {})};
+}
+
+function requireVisionAuditReferences(payload: JsonObject): void {
+  const owner = object(payload.vision_continuation_audit);
+  const chained = Object.hasOwn(owner, "projection_ref") || Object.hasOwn(object(owner.vision_gap_judge), "projection_ref");
+  if (chained) throw new EffectRuntimeRequestError("Vision audit reference requires a complete inline root owner");
+  mapNestedVisionAudits(payload, audit => {
+    if (Object.hasOwn(audit, "projection_ref") || Object.hasOwn(object(audit.vision_gap_judge), "projection_ref")) {
+      if (!Object.keys(owner).length ||
+          JSON.stringify(canonicalValue(audit)) !== JSON.stringify(canonicalValue(visionAuditReference(owner)))) {
+        throw new EffectRuntimeRequestError("Vision audit reference must match its complete inline root owner without conflicting facts");
+      }
+    }
+    return audit;
+  });
+}
+
+function referenceVisionContinuation(payload: JsonObject,
+    identity = contextSourceIdentity(payload)): JsonObject {
+  requireVisionAuditReferences(payload);
+  const owner = object(payload.vision_continuation_audit);
+  if (!Object.keys(owner).length) return payload;
+  const encoded = JSON.stringify(canonicalValue(owner)), reference = visionAuditReference(owner);
+  // A short audit is cheaper inline. Do not create a reference merely to count sharing.
+  if (Buffer.byteLength(encoded, "utf8") <= Buffer.byteLength(JSON.stringify(reference), "utf8")) return payload;
+  const source = identity(owner, ["vision_continuation_audit"]);
+  return mapNestedVisionAudits(payload, (audit, path) =>
+    identity(audit, path) === source ? structuredClone(reference) : audit);
+}
+
 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);
@@ -486,6 +550,7 @@ function replanView(request: JsonObject): JsonObject {
   if (request.source_hash !== canonicalHash(payload) || payload.ok !== true) {
     throw new EffectRuntimeRequestError("Replan read requires the original successful decision and matching source hash");
   }
+  requireVisionAuditReferences(payload);
   const packet = object(payload.replan_action_packet);
   const obligation = object(payload.autonomous_replan_obligation);
   const context = object(obligation.replan_context);
@@ -1485,6 +1550,8 @@ export function evaluateTurnEnvelope(value: unknown): JsonObject {
   if (operation === "reference_context") {
     const source = requireJsonObject(request.payload, "quota decision");
     let payload = source;
+    if (request.vision_audit === true) payload = referenceVisionContinuation(source,
+      contextSourceIdentity(source, request.source_number_tokens));
     if (jsonObject(request.diagnostics)) payload = projectQuotaDiagnostics(payload, request.diagnostics as JsonObject);
     if (request.task_title === true) payload = referenceCurrentTaskTitle(payload);
     if (request.explore_guidance === true) payload = referenceExploreGuidance(payload);