diff --git a/docs/architecture/rfcs/controller-executor-unified-contract-v0.md b/docs/architecture/rfcs/controller-executor-unified-contract-v0.md index 57601013b0..3655006cff 100644 --- a/docs/architecture/rfcs/controller-executor-unified-contract-v0.md +++ b/docs/architecture/rfcs/controller-executor-unified-contract-v0.md @@ -130,6 +130,21 @@ 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 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 | | --- | --- | --- | | `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..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,6 +104,17 @@ Quota capture 与 Turn journal capture 都调用 context-delivery owner。C1 应收敛剩余的组合逻辑与消费者假设，不另建 capture service， 也不重写已经共享的差分与 native-resume 逻辑。 +Journal 持久化是独立的来源保真前置。本次提议由 Python adapter 复用 context +delivery 的稀疏数值 token 收集器；两个 TypeScript 消费者复用同一来源数值校验器。 +Journal 继续传递 JSON 对象，仅为整值浮点、带符号零和非安全整数增加元数据。 +Journal owner 在原子发布前恢复这些数值，并在重放与不可变 plan 比较中保留其 +区别。普通原生对象保留已有编码，不改写历史回执。 + +中间的完整 JSON 字符串方案，在合成的 155 万字符带引号正文上超过既有 2 MiB +请求边界；原对象路径能够接收同一负载。保留这一负面结果：稀疏替代须在相同 +负载与上限下比较，包括数值密集时的元数据开销。该前置或组件字节减少，都不 +构成统一 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..c4c0bff3c8 100644 --- a/docs/reference/protocols/loopx-turn-v0.md +++ b/docs/reference/protocols/loopx-turn-v0.md @@ -516,6 +516,24 @@ 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 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 `null`, and adds `journal_observation` from this same read-only TS owner. Its 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 730302f7bb..0518c3172b 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, @@ -18,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, @@ -63,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[]; @@ -268,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"); } @@ -286,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"); } @@ -318,7 +349,27 @@ export async function commitTurnJournal( if (!isAbsolute(path)) { throw new EffectRuntimeRequestError("Turn journal path must be absolute"); } - const journal = asObject(params.journal); + 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); + } + // 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( @@ -336,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 ( @@ -354,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, @@ -363,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; } @@ -373,7 +425,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..7fadc4f8a0 100644 --- a/loopx/control_plane/turn_driver/turn_journal_runtime.py +++ b/loopx/control_plane/turn_driver/turn_journal_runtime.py @@ -3,7 +3,7 @@ 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 = ( @@ -175,6 +175,11 @@ def write_turn_journal( "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 11d86eeac4..44ec4fc97e 100644 --- a/tests/control_plane/test_turn_journal_goal_instance.py +++ b/tests/control_plane/test_turn_journal_goal_instance.py @@ -27,6 +27,57 @@ 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 + + +@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 d484160691..9830852805 100644 --- a/tests/control_plane_ts/turn_journal_effects.test.ts +++ b/tests/control_plane_ts/turn_journal_effects.test.ts @@ -83,6 +83,104 @@ function sourceJournal( return snapshot; } +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 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); + 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: 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); + } finally { + await rm(root, {recursive: true, force: true}); + } +}); + +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: {...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}); + } + }); +} + +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(