diff --git a/CHANGELOG.md b/CHANGELOG.md index e6bf2c7..ab8268c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,7 @@ All notable changes to Lobster will be documented in this file. - Require Node.js 22 or newer for the npm package, matching release CI. - Write Lobster state files atomically while preserving restricted file modes, preventing truncated resume/session state after process termination. Thanks to [@KrasimirKralev](https://github.com/KrasimirKralev) (Issues [#108](https://github.com/openclaw/lobster/issues/108), [#109](https://github.com/openclaw/lobster/issues/109), PR [#110](https://github.com/openclaw/lobster/pull/110)). +- Harden LLM cache files, diff snapshots, and approval ID indexes against truncated JSON after process termination. Disposable cache/snapshot corruption now recovers as a miss, authoritative resume state still surfaces malformed JSON, and approval short-ID indexes are published atomically without overwriting existing mappings. Thanks to [@TurboTheTurtle](https://github.com/TurboTheTurtle) (Issues [#111](https://github.com/openclaw/lobster/issues/111), [#112](https://github.com/openclaw/lobster/issues/112), [#113](https://github.com/openclaw/lobster/issues/113), PR [#114](https://github.com/openclaw/lobster/pull/114)). - Fix `timeout_ms` + `retry` so per-attempt timeouts retry as documented while external workflow cancellation still stops immediately. Thanks to [@KrasimirKralev](https://github.com/KrasimirKralev) (PR [#106](https://github.com/openclaw/lobster/pull/106)). ## 2026.5.22 diff --git a/src/commands/stdlib/llm_invoke.ts b/src/commands/stdlib/llm_invoke.ts index 06a6c29..ad15ce3 100644 --- a/src/commands/stdlib/llm_invoke.ts +++ b/src/commands/stdlib/llm_invoke.ts @@ -4,7 +4,14 @@ import { createHash } from "node:crypto"; import { Ajv } from "ajv"; import type { ErrorObject } from "ajv"; -import { readStateJson, writeStateJson, stableStringify } from "../../state/store.js"; +import { + ensureDirectory, + isJsonSyntaxError, + readStateJson, + stableStringify, + writeFileAtomic, + writeStateJson, +} from "../../state/store.js"; import { createCompileCached } from "../../validation.js"; import type { LobsterCommand } from "../types.js"; @@ -378,7 +385,7 @@ async function runLlmInvoke({ }); if (stateKey && !forceRefresh) { - const stored = await readStateJson({ env, key: stateKey }).catch(() => null); + const stored = await readReusableLlmState(env, stateKey); const reused = pickReusableState(stored, cacheKey, config.stateType); if (reused) { return { @@ -887,6 +894,15 @@ async function persistOutputs({ await writeStateJson({ env, key: stateKey, value: record }); } +async function readReusableLlmState(env: any, stateKey: string) { + try { + return await readStateJson({ env, key: stateKey }); + } catch (err: any) { + if (isJsonSyntaxError(err)) return null; + throw err; + } +} + function pickReusableState(stored: any, cacheKey: string, stateType: string) { if (!stored || typeof stored !== "object") return null; if (stored.type !== stateType) return null; @@ -908,9 +924,12 @@ async function readCacheEntry( const filePath = path.join(getCacheDir(env), cacheNamespace, `${key}.json`); try { const text = await fsp.readFile(filePath, "utf8"); - return JSON.parse(text) as CacheEntry; + const parsed = JSON.parse(text) as Partial; + if (parsed?.cacheKey !== key || !Array.isArray(parsed.items)) return null; + return parsed as CacheEntry; } catch (err: any) { if (err?.code === "ENOENT") return null; + if (isJsonSyntaxError(err)) return null; throw err; } } @@ -922,11 +941,11 @@ async function writeCacheEntry( cacheNamespace: string, ) { const dir = path.join(getCacheDir(env), cacheNamespace); - await fsp.mkdir(dir, { recursive: true }); + await ensureDirectory(dir); const filePath = path.join(dir, `${key}.json`); - await fsp.writeFile( + await writeFileAtomic( filePath, - JSON.stringify({ items, cacheKey: key, storedAt: new Date().toISOString() }, null, 2), + JSON.stringify({ items, cacheKey: key, storedAt: new Date().toISOString() }, null, 2) + "\n", ); } diff --git a/src/commands/stdlib/state.ts b/src/commands/stdlib/state.ts index eacfba7..e1e6e64 100644 --- a/src/commands/stdlib/state.ts +++ b/src/commands/stdlib/state.ts @@ -1,6 +1,6 @@ import { promises as fsp } from "node:fs"; -import { defaultStateDir, keyToPath, writeFileAtomic } from "../../state/store.js"; +import { defaultStateDir, ensureDirectory, keyToPath, writeFileAtomic } from "../../state/store.js"; export const stateGetCommand = { name: "state.get", @@ -69,7 +69,7 @@ export const stateSetCommand = { const stateDir = defaultStateDir(ctx.env); const filePath = keyToPath(stateDir, key); - await fsp.mkdir(stateDir, { recursive: true }); + await ensureDirectory(stateDir); await writeFileAtomic(filePath, JSON.stringify(value, null, 2) + "\n"); return { output: asStream([value]) }; diff --git a/src/pipeline_resume_state.ts b/src/pipeline_resume_state.ts index 8a26d54..ea2a37c 100644 --- a/src/pipeline_resume_state.ts +++ b/src/pipeline_resume_state.ts @@ -52,7 +52,7 @@ export type PipelineToolRunResolution = items: unknown[]; preview?: string; resumeToken: string; - approvalId: string; + approvalId?: string; }; requiresInput: null; } @@ -108,7 +108,7 @@ export async function finalizePipelineToolRun(params: { await cleanupApprovalIndexByStateKey({ env: params.env, stateKey: params.previousStateKey }); await deleteStateJson({ env: params.env, key: params.previousStateKey }); } - let approvalId: string; + let approvalId: string | null; try { approvalId = await createApprovalIndex({ env: params.env, stateKey: nextStateKey }); } catch (err) { @@ -127,7 +127,7 @@ export async function finalizePipelineToolRun(params: { requiresApproval: { ...approval, resumeToken, - approvalId, + ...(approvalId ? { approvalId } : null), }, requiresInput: null, }; diff --git a/src/sdk/primitives/diff.ts b/src/sdk/primitives/diff.ts index 7470057..f400c58 100644 --- a/src/sdk/primitives/diff.ts +++ b/src/sdk/primitives/diff.ts @@ -17,6 +17,7 @@ import { promises as fsp } from "node:fs"; import os from "node:os"; import path from "node:path"; +import { ensureDirectory, isJsonSyntaxError, writeFileAtomic } from "../../state/store.js"; /** * Get the state directory @@ -103,7 +104,7 @@ export function diffLast(key, options: any = {}) { const text = await fsp.readFile(filePath, "utf8"); before = JSON.parse(text); } catch (err) { - if (err?.code !== "ENOENT") { + if (err?.code !== "ENOENT" && !isJsonSyntaxError(err)) { throw err; } } @@ -112,8 +113,8 @@ export function diffLast(key, options: any = {}) { const changed = stableStringify(before) !== stableStringify(value); // Store new value - await fsp.mkdir(stateDir, { recursive: true }); - await fsp.writeFile(filePath, JSON.stringify(value, null, 2) + "\n", "utf8"); + await ensureDirectory(stateDir); + await writeFileAtomic(filePath, JSON.stringify(value, null, 2) + "\n"); // Build result const result = { @@ -159,7 +160,7 @@ export async function diffAndStoreValue(key, value, ctx = {}) { const text = await fsp.readFile(filePath, "utf8"); before = JSON.parse(text); } catch (err) { - if (err?.code !== "ENOENT") { + if (err?.code !== "ENOENT" && !isJsonSyntaxError(err)) { throw err; } } @@ -168,8 +169,8 @@ export async function diffAndStoreValue(key, value, ctx = {}) { const changed = stableStringify(before) !== stableStringify(value); // Store new value - await fsp.mkdir(stateDir, { recursive: true }); - await fsp.writeFile(filePath, JSON.stringify(value, null, 2) + "\n", "utf8"); + await ensureDirectory(stateDir); + await writeFileAtomic(filePath, JSON.stringify(value, null, 2) + "\n"); return { before, after: value, changed }; } diff --git a/src/state/store.ts b/src/state/store.ts index 8d787d7..3810da3 100644 --- a/src/state/store.ts +++ b/src/state/store.ts @@ -33,6 +33,89 @@ export function stableStringify(value) { }); } +type AtomicWriteOptions = { + renameFile?: typeof fsp.rename; + syncParentDir?: (filePath: string) => Promise; +}; + +type AtomicExclusiveWriteOptions = { + linkFile?: typeof fsp.link; + syncParentDir?: (filePath: string) => Promise; +}; + +function isDirectorySyncUnsupportedError(err: any): boolean { + return [ + "EACCES", + "EBADF", + "EINVAL", + "EISDIR", + "ENOSYS", + "ENOTSUP", + "EOPNOTSUPP", + "EPERM", + ].includes(err?.code); +} + +async function syncParentDir(filePath: string) { + await syncDirectory(path.dirname(filePath)); +} + +async function syncDirectory(dir: string) { + let handle; + try { + handle = await fsp.open(dir, "r"); + } catch (err) { + if (isDirectorySyncUnsupportedError(err)) return; + throw err; + } + + try { + await handle.sync(); + } catch (err) { + if (!isDirectorySyncUnsupportedError(err)) throw err; + } finally { + if (handle) await handle.close().catch(() => {}); + } +} + +async function syncCreatedDirectoryChain(firstCreated: string, finalDir: string) { + const final = path.resolve(finalDir); + let current = path.resolve(firstCreated); + + await syncDirectory(path.dirname(current)); + while (current !== final) { + await syncDirectory(current); + const relative = path.relative(current, final); + const next = relative.split(path.sep)[0]; + if (!next || next === "..") break; + current = path.join(current, next); + } +} + +export async function ensureDirectory(dir: string) { + const created = await fsp.mkdir(dir, { recursive: true }); + if (created) await syncCreatedDirectoryChain(created, dir); +} + +export function isJsonSyntaxError(err) { + return err instanceof SyntaxError; +} + +function isLinkUnsupportedError(err: any): boolean { + return ["ENOSYS", "ENOTSUP", "EOPNOTSUPP", "EPERM", "EXDEV"].includes(err?.code); +} + +export function isAtomicExclusiveUnsupportedError(err: any): boolean { + return err?.code === "ENOTSUP" && err?.cause && isLinkUnsupportedError(err.cause); +} + +function isOptionalApprovalIndexPersistenceError(err: any): boolean { + return ( + isAtomicExclusiveUnsupportedError(err) || + ["EACCES", "EDQUOT", "EIO", "ENOSPC", "EPERM", "EROFS"].includes(err?.code) + ); +} + /** * Write a file atomically: stage to a sibling temp file, fsync, then rename * over the target. `rename(2)` is atomic on a single filesystem, so a reader @@ -42,7 +125,9 @@ export function stableStringify(value) { * power loss. New state files are private by default; existing file modes are * preserved across replacement. The temp file is removed on any failed path. */ -export async function writeFileAtomic(filePath, data) { +export async function writeFileAtomic(filePath, data, options: AtomicWriteOptions = {}) { + const renameFile = options.renameFile ?? fsp.rename; + const syncDir = options.syncParentDir ?? syncParentDir; const dir = path.dirname(filePath); const tmpPath = path.join( dir, @@ -59,11 +144,12 @@ export async function writeFileAtomic(filePath, data) { } handle = await fsp.open(tmpPath, "wx", mode); await handle.writeFile(data, "utf8"); + await handle.chmod(mode); await handle.sync(); await handle.close(); handle = undefined; - await fsp.chmod(tmpPath, mode); - await fsp.rename(tmpPath, filePath); + await renameFile(tmpPath, filePath); + await syncDir(filePath); cleanup = false; } finally { if (handle) await handle.close().catch(() => {}); @@ -71,6 +157,51 @@ export async function writeFileAtomic(filePath, data) { } } +export async function writeFileAtomicExclusive( + filePath, + data, + options: AtomicExclusiveWriteOptions = {}, +) { + const linkFile = options.linkFile ?? fsp.link; + const syncDir = options.syncParentDir ?? syncParentDir; + const dir = path.dirname(filePath); + const tmpPath = path.join( + dir, + `.${path.basename(filePath)}.${randomBytes(6).toString("hex")}.tmp`, + ); + let handle; + try { + handle = await fsp.open(tmpPath, "wx", 0o600); + await handle.writeFile(data, "utf8"); + await handle.chmod(0o600); + await handle.sync(); + await handle.close(); + handle = undefined; + try { + await linkFile(tmpPath, filePath); + } catch (err) { + if (!isLinkUnsupportedError(err)) throw err; + const unsupported = new Error( + "Atomic exclusive file creation requires hard-link support on this filesystem", + ); + (unsupported as NodeJS.ErrnoException).code = "ENOTSUP"; + (unsupported as Error).cause = err; + throw unsupported; + } + try { + await fsp.unlink(tmpPath); + await syncDir(filePath); + } catch (err) { + await fsp.unlink(filePath).catch(() => {}); + await syncDir(filePath).catch(() => {}); + throw err; + } + } finally { + if (handle) await handle.close().catch(() => {}); + await fsp.rm(tmpPath, { force: true }).catch(() => {}); + } +} + export async function readStateJson({ env, key }) { const stateDir = defaultStateDir(env); const filePath = keyToPath(stateDir, key); @@ -88,7 +219,7 @@ export async function writeStateJson({ env, key, value }) { const stateDir = defaultStateDir(env); const filePath = keyToPath(stateDir, key); - await fsp.mkdir(stateDir, { recursive: true }); + await ensureDirectory(stateDir); await writeFileAtomic(filePath, JSON.stringify(value, null, 2) + "\n"); } @@ -124,20 +255,22 @@ export async function writeApprovalIndex({ env, stateKey, approvalId, + options, }: { env: Record; stateKey: string; approvalId: string; + options?: AtomicExclusiveWriteOptions; }) { const stateDir = defaultStateDir(env); const safe = sanitizeApprovalId(approvalId); if (!safe) return; - await fsp.mkdir(stateDir, { recursive: true }); + await ensureDirectory(stateDir); const indexPath = path.join(stateDir, `approval_${safe}.json`); - await fsp.writeFile( + await writeFileAtomicExclusive( indexPath, JSON.stringify({ stateKey, createdAt: new Date().toISOString() }) + "\n", - { encoding: "utf8", flag: "wx", mode: 0o600 }, + options, ); } @@ -147,17 +280,20 @@ export async function writeApprovalIndex({ export async function createApprovalIndex({ env, stateKey, + options, }: { env: Record; stateKey: string; -}) { + options?: AtomicExclusiveWriteOptions; +}): Promise { for (let attempt = 0; attempt < 16; attempt++) { const approvalId = generateApprovalId(); try { - await writeApprovalIndex({ env, stateKey, approvalId }); + await writeApprovalIndex({ env, stateKey, approvalId, options }); return approvalId; } catch (err: any) { if (err?.code === "EEXIST") continue; + if (isOptionalApprovalIndexPersistenceError(err)) return null; throw err; } } @@ -185,6 +321,7 @@ export async function findStateKeyByApprovalId({ return typeof data?.stateKey === "string" ? data.stateKey : null; } catch (err: any) { if (err?.code === "ENOENT") return null; + if (isJsonSyntaxError(err)) return null; throw err; } } @@ -247,7 +384,10 @@ export async function cleanupApprovalIndexByStateKey({ } export async function diffAndStore({ env, key, value }) { - const before = await readStateJson({ env, key }); + const before = await readStateJson({ env, key }).catch((err) => { + if (isJsonSyntaxError(err)) return null; + throw err; + }); const changed = stableStringify(before) !== stableStringify(value); await writeStateJson({ env, key, value }); return { before, after: value, changed }; diff --git a/src/workflows/file.ts b/src/workflows/file.ts index a2e8c98..64334fd 100644 --- a/src/workflows/file.ts +++ b/src/workflows/file.ts @@ -1293,7 +1293,7 @@ export async function runWorkflowFile({ createdAt: new Date().toISOString(), }); - let approvalId: string; + let approvalId: string | null; try { approvalId = await createApprovalIndex({ env: ctx.env, stateKey }); } catch (err) { @@ -1318,7 +1318,7 @@ export async function runWorkflowFile({ requiresApproval: { ...approval, resumeToken, - approvalId, + ...(approvalId ? { approvalId } : null), }, }; } diff --git a/test/approval_id.test.ts b/test/approval_id.test.ts index 8b9ed4d..ce7ebb8 100644 --- a/test/approval_id.test.ts +++ b/test/approval_id.test.ts @@ -213,3 +213,19 @@ test("approval index writes never overwrite an existing approval ID mapping", as const resolved = await findStateKeyByApprovalId({ env, approvalId: "deadbeef" }); assert.equal(resolved, "workflow_resume_original"); }); + +test("corrupt approval index is treated as expired instead of crashing (#113)", async () => { + const tmpDir = await fsp.mkdtemp(path.join(os.tmpdir(), "lobster-aid-corrupt-")); + const stateDir = path.join(tmpDir, "state"); + const env = { LOBSTER_STATE_DIR: stateDir }; + await fsp.mkdir(stateDir, { recursive: true }); + await fsp.writeFile(path.join(stateDir, "approval_deadbeef.json"), '{"stateKey"', "utf8"); + + const resolved = await findStateKeyByApprovalId({ env, approvalId: "deadbeef" }); + assert.equal(resolved, null); + + const resumed = runCli(["resume", "--id", "deadbeef", "--approve", "yes"], env); + const json = JSON.parse(resumed.stdout); + assert.equal(json.ok, false); + assert.match(json.error?.message ?? "", /not found or expired/); +}); diff --git a/test/llm_task_invoke.test.ts b/test/llm_task_invoke.test.ts index 42410c7..69fa2b8 100644 --- a/test/llm_task_invoke.test.ts +++ b/test/llm_task_invoke.test.ts @@ -1,7 +1,7 @@ import test from "node:test"; import assert from "node:assert/strict"; import http from "node:http"; -import { mkdtemp, rm } from "node:fs/promises"; +import { mkdtemp, readdir, readFile, rm, stat, writeFile } from "node:fs/promises"; import { tmpdir } from "node:os"; import path from "node:path"; @@ -298,6 +298,87 @@ test("llm_task.invoke reuses file cache when URL unavailable", async () => { } }); +test("llm_task.invoke treats corrupt file cache as a miss and rewrites it atomically (#111)", async () => { + const cacheDir = await mkdtemp(path.join(tmpdir(), "lobster-cache-corrupt-")); + const registry = createDefaultRegistry(); + const cmd = registry.get("llm_task.invoke"); + assert.ok(cmd); + + let calls = 0; + const server = http.createServer((req, res) => { + if (req.method !== "POST" || req.url !== "/tools/invoke") { + res.writeHead(404); + res.end("not found"); + return; + } + calls += 1; + res.writeHead(200, { "content-type": "application/json" }); + res.end( + JSON.stringify({ + ok: true, + result: { + ok: true, + result: { runId: `cache_repair_${calls}`, output: { text: `fresh ${calls}` } }, + }, + }), + ); + }); + await new Promise((resolve) => server.listen(0, resolve)); + const addr = server.address(); + const port = typeof addr === "object" && addr ? addr.port : 0; + const ctxEnv = { LOBSTER_CACHE_DIR: cacheDir, CLAWD_URL: `http://localhost:${port}` }; + + try { + const args = { _: [], model: "claude", prompt: "Repair corrupt cache" }; + const first = await cmd.run({ + input: streamOf([]), + args, + ctx: baseCtx(ctxEnv, registry), + } as any); + const firstItems = await collect(first.output!); + assert.equal(firstItems[0].runId, "cache_repair_1"); + assert.equal(calls, 1); + + const namespaceDir = path.join(cacheDir, "llm_task.invoke"); + const cacheFiles = (await readdir(namespaceDir)).filter((name) => name.endsWith(".json")); + assert.equal(cacheFiles.length, 1); + const cachePath = path.join(namespaceDir, cacheFiles[0]); + assert.equal((await stat(cachePath)).mode & 0o777, 0o600); + await writeFile(cachePath, '{"items"', "utf8"); + + const second = await cmd.run({ + input: streamOf([]), + args, + ctx: baseCtx(ctxEnv, registry), + } as any); + const secondItems = await collect(second.output!); + assert.equal(secondItems[0].runId, "cache_repair_2"); + assert.equal(secondItems[0].source, "clawd"); + assert.equal(secondItems[0].cached, false); + assert.equal(calls, 2); + + const repaired = JSON.parse(await readFile(cachePath, "utf8")); + assert.equal(repaired.items[0].runId, "cache_repair_2"); + + await writeFile( + cachePath, + JSON.stringify({ cacheKey: repaired.cacheKey, items: null }), + "utf8", + ); + const third = await cmd.run({ + input: streamOf([]), + args, + ctx: baseCtx(ctxEnv, registry), + } as any); + const thirdItems = await collect(third.output!); + assert.equal(thirdItems[0].runId, "cache_repair_3"); + assert.equal(calls, 3); + } finally { + await rm(cacheDir, { recursive: true, force: true }); + await closeServer(server); + } +}); + test("llm_task.invoke uses CLAWD_URL (/tools/invoke) without requiring --url/--model", async () => { const registry = createDefaultRegistry(); const cmd = registry.get("llm_task.invoke"); diff --git a/test/state.test.ts b/test/state.test.ts index 455b08b..44e4654 100644 --- a/test/state.test.ts +++ b/test/state.test.ts @@ -5,8 +5,16 @@ import path from "node:path"; import { mkdtempSync, promises as fsp } from "node:fs"; import { createDefaultRegistry } from "../src/commands/registry.js"; import { runPipeline } from "../src/runtime.js"; +import { diffLast, diffAndStoreValue } from "../src/sdk/primitives/diff.js"; import { stateSet, readState, writeState } from "../src/sdk/primitives/state.js"; -import { writeStateJson, readStateJson, writeFileAtomic } from "../src/state/store.js"; +import { + createApprovalIndex, + diffAndStore, + writeStateJson, + readStateJson, + writeFileAtomic, + writeFileAtomicExclusive, +} from "../src/state/store.js"; function streamOf(items) { return (async function* () { @@ -142,6 +150,215 @@ test("writeFileAtomic removes temp files when replacement fails", async () => { assert.deepEqual(leftovers, []); }); +test("writeFileAtomic leaves existing target untouched when publish fails", async () => { + const tmp = mkdtempSync(path.join(os.tmpdir(), "lobster-atomic-fault-")); + const target = path.join(tmp, "state.json"); + await fsp.writeFile(target, '{"old":true}\n', { mode: 0o600 }); + const fault = Object.assign(new Error("rename failed"), { code: "EIO" }); + + await assert.rejects( + () => + writeFileAtomic(target, '{"new":true}\n', { + async renameFile() { + throw fault; + }, + }), + (err: NodeJS.ErrnoException) => err?.code === "EIO", + ); + + assert.equal(await fsp.readFile(target, "utf8"), '{"old":true}\n'); + const leftovers = (await fsp.readdir(tmp)).filter((f) => f.includes(".tmp")); + assert.deepEqual(leftovers, []); +}); + +test("writeFileAtomic propagates parent directory sync failures", async () => { + const tmp = mkdtempSync(path.join(os.tmpdir(), "lobster-atomic-dir-sync-")); + const target = path.join(tmp, "state.json"); + const fault = Object.assign(new Error("dir sync failed"), { code: "EIO" }); + + await assert.rejects( + () => + writeFileAtomic(target, '{"ok":true}\n', { + async syncParentDir() { + throw fault; + }, + }), + (err: NodeJS.ErrnoException) => err?.code === "EIO", + ); + + assert.equal(await fsp.readFile(target, "utf8"), '{"ok":true}\n'); + const leftovers = (await fsp.readdir(tmp)).filter((f) => f.includes(".tmp")); + assert.deepEqual(leftovers, []); +}); + +test("readStateJson surfaces malformed authoritative state", async () => { + const tmp = mkdtempSync(path.join(os.tmpdir(), "lobster-state-corrupt-")); + const env = { LOBSTER_STATE_DIR: tmp }; + await fsp.writeFile(path.join(tmp, "resume.json"), '{"partial"', "utf8"); + + await assert.rejects(() => readStateJson({ env, key: "resume" }), SyntaxError); +}); + +test("writeFileAtomicExclusive creates private files without replacing existing targets", async () => { + const tmp = mkdtempSync(path.join(os.tmpdir(), "lobster-atomic-exclusive-")); + const target = path.join(tmp, "approval_deadbeef.json"); + + await writeFileAtomicExclusive(target, '{"stateKey":"original"}\n'); + assert.equal((await fsp.stat(target)).mode & 0o777, 0o600); + + await assert.rejects( + () => writeFileAtomicExclusive(target, '{"stateKey":"replacement"}\n'), + (err: NodeJS.ErrnoException) => err?.code === "EEXIST", + ); + assert.equal(await fsp.readFile(target, "utf8"), '{"stateKey":"original"}\n'); + + const leftovers = (await fsp.readdir(tmp)).filter((f) => f.includes(".tmp")); + assert.deepEqual(leftovers, []); +}); + +test("writeFileAtomicExclusive removes temp link before final directory sync", async () => { + const tmp = mkdtempSync(path.join(os.tmpdir(), "lobster-atomic-exclusive-sync-order-")); + const target = path.join(tmp, "approval_deadbeef.json"); + let filesAtSync: string[] = []; + + await writeFileAtomicExclusive(target, '{"stateKey":"original"}\n', { + async syncParentDir() { + filesAtSync = await fsp.readdir(tmp); + }, + }); + + assert.equal(await fsp.readFile(target, "utf8"), '{"stateKey":"original"}\n'); + assert.ok(filesAtSync.includes("approval_deadbeef.json")); + assert.deepEqual( + filesAtSync.filter((file) => file.includes(".tmp")), + [], + ); +}); + +test("writeFileAtomicExclusive rejects unsupported hard links without a partial target", async () => { + const tmp = mkdtempSync(path.join(os.tmpdir(), "lobster-atomic-exclusive-unsupported-")); + const target = path.join(tmp, "approval_deadbeef.json"); + const unsupported = Object.assign(new Error("operation not supported"), { code: "ENOTSUP" }); + const options = { + async linkFile() { + throw unsupported; + }, + }; + + await assert.rejects( + () => writeFileAtomicExclusive(target, '{"stateKey":"original"}\n', options), + (err: NodeJS.ErrnoException) => err?.code === "ENOTSUP", + ); + await assert.rejects( + () => fsp.stat(target), + (err: NodeJS.ErrnoException) => err?.code === "ENOENT", + ); + + const leftovers = (await fsp.readdir(tmp)).filter((f) => f.includes(".tmp")); + assert.deepEqual(leftovers, []); +}); + +test("writeFileAtomicExclusive removes published target when parent directory sync fails", async () => { + const tmp = mkdtempSync(path.join(os.tmpdir(), "lobster-atomic-exclusive-dir-sync-")); + const target = path.join(tmp, "approval_deadbeef.json"); + const fault = Object.assign(new Error("dir sync failed"), { code: "EIO" }); + + await assert.rejects( + () => + writeFileAtomicExclusive(target, '{"stateKey":"original"}\n', { + async syncParentDir() { + throw fault; + }, + }), + (err: NodeJS.ErrnoException) => err?.code === "EIO", + ); + await assert.rejects( + () => fsp.stat(target), + (err: NodeJS.ErrnoException) => err?.code === "ENOENT", + ); + + const leftovers = (await fsp.readdir(tmp)).filter((f) => f.includes(".tmp")); + assert.deepEqual(leftovers, []); +}); + +test("createApprovalIndex omits short ID when atomic exclusive publish is unsupported", async () => { + const tmp = mkdtempSync(path.join(os.tmpdir(), "lobster-approval-index-unsupported-")); + const env = { LOBSTER_STATE_DIR: tmp }; + const unsupported = Object.assign(new Error("operation not supported"), { code: "ENOTSUP" }); + + const approvalId = await createApprovalIndex({ + env, + stateKey: "workflow_resume_1", + options: { + async linkFile() { + throw unsupported; + }, + }, + }); + + assert.equal(approvalId, null); + const files = await fsp.readdir(tmp); + assert.deepEqual(files, []); +}); + +test("createApprovalIndex omits short ID when approval index durability fails", async () => { + const tmp = mkdtempSync(path.join(os.tmpdir(), "lobster-approval-index-sync-fails-")); + const env = { LOBSTER_STATE_DIR: tmp }; + const fault = Object.assign(new Error("dir sync failed"), { code: "EIO" }); + + const approvalId = await createApprovalIndex({ + env, + stateKey: "workflow_resume_1", + options: { + async syncParentDir() { + throw fault; + }, + }, + }); + + assert.equal(approvalId, null); + const files = await fsp.readdir(tmp); + assert.deepEqual(files, []); +}); + +test("diffAndStore treats corrupt previous state as a miss and rewrites atomically (#112)", async () => { + const tmp = mkdtempSync(path.join(os.tmpdir(), "lobster-diff-corrupt-")); + const env = { LOBSTER_STATE_DIR: tmp }; + await fsp.writeFile(path.join(tmp, "snapshot.json"), '{"partial"', "utf8"); + + const result = await diffAndStore({ env, key: "snapshot", value: { ok: true } }); + + assert.equal(result.before, null); + assert.equal(result.changed, true); + assert.deepEqual(await readStateJson({ env, key: "snapshot" }), { ok: true }); +}); + +test("SDK diff primitives treat corrupt previous state as a miss (#112)", async () => { + const tmp = mkdtempSync(path.join(os.tmpdir(), "lobster-sdk-diff-corrupt-")); + const ctx = { env: { LOBSTER_STATE_DIR: tmp } }; + await fsp.writeFile(path.join(tmp, "sdk-snapshot.json"), '{"partial"', "utf8"); + + const direct = await diffAndStoreValue("sdk-snapshot", { next: true }, ctx); + assert.equal(direct.before, null); + assert.equal(direct.changed, true); + + await fsp.writeFile(path.join(tmp, "stage-snapshot.json"), '{"partial"', "utf8"); + const stage = diffLast("stage-snapshot"); + const result = await stage.run({ input: streamOf([{ next: true }]), ctx }); + const output = []; + for await (const item of result.output) output.push(item); + + assert.deepEqual(output, [ + { + kind: "diff.last", + key: "stage-snapshot", + changed: true, + before: null, + after: { next: true }, + }, + ]); +}); + test("SDK stateSet/readState is atomic under concurrent reads (#109)", async () => { const tmp = mkdtempSync(path.join(os.tmpdir(), "lobster-atomic-sdk-")); const ctx = { env: { LOBSTER_STATE_DIR: tmp } };