From 046c5aceef69bf5f562b6ed874e7753a352a47cc Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sun, 8 Nov 2026 00:50:45 +0800 Subject: [PATCH] Reuse verified sources in staged replan reads Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- loopx/control_plane/quota/turn_envelope.ts | 53 +++++++++---------- .../control_plane/test_replan_staged_read.py | 31 +++++++++++ 2 files changed, 56 insertions(+), 28 deletions(-) diff --git a/loopx/control_plane/quota/turn_envelope.ts b/loopx/control_plane/quota/turn_envelope.ts index c8961823a7..c0099bbcfe 100644 --- a/loopx/control_plane/quota/turn_envelope.ts +++ b/loopx/control_plane/quota/turn_envelope.ts @@ -5,6 +5,7 @@ import {contextProjectionChanges, contextSourceIdentity, projectQuotaDiagnostics import type { LoopXTurnRouteValue } from "../turn_driver/turn_contract_generated.ts"; import { EffectiveAction, type QuotaEffectiveActionValue } from "./effective_action.generated.ts"; import { createHash } from "node:crypto"; +import {authorityUnicodeCompare} from "../coordination/authority_store_codec.ts"; import { ENVELOPED_SHA256_PATTERN } from "../content_digest.ts"; import {omitUnboundExploreResultWriteback, referenceExploreResultWriteback} from "../capabilities/explore_turn_context.ts"; import {prepareContextDelivery, restoreContextDelivery} from "./context_delivery.ts"; @@ -210,21 +211,10 @@ function canonicalValue(value: unknown): unknown { if (typeof value !== "object" || value === null) return value; const source = value as JsonObject; return Object.fromEntries( - Object.keys(source).sort(comparePythonUnicode).map((key) => [key, canonicalValue(source[key])]), + Object.keys(source).sort(authorityUnicodeCompare).map((key) => [key, canonicalValue(source[key])]), ); } -function comparePythonUnicode(left: string, right: string): number { - const leftPoints = Array.from(left, (item) => item.codePointAt(0) ?? 0); - const rightPoints = Array.from(right, (item) => item.codePointAt(0) ?? 0); - const sharedLength = Math.min(leftPoints.length, rightPoints.length); - for (let index = 0; index < sharedLength; index += 1) { - const difference = leftPoints[index] - rightPoints[index]; - if (difference !== 0) return difference; - } - return leftPoints.length - rightPoints.length; -} - function canonicalHash(value: unknown): string { return `sha256:${createHash("sha256").update(JSON.stringify(canonicalValue(value)), "utf8").digest("hex")}`; } @@ -579,7 +569,7 @@ function selectedTodo(payload: JsonObject, recommendedAction: string | null): Js /** These reads stay on one captured observation; they never re-enter the guard. */ function replanReadCommands(payload: JsonObject, path = payload.replan_read_path, - argv = payload.replan_read_argv ?? ["loopx"]): JsonObject | null { + argv = payload.replan_read_argv ?? ["loopx"], verifiedSourceHash?: string): JsonObject | null { const packet = object(payload.replan_action_packet); if (typeof path !== "string" || !path.trim() || path.includes("\0") || payload.ok !== true || payload.effective_action === EffectiveAction.GOVERNED_CAPABILITY_INTENT || @@ -590,7 +580,7 @@ function replanReadCommands(payload: JsonObject, path = payload.replan_read_path throw new EffectRuntimeRequestError("captured replan CLI route is malformed"); } const command = `${argv.map(arg => shellQuote(arg as string)).join(" ")} --format json quota replan-view --decision-file ${shellQuote(path)}` + - ` --source-hash ${canonicalHash(payload)}`; + ` --source-hash ${verifiedSourceHash ?? canonicalHash(payload)}`; return {review: `${command} --replan-stage review`, writeback: `${command} --replan-stage writeback`, acceptance: `${command} --replan-stage acceptance`, explore: `${command} --replan-stage explore`}; } @@ -612,6 +602,16 @@ function deferredReplanWriteback(writeback: JsonObject): JsonObject { function replanGoalContext(payload: JsonObject, reads: JsonObject | null, coreRef = "$.core_goal") { const context = object(object(payload.autonomous_replan_obligation).replan_context); const outcomes = new Map(); + const refsByOutcome = new WeakMap(); + function indexOutcome(value: unknown): string { + const outcome = requireJsonObject(value, "captured acceptance outcome"); + // Shared carriers can resolve to the same original object. Distinct + // objects are still checked, even when their IDs or revisions coincide. + const ref = refsByOutcome.get(outcome) ?? canonicalHash(outcome); + refsByOutcome.set(outcome, ref); + outcomes.set(ref, outcome); + return ref; + } function indexedGoal(value: unknown): JsonObject { const core = {...object(value)}; if (!("acceptance_contract" in core) && "acceptance_contract_ref" in core) { @@ -622,12 +622,7 @@ function replanGoalContext(payload: JsonObject, reads: JsonObject | null, coreRe 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; - }); + const refs = requirements.outcomes.map(indexOutcome); const {outcomes: _outcomes, ...metadata} = requirements; delete acceptance.outcome_requirements; return {...core, acceptance_contract: acceptance, @@ -693,7 +688,8 @@ function replanWorkContext(payload: JsonObject, context: JsonObject, reads = rep function replanView(request: JsonObject): JsonObject { const payload = requireJsonObject(request.payload, "captured replan decision"); - if (request.source_hash !== canonicalHash(payload) || payload.ok !== true) { + const sourceHash = canonicalHash(payload); + if (request.source_hash !== sourceHash || payload.ok !== true) { throw new EffectRuntimeRequestError("Replan read requires the original successful decision and matching source hash"); } requireVisionAuditReferences(payload); @@ -710,7 +706,7 @@ function replanView(request: JsonObject): JsonObject { // A native journal binds its path after planning. The CLI supplies that same // physical source route without modifying the captured observation or hash. const path = request.captured_decision_path ?? payload.replan_read_path; - const reads = replanReadCommands(payload, path, payload.replan_read_argv ?? request.reader_argv ?? ["loopx"]); + const reads = replanReadCommands(payload, path, payload.replan_read_argv ?? request.reader_argv ?? ["loopx"], sourceHash); const sourceRead = reads ? {source_read: `cat -- ${shellQuote(path as string)}`} : {}; if (request.stage === "acceptance") { const {outcomes} = replanGoalContext(payload, reads); @@ -732,7 +728,7 @@ function replanView(request: JsonObject): JsonObject { delete authoring.minimal_example; contract.vision_authoring = authoring; } - const writeback = actionProjection(payload, false).writeback; + const writeback = actionProjection(payload, false, sourceHash).writeback; const plan = object(object(writeback).settlement_plan); const identity = object(plan.identity); if (typeof identity.turn_instance_id === "string") { @@ -1100,7 +1096,7 @@ function contractCapsule(payload: JsonObject): JsonObject { return capsule; } -function actionProjection(payload: JsonObject, deferSettlement = true): JsonObject { +function actionProjection(payload: JsonObject, deferSettlement = true, verifiedSourceHash?: string): JsonObject { const agentIdentity = object(payload.agent_identity); const semanticAgentId = scalarString( payload.agent_id || agentIdentity.agent_id, @@ -1117,12 +1113,14 @@ function actionProjection(payload: JsonObject, deferSettlement = true): JsonObje ? projectPendingCapabilityIntent(payload.pending_capability_intent) : null; // A governed capability action has already won the live decision. Stale // replan/host-reentry projections must not replace its exact command. - const replanPacket = capabilityIntent ? null : replanActionPacket(payload); + const reads = capabilityIntent ? null + : replanReadCommands(payload, payload.replan_read_path, payload.replan_read_argv ?? ["loopx"], verifiedSourceHash); + const replanPacket = capabilityIntent ? null : replanActionPacket(payload, reads); const replanSettlementOnly = object(payload.replan_action_packet).settlement_only === true; const recommendedAction = replanPacket ? replanSettlementOnly ? scalarString(object(object(payload.replan_action_packet).writeback_contract).rule, "replan settlement rule") - : replanReadCommands(payload) ? "read replan_action_packet.review, decide from state, then follow review.next" + : reads ? "read replan_action_packet.review, decide from state, then follow review.next" : "apply replan_action_packet and emit one required semantic outcome" : text(turn.observation.recommended_action || payload.recommended_action, 480); // The signed host action is executable authority, not a display summary. @@ -1131,7 +1129,7 @@ function actionProjection(payload: JsonObject, deferSettlement = true): JsonObje const action: JsonObject = { recommended_action: recommendedAction, primary_action: replanPacket && !replanSettlementOnly - ? replanReadCommands(payload)?.review ?? "produce one required semantic outcome" + ? reads?.review ?? "produce one required semantic outcome" : primaryAction || null, must_attempt: Boolean(agentChannel.must_attempt), delivery_allowed: Boolean(agentChannel.delivery_allowed), @@ -1142,7 +1140,6 @@ function actionProjection(payload: JsonObject, deferSettlement = true): JsonObje if (turn.observation.action_portfolio !== null) { action.action_portfolio = { ...turn.observation.action_portfolio }; } - 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"); diff --git a/tests/control_plane/test_replan_staged_read.py b/tests/control_plane/test_replan_staged_read.py index 05df95313d..22d64e7ce4 100644 --- a/tests/control_plane/test_replan_staged_read.py +++ b/tests/control_plane/test_replan_staged_read.py @@ -764,6 +764,37 @@ def test_explore_uses_its_own_incremental_budget_and_real_result_log(tmp_path): assert json_bytes(writer) <= 16 * 1024 +def test_historical_replan_reader_keeps_unicode_capture_hash(tmp_path): + from hashlib import sha256 + + source = { + "ok": True, "goal_id": "fixture-goal", "agent_id": "fixture-agent", + "replan_action_packet": {"obligation_id": "fixture-replan"}, + "autonomous_replan_obligation": {"replan_context": { + "obligation_id": "fixture-replan", + "core_goal": {"future_fields": { + "\U00010000": "Supplementary", "\ue000": "BMP", "a\u00e9": "Prefix", "a": "ASCII", + }}, + }}, + } + capture = tmp_path / "decision.json" + original = json.dumps(source, ensure_ascii=False, indent=2).encode() + capture.write_bytes(original) + # Python sorts Unicode code points, independently of JS's UTF-16 default. + canonical = json.dumps(source, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode() + digest = f"sha256:{sha256(canonical).hexdigest()}" + result = subprocess.run([sys.executable, "-m", "loopx.entrypoint", "--format", "json", "quota", "replan-view", + "--decision-file", str(capture), "--source-hash", digest, "--replan-stage", "review"], + capture_output=True, text=True, timeout=15) + assert result.returncode == 0, result.stdout + review = json.loads(result.stdout) + assert review["source_hash"] == digest + assert review["core_goal"] == source["autonomous_replan_obligation"]["replan_context"]["core_goal"] + writer = shlex.split(review["next"]["writeback"]) + assert writer[writer.index("--source-hash") + 1] == digest + assert capture.read_bytes() == original + + def test_replan_reader_rejects_replaced_entries_and_recovers_original(tmp_path): source = _quota_payload() original = tmp_path / "decision.json"