fix(state): write state atomically

Fixes #108 and #109.

- Replace direct state writes with same-directory atomic temp-file writes.
- Preserve existing file modes and create new state files as 0600.
- Clean up temp files on failed replacement paths.
- Add state/SDK regression coverage plus changelog entry.

Proof:
- pnpm run typecheck
- node --test dist/test/state.test.js dist/test/resume.test.js dist/test/multi_approval_resume.test.js dist/test/approve_preview.test.js
- pnpm run lint
- pnpm run test
- built SDK write/read proof preserved 0600 across replacement
- autoreview clean: no accepted/actionable findings

Co-authored-by: Krasimir Kralev <krasi@idrobots.com>

CHANGELOG.md

@@ -4,6 +4,7 @@ All notable changes to Lobster will be documented in this file.
 
 ## Unreleased
 
+- 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)).
 - 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/state.ts

@@ -1,6 +1,6 @@
 import { promises as fsp } from "node:fs";
 
-import { defaultStateDir, keyToPath } from "../../state/store.js";
+import { defaultStateDir, keyToPath, writeFileAtomic } from "../../state/store.js";
 
 export const stateGetCommand = {
   name: "state.get",
@@ -70,7 +70,7 @@ export const stateSetCommand = {
     const filePath = keyToPath(stateDir, key);
 
     await fsp.mkdir(stateDir, { recursive: true });
-    await fsp.writeFile(filePath, JSON.stringify(value, null, 2) + "\n", "utf8");
+    await writeFileAtomic(filePath, JSON.stringify(value, null, 2) + "\n");
 
     return { output: asStream([value]) };
   },

src/sdk/primitives/state.ts

@@ -15,10 +15,50 @@
  *   .pipe(stateSet('my-key'));
  */
 
+import { randomBytes } from "node:crypto";
 import { promises as fsp } from "node:fs";
 import os from "node:os";
 import path from "node:path";
 
+/**
+ * Write a file atomically (stage to a sibling temp file, fsync, then rename).
+ * `rename(2)` is atomic on a single filesystem, so a concurrent reader or a
+ * crash never observes a truncated/partial file. Plain `fsp.writeFile`
+ * truncates the target up front, leaving a corruption window on SIGKILL/OOM/
+ * power loss. New state files are private by default; existing file modes are
+ * preserved across replacement. Kept local to keep the SDK self-contained.
+ * @param {string} filePath
+ * @param {string} data
+ */
+async function writeFileAtomic(filePath, data) {
+  const dir = path.dirname(filePath);
+  const tmpPath = path.join(
+    dir,
+    `.${path.basename(filePath)}.${randomBytes(6).toString("hex")}.tmp`,
+  );
+  let mode = 0o600;
+  let handle;
+  let cleanup = true;
+  try {
+    try {
+      mode = (await fsp.stat(filePath)).mode & 0o777;
+    } catch (err) {
+      if (err?.code !== "ENOENT") throw err;
+    }
+    handle = await fsp.open(tmpPath, "wx", mode);
+    await handle.writeFile(data, "utf8");
+    await handle.sync();
+    await handle.close();
+    handle = undefined;
+    await fsp.chmod(tmpPath, mode);
+    await fsp.rename(tmpPath, filePath);
+    cleanup = false;
+  } finally {
+    if (handle) await handle.close().catch(() => {});
+    if (cleanup) await fsp.rm(tmpPath, { force: true }).catch(() => {});
+  }
+}
+
 /**
  * Get the state directory
  * @param {Object} ctx
@@ -116,7 +156,7 @@ export function stateSet(key) {
       const filePath = keyToPath(stateDir, key);
 
       await fsp.mkdir(stateDir, { recursive: true });
-      await fsp.writeFile(filePath, JSON.stringify(value, null, 2) + "\n", "utf8");
+      await writeFileAtomic(filePath, JSON.stringify(value, null, 2) + "\n");
 
       // Pass through the value
       return {
@@ -174,5 +214,5 @@ export async function writeState(key, value, ctx = {}) {
   const filePath = keyToPath(stateDir, key);
 
   await fsp.mkdir(stateDir, { recursive: true });
-  await fsp.writeFile(filePath, JSON.stringify(value, null, 2) + "\n", "utf8");
+  await writeFileAtomic(filePath, JSON.stringify(value, null, 2) + "\n");
 }

src/state/store.ts

@@ -33,6 +33,44 @@ export function stableStringify(value) {
   });
 }
 
+/**
+ * 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
+ * (or a crash) never observes a truncated/partial file — it sees either the
+ * complete old content or the complete new content. Plain `fsp.writeFile`
+ * truncates the target up front, leaving a corruption window on SIGKILL/OOM/
+ * 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) {
+  const dir = path.dirname(filePath);
+  const tmpPath = path.join(
+    dir,
+    `.${path.basename(filePath)}.${randomBytes(6).toString("hex")}.tmp`,
+  );
+  let mode = 0o600;
+  let handle;
+  let cleanup = true;
+  try {
+    try {
+      mode = (await fsp.stat(filePath)).mode & 0o777;
+    } catch (err) {
+      if (err?.code !== "ENOENT") throw err;
+    }
+    handle = await fsp.open(tmpPath, "wx", mode);
+    await handle.writeFile(data, "utf8");
+    await handle.sync();
+    await handle.close();
+    handle = undefined;
+    await fsp.chmod(tmpPath, mode);
+    await fsp.rename(tmpPath, filePath);
+    cleanup = false;
+  } finally {
+    if (handle) await handle.close().catch(() => {});
+    if (cleanup) await fsp.rm(tmpPath, { force: true }).catch(() => {});
+  }
+}
+
 export async function readStateJson({ env, key }) {
   const stateDir = defaultStateDir(env);
   const filePath = keyToPath(stateDir, key);
@@ -51,7 +89,7 @@ export async function writeStateJson({ env, key, value }) {
   const filePath = keyToPath(stateDir, key);
 
   await fsp.mkdir(stateDir, { recursive: true });
-  await fsp.writeFile(filePath, JSON.stringify(value, null, 2) + "\n", "utf8");
+  await writeFileAtomic(filePath, JSON.stringify(value, null, 2) + "\n");
 }
 
 export async function deleteStateJson({ env, key }) {

test/state.test.ts

@@ -2,9 +2,11 @@ import test from "node:test";
 import assert from "node:assert/strict";
 import os from "node:os";
 import path from "node:path";
-import { mkdtempSync } from "node:fs";
+import { mkdtempSync, promises as fsp } from "node:fs";
 import { createDefaultRegistry } from "../src/commands/registry.js";
 import { runPipeline } from "../src/runtime.js";
+import { stateSet, readState, writeState } from "../src/sdk/primitives/state.js";
+import { writeStateJson, readStateJson, writeFileAtomic } from "../src/state/store.js";
 
 function streamOf(items) {
   return (async function* () {
@@ -73,3 +75,135 @@ test("state.get returns null for missing key", async () => {
 
   assert.deepEqual(output.items, [null]);
 });
+
+// --- Atomic-write behavior proofs (issues #108, #109) ---
+//
+// Plain fsp.writeFile truncates the target before writing, so a concurrent
+// reader (or a crash mid-write) can observe an empty/partial file and fail to
+// JSON.parse it. These tests drive many large writes while reading in parallel
+// and assert the reader NEVER sees a truncated value. They fail against the
+// pre-fix non-atomic writeFile and pass with writeFileAtomic (stage + rename).
+
+test("writeStateJson is atomic: concurrent reads never observe truncated state (#108)", async () => {
+  const tmp = mkdtempSync(path.join(os.tmpdir(), "lobster-atomic-store-"));
+  const env = { LOBSTER_STATE_DIR: tmp };
+  const key = "pipeline-resume";
+  const payload = "x".repeat(256 * 1024); // large enough that writeFile is not instantaneous
+
+  await writeStateJson({ env, key, value: { payload, n: 0 } });
+
+  let readErrors = 0;
+  let partialReads = 0;
+  const reader = (async () => {
+    for (let i = 0; i < 500; i++) {
+      try {
+        const v = await readStateJson({ env, key });
+        if (!v || v.payload !== payload) partialReads++;
+      } catch {
+        readErrors++; // JSON.parse on truncated content throws SyntaxError
+      }
+    }
+  })();
+  const writer = (async () => {
+    for (let n = 1; n <= 150; n++) {
+      await writeStateJson({ env, key, value: { payload, n } });
+    }
+  })();
+  await Promise.all([reader, writer]);
+
+  assert.equal(readErrors, 0, "reader must never hit a parse/IO error mid-write");
+  assert.equal(partialReads, 0, "reader must never observe truncated/empty state");
+
+  const leftovers = (await fsp.readdir(tmp)).filter((f) => f.includes(".tmp"));
+  assert.deepEqual(leftovers, [], "atomic write must not leave temp files behind");
+});
+
+test("writeFileAtomic creates private files and preserves existing modes", async () => {
+  const tmp = mkdtempSync(path.join(os.tmpdir(), "lobster-atomic-mode-"));
+  const freshPath = path.join(tmp, "fresh.json");
+  const existingPath = path.join(tmp, "existing.json");
+
+  await writeFileAtomic(freshPath, '{"ok":true}\n');
+  assert.equal((await fsp.stat(freshPath)).mode & 0o777, 0o600);
+
+  await fsp.writeFile(existingPath, '{"old":true}\n', { mode: 0o640 });
+  await fsp.chmod(existingPath, 0o640);
+  await writeFileAtomic(existingPath, '{"ok":true}\n');
+  assert.equal((await fsp.stat(existingPath)).mode & 0o777, 0o640);
+});
+
+test("writeFileAtomic removes temp files when replacement fails", async () => {
+  const tmp = mkdtempSync(path.join(os.tmpdir(), "lobster-atomic-cleanup-"));
+  const targetDir = path.join(tmp, "state.json");
+  await fsp.mkdir(targetDir);
+
+  await assert.rejects(() => writeFileAtomic(targetDir, '{"ok":true}\n'));
+  const leftovers = (await fsp.readdir(tmp)).filter((f) => f.includes(".tmp"));
+  assert.deepEqual(leftovers, []);
+});
+
+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 } };
+  const key = "sdk-state";
+  const payload = "y".repeat(256 * 1024);
+
+  const writeOnce = async (n: number) => {
+    const prim = stateSet(key);
+    const input = (async function* () {
+      yield { payload, n };
+    })();
+    const res = await prim.run({ input, ctx });
+    for await (const _ of res.output) {
+      void _;
+    }
+  };
+
+  await writeOnce(0);
+
+  let readErrors = 0;
+  let partialReads = 0;
+  const reader = (async () => {
+    for (let i = 0; i < 500; i++) {
+      try {
+        const v = await readState(key, ctx);
+        if (!v || v.payload !== payload) partialReads++;
+      } catch {
+        readErrors++;
+      }
+    }
+  })();
+  const writer = (async () => {
+    for (let n = 1; n <= 120; n++) {
+      await writeOnce(n);
+    }
+  })();
+  await Promise.all([reader, writer]);
+
+  assert.equal(readErrors, 0, "SDK reader must never hit a parse/IO error mid-write");
+  assert.equal(partialReads, 0, "SDK reader must never observe truncated/empty state");
+});
+
+test("SDK writeState preserves restricted state-file mode", async () => {
+  const tmp = mkdtempSync(path.join(os.tmpdir(), "lobster-atomic-sdk-mode-"));
+  const ctx = { env: { LOBSTER_STATE_DIR: tmp } };
+  const filePath = path.join(tmp, "sdk-state.json");
+  await fsp.mkdir(tmp, { recursive: true });
+  await fsp.writeFile(filePath, '{"old":true}\n', { mode: 0o600 });
+  await fsp.chmod(filePath, 0o600);
+
+  await writeState("sdk-state", { ok: true }, ctx);
+
+  assert.equal((await fsp.stat(filePath)).mode & 0o777, 0o600);
+  assert.deepEqual(await readState("sdk-state", ctx), { ok: true });
+});
+
+test("SDK writeState removes temp files when replacement fails", async () => {
+  const tmp = mkdtempSync(path.join(os.tmpdir(), "lobster-atomic-sdk-cleanup-"));
+  const ctx = { env: { LOBSTER_STATE_DIR: tmp } };
+  await fsp.mkdir(path.join(tmp, "sdk-state.json"));
+
+  await assert.rejects(() => writeState("sdk-state", { ok: true }, ctx));
+  const leftovers = (await fsp.readdir(tmp)).filter((f) => f.includes(".tmp"));
+  assert.deepEqual(leftovers, []);
+});