fix(quota): resolve complete replan authoring through one typed owner

docs/guides/context-delivery.md

@@ -75,6 +75,14 @@ One shared preparation call does not prove that every final caller has migrated.
 | Saved-file delta | Verified immutable file versions with compatible identity | Model retention, comprehension or fresh permission |
 | Provider prompt cache | Provider-specific processing/billing reuse of input | Source correctness, task success or LoopX net savings |
 
+Replan authoring uses the same-response form: CLI display can reference the
+complete `vision_authoring` in a durable-writeback step. The existing TypeScript
+Turn owner chooses an exact owner and resolves it for ordinary and captured
+readers. Dangling references, reference chains, conflicting inline values and
+references to another kind of step are rejected. A small or different value
+stays inline. Sharing neither changes the original settlement identity nor
+establishes cross-Turn retention or net cost savings.
+
 Semantic summarization is another operation: it can change wording or meaning.
 The exact codec is not a semantic summary. A controller-authored brief remains
 a separate proposed workflow with its own retained-requirement and outcome

docs/guides/context-delivery.zh-CN.md

@@ -62,6 +62,12 @@ flowchart TD
 | 保存文件差分 | 身份兼容、经过校验的不可变文件版本 | 模型保留、理解或新鲜权限 |
 | Provider prompt cache | Provider 自己的输入处理／计费复用 | 来源正确、任务成功或 LoopX 净收益 |
 
+重规划 authoring 使用同响应引用：CLI 展示可指向 durable-writeback 步骤中完整的
+`vision_authoring`。现有 TypeScript Turn owner 选择精确一致的正文，并为普通及捕获
+读取器展开它。悬空引用、引用链、与引用冲突的内联值，以及指向其他种类步骤的引用
+都必须拒绝；很小或不同的值保持内联。共享不改变原结算身份，也不证明跨 Turn 保留
+或整体成本节省。
+
 语义摘要是另一种操作，会改变措辞，可能改变含义。精确 codec 不做语义摘要。
 Controller 编写 brief 仍是单独提议的工作流，须验收保留要求与实际成果；更短的文字或
 provider cache 命中都不能替它通过。

loopx/control_plane/quota/cli_projection.py

@@ -23,9 +23,6 @@
     "quota_cli_vision_continuation_compaction_v0"
 )
 QUOTA_CLI_VISION_DETAIL_COMMAND = "quota should-run --include-detail vision"
-QUOTA_CLI_REPLAN_ACTION_COMPACTION_SCHEMA_VERSION = (
-    "quota_cli_replan_action_compaction_v0"
-)
 QUOTA_CLI_MONITOR_POLL_DECISION_COMPACTION_SCHEMA_VERSION = (
     "quota_cli_monitor_poll_decision_compaction_v0"
 )
@@ -94,16 +91,6 @@
 )
 
 
-def _json_content_key(value: Any) -> str:
-    """Compare original JSON values for sharing, including types and signed zero.
-
-    This is local serialization IO, not another quota or context-selection rule.
-    Python equality merges booleans with integers and integral floats; those
-    distinct source facts must not borrow each other's in-response owner.
-    """
-    return json.dumps(value, sort_keys=True, ensure_ascii=False, allow_nan=False)
-
-
 def _compact_monitor_poll_interaction_contract(
     contract: Any,
 ) -> dict[str, Any] | None:
@@ -248,49 +235,6 @@ def _compact_vision_continuation_audit(
     return compact
 
 
-def _reference_inline_replan_authoring(
-    packet: dict[str, Any], *, payload: dict[str, Any]
-) -> dict[str, Any]:
-    """Reference an identical executable input already present in this response."""
-
-    writeback = packet.get("writeback_contract")
-    if not isinstance(writeback, dict) or not isinstance(
-        writeback.get("vision_authoring"), dict
-    ):
-        return packet
-    interaction = payload.get("interaction_contract")
-    cli = interaction.get("cli_channel") if isinstance(interaction, dict) else None
-    plan = cli.get("settlement_plan") if isinstance(cli, dict) else None
-    steps = plan.get("ordered_steps") if isinstance(plan, dict) else None
-    if not isinstance(steps, list):
-        return packet
-    authoring_key = _json_content_key(writeback["vision_authoring"])
-    index = next(
-        (index for index, step in enumerate(steps)
-         if isinstance(step, dict) and step.get("kind") == "durable_writeback"
-         and _json_content_key(step.get("vision_authoring")) == authoring_key),
-        None,
-    )
-    if index is None:
-        return packet
-    reference = (
-        "$.interaction_contract.cli_channel.settlement_plan.ordered_steps"
-        f"[{index}].vision_authoring"
-    )
-    compact_writeback = dict(writeback)
-    compact_writeback.pop("vision_authoring")
-    compact_writeback["vision_authoring_ref"] = reference
-    compact = dict(packet)
-    compact["writeback_contract"] = compact_writeback
-    compact["payload_compaction"] = {
-        "schema_version": QUOTA_CLI_REPLAN_ACTION_COMPACTION_SCHEMA_VERSION,
-        "mode": "compact_hot_path",
-        "compacted_fields": ["writeback_contract.vision_authoring"],
-        "projection_ref": reference,
-    }
-    return compact
-
-
 def _deduplicate_next_action_warning(
     warning: dict[str, Any],
     *,
@@ -441,6 +385,7 @@ def _promote_interaction_contract(payload: dict[str, Any]) -> dict[str, Any]:
 def _reference_context(
     payload: dict[str, Any], *, task_title: bool = True, explore_guidance: bool = True,
     goal_acceptance: bool = False, vision_audit: bool = False,
+    replan_authoring: bool = False,
     diagnostics: dict[str, Any] | None = None,
 ) -> dict[str, Any]:
     """One coarse transport call for the existing typed context owner."""
@@ -452,7 +397,7 @@ def _reference_context(
     if not needs_context and (diagnostics is None or (
         diagnostics.get("agent_detail") is True and diagnostics.get("user_detail") is True
     )):
-        if not goal_acceptance and not vision_audit:
+        if not goal_acceptance and not vision_audit and not replan_authoring:
             return payload
         if goal_acceptance:
             # A reference can only be created or relocated when two literal
@@ -469,7 +414,7 @@ def contains_contract(value: Any, *path: str) -> bool:
                 contains_contract(payload, "autonomous_replan_obligation", "replan_context", "core_goal", "acceptance_contract"),
                 contains_contract(payload, "agent_todo_summary", "goal_acceptance_contract"),
             )
-            if sum(carriers) < 2 and not vision_audit:
+            if sum(carriers) < 2 and not vision_audit and not replan_authoring:
                 return payload
     from .turn_envelope import _turn_envelope_result
 
@@ -480,6 +425,8 @@ def contains_contract(value: Any, *path: str) -> bool:
         "todo_summary_projection", "planning_horizon", "autonomous_replan_obligation",
         "interaction_contract", "agent_lane_next_action", "selected_todo",
     ) if key in payload}
+    if replan_authoring and "replan_action_packet" in payload:
+        source["replan_action_packet"] = payload["replan_action_packet"]
     if (
         isinstance(contract, dict)
         and isinstance(agent, dict)
@@ -512,6 +459,7 @@ def contains_contract(value: Any, *path: str) -> bool:
         task_title=task_title, explore_guidance=explore_guidance,
         goal_acceptance=goal_acceptance,
         **({"vision_audit": True} if vision_audit else {}),
+        **({"replan_authoring": True} if replan_authoring else {}),
         **({"diagnostics": diagnostics} if diagnostics is not None else {}))
     return _apply_context_projection_changes(payload, result)
 
@@ -591,13 +539,9 @@ def compact_quota_should_run_cli_payload(
     vision_audit = payload.get("vision_continuation_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(
-            replan_action, payload=compact
-        )
-        if compact_replan_action is not replan_action:
-            compact = dict(compact)
-            compact["replan_action_packet"] = compact_replan_action
+    writeback = replan_action.get("writeback_contract") if isinstance(replan_action, dict) else None
+    share_authoring = (not include_vision_detail and isinstance(writeback, dict)
+        and bool({"vision_authoring", "vision_authoring_ref"}.intersection(writeback)))
     if not include_todo_summary_detail:
         warning = payload.get("next_action_projection_warning")
         if isinstance(warning, dict):
@@ -629,7 +573,7 @@ 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,
-        goal_acceptance=True, vision_audit=share_audit,
+        goal_acceptance=True, vision_audit=share_audit, replan_authoring=share_authoring,
         explore_guidance=not include_vision_detail, diagnostics=diagnostics)
     if share_audit and isinstance(vision_audit, dict):
         compact = {**compact, "vision_continuation_audit": _compact_vision_continuation_audit(vision_audit)}

loopx/control_plane/quota/turn_envelope.py

@@ -57,7 +57,7 @@ def _turn_envelope_result(
 
     request = {"operation": operation, **params}
     if operation == "reference_context" and (
-        params.get("goal_acceptance") or params.get("vision_audit")
+        params.get("goal_acceptance") or params.get("vision_audit") or params.get("replan_authoring")
     ):
         tokens = source_number_tokens(params["payload"])
         if tokens:

loopx/control_plane/quota/turn_envelope.ts

@@ -228,6 +228,50 @@ function canonicalHash(value: unknown): string {
   return `sha256:${createHash("sha256").update(JSON.stringify(canonicalValue(value)), "utf8").digest("hex")}`;
 }
 
+function replanAuthoringContract(payload: JsonObject): JsonObject {
+  const contract = object(object(payload.replan_action_packet).writeback_contract);
+  if (!Object.hasOwn(contract, "vision_authoring_ref")) return contract;
+  const reference = contract.vision_authoring_ref;
+  const match = typeof reference === "string"
+    ? /^\$\.interaction_contract\.cli_channel\.settlement_plan\.ordered_steps\[(0|[1-9]\d*)\]\.vision_authoring$/.exec(reference) : null;
+  const steps = object(object(object(payload.interaction_contract).cli_channel).settlement_plan).ordered_steps;
+  const step = object(match && Array.isArray(steps) ? steps[Number(match[1])] : null);
+  if (Object.hasOwn(contract, "vision_authoring") || step.kind !== "durable_writeback" ||
+      Object.hasOwn(step, "vision_authoring_ref") || !Object.keys(object(step.vision_authoring)).length) {
+    throw new EffectRuntimeRequestError("Replan authoring reference must resolve to one complete inline durable-writeback owner");
+  }
+  delete contract.vision_authoring_ref;
+  return {...contract, vision_authoring: step.vision_authoring};
+}
+
+function referenceReplanAuthoring(payload: JsonObject,
+    identity = contextSourceIdentity(payload)): JsonObject {
+  const packet = object(payload.replan_action_packet);
+  const contract = object(packet.writeback_contract);
+  if (Object.hasOwn(contract, "vision_authoring_ref")) {
+    replanAuthoringContract(payload);
+    return payload;
+  }
+  const authoring = object(contract.vision_authoring);
+  const steps = object(object(object(payload.interaction_contract).cli_channel).settlement_plan).ordered_steps;
+  if (!Object.keys(authoring).length || !Array.isArray(steps)) return payload;
+  const encoded = JSON.stringify(canonicalValue(authoring));
+  const source = identity(authoring, ["replan_action_packet", "writeback_contract", "vision_authoring"]);
+  const index = steps.findIndex((value, index) => {
+    const step = object(value);
+    return step.kind === "durable_writeback" && !Object.hasOwn(step, "vision_authoring_ref") &&
+      jsonObject(step.vision_authoring) !== null &&
+      identity(step.vision_authoring, ["interaction_contract", "cli_channel", "settlement_plan",
+        "ordered_steps", index, "vision_authoring"]) === source;
+  });
+  if (index < 0) return payload;
+  const reference = `$.interaction_contract.cli_channel.settlement_plan.ordered_steps[${index}].vision_authoring`;
+  if (Buffer.byteLength(encoded, "utf8") <= Buffer.byteLength(reference, "utf8") + 8) return payload;
+  delete contract.vision_authoring;
+  return {...payload, replan_action_packet: {...packet, writeback_contract: {
+    ...contract, vision_authoring_ref: reference}}};
+}
+
 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;
@@ -647,6 +691,7 @@ function replanView(request: JsonObject): JsonObject {
     throw new EffectRuntimeRequestError("Replan read requires the original successful decision and matching source hash");
   }
   requireVisionAuditReferences(payload);
+  replanAuthoringContract(payload);
   const packet = object(payload.replan_action_packet);
   const obligation = object(payload.autonomous_replan_obligation);
   const context = object(obligation.replan_context);
@@ -675,7 +720,7 @@ function replanView(request: JsonObject): JsonObject {
     // The current nested review syntax belongs to authoring, not duplicate
     // decision evidence. Top-level required_fields cannot describe that shape.
     for (const key of ["decision_followthrough", "explore_basis", "decision_value_review"]) delete stride[key];
-    const contract = {...object(packet.writeback_contract)};
+    const contract = replanAuthoringContract(payload);
     if (contract.vision_authoring) {
       const authoring = {...object(contract.vision_authoring)};
       delete authoring.minimal_example;
@@ -760,6 +805,7 @@ function replanView(request: JsonObject): JsonObject {
 function replanActionPacket(payload: JsonObject, reads = replanReadCommands(payload)): JsonObject | null {
   const source = object(payload.replan_action_packet);
   if (Object.keys(source).length === 0) return null;
+  const writeback = replanAuthoringContract(payload);
   if (reads) return {...compactFields(source, ["schema_version", "decision", "obligation_id", "required_outcome", "allowed_terminal"]),
     review: reads.review, writeback: reads.writeback,
     guidance: "Read review state and applicable acceptance/evidence; choose the next work, then follow review.next for Todo selection or writeback. Reads grant no authority and require no acknowledgement."};
@@ -770,7 +816,6 @@ function replanActionPacket(payload: JsonObject, reads = replanReadCommands(payl
   // The typed replan owner supplies bounded instructions; do not truncate their
   // authority/stop qualifiers through generic diagnostic compaction.
   if (Array.isArray(source.planning_guidance)) compact.planning_guidance = source.planning_guidance;
-  const writeback = object(source.writeback_contract);
   if (writeback.vision_authoring) compact.writeback_contract = writeback;
   return Object.keys(compact).length > 0 ? compact : null;
 }
@@ -1660,6 +1705,17 @@ export function evaluateTurnEnvelope(value: unknown): JsonObject {
     if (jsonObject(request.diagnostics)) payload = projectQuotaDiagnostics(payload, request.diagnostics as JsonObject);
     if (request.goal_acceptance === true) payload = referenceGoalAcceptance(payload,
       contextSourceIdentity(source, request.source_number_tokens));
+    if (request.replan_authoring === true) {
+      const shared = referenceReplanAuthoring(payload, contextSourceIdentity(source, request.source_number_tokens));
+      if (shared !== payload) {
+        const packet = object(shared.replan_action_packet);
+        payload = {...shared, replan_action_packet: {...packet, payload_compaction: {
+          schema_version: "quota_cli_replan_action_compaction_v0", mode: "compact_hot_path",
+          compacted_fields: ["writeback_contract.vision_authoring"],
+          projection_ref: object(packet.writeback_contract).vision_authoring_ref,
+        }}};
+      }
+    }
     if (request.task_title === true) payload = referenceCurrentTaskTitle(payload);
     if (request.explore_guidance === true) payload = referenceExploreGuidance(payload);
     if (request.refresh_result === true) payload = referenceExploreResultDelivery(payload, request.source_number_tokens);

tests/control_plane/canonical_authority_fixture.py

@@ -2,6 +2,8 @@
 
 This fixture runs the real FileAuthorityStore in a Node process. It deliberately
 does not claim that a shadow qualification can promote canonical authority.
+Resolve TypeScript modules from the imported package so source and installed
+wheel tests exercise the same Python and TypeScript build.
 """
 from __future__ import annotations
 
@@ -12,6 +14,8 @@
 import subprocess
 import tempfile
 
+import loopx
+
 
 def assert_todo_read_unchanged(current, previous):
     """Compare complete readback except validated evaluator/observation clocks."""
@@ -56,10 +60,10 @@ def without_resume_observation_time(value):
 
 
 def initialize_canonical_authority(runtime_root: Path, goal_id: str, projection: dict, *, state_path: Path, provider: str = "file") -> dict:
-    repository = Path(__file__).resolve().parents[2]
-    module = repository / "loopx/control_plane/coordination/file_authority_store.ts"
-    fence_module = repository / "loopx/control_plane/coordination/legacy_writer_fence.ts"
-    codec_module = repository / "loopx/control_plane/coordination/authority_store_codec.ts"
+    package = Path(loopx.__file__).resolve().parent
+    module = package / "control_plane/coordination/file_authority_store.ts"
+    fence_module = package / "control_plane/coordination/legacy_writer_fence.ts"
+    codec_module = package / "control_plane/coordination/authority_store_codec.ts"
     script = (
         f"import {{FileAuthorityStore}} from {json.dumps(module.as_uri())};"
         f"import {{selectLocalSqliteAuthority,openLocalAuthorityStore}} from {json.dumps((module.parent / 'local_authority_provider.ts').as_uri())};"
@@ -107,7 +111,7 @@ def advance_canonical_fixture_authority(
     Consumer fixtures may author a later synthetic state, but must not replace
     storage bytes, reset lineage, re-import Markdown or re-engage its fence.
     """
-    module = Path(__file__).resolve().parents[2] / "loopx/control_plane/coordination/local_authority_provider.ts"
+    module = Path(loopx.__file__).resolve().parent / "control_plane/coordination/local_authority_provider.ts"
     script = (
         f"import {{openLocalAuthorityStore}} from {json.dumps(module.as_uri())};"
         "let input='';for await(const chunk of process.stdin)input+=chunk;"