fix: harden disposable persistence writes

Harden LLM cache files, diff snapshots, and approval ID indexes against truncated or malformed JSON after process termination.

Disposable cache and snapshot corruption now recovers as a miss; authoritative resume state still surfaces malformed JSON. Approval short-ID indexes use atomic no-overwrite publication and degrade to full resume-token approval when the index cannot be published durably.

Closes #111.
Closes #112.
Closes #113.

Co-authored-by: Andy Ye <35905412+TurboTheTurtle@users.noreply.github.com>

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

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<CacheEntry>;
+    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",
   );
 }
 

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]) };

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,
     };

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 };
 }

src/state/store.ts

@@ -33,6 +33,89 @@ export function stableStringify(value) {
   });
 }
 
+type AtomicWriteOptions = {
+  renameFile?: typeof fsp.rename;
+  syncParentDir?: (filePath: string) => Promise<void>;
+};
+
+type AtomicExclusiveWriteOptions = {
+  linkFile?: typeof fsp.link;
+  syncParentDir?: (filePath: string) => Promise<void>;
+};
+
+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<string, string | undefined>;
   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<string, string | undefined>;
   stateKey: string;
-}) {
+  options?: AtomicExclusiveWriteOptions;
+}): Promise<string | null> {
   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 };

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),
             },
           };
         }

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/);
+});

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<void>((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");

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 } };