From 7f9b6fa983af023ecb0c18957445268bc4d0a072 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sun, 8 Nov 2026 06:08:18 +0800 Subject: [PATCH] Preserve captured numbers in staged replan reads Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- loopx/control_plane/quota/cli_projection.py | 39 +-- .../quota/diagnostic_projection.ts | 24 +- loopx/control_plane/quota/turn_envelope.py | 57 ++++- loopx/control_plane/quota/turn_envelope.ts | 228 ++++++++++++------ .../test_replan_host_context_projection.py | 83 +++++++ 5 files changed, 310 insertions(+), 121 deletions(-) diff --git a/loopx/control_plane/quota/cli_projection.py b/loopx/control_plane/quota/cli_projection.py index 714d756976..0cea6dbf8e 100644 --- a/loopx/control_plane/quota/cli_projection.py +++ b/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) diff --git a/loopx/control_plane/quota/diagnostic_projection.ts b/loopx/control_plane/quota/diagnostic_projection.ts index b9e9496f69..b3128d5e41 100644 --- a/loopx/control_plane/quota/diagnostic_projection.ts +++ b/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>; +/** 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(); + 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}; } diff --git a/loopx/control_plane/quota/turn_envelope.py b/loopx/control_plane/quota/turn_envelope.py index 421ac1872f..ea44fecdba 100644 --- a/loopx/control_plane/quota/turn_envelope.py +++ b/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,6 +50,42 @@ 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, @@ -57,10 +93,17 @@ def _turn_envelope_result( """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 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 +113,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) @@ -197,8 +242,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, @@ -224,8 +267,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)) @@ -246,8 +287,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) diff --git a/loopx/control_plane/quota/turn_envelope.ts b/loopx/control_plane/quota/turn_envelope.ts index 363dad69d5..5c62588f3f 100644 --- a/loopx/control_plane/quota/turn_envelope.ts +++ b/loopx/control_plane/quota/turn_envelope.ts @@ -1,7 +1,8 @@ import {EXECUTION_PREPARATION_SCHEMA_VERSION} from "../work_items/execution_preparation.ts"; import {capturedContextSelection} from "./context_selection.ts"; import {contextReadSectionDigests, initialContextReadPointers, prepareProgressiveContext, prepareProgressiveContextDelivery, progressiveContextPlan} from "./progressive_context.ts"; -import {contextProjectionChanges, contextSourceIdentity, projectQuotaDiagnostics} from "./diagnostic_projection.ts"; +import {contextProjectionChanges, contextProjectionIdentity, contextProjectionRetainFields, + contextSourceIdentity, projectQuotaDiagnostics, type ProjectionSourceValues} 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"; @@ -229,8 +230,25 @@ 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); +type ReplanValueIdentity = (value: unknown, path: (string | number)[]) => string; +type ReplanReadPrecision = { + identity: ReplanValueIdentity; + retain: ReturnType; + equalProjection: (left: JsonObject, right: JsonObject) => boolean; + values: ProjectionSourceValues; +}; + +function replanReadPrecision(source: JsonObject, supplied: unknown): ReplanReadPrecision | undefined { + if (supplied === undefined) return undefined; + const identity = contextSourceIdentity(source, supplied); + const values: ProjectionSourceValues = new WeakMap(); + const projectedIdentity = (view: JsonObject) => contextProjectionIdentity(source, view, supplied, values); + return {identity, values, retain: contextProjectionRetainFields(source, values), + equalProjection: (left, right) => projectedIdentity(left) === projectedIdentity(right)}; +} + +function replanAuthoringContract(payload: JsonObject, precision?: ReplanReadPrecision): JsonObject { + const contract = jsonObject(jsonObject(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" @@ -241,8 +259,9 @@ function replanAuthoringContract(payload: JsonObject): JsonObject { 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}; + const {vision_authoring_ref: _reference, ...fields} = contract; + const resolved = {...fields, vision_authoring: step.vision_authoring}; + return precision?.retain(resolved, contract) ?? resolved; } function referenceReplanAuthoring(payload: JsonObject, @@ -609,53 +628,89 @@ function deferredReplanWriteback(writeback: JsonObject): JsonObject { /** Preserve the captured Goal and work horizon for review and the signed * executor envelope, indexing cumulative outcome details by exact content. */ -function replanGoalContext(payload: JsonObject, reads: JsonObject | null, coreRef = "$.core_goal") { +function replanGoalContext(payload: JsonObject, reads: JsonObject | null, coreRef = "$.core_goal", + precision?: ReplanReadPrecision) { const context = object(object(payload.autonomous_replan_obligation).replan_context); - const outcomes = new Map(); - function indexedGoal(value: unknown): JsonObject { - const core = {...object(value)}; + const retain = precision?.retain ?? ((view: JsonObject) => view); + function resolvedGoal(value: unknown): JsonObject { + const original = jsonObject(value) ?? {}; + const core = retain({...original}, original); if (!("acceptance_contract" in core) && "acceptance_contract_ref" in core) { const acceptance = inlineAcceptance(payload, core.acceptance_contract_ref); delete core.acceptance_contract_ref; core.acceptance_contract = acceptance; } - const acceptance = {...object(core.acceptance_contract)}; - const requirements = object(acceptance.outcome_requirements); - if (!Array.isArray(requirements.outcomes) || !requirements.outcomes.length || !reads) return core; - const refs = requirements.outcomes.map(value => { - const outcome = requireJsonObject(value, "captured acceptance outcome"); - const ref = canonicalHash(outcome); - outcomes.set(ref, outcome); - return ref; - }); + return core; + } + function capturedOutcomes(core: JsonObject): JsonObject[] { + const requirements = object(object(core.acceptance_contract).outcome_requirements); + return reads && Array.isArray(requirements.outcomes) + ? requirements.outcomes.map(value => requireJsonObject(value, "captured acceptance outcome")) : []; + } + const resolvedCore = resolvedGoal(context.core_goal); + const originalHorizon = jsonObject(payload.planning_horizon) ?? {}; + const horizon = retain({...originalHorizon}, originalHorizon); + const resolvedHorizon = horizon.core_goal ? resolvedGoal(horizon.core_goal) : null; + const coreOutcomes = capturedOutcomes(resolvedCore); + const horizonOutcomes = resolvedHorizon ? capturedOutcomes(resolvedHorizon) : []; + const refs = new Map(); + const legacyGroups = new Map(); + for (const outcome of [...coreOutcomes, ...horizonOutcomes]) { + if (refs.has(outcome)) continue; + const legacy = canonicalHash(outcome); + refs.set(outcome, legacy); + if (precision) { + const group = legacyGroups.get(legacy); + if (group) group.push(outcome); + else legacyGroups.set(legacy, [outcome]); + } + } + // Keep historical references where they identify exactly one outcome. + // When number decoding collapsed different outcomes, index each original + // body and reject the old ambiguous reference on the acceptance stage. + const ambiguous = new Set(); + if (precision) for (const [legacy, group] of legacyGroups) { + if (group.length < 2) continue; + const exact = group.map(outcome => precision.identity(outcome, [])); + if (new Set(exact).size < 2) continue; + ambiguous.add(legacy); + group.forEach((outcome, index) => refs.set(outcome, canonicalHash([legacy, exact[index]]))); + } + const outcomes = new Map(); + for (const [outcome, ref] of refs) outcomes.set(ref, outcome); + function indexedGoal(core: JsonObject, captured: JsonObject[]): JsonObject { + if (!captured.length) return core; + const originalAcceptance = jsonObject(core.acceptance_contract) ?? {}; + const acceptance = retain({...originalAcceptance}, originalAcceptance); + const requirements = jsonObject(acceptance.outcome_requirements) ?? {}; const {outcomes: _outcomes, ...metadata} = requirements; delete acceptance.outcome_requirements; - return {...core, acceptance_contract: acceptance, - outcome_requirements_index: {...metadata, outcome_refs: refs}}; - } - const core = indexedGoal(context.core_goal); - const horizon = {...object(payload.planning_horizon)}; - if (horizon.core_goal) { - const horizonCore = indexedGoal(horizon.core_goal); - if (canonicalHash(horizonCore) === canonicalHash(core)) { + return retain({...core, acceptance_contract: acceptance, + outcome_requirements_index: retain({...metadata, outcome_refs: captured.map(outcome => refs.get(outcome)!)}, + requirements)}, core); + } + const core = indexedGoal(resolvedCore, coreOutcomes); + if (resolvedHorizon) { + const horizonCore = indexedGoal(resolvedHorizon, horizonOutcomes); + if (canonicalHash(horizonCore) === canonicalHash(core) && + (precision?.equalProjection(horizonCore, core) ?? true)) { delete horizon.core_goal; horizon.core_goal_ref = coreRef; } else horizon.core_goal = horizonCore; } const acceptanceIndex = outcomes.size ? { read_action_template: `${reads!.acceptance} --outcome-ref {outcome_ref}`, - outcomes: [...outcomes].map(([ref, outcome]) => ({outcome_ref: ref, id: outcome.id, + outcomes: [...outcomes].map(([ref, outcome]) => retain({outcome_ref: ref, id: outcome.id, objective: outcome.objective, revision: outcome.revision, - requirement_count: Array.isArray(outcome.requirements) ? outcome.requirements.length : null})), + requirement_count: Array.isArray(outcome.requirements) ? outcome.requirements.length : null}, + outcome, ["id", "objective", "revision"])), } : null; - return {core, horizon, outcomes, acceptanceIndex}; + return {core, horizon, outcomes, acceptanceIndex, ambiguous}; } /** Move only an identical acceptance copy behind the existing captured read. * Different revisions, other source fields and unavailable context stay intact. */ -type ReplanValueIdentity = (value: unknown, path: (string | number)[]) => string; - function replanWorkContext(payload: JsonObject, context: JsonObject, reads = replanReadCommands(payload), identity: ReplanValueIdentity = canonicalHash): JsonObject { context = resolveExploreGuidance(payload, context); @@ -696,11 +751,18 @@ 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"); } + const precision = replanReadPrecision(payload, request.source_number_tokens); + const view = projectReplanView(request, payload, precision); + return precision ? contextProjectionChanges(payload, view, precision.values, {}) : view; +} + +function projectReplanView(request: JsonObject, payload: JsonObject, precision?: ReplanReadPrecision): JsonObject { + const retain = precision?.retain ?? ((view: JsonObject) => view); requireVisionAuditReferences(payload); - replanAuthoringContract(payload); - const packet = object(payload.replan_action_packet); - const obligation = object(payload.autonomous_replan_obligation); - const context = object(obligation.replan_context); + const originalContract = replanAuthoringContract(payload, precision); + const packet = jsonObject(payload.replan_action_packet) ?? {}; + const obligation = jsonObject(payload.autonomous_replan_obligation) ?? {}; + const context = jsonObject(obligation.replan_context) ?? {}; if (!packet.obligation_id || packet.settlement_only === true || packet.obligation_id !== context.obligation_id) { throw new EffectRuntimeRequestError("Replan read requires a matching decision context"); } @@ -713,22 +775,28 @@ function replanView(request: JsonObject): JsonObject { const reads = replanReadCommands(payload, path, payload.replan_read_argv ?? request.reader_argv ?? ["loopx"]); const sourceRead = reads ? {source_read: `cat -- ${shellQuote(path as string)}`} : {}; if (request.stage === "acceptance") { - const {outcomes} = replanGoalContext(payload, reads); - const outcome = outcomes.get(String(request.outcome_ref)); + const {outcomes, ambiguous} = replanGoalContext(payload, reads, undefined, precision); + const ref = String(request.outcome_ref); + if (ambiguous.has(ref)) throw new EffectRuntimeRequestError( + "The old captured outcome reference is ambiguous. Read the review stage for the distinct references."); + const outcome = outcomes.get(ref); if (!outcome) throw new EffectRuntimeRequestError("acceptance read requires an exact captured outcome reference"); return {...common, outcome_ref: request.outcome_ref, outcome}; } - if (request.stage === "explore") return {...common, explore_basis: context.explore_basis ?? null}; + if (request.stage === "explore") + return retain({...common, explore_basis: context.explore_basis ?? null}, context, ["explore_basis"]); if (request.stage === "writeback") { // Preserve executable shape and all settlement rules; decision advice and // duplicate context are available in the unchanged full captured source. - const stride = {...object(packet.stride_control)}; + const originalStride = jsonObject(packet.stride_control) ?? {}; + const stride = retain({...originalStride}, originalStride); // 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 = replanAuthoringContract(payload); + const contract = {...originalContract}; if (contract.vision_authoring) { - const authoring = {...object(contract.vision_authoring)}; + const originalAuthoring = jsonObject(contract.vision_authoring) ?? {}; + const authoring = retain({...originalAuthoring}, originalAuthoring); delete authoring.minimal_example; contract.vision_authoring = authoring; } @@ -750,7 +818,7 @@ function replanView(request: JsonObject): JsonObject { } const steps = Array.isArray(plan.ordered_steps) ? plan.ordered_steps.map(object) : []; const refresh = steps.find(step => step.kind === "durable_writeback"); - return {...common, ...sourceRead, writeback_contract: contract, + return retain({...common, ...sourceRead, writeback_contract: retain(contract, originalContract), required_outcome: packet.required_outcome, allowed_terminal: packet.allowed_terminal, ...(packet.stride_control ? {stride_control: stride} : {}), writeback, @@ -758,21 +826,27 @@ function replanView(request: JsonObject): JsonObject { settlement_command: `${refresh.command_template} --settle-turn`, settlement_usage: "After validation, fill this refresh command from the actual result. It follows only the verified owed quota step on this same binding. " + "Do not also execute a quota step after settled readback; retain the original plan for completion, blocked/no-spend closeout and failure recovery.", - } : {})}; + } : {})}, packet, ["required_outcome", "allowed_terminal"]); } if (request.stage !== "review") throw new EffectRuntimeRequestError("unsupported replan read stage"); - const {core, horizon, acceptanceIndex} = replanGoalContext(payload, reads); + const {core, horizon, acceptanceIndex} = replanGoalContext(payload, reads, undefined, precision); const cli = object(object(payload.interaction_contract).cli_channel); // Index every captured outcome, including differing context/horizon revisions. // Full-content equality alone shares a row; identical IDs do not erase edits. - const coverage = Array.isArray(context.coverage_ledger) ? context.coverage_ledger.map(object) : []; - const follow = {...object(context.decision_followthrough)}; + const coverage = Array.isArray(context.coverage_ledger) ? context.coverage_ledger.map(value => + jsonObject(value) ?? {}) : []; + const originalFollow = jsonObject(context.decision_followthrough) ?? {}; + const follow = {...originalFollow}; delete follow.guidance; + const followFields = Object.keys(follow); if (Object.hasOwn(follow, "recorded_progress_count")) { follow.observation_count = follow.recorded_progress_count; delete follow.recorded_progress_count; - } - return {...common, + retain(follow, originalFollow, followFields.filter(field => field !== "observation_count")); + retain(follow, originalFollow, [["recorded_progress_count", "observation_count"]]); + } else retain(follow, originalFollow, followFields); + const explore = jsonObject(context.explore_basis) ?? {}; + const review = retain({...common, core_goal: core, planning_guidance: packet.planning_guidance ?? [], ...(acceptanceIndex ? {acceptance_index: acceptanceIndex} : {}), @@ -780,16 +854,17 @@ function replanView(request: JsonObject): JsonObject { decision_followthrough: Object.keys(follow).length ? follow : null, planning_horizon: payload.planning_horizon == null ? null : horizon, action_portfolio: payload.action_portfolio ?? null, - ...(context.explore_basis ? {capabilities: {explore: { - availability: object(context.explore_basis).availability, revision: object(context.explore_basis).revision, + ...(context.explore_basis ? {capabilities: {explore: retain({ + availability: explore.availability, revision: explore.revision, read_action: reads?.explore ?? null, - }}} : {}), + }, explore, ["availability", "revision"])}} : {}), evidence: (Array.isArray(context.evidence) ? context.evidence : []).map(value => { - const row = object(value); + const row = jsonObject(value) ?? {}; const observation = coverage.find(item => item.fingerprint === row.coverage_ref); - return {...compactFields(row, ["evidence_ref", "generated_at", "delivery_outcome", "coverage_ref", + const item = retain({...compactFields(row, ["evidence_ref", "generated_at", "delivery_outcome", "coverage_ref", "occurrences", "first_observed_at", "read_action"]), - ...(observation ? {result_class: observation.result_class} : {})}; + ...(observation ? {result_class: observation.result_class} : {})}, row); + return observation ? retain(item, observation, ["result_class"]) : item; }), evidence_count: context.evidence_count, coverage_count: context.coverage_count, @@ -805,7 +880,10 @@ function replanView(request: JsonObject): JsonObject { next: cli.selection_required === true ? {selection_required: true, selection_command_ref: "$.interaction_contract.cli_channel.selection_command"} : {writeback: reads?.writeback ?? null}, - }; + }, context, ["evidence_count", "coverage_count", "coverage_truncated", "omitted_evidence", + "evidence_read_action_template", "uncovered_frontier"]); + retain(review, obligation, ["triggers"]); + return retain(review, packet, ["planning_guidance"]); } function replanActionPacket(payload: JsonObject, reads = replanReadCommands(payload)): JsonObject | null { @@ -1100,7 +1178,7 @@ function contractCapsule(payload: JsonObject): JsonObject { return capsule; } -function actionProjection(payload: JsonObject, deferSettlement = true): JsonObject { +function actionProjection(payload: JsonObject, deferSettlement = true, precision?: ReplanReadPrecision): JsonObject { const agentIdentity = object(payload.agent_identity); const semanticAgentId = scalarString( payload.agent_id || agentIdentity.agent_id, @@ -1145,7 +1223,7 @@ function actionProjection(payload: JsonObject, deferSettlement = true): JsonObje const reads = capabilityIntent ? null : replanReadCommands(payload); if (turn.observation.planning_horizon !== null) action.planning_horizon = {...turn.observation.planning_horizon}; if (reads && deferSettlement) { - const goal = replanGoalContext(payload, reads, "$.action.core_goal"); + const goal = replanGoalContext(payload, reads, "$.action.core_goal", precision); action.core_goal = goal.core; if (turn.observation.planning_horizon !== null) action.planning_horizon = goal.horizon; } @@ -1157,7 +1235,7 @@ function actionProjection(payload: JsonObject, deferSettlement = true): JsonObje if (nextCliActions.length === 0 && Array.isArray(cliChannel.next_cli_actions)) { nextCliActions = [...cliChannel.next_cli_actions].map(pythonString); } - const settlementPlan = object(cliChannel.settlement_plan); + const settlementPlan = jsonObject(cliChannel.settlement_plan) ?? {}; const hasSettlementPlan = Object.keys(settlementPlan).length > 0; const recovery = object(payload.unsettled_host_turn_recovery); const receiptBoundSettlementOnly = recovery.scope === "current_turn" && recovery.repair === "quota_spend"; @@ -1206,8 +1284,8 @@ function actionProjection(payload: JsonObject, deferSettlement = true): JsonObje // loses the effect identity, conditional closeout and host/agent boundary. if (hasSettlementPlan) writeback.settlement_plan = settlementPlan; for (const field of ["replan_settlement_contract", "delivery_workspace_causality"]) { - const value = object(cliChannel[field]); - if (Object.keys(value).length > 0) writeback[field] = value; + const value = jsonObject(cliChannel[field]); + if (value && Object.keys(value).length > 0) writeback[field] = value; } const projection: JsonObject = { agent_id: semanticAgentId, @@ -1264,8 +1342,8 @@ function compactPeerActivation(contract: JsonObject, detailRef: string): JsonObj }; } -function turnActionProjection(payload: JsonObject): JsonObject { - const projection = actionProjection(payload); +function turnActionProjection(payload: JsonObject, precision?: ReplanReadPrecision): JsonObject { + const projection = actionProjection(payload, true, precision); const orchestration = object(projection.task_orchestration_contract); if (Object.keys(orchestration).length > 0) { const peer = object(orchestration.peer_activation_diagnostic); @@ -1407,10 +1485,10 @@ export function turnEnvelopeActionSignatureDocument(value: unknown): JsonObject return signature; } -export function quotaActionSignatureDocument(value: unknown): JsonObject { +export function quotaActionSignatureDocument(value: unknown, supplied?: unknown): JsonObject { const payload = requireJsonObject(value, "quota payload"); return turnEnvelopeActionSignatureDocument( - turnActionProjection(payload), + turnActionProjection(payload, replanReadPrecision(payload, supplied)), ); } @@ -1475,7 +1553,7 @@ export function buildTurnEnvelope(value: unknown): JsonObject { object(payload.agent_identity).agent_id, "quota payload agent_identity.agent_id", ).trim() || null; - const actionProjectionValue = turnActionProjection(payload); + const actionProjectionValue = turnActionProjection(payload, replanReadPrecision(payload, request.source_number_tokens)); const envelope: JsonObject = { ok: Boolean(payload.ok), schema_version: TURN_ENVELOPE_SCHEMA_VERSION, @@ -1531,7 +1609,8 @@ function bindTurnDecisionCapture(request: JsonObject): JsonObject { const capture = object(coldPath(payload, String(envelope.agent_id), "", path).captured_decision); const readerArgv = detail.replan_read_argv ?? request.reader_argv; const reads = readerArgv ? replanReadCommands(payload, path, readerArgv) : null; - const goal = reads ? replanGoalContext(payload, reads) : null; + const goal = reads ? replanGoalContext(payload, reads, undefined, + replanReadPrecision(payload, request.source_number_tokens)) : null; const bound = {...envelope, detail_ref: {...detail, full_decision: `cat -- ${shellQuote(path)}`, ...(reads ? {replan_action_packet: replanActionPacket(payload, reads), replan_read_argv: readerArgv, @@ -1646,7 +1725,7 @@ function capturedDecisionSelection(request: JsonObject): JsonObject { } function replanCliView(payload: JsonObject, identity: ReplanValueIdentity = canonicalHash, - reads = replanReadCommands(payload)): JsonObject { + reads = replanReadCommands(payload), precision?: ReplanReadPrecision): JsonObject { if (!reads) return payload; const obligation = {...object(payload.autonomous_replan_obligation)}; delete obligation.replan_context; @@ -1654,7 +1733,7 @@ function replanCliView(payload: JsonObject, identity: ReplanValueIdentity = cano const cli = object(interaction.cli_channel); // Defer only authoring/settlement format, not admission or stop facts. // This default captured path needs no optional native delta transport. - const goal = replanGoalContext(payload, reads); + const goal = replanGoalContext(payload, reads, undefined, precision); const summary = object(payload.agent_todo_summary); const originalCore = object(object(object(payload.autonomous_replan_obligation).replan_context).core_goal); // Reuse the same captured acceptance index for an exact summary copy. @@ -1721,9 +1800,10 @@ function nativeReplanPresentation(request: JsonObject): JsonObject | undefined { || typeof packet.obligation_id !== "string" || packet.obligation_id !== review.obligation_id || packet.review !== undefined || packet.writeback !== undefined) return undefined; - const identity = contextSourceIdentity(request, request.source_number_tokens); + const precision = replanReadPrecision(request, request.source_number_tokens); + const identity = precision?.identity ?? contextSourceIdentity(request); const view = replanCliView(source, (value, field) => identity(value, ["context", ...field]), - replanReadCommands(source, path)); + replanReadCommands(source, path), precision); return view === source ? undefined : {staged_context: view, deferred: [], instructions: []}; } @@ -1737,7 +1817,11 @@ export function evaluateTurnEnvelope(value: unknown): JsonObject { const operation = request.operation; if (operation === "replan_view") return replanView(request); if (operation === "context_select") return capturedDecisionSelection(request); - if (operation === "replan_cli_view") return replanCliView(requireJsonObject(request.payload, "replan CLI payload")); + if (operation === "replan_cli_view") { + const payload = requireJsonObject(request.payload, "replan CLI payload"); + const precision = replanReadPrecision(payload, request.source_number_tokens); + return replanCliView(payload, precision?.identity, undefined, precision); + } if (operation === "selected_work_todo") { const source = requireJsonObject(request.payload, "Turn work selection source"); const historical = scalarString(request.historical_primary_text, "historical primary Todo"); @@ -1809,7 +1893,7 @@ export function evaluateTurnEnvelope(value: unknown): JsonObject { return turnEnvelopeActionSignatureDocument(request.envelope); } if (operation === "quota_signature") { - return quotaActionSignatureDocument(request.payload); + return quotaActionSignatureDocument(request.payload, request.source_number_tokens); } throw new EffectRuntimeRequestError("turn envelope operation is unsupported"); } diff --git a/tests/control_plane/test_replan_host_context_projection.py b/tests/control_plane/test_replan_host_context_projection.py index 5e8ebe657e..542f342412 100644 --- a/tests/control_plane/test_replan_host_context_projection.py +++ b/tests/control_plane/test_replan_host_context_projection.py @@ -1,6 +1,11 @@ from __future__ import annotations import copy +import hashlib +import json +import shlex +import subprocess +import sys import pytest @@ -114,6 +119,84 @@ def _quota_payload(*, include_legacy_receipt: bool = False, stride_strategy: str ) +def test_staged_replan_preserves_source_numbers_and_distinct_outcome_refs(tmp_path) -> None: + def encoded(value: object) -> str: + return json.dumps(value, sort_keys=True, separators=(",", ":")) + + def legacy_ref(normalized: dict[str, object]) -> str: + return "sha256:" + hashlib.sha256(encoded(normalized).encode()).hexdigest() + + capture = tmp_path / "decision.json" + source = _quota_payload() + source["replan_read_path"] = str(capture) + exact = [1.0, -0.0, 2**53 + 1] + rounded = [1, 0, 2**53] + outcomes = [ + {"id": "outcome", "revision": 1, "requirements": [{"values": values}]} + for values in (exact, rounded) + ] + separate = {**outcomes[0], "id": "unambiguous"} + context = source["autonomous_replan_obligation"]["replan_context"] + core = { + "evidence_value": 2**53 + 1, + "acceptance_contract": {"outcome_requirements": {"outcomes": [*outcomes, separate]}}, + } + context["core_goal"] = core + context["explore_basis"] = {"availability": "available", "revision": -0.0, "values": exact} + context["decision_followthrough"] = {"recorded_progress_count": 2**53 + 1} + context["evidence"][0]["occurrences"] = 1.0 + source["planning_horizon"] = {"core_goal": {**core, "evidence_value": 2**53}} + source["replan_action_packet"]["writeback_contract"] = { + "vision_authoring": {"sample_count": 1.0}, + "captured_scalar": -0.0, + "captured_values": exact, + } + raw = encoded(source).encode() + capture.write_bytes(raw) + envelope = build_turn_envelope(source) + digest = envelope["action_signature"]["source_decision_hash"] + registry = tmp_path / "absent-registry.json" + runtime = tmp_path / "absent-runtime" + cli = [sys.executable, "-m", "loopx.entrypoint", "--registry", str(registry), + "--runtime-root", str(runtime), "--format", "json"] + + def read(*args: str, expected: int = 0): + process = subprocess.run([*cli, *args], capture_output=True, text=True, timeout=15) + result = json.loads(process.stdout) + assert process.returncode == expected, result + return result + + route = ("quota", "replan-view", "--decision-file", str(capture), "--source-hash", digest) + review = read(*route, "--replan-stage", "review") + assert review["core_goal"]["evidence_value"] == 2**53 + 1 + assert review["planning_horizon"]["core_goal"]["evidence_value"] == 2**53 + assert review["decision_followthrough"]["observation_count"] == 2**53 + 1 + assert encoded(review["evidence"][0]["occurrences"]) == "1.0" + assert encoded(review["capabilities"]["explore"]["revision"]) == "-0.0" + index = review["acceptance_index"] + refs = [row["outcome_ref"] for row in index["outcomes"]] + assert len(set(refs)) == 3 + assert envelope["action"]["core_goal"]["outcome_requirements_index"]["outcome_refs"] == refs + for ref, outcome in zip(refs, [*outcomes, separate], strict=True): + detail = shlex.split(index["read_action_template"].replace("{outcome_ref}", ref)) + assert detail[detail.index("quota"):][:2] == ["quota", "replan-view"] + assert encoded(read(*detail[detail.index("quota"):])["outcome"]) == encoded(outcome) + + old_ambiguous = read(*route, "--replan-stage", "acceptance", "--outcome-ref", + legacy_ref(outcomes[1]), expected=1) + assert not old_ambiguous["ok"] and "ambiguous" in old_ambiguous["reason"] + old_separate = legacy_ref({**outcomes[1], "id": "unambiguous"}) + assert refs[-1] == old_separate + assert encoded(read(*route, "--replan-stage", "acceptance", "--outcome-ref", old_separate)["outcome"]) == encoded(separate) + + explore = read(*route, "--replan-stage", "explore") + assert encoded(explore["explore_basis"]) == encoded(context["explore_basis"]) + writer = read(*route, "--replan-stage", "writeback") + assert encoded(writer["writeback_contract"]) == encoded(source["replan_action_packet"]["writeback_contract"]) + assert capture.read_bytes() == raw + assert not registry.exists() and not runtime.exists() + + def test_quota_delivers_coverage_context_and_minimal_replan_action() -> None: payload = _quota_payload()