Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 19 additions & 2 deletions loopx/control_plane/turn_driver/turn_journal_effects.ts
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@ import {
} from "../effect_runtime_errors.ts";
import {
atomicWriteJson,
atomicWriteText,
claimFileMutationLock,
mutationLockOwner,
releaseFileMutationLock,
Expand Down Expand Up @@ -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(
Expand Down Expand Up @@ -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,
Expand Down
5 changes: 4 additions & 1 deletion loopx/control_plane/turn_driver/turn_journal_runtime.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
from __future__ import annotations

import json
from collections.abc import Mapping
from typing import Any

Expand Down Expand Up @@ -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:
Expand Down
13 changes: 13 additions & 0 deletions tests/control_plane/test_turn_journal_goal_instance.py
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand Down
35 changes: 35 additions & 0 deletions tests/control_plane_ts/turn_journal_effects.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand Down
Loading