From aa244feaf8392e895c27b6653b8db06c6fbe9314 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Thu, 29 Oct 2026 18:07:05 +0800 Subject: [PATCH 1/3] fix(turn): preserve original numeric evidence in journals Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../turn_driver/turn_journal_effects.ts | 21 +++++++++-- .../turn_driver/turn_journal_runtime.py | 5 ++- .../test_turn_journal_goal_instance.py | 13 +++++++ .../turn_journal_effects.test.ts | 35 +++++++++++++++++++ 4 files changed, 71 insertions(+), 3 deletions(-) diff --git a/loopx/control_plane/turn_driver/turn_journal_effects.ts b/loopx/control_plane/turn_driver/turn_journal_effects.ts index 730302f7bb..0e6e588b8c 100644 --- a/loopx/control_plane/turn_driver/turn_journal_effects.ts +++ b/loopx/control_plane/turn_driver/turn_journal_effects.ts @@ -10,6 +10,7 @@ import { } from "../effect_runtime_errors.ts"; import { atomicWriteJson, + atomicWriteText, claimFileMutationLock, mutationLockOwner, releaseFileMutationLock, @@ -318,7 +319,22 @@ export async function commitTurnJournal( if (!isAbsolute(path)) { throw new EffectRuntimeRequestError("Turn journal path must be absolute"); } - const journal = asObject(params.journal); + // Serialized input is an IO carrier, not a second transition owner. Validate + // its decoded facts below, then publish those same source bytes atomically. + // Existing native object callers keep their historical encoding. + const originalJson = typeof params.journal === "string" ? params.journal : null; + let value: unknown = params.journal; + if (originalJson !== null) { + try { + value = JSON.parse(originalJson, (_key, item: unknown) => { + if (typeof item === "number" && !Number.isFinite(item)) throw new Error("non-finite number"); + return item; + }); + } catch { + throw new EffectRuntimeRequestError("Turn journal requires valid finite JSON", "journal_snapshot_invalid"); + } + } + const journal = asObject(value); const goalBinding = parseTurnJournalGoalBinding(journal); if (goalBinding.kind === "invalid") { throw new EffectRuntimeRequestError( @@ -373,7 +389,8 @@ export async function commitTurnJournal( )) { conflict("A new Turn journal must begin in progress with no completed phases"); } - await atomicWriteJson(path, journal); + if (originalJson === null) await atomicWriteJson(path, journal); + else await atomicWriteText(path, `${originalJson}\n`); return { ok: true, appended: true, diff --git a/loopx/control_plane/turn_driver/turn_journal_runtime.py b/loopx/control_plane/turn_driver/turn_journal_runtime.py index 226723d914..81c86609bd 100644 --- a/loopx/control_plane/turn_driver/turn_journal_runtime.py +++ b/loopx/control_plane/turn_driver/turn_journal_runtime.py @@ -1,5 +1,6 @@ from __future__ import annotations +import json from collections.abc import Mapping from typing import Any @@ -172,7 +173,9 @@ def write_turn_journal( request: dict[str, Any] = { "path": path, - "journal": dict(journal), + # TS owns validation, transitions and publication. Carry the original + # JSON as text so its parser/stringifier cannot round source evidence. + "journal": json.dumps(dict(journal), ensure_ascii=False, separators=(",", ":"), allow_nan=False), "expected_effect_id": expected_effect_id, } if source_admission is not None: diff --git a/tests/control_plane/test_turn_journal_goal_instance.py b/tests/control_plane/test_turn_journal_goal_instance.py index 11d86eeac4..1601eeea3b 100644 --- a/tests/control_plane/test_turn_journal_goal_instance.py +++ b/tests/control_plane/test_turn_journal_goal_instance.py @@ -27,6 +27,19 @@ TURN_B = f"sha256:{'b' * 64}" +@pytest.mark.parametrize("number", [1, 1.0, -0.0, 2**60 + 1, float(2**60), 1e-7]) +def test_journal_retains_original_numbers_through_native_commit_and_replay(tmp_path, number): + journal = _journal(None, TURN_A) + journal["plan"]["original_evidence"] = {"values": [number, {"number": number}], "text": "1.0"} + path = tmp_path / "turn.json" + write_turn_journal_checkpoint(path, journal) + expected = json.dumps(journal["plan"]["original_evidence"], sort_keys=True) + assert json.dumps(json.loads(path.read_text())["plan"]["original_evidence"], sort_keys=True) == expected + original = path.read_bytes() + write_turn_journal_checkpoint(path, journal) + assert path.read_bytes() == original + + def _registry_payload(path: Path, instance_id: str) -> dict[str, Any]: return { "schema_version": "0.2", diff --git a/tests/control_plane_ts/turn_journal_effects.test.ts b/tests/control_plane_ts/turn_journal_effects.test.ts index d484160691..14c6f5e9e5 100644 --- a/tests/control_plane_ts/turn_journal_effects.test.ts +++ b/tests/control_plane_ts/turn_journal_effects.test.ts @@ -83,6 +83,41 @@ function sourceJournal( return snapshot; } +test("serialized journal preserves numeric source bytes and retains transition and lifetime checks", async () => { + const root = await mkdtemp(join(tmpdir(), "loopx-journal-source-")); + const path = join(root, "turn.json"); + try { + const encoded = JSON.stringify(journal()).replace(/}$/, ',"source":[1.0,-0.0,1152921504606846977]}'); + const written = await commitTurnJournal({path, journal: encoded, expected_effect_id: effectId()}); + assert.equal(written.appended, true); + assert.equal(await readFile(path, "utf8"), encoded + "\n"); + assert.equal((await commitTurnJournal({path, journal: encoded})).replayed, true); + await assert.rejects(commitTurnJournal({path, journal: JSON.stringify(journal("committed", phases))}), + /skip transaction checkpoints/); + await assert.rejects(commitTurnJournal({path: join(root, "scoped.json"), + journal: JSON.stringify(sourceJournal(instanceA, turnKey))}), /require source admission/); + await assert.rejects(commitTurnJournal({path, journal: encoded, expected_effect_id: "other-effect"}), + /expected settlement effect/); + assert.equal(await readFile(path, "utf8"), encoded + "\n"); + } finally { + await rm(root, {recursive: true, force: true}); + } +}); + +for (const encoded of ['{"broken":', 'null', '[]', '1', + JSON.stringify(journal()).replace(/}$/, ',"source":1e400}')]) { + test(`serialized journal rejects invalid input before writing: ${encoded.slice(0, 24)}`, async () => { + const root = await mkdtemp(join(tmpdir(), "loopx-journal-invalid-")); + const path = join(root, "turn.json"); + try { + await assert.rejects(commitTurnJournal({path, journal: encoded}), /JSON|schema/); + await assert.rejects(readFile(path), {code: "ENOENT"}); + } finally { + await rm(root, {recursive: true, force: true}); + } + }); +} + function sourceGuardPath(registryPath: string, goalId: string): string { const alias = createHash("sha256").update(goalId, "utf8").digest("hex"); return join( From 7462c126018db9687bfb16f38d4a7f6dcabd8de2 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Fri, 30 Oct 2026 07:13:57 +0800 Subject: [PATCH 2/3] docs(turn): state journal source preservation and transport cost Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../controller-executor-unified-contract-v0.md | 9 +++++++++ ...ntroller-executor-unified-contract-v0.zh-CN.md | 7 +++++++ docs/reference/protocols/loopx-turn-v0.md | 15 +++++++++++++++ 3 files changed, 31 insertions(+) diff --git a/docs/architecture/rfcs/controller-executor-unified-contract-v0.md b/docs/architecture/rfcs/controller-executor-unified-contract-v0.md index 57601013b0..66d42cfc7d 100644 --- a/docs/architecture/rfcs/controller-executor-unified-contract-v0.md +++ b/docs/architecture/rfcs/controller-executor-unified-contract-v0.md @@ -130,6 +130,15 @@ the existing context-delivery owner. C1 must converge the remaining composition and consumer assumptions, rather than introduce another capture service or rewrite already shared delta/native-resume logic. +Journal persistence is an independent source-fidelity prerequisite. The proposed +Python adapter carries finite original JSON text through the existing TypeScript +journal owner, which retains its transition, lifetime, effect and lock checks +before atomic publication. Integral floats, signed zero and unsafe integers +retain their source values; native object callers keep their existing encoding. +Historical hashes and receipts retain their codecs. JSON-string escaping adds +request bytes, so qualify matched transport costs under unchanged limits. This +slice does not qualify the unified writer/consumer cutover or net cost benefit. + | Current source / obligation | Present envelope treatment | Required C1 disposition | | --- | --- | --- | | `interaction_contract.agent_channel.work_context` and pending required reads | `work_context` plus `required_reads`; identical inline material can be referenced | One complete requirement/evidence source with resolvable references; preserve pending reads and their completion conditions | diff --git a/docs/architecture/rfcs/controller-executor-unified-contract-v0.zh-CN.md b/docs/architecture/rfcs/controller-executor-unified-contract-v0.zh-CN.md index 61e1518567..2a3a545030 100644 --- a/docs/architecture/rfcs/controller-executor-unified-contract-v0.zh-CN.md +++ b/docs/architecture/rfcs/controller-executor-unified-contract-v0.zh-CN.md @@ -104,6 +104,13 @@ Quota capture 与 Turn journal capture 都调用 context-delivery owner。C1 应收敛剩余的组合逻辑与消费者假设,不另建 capture service, 也不重写已经共享的差分与 native-resume 逻辑。 +Journal 持久化是独立的来源保真前置。本次提议由 Python adapter 传递有限数值的 +原始 JSON 文本,既有 TypeScript journal owner 在原子发布前继续执行状态转换、 +生命周期、effect 与锁检查。整值浮点、带符号零和非安全整数保留来源值;原生对象 +调用方保留已有编码。历史 hash 与回执保留原 codec。JSON 字符串转义会增加请求 +字节,须在原有限制下比较相同负载的传输成本;该切片不构成统一 writer/consumer +切换或净成本收益的资格。 + | 当前来源 / 义务 | 当前 envelope 处理 | C1 必须保留的归属与语义 | | --- | --- | --- | | `interaction_contract.agent_channel.work_context` 与未完成必读 | `work_context` 加 `required_reads`;相同内联材料可改用引用 | 一份完整要求/证据来源与可解析引用;保留未完成必读及其完成条件 | diff --git a/docs/reference/protocols/loopx-turn-v0.md b/docs/reference/protocols/loopx-turn-v0.md index 8c29f31e68..051a5b8adb 100644 --- a/docs/reference/protocols/loopx-turn-v0.md +++ b/docs/reference/protocols/loopx-turn-v0.md @@ -516,6 +516,21 @@ completed effects remain proved; other effects are unknown, not `false`. The writer's empty-map rejection is retained: no pending intent uses an absent field; committed history is carried by completed checkpoints, not retained intents. +Python-origin journal snapshots carry finite original JSON text through that +same TypeScript writer. The owner validates the decoded snapshot and publishes +the original text atomically, preserving integral floats, signed zero and large +integers in saved evidence. Direct object callers retain their existing JSON +encoding; historical hash and receipt codecs are unchanged. Replay preserves the +original file. This remains one RPC under the existing 2 MiB request limit; +string escaping increases transport bytes and reduces available source headroom. +It is a source-fidelity repair, not a latency or long-horizon benefit result. + +Python 来源的 journal 快照通过同一 TypeScript writer 传递有限数值的原始 JSON +文本。owner 校验解码后的快照,再原子发布原文,保留证据中的整值浮点、带符号零 +和大整数。直接对象调用方保留已有 JSON 编码,历史 hash 和回执 codec 不变;重放 +保留原文件。仍为一次 RPC,沿用 2 MiB 请求上限;字符串转义增加传输字节、减少 +可容纳的来源正文。这是来源保真修复,不代表延迟或长程收益已得到验证。 + If executing `run-once` raises unexpectedly, its error response retains the original error and resume key, marks uncertain current-invocation `effects` as `null`, and adds `journal_observation` from this same read-only TS owner. Its From aafc2d29b93ccd8d34a8bc7d2e2a057c92451543 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Fri, 30 Oct 2026 07:47:13 +0800 Subject: [PATCH 3/3] fix: preserve journal source numbers with shared sparse IO metadata Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- ...controller-executor-unified-contract-v0.md | 20 ++-- ...ller-executor-unified-contract-v0.zh-CN.md | 16 ++-- docs/reference/protocols/loopx-turn-v0.md | 31 +++--- loopx/control_plane/effect_runtime.py | 26 +++++ .../quota/diagnostic_projection.ts | 28 +----- loopx/control_plane/quota/turn_envelope.py | 29 +----- loopx/control_plane/runtime_decode.ts | 33 +++++++ .../turn_driver/turn_journal_effects.ts | 96 +++++++++++++------ .../turn_driver/turn_journal_runtime.py | 12 ++- .../test_turn_journal_goal_instance.py | 38 ++++++++ .../turn_journal_effects.test.ts | 89 ++++++++++++++--- 11 files changed, 290 insertions(+), 128 deletions(-) diff --git a/docs/architecture/rfcs/controller-executor-unified-contract-v0.md b/docs/architecture/rfcs/controller-executor-unified-contract-v0.md index 66d42cfc7d..3655006cff 100644 --- a/docs/architecture/rfcs/controller-executor-unified-contract-v0.md +++ b/docs/architecture/rfcs/controller-executor-unified-contract-v0.md @@ -131,13 +131,19 @@ and consumer assumptions, rather than introduce another capture service or rewrite already shared delta/native-resume logic. Journal persistence is an independent source-fidelity prerequisite. The proposed -Python adapter carries finite original JSON text through the existing TypeScript -journal owner, which retains its transition, lifetime, effect and lock checks -before atomic publication. Integral floats, signed zero and unsafe integers -retain their source values; native object callers keep their existing encoding. -Historical hashes and receipts retain their codecs. JSON-string escaping adds -request bytes, so qualify matched transport costs under unchanged limits. This -slice does not qualify the unified writer/consumer cutover or net cost benefit. +Python adapter reuses context delivery's sparse number-token collector; both +TypeScript consumers reuse one source-number validator. Journals remain JSON +objects, with metadata only for integral floats, signed zero and unsafe integers. +The journal owner restores these numbers before atomic publication and retains +their distinctions in replay and immutable-plan comparisons. Ordinary native +objects retain their encoding; historical receipts are not rewritten. + +The interim whole-JSON-string carrier failed the existing 2 MiB request boundary +for a synthetic 1.55-million-character quoted body accepted by the object path. +Keep that negative result: qualify the sparse replacement against the same +workload and limit, including number-dense overhead. Neither this prerequisite +nor a component byte saving qualifies the unified writer/consumer cutover or +long-horizon net benefit. | Current source / obligation | Present envelope treatment | Required C1 disposition | | --- | --- | --- | diff --git a/docs/architecture/rfcs/controller-executor-unified-contract-v0.zh-CN.md b/docs/architecture/rfcs/controller-executor-unified-contract-v0.zh-CN.md index 2a3a545030..23516e0c16 100644 --- a/docs/architecture/rfcs/controller-executor-unified-contract-v0.zh-CN.md +++ b/docs/architecture/rfcs/controller-executor-unified-contract-v0.zh-CN.md @@ -104,12 +104,16 @@ Quota capture 与 Turn journal capture 都调用 context-delivery owner。C1 应收敛剩余的组合逻辑与消费者假设,不另建 capture service, 也不重写已经共享的差分与 native-resume 逻辑。 -Journal 持久化是独立的来源保真前置。本次提议由 Python adapter 传递有限数值的 -原始 JSON 文本,既有 TypeScript journal owner 在原子发布前继续执行状态转换、 -生命周期、effect 与锁检查。整值浮点、带符号零和非安全整数保留来源值;原生对象 -调用方保留已有编码。历史 hash 与回执保留原 codec。JSON 字符串转义会增加请求 -字节,须在原有限制下比较相同负载的传输成本;该切片不构成统一 writer/consumer -切换或净成本收益的资格。 +Journal 持久化是独立的来源保真前置。本次提议由 Python adapter 复用 context +delivery 的稀疏数值 token 收集器;两个 TypeScript 消费者复用同一来源数值校验器。 +Journal 继续传递 JSON 对象,仅为整值浮点、带符号零和非安全整数增加元数据。 +Journal owner 在原子发布前恢复这些数值,并在重放与不可变 plan 比较中保留其 +区别。普通原生对象保留已有编码,不改写历史回执。 + +中间的完整 JSON 字符串方案,在合成的 155 万字符带引号正文上超过既有 2 MiB +请求边界;原对象路径能够接收同一负载。保留这一负面结果:稀疏替代须在相同 +负载与上限下比较,包括数值密集时的元数据开销。该前置或组件字节减少,都不 +构成统一 writer/consumer 切换或长程净收益的资格。 | 当前来源 / 义务 | 当前 envelope 处理 | C1 必须保留的归属与语义 | | --- | --- | --- | diff --git a/docs/reference/protocols/loopx-turn-v0.md b/docs/reference/protocols/loopx-turn-v0.md index 051a5b8adb..c4c0bff3c8 100644 --- a/docs/reference/protocols/loopx-turn-v0.md +++ b/docs/reference/protocols/loopx-turn-v0.md @@ -516,20 +516,23 @@ completed effects remain proved; other effects are unknown, not `false`. The writer's empty-map rejection is retained: no pending intent uses an absent field; committed history is carried by completed checkpoints, not retained intents. -Python-origin journal snapshots carry finite original JSON text through that -same TypeScript writer. The owner validates the decoded snapshot and publishes -the original text atomically, preserving integral floats, signed zero and large -integers in saved evidence. Direct object callers retain their existing JSON -encoding; historical hash and receipt codecs are unchanged. Replay preserves the -original file. This remains one RPC under the existing 2 MiB request limit; -string escaping increases transport bytes and reduces available source headroom. -It is a source-fidelity repair, not a latency or long-horizon benefit result. - -Python 来源的 journal 快照通过同一 TypeScript writer 传递有限数值的原始 JSON -文本。owner 校验解码后的快照,再原子发布原文,保留证据中的整值浮点、带符号零 -和大整数。直接对象调用方保留已有 JSON 编码,历史 hash 和回执 codec 不变;重放 -保留原文件。仍为一次 RPC,沿用 2 MiB 请求上限;字符串转义增加传输字节、减少 -可容纳的来源正文。这是来源保真修复,不代表延迟或长程收益已得到验证。 +Python-origin journal snapshots remain JSON objects through that same TypeScript +writer. They reuse context delivery's sparse numeric IO metadata and validator +to preserve integral floats, signed zero and unsafe integers before atomic +publication. Replay and immutable-plan comparison distinguish these source +numbers; refusals retain the original file. Ordinary objects without metadata +retain their encoding and operation identity; historical receipts are not +rewritten. This remains one RPC under the existing 2 MiB request limit and avoids +double escaping evidence text. Metadata still costs bytes for number-dense +sources. Source fidelity and transport checks do not establish latency or +long-horizon benefit. + +Python 来源的 journal 快照仍经同一 TypeScript writer 传递 JSON 对象,复用 +context delivery 的稀疏数值 IO 元数据与校验器,在原子发布前保留整值浮点、带 +符号零和非安全整数。重放与不可变 plan 比较区分这些来源数值;拒绝时保留原 +文件。无元数据的普通对象保留已有编码与操作身份,不改写历史回执。仍为一次 +RPC,沿用 2 MiB 请求上限,避免证据正文的双重转义;数值密集来源仍有元数据 +字节成本。来源保真与传输检查不代表延迟或长程收益已得到验证。 If executing `run-once` raises unexpectedly, its error response retains the original error and resume key, marks uncertain current-invocation `effects` as diff --git a/loopx/control_plane/effect_runtime.py b/loopx/control_plane/effect_runtime.py index 2c999f4360..dd8a9684e1 100644 --- a/loopx/control_plane/effect_runtime.py +++ b/loopx/control_plane/effect_runtime.py @@ -2,6 +2,7 @@ import hashlib import json +import math import os import secrets import socket @@ -86,6 +87,31 @@ _RuntimeSourceSnapshot = tuple[tuple[str, int, int, int], ...] +def source_number_tokens(value: object) -> list[list[Any]]: + """Carry numeric IO spellings that a JavaScript number cannot distinguish. + + This is transport metadata, not a hash codec or a projection policy. + Fractional doubles and safe integers keep their existing object transport. + """ + tokens: list[list[Any]] = [] + + def visit(item: object, path: list[str | int]) -> None: + if isinstance(item, float) and not math.isfinite(item): + raise ValueError("Context source numbers must be finite") + if (isinstance(item, float) and item.is_integer() + or type(item) is int and abs(item) > 2**53 - 1): + tokens.append([path, json.dumps(item, allow_nan=False)]) + elif isinstance(item, Mapping): + for key, child in item.items(): + visit(child, [*path, key]) + elif isinstance(item, (list, tuple)): + for index, child in enumerate(item): + visit(child, [*path, index]) + + visit(value, []) + return tokens + + class _RuntimeSourceChanged(RuntimeError): def __init__(self, snapshot: _RuntimeSourceSnapshot) -> None: super().__init__("runtime source changed while hashing") diff --git a/loopx/control_plane/quota/diagnostic_projection.ts b/loopx/control_plane/quota/diagnostic_projection.ts index 0b3ddca22d..b9e9496f69 100644 --- a/loopx/control_plane/quota/diagnostic_projection.ts +++ b/loopx/control_plane/quota/diagnostic_projection.ts @@ -5,10 +5,9 @@ * decision. This projection grants no admission or execution authority. */ import type {JsonObject} from "../effect_program.ts"; -import {jsonObject} from "../runtime_decode.ts"; +import {jsonObject, sourceNumberTokens} from "../runtime_decode.ts"; import {TODO_CANONICAL_READ_RECORD_FIELDS} from "../coordination/coordination_state_contract.ts"; import {canonicalAuthorityJson, authorityUnicodeCompare} from "../coordination/authority_store_codec.ts"; -import {EffectRuntimeRequestError} from "../effect_runtime_errors.ts"; type ProjectionPath = (string | number)[]; type ProjectionChange = {path: ProjectionPath; value: unknown} | @@ -33,30 +32,7 @@ function contextObjectPaths(source: JsonObject): WeakMap export function contextSourceIdentity(source: JsonObject, supplied?: unknown): (value: unknown, path: ProjectionPath) => string { if (supplied === undefined) return value => JSON.stringify(canonicalAuthorityJson(value)); - if (!Array.isArray(supplied)) throw new EffectRuntimeRequestError("context source number tokens must be an array"); - const tokens: {path: ProjectionPath; token: string}[] = [], seen = new Set(); - for (const entry of supplied) { - if (!Array.isArray(entry) || entry.length !== 2 || !Array.isArray(entry[0]) || !entry[0].length || - typeof entry[1] !== "string" || !/^-?(?:0|[1-9]\d*)(?:\.\d+)?(?:[eE][+-]?\d+)?$/.test(entry[1])) { - throw new EffectRuntimeRequestError("invalid context source number token"); - } - const path: ProjectionPath = entry[0], token: string = entry[1]; - let value: unknown = source; - for (const key of path) { - if (value === null || typeof value !== "object" || - !(Array.isArray(value) ? typeof key === "number" && Number.isSafeInteger(key) && key >= 0 - : typeof key === "string") || !Object.hasOwn(value, key)) { - throw new EffectRuntimeRequestError("context source number token path is unavailable"); - } - value = (value as Record)[key]; - } - const key = JSON.stringify(path); - if (seen.has(key) || typeof value !== "number" || !Number.isFinite(value) || !Object.is(JSON.parse(token), value)) { - throw new EffectRuntimeRequestError("context source number token differs from input"); - } - seen.add(key); - tokens.push({path, token}); - } + const tokens = sourceNumberTokens(source, supplied); const paths = contextObjectPaths(source); return (value, path) => { const original = value !== null && typeof value === "object" ? paths.get(value) : undefined; diff --git a/loopx/control_plane/quota/turn_envelope.py b/loopx/control_plane/quota/turn_envelope.py index 1428f3d710..577c96b0f3 100644 --- a/loopx/control_plane/quota/turn_envelope.py +++ b/loopx/control_plane/quota/turn_envelope.py @@ -3,12 +3,11 @@ from __future__ import annotations import json -import math from collections.abc import Mapping from dataclasses import dataclass from typing import Any -from ..effect_runtime import EffectRuntimeRejected, effect_runtime_result +from ..effect_runtime import EffectRuntimeRejected, effect_runtime_result, source_number_tokens from ..scheduler.execution_context import ( SchedulerExecutionContextResolution, render_scheduler_execution_args, @@ -57,7 +56,7 @@ def _turn_envelope_result( request = {"operation": operation, **params} if operation in {"context_delivery", "context_recovery"}: - tokens = _source_number_tokens(params) + tokens = source_number_tokens(params) if tokens: request["source_number_tokens"] = tokens try: @@ -74,30 +73,6 @@ def _turn_envelope_result( return dict(result) -def _source_number_tokens(value: object) -> list[list[Any]]: - """Transport sparse numeric source identity; no carrier or sharing policy.""" - tokens: list[list[Any]] = [] - - def visit(item: object, path: list[str | int]) -> None: - if isinstance(item, float) and not math.isfinite(item): - raise ValueError("Context source numbers must be finite") - # Fractional doubles retain their identity through JSON/JavaScript. - # Only integral floats (including signed zero) and unsafe integers - # can become indistinguishable from a different Python source number. - if (isinstance(item, float) and item.is_integer() - or type(item) is int and abs(item) > 2**53 - 1): - tokens.append([path, json.dumps(item, allow_nan=False)]) - elif isinstance(item, Mapping): - for key, child in item.items(): - visit(child, [*path, key]) - elif isinstance(item, (list, tuple)): - for index, child in enumerate(item): - visit(child, [*path, index]) - - visit(value, []) - return tokens - - def assert_turn_envelope_consumer_version(envelope: Mapping[str, Any]) -> None: """Use the typed wire-version gate before any Host consumes this packet.""" _turn_envelope_result("consumer_version", envelope=dict(envelope)) diff --git a/loopx/control_plane/runtime_decode.ts b/loopx/control_plane/runtime_decode.ts index 975b7766aa..805070f999 100644 --- a/loopx/control_plane/runtime_decode.ts +++ b/loopx/control_plane/runtime_decode.ts @@ -14,6 +14,39 @@ export function requireJsonObject(value: unknown, label: string): JsonObject { return result; } +/** Sparse IO spellings, validated against the already decoded source. + * Shared by context identity and durable journal publication; no carrier, + * authority or projection decision is made here. */ +export function sourceNumberTokens(source: JsonObject, supplied?: unknown): + {path: (string | number)[]; token: string}[] { + if (supplied === undefined) return []; + if (!Array.isArray(supplied)) throw new EffectRuntimeRequestError("context source number tokens must be an array"); + const tokens: {path: (string | number)[]; token: string}[] = [], seen = new Set(); + for (const entry of supplied) { + if (!Array.isArray(entry) || entry.length !== 2 || !Array.isArray(entry[0]) || !entry[0].length || + typeof entry[1] !== "string" || !/^-?(?:0|[1-9]\d*)(?:\.\d+)?(?:[eE][+-]?\d+)?$/.test(entry[1])) { + throw new EffectRuntimeRequestError("invalid context source number token"); + } + const path: (string | number)[] = entry[0], token: string = entry[1]; + let value: unknown = source; + for (const key of path) { + if (value === null || typeof value !== "object" || + !(Array.isArray(value) ? typeof key === "number" && Number.isSafeInteger(key) && key >= 0 + : typeof key === "string") || !Object.hasOwn(value, key)) { + throw new EffectRuntimeRequestError("context source number token path is unavailable"); + } + value = (value as Record)[key]; + } + const key = JSON.stringify(path); + if (seen.has(key) || typeof value !== "number" || !Number.isFinite(value) || !Object.is(JSON.parse(token), value)) { + throw new EffectRuntimeRequestError("context source number token differs from input"); + } + seen.add(key); + tokens.push({path, token}); + } + return tokens; +} + export function requireBoolean(value: unknown, label: string): boolean { if (typeof value !== "boolean") { throw new EffectRuntimeRequestError(`${label} must be a boolean`); diff --git a/loopx/control_plane/turn_driver/turn_journal_effects.ts b/loopx/control_plane/turn_driver/turn_journal_effects.ts index 0e6e588b8c..0518c3172b 100644 --- a/loopx/control_plane/turn_driver/turn_journal_effects.ts +++ b/loopx/control_plane/turn_driver/turn_journal_effects.ts @@ -19,7 +19,7 @@ import { } from "../effect_runtime_io.ts"; import { decideFirstPartyHostRuntime } from "../goals/first_party_host_runtime.ts"; import { parseExactGoalRef } from "../goals/goal_instance_identity.ts"; -import { requireNonEmptyString as requiredString } from "../runtime_decode.ts"; +import { requireNonEmptyString as requiredString, sourceNumberTokens } from "../runtime_decode.ts"; import { preparedAttemptViolation } from "./turn_journal_attempt_contract.ts"; import { interpretTurnJournalEffect, @@ -64,26 +64,54 @@ function sha256(value: string): string { return createHash("sha256").update(value).digest("hex"); } -function stableValue(value: unknown): unknown { - if (Array.isArray(value)) return value.map(stableValue); +type SourceNumbers = WeakMap>; + +function rawNumber(token: string): unknown { + const rawJSON = (JSON as typeof JSON & {rawJSON?: (text: string) => unknown}).rawJSON; + if (!rawJSON) throw new EffectRuntimeRequestError("Turn journal number preservation requires JSON.rawJSON", "journal_snapshot_invalid"); + return rawJSON(token); +} + +function stableValue(value: unknown, numbers?: SourceNumbers): unknown { if (typeof value !== "object" || value === null) return value; + const child = (key: string, item: unknown): unknown => { + const token = numbers?.get(value)?.get(key); + return token === undefined ? stableValue(item, numbers) : rawNumber(token); + }; + if (Array.isArray(value)) return value.map((item, index) => child(String(index), item)); return Object.fromEntries( Object.entries(value as Record) .sort(([left], [right]) => left.localeCompare(right)) - .map(([key, child]) => [key, stableValue(child)]), + .map(([key, item]) => [key, child(key, item)]), ); } -function operationId(journal: JsonObject): string { - return `sha256:${sha256(JSON.stringify(stableValue(journal)))}`; +function operationId(journal: JsonObject, numbers?: SourceNumbers): string { + return `sha256:${sha256(JSON.stringify(stableValue(journal, numbers)))}`; } -function journalWithoutRecoveryAudit(journal: JsonObject): JsonObject { +function journalWithoutRecoveryAudit(journal: JsonObject, numbers?: SourceNumbers): JsonObject { const projected = { ...journal }; + const fields = numbers?.get(journal); + if (fields) numbers?.set(projected, fields); delete projected.recovery_audit; return projected; } +function parseJournal(encoded: string): {journal: JsonObject; numbers: SourceNumbers} { + const numbers: SourceNumbers = new WeakMap(); + const journal = asObject(JSON.parse(encoded, function (key, value: unknown, context?: {source?: string}) { + if (typeof value === "number" && Number.isInteger(value) && + context?.source !== undefined && context.source !== JSON.stringify(value)) { + const fields = numbers.get(this) ?? new Map(); + fields.set(key, context.source); + numbers.set(this, fields); + } + return value; + })); + return {journal, numbers}; +} + interface JournalState { status: string; completedPhases: string[]; @@ -269,15 +297,17 @@ function requireJournalTransition( incoming: JsonObject, previous: JournalState, next: JournalState, + previousNumbers?: SourceNumbers, + nextNumbers?: SourceNumbers, ): void { - if (operationId(asObject(existing.plan)) !== operationId(asObject(incoming.plan))) { + if (operationId(asObject(existing.plan), previousNumbers) !== operationId(asObject(incoming.plan), nextNumbers)) { conflict("Turn journal transaction plan is immutable", "journal_plan_conflict"); } if (terminalStatuses.has(previous.status)) { const auditOnly = previous.status === next.status && - operationId(journalWithoutRecoveryAudit(existing)) === - operationId(journalWithoutRecoveryAudit(incoming)); + operationId(journalWithoutRecoveryAudit(existing, previousNumbers), previousNumbers) === + operationId(journalWithoutRecoveryAudit(incoming, nextNumbers), nextNumbers); if (auditOnly) return; conflict("Terminal Turn journal tombstones are immutable"); } @@ -287,8 +317,8 @@ function requireJournalTransition( if ( previous.status === next.status && ["failed", "scheduler_action_required"].includes(previous.status) && - operationId(journalWithoutRecoveryAudit(existing)) !== - operationId(journalWithoutRecoveryAudit(incoming)) + operationId(journalWithoutRecoveryAudit(existing, previousNumbers), previousNumbers) !== + operationId(journalWithoutRecoveryAudit(incoming, nextNumbers), nextNumbers) ) { conflict("Settled Turn journal state only permits recovery-audit completion"); } @@ -319,22 +349,27 @@ export async function commitTurnJournal( if (!isAbsolute(path)) { throw new EffectRuntimeRequestError("Turn journal path must be absolute"); } - // Serialized input is an IO carrier, not a second transition owner. Validate - // its decoded facts below, then publish those same source bytes atomically. - // Existing native object callers keep their historical encoding. - const originalJson = typeof params.journal === "string" ? params.journal : null; - let value: unknown = params.journal; - if (originalJson !== null) { - try { - value = JSON.parse(originalJson, (_key, item: unknown) => { - if (typeof item === "number" && !Number.isFinite(item)) throw new Error("non-finite number"); - return item; - }); - } catch { - throw new EffectRuntimeRequestError("Turn journal requires valid finite JSON", "journal_snapshot_invalid"); + const source = asObject(params.journal); + const tokens = sourceNumberTokens(source, params.source_number_tokens); + let originalJson: string | null = null; + if (tokens.length) { + const numbers = new WeakMap>(); + for (const {path, token} of tokens) { + let parent: Record = source; + for (const key of path.slice(0, -1)) parent = parent[key] as typeof parent; + const fields = numbers.get(parent) ?? new Map(); + fields.set(String(path.at(-1)), token); + numbers.set(parent, fields); } + originalJson = JSON.stringify(source, function (key, value: unknown) { + const token = numbers.get(this)?.get(key); + return token === undefined ? value : rawNumber(token); + }, 2); } - const journal = asObject(value); + // Validate and publish the same isolated snapshot when source spellings are + // supplied. Native objects without metadata retain their historical path. + const {journal, numbers} = originalJson === null + ? {journal: source, numbers: undefined} : parseJournal(originalJson); const goalBinding = parseTurnJournalGoalBinding(journal); if (goalBinding.kind === "invalid") { throw new EffectRuntimeRequestError( @@ -352,13 +387,14 @@ export async function commitTurnJournal( "Turn journal does not carry the expected settlement effect", ); } - const incomingOperationId = operationId(journal); + const incomingOperationId = operationId(journal, numbers); const commit = async (): Promise => await withFileMutationLock(path, async () => { let existing: JsonObject | null = null; try { const encoded = await readFile(path, "utf8"); - existing = asObject(JSON.parse(encoded)); + const previous = parseJournal(encoded); + existing = previous.journal; const existingState = requireJournalState(existing); const existingEffectId = existingState.effectId; if ( @@ -370,7 +406,7 @@ export async function commitTurnJournal( "journal_effect_conflict", ); } - if (operationId(existing) === incomingOperationId) { + if (operationId(existing, previous.numbers) === incomingOperationId) { return { ok: true, appended: false, @@ -379,7 +415,7 @@ export async function commitTurnJournal( operation_id: incomingOperationId, }; } - requireJournalTransition(existing, journal, existingState, incomingState); + requireJournalTransition(existing, journal, existingState, incomingState, previous.numbers, numbers); } catch (error) { if ((error as NodeJS.ErrnoException).code !== "ENOENT") throw error; } diff --git a/loopx/control_plane/turn_driver/turn_journal_runtime.py b/loopx/control_plane/turn_driver/turn_journal_runtime.py index 81c86609bd..7fadc4f8a0 100644 --- a/loopx/control_plane/turn_driver/turn_journal_runtime.py +++ b/loopx/control_plane/turn_driver/turn_journal_runtime.py @@ -1,10 +1,9 @@ from __future__ import annotations -import json from collections.abc import Mapping from typing import Any -from ..effect_runtime import effect_runtime_result +from ..effect_runtime import effect_runtime_result, source_number_tokens TURN_JOURNAL_INTERPRETATION_REQUEST_SCHEMA_VERSION = ( @@ -173,11 +172,14 @@ def write_turn_journal( request: dict[str, Any] = { "path": path, - # TS owns validation, transitions and publication. Carry the original - # JSON as text so its parser/stringifier cannot round source evidence. - "journal": json.dumps(dict(journal), ensure_ascii=False, separators=(",", ":"), allow_nan=False), + "journal": dict(journal), "expected_effect_id": expected_effect_id, } + # Reuse context-delivery's sparse IO metadata. Keep the ordinary object + # carrier so quoted task/evidence text is not escaped a second time. + tokens = source_number_tokens(journal) + if tokens: + request["source_number_tokens"] = tokens if source_admission is not None: request["source_admission"] = dict(source_admission) payload = effect_runtime_result( diff --git a/tests/control_plane/test_turn_journal_goal_instance.py b/tests/control_plane/test_turn_journal_goal_instance.py index 1601eeea3b..44ec4fc97e 100644 --- a/tests/control_plane/test_turn_journal_goal_instance.py +++ b/tests/control_plane/test_turn_journal_goal_instance.py @@ -40,6 +40,44 @@ def test_journal_retains_original_numbers_through_native_commit_and_replay(tmp_p assert path.read_bytes() == original +@pytest.mark.parametrize("before,after", [(1, 1.0), (-0.0, 0), (2**60 + 1, 2**60 + 2), (float(2**60), 2**60)]) +def test_journal_immutable_plan_distinguishes_original_numeric_facts(tmp_path, before, after): + journal = _journal(None, TURN_A) + journal["plan"]["measurement"] = before + path = tmp_path / "turn.json" + write_turn_journal_checkpoint(path, journal) + original = path.read_bytes() + journal["plan"]["measurement"] = after + with pytest.raises(EffectRuntimeConflict, match="plan is immutable"): + write_turn_journal_checkpoint(path, journal) + assert path.read_bytes() == original + + +def test_journal_preserves_numbers_without_double_escaping_large_quoted_evidence(tmp_path): + journal = _journal(None, TURN_A) + evidence = {"body": ('line "value"\n' * 130_000)[:1_550_000], "number": 2**60 + 1} + journal["plan"]["original_evidence"] = evidence + path = tmp_path / "turn.json" + write_turn_journal_checkpoint(path, journal) + assert json.loads(path.read_text())["plan"]["original_evidence"] == evidence + + +@pytest.mark.parametrize("number", [float("nan"), float("inf"), float("-inf")]) +def test_journal_nonfinite_source_is_rejected_before_transport(tmp_path, monkeypatch, number): + from loopx.control_plane.turn_driver import turn_journal_runtime + + def unexpected_request(*args, **kwargs): + pytest.fail("nonfinite journal reached the runtime") + + monkeypatch.setattr(turn_journal_runtime, "effect_runtime_result", unexpected_request) + journal = _journal(None, TURN_A) + journal["plan"]["measurement"] = number + path = tmp_path / "turn.json" + with pytest.raises(ValueError, match="finite"): + write_turn_journal_checkpoint(path, journal) + assert not path.exists() + + def _registry_payload(path: Path, instance_id: str) -> dict[str, Any]: return { "schema_version": "0.2", diff --git a/tests/control_plane_ts/turn_journal_effects.test.ts b/tests/control_plane_ts/turn_journal_effects.test.ts index 14c6f5e9e5..9830852805 100644 --- a/tests/control_plane_ts/turn_journal_effects.test.ts +++ b/tests/control_plane_ts/turn_journal_effects.test.ts @@ -83,34 +83,46 @@ function sourceJournal( return snapshot; } -test("serialized journal preserves numeric source bytes and retains transition and lifetime checks", async () => { +test("sparse journal numbers survive publication, replay and existing guards", async () => { const root = await mkdtemp(join(tmpdir(), "loopx-journal-source-")); const path = join(root, "turn.json"); try { - const encoded = JSON.stringify(journal()).replace(/}$/, ',"source":[1.0,-0.0,1152921504606846977]}'); - const written = await commitTurnJournal({path, journal: encoded, expected_effect_id: effectId()}); + const snapshot = {...journal(), source: [1, -0, 2 ** 60], literal: "1.0 and 1152921504606846977"}; + const source_number_tokens = [ + [["source", 0], "1.0"], [["source", 1], "-0.0"], [["source", 2], "1152921504606846977"], + ]; + const written = await commitTurnJournal({path, journal: snapshot, source_number_tokens, expected_effect_id: effectId()}); assert.equal(written.appended, true); - assert.equal(await readFile(path, "utf8"), encoded + "\n"); - assert.equal((await commitTurnJournal({path, journal: encoded})).replayed, true); - await assert.rejects(commitTurnJournal({path, journal: JSON.stringify(journal("committed", phases))}), + const encoded = await readFile(path, "utf8"); + assert.match(encoded, /1\.0,\n\s+-0\.0,\n\s+1152921504606846977/); + assert.equal(JSON.parse(encoded).literal, snapshot.literal); + assert.ok(!encoded.includes("source_number_tokens")); + const replay = await commitTurnJournal({path, journal: snapshot, source_number_tokens}); + assert.equal(replay.replayed, true); + assert.equal(replay.operation_id, written.operation_id); + await assert.rejects(commitTurnJournal({path, journal: journal("committed", phases)}), /skip transaction checkpoints/); await assert.rejects(commitTurnJournal({path: join(root, "scoped.json"), - journal: JSON.stringify(sourceJournal(instanceA, turnKey))}), /require source admission/); - await assert.rejects(commitTurnJournal({path, journal: encoded, expected_effect_id: "other-effect"}), + journal: sourceJournal(instanceA, turnKey), source_number_tokens: []}), /require source admission/); + await assert.rejects(commitTurnJournal({path, journal: snapshot, source_number_tokens, expected_effect_id: "other-effect"}), /expected settlement effect/); - assert.equal(await readFile(path, "utf8"), encoded + "\n"); + assert.equal(await readFile(path, "utf8"), encoded); } finally { await rm(root, {recursive: true, force: true}); } }); -for (const encoded of ['{"broken":', 'null', '[]', '1', - JSON.stringify(journal()).replace(/}$/, ',"source":1e400}')]) { - test(`serialized journal rejects invalid input before writing: ${encoded.slice(0, 24)}`, async () => { +for (const tokens of [null, {}, [[[], "1"]], [[["missing"], "1"]], + [[["source"], "2"]], [[["source"], "NaN"]], [[["source"], "01"]], + [[["source"], "1"], [["source"], "1.0"]], [[["literal"], "1"]], + [[["rows", "0"], "1.0"]], [[["source"], "-0.0"]]]) { + test(`journal rejects invalid numeric metadata without writing: ${JSON.stringify(tokens)}`, async () => { const root = await mkdtemp(join(tmpdir(), "loopx-journal-invalid-")); const path = join(root, "turn.json"); try { - await assert.rejects(commitTurnJournal({path, journal: encoded}), /JSON|schema/); + await assert.rejects(commitTurnJournal({path, + journal: {...journal(), source: 1, literal: "1", rows: [1]}, source_number_tokens: tokens}), + /context source number token/); await assert.rejects(readFile(path), {code: "ENOENT"}); } finally { await rm(root, {recursive: true, force: true}); @@ -118,6 +130,57 @@ for (const encoded of ['{"broken":', 'null', '[]', '1', }); } +test("ordinary object journals retain native bytes and operation identity with no numeric metadata", async () => { + const root = await mkdtemp(join(tmpdir(), "loopx-journal-native-")); + try { + const snapshot = journal(), path = join(root, "turn.json"); + const initial = await commitTurnJournal({path, journal: snapshot}); + const replay = await commitTurnJournal({path, journal: snapshot, source_number_tokens: []}); + assert.equal(await readFile(path, "utf8"), JSON.stringify(snapshot, null, 2) + "\n"); + assert.equal(replay.operation_id, initial.operation_id); + assert.equal(replay.replayed, true); + } finally { + await rm(root, {recursive: true, force: true}); + } +}); + +test("numeric source differences are not mistaken for replay or immutable-plan equality", async () => { + const root = await mkdtemp(join(tmpdir(), "loopx-journal-number-identity-")); + try { + const snapshot = journal(), path = join(root, "turn.json"); + (snapshot.plan as Record).measurement = 1; + const initial = await commitTurnJournal({path, journal: snapshot, + source_number_tokens: [[["plan", "measurement"], "1.0"]]}); + await assert.rejects(commitTurnJournal({path, journal: snapshot}), /plan is immutable/); + const before = await readFile(path, "utf8"); + assert.match(before, /"measurement": 1\.0/); + const changed = {...snapshot, measurement: 1}; + const next = await commitTurnJournal({path, journal: changed, + source_number_tokens: [[["plan", "measurement"], "1.0"], [["measurement"], "1.0"]]}); + assert.equal(next.appended, true); + assert.notEqual(next.operation_id, initial.operation_id); + } finally { + await rm(root, {recursive: true, force: true}); + } +}); + +test("missing raw-number support fails before writing and leaves ordinary objects available", async () => { + const root = await mkdtemp(join(tmpdir(), "loopx-journal-number-support-")); + const descriptor = Object.getOwnPropertyDescriptor(JSON, "rawJSON"); + try { + Object.defineProperty(JSON, "rawJSON", {value: undefined, configurable: true}); + const path = join(root, "source.json"); + await assert.rejects(commitTurnJournal({path, journal: {...journal(), value: 1}, + source_number_tokens: [[["value"], "1.0"]]}), /requires JSON.rawJSON/); + await assert.rejects(readFile(path), {code: "ENOENT"}); + assert.equal((await commitTurnJournal({path: join(root, "native.json"), journal: journal()})).ok, true); + } finally { + if (descriptor) Object.defineProperty(JSON, "rawJSON", descriptor); + else Reflect.deleteProperty(JSON, "rawJSON"); + await rm(root, {recursive: true, force: true}); + } +}); + function sourceGuardPath(registryPath: string, goalId: string): string { const alias = createHash("sha256").update(goalId, "utf8").digest("hex"); return join(