fix(auto-reply): RawBody commands + locked session updates (#643)

src/config/sessions.test.ts

@@ -11,6 +11,7 @@ import {
   resolveSessionTranscriptPath,
   resolveSessionTranscriptsDir,
   updateLastRoute,
+  updateSessionStoreEntry,
 } from "./sessions.js";
 
 describe("sessions", () => {
@@ -187,4 +188,52 @@ describe("sessions", () => {
       }
     }
   });
+
+  it("updateSessionStoreEntry merges concurrent patches", async () => {
+    const mainSessionKey = "agent:main:main";
+    const dir = await fs.mkdtemp(path.join(os.tmpdir(), "clawdbot-sessions-"));
+    const storePath = path.join(dir, "sessions.json");
+    await fs.writeFile(
+      storePath,
+      JSON.stringify(
+        {
+          [mainSessionKey]: {
+            sessionId: "sess-1",
+            updatedAt: 123,
+            thinkingLevel: "low",
+          },
+        },
+        null,
+        2,
+      ),
+      "utf-8",
+    );
+
+    const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));
+    await Promise.all([
+      updateSessionStoreEntry({
+        storePath,
+        sessionKey: mainSessionKey,
+        update: async () => {
+          await sleep(50);
+          return { modelOverride: "anthropic/claude-opus-4-5" };
+        },
+      }),
+      updateSessionStoreEntry({
+        storePath,
+        sessionKey: mainSessionKey,
+        update: async () => {
+          await sleep(10);
+          return { thinkingLevel: "high" };
+        },
+      }),
+    ]);
+
+    const store = loadSessionStore(storePath);
+    expect(store[mainSessionKey]?.modelOverride).toBe(
+      "anthropic/claude-opus-4-5",
+    );
+    expect(store[mainSessionKey]?.thinkingLevel).toBe("high");
+    await expect(fs.stat(`${storePath}.lock`)).rejects.toThrow();
+  });
 });

src/config/sessions.ts

@@ -139,11 +139,14 @@ export function mergeSessionEntry(
 ): SessionEntry {
   const sessionId =
     patch.sessionId ?? existing?.sessionId ?? crypto.randomUUID();
-  const updatedAt = patch.updatedAt ?? existing?.updatedAt ?? Date.now();
+  const updatedAt = Math.max(
+    existing?.updatedAt ?? 0,
+    patch.updatedAt ?? 0,
+    Date.now(),
+  );
   if (!existing) return { ...patch, sessionId, updatedAt };
   return { ...existing, ...patch, sessionId, updatedAt };
 }
-
 export type GroupKeyResolution = {
   key: string;
   legacyKey?: string;
@@ -487,6 +490,92 @@ export async function saveSessionStore(
   }
 }
 
+type SessionStoreLockOptions = {
+  timeoutMs?: number;
+  pollIntervalMs?: number;
+  staleMs?: number;
+};
+
+async function withSessionStoreLock<T>(
+  storePath: string,
+  fn: () => Promise<T>,
+  opts: SessionStoreLockOptions = {},
+): Promise<T> {
+  const timeoutMs = opts.timeoutMs ?? 10_000;
+  const pollIntervalMs = opts.pollIntervalMs ?? 25;
+  const staleMs = opts.staleMs ?? 30_000;
+  const lockPath = `${storePath}.lock`;
+  const startedAt = Date.now();
+
+  await fs.promises.mkdir(path.dirname(storePath), { recursive: true });
+
+  while (true) {
+    try {
+      const handle = await fs.promises.open(lockPath, "wx");
+      try {
+        await handle.writeFile(
+          JSON.stringify({ pid: process.pid, startedAt: Date.now() }),
+          "utf-8",
+        );
+      } catch {
+        // best-effort
+      }
+      await handle.close();
+      break;
+    } catch (err) {
+      const code =
+        err && typeof err === "object" && "code" in err
+          ? String((err as { code?: unknown }).code)
+          : null;
+      if (code !== "EEXIST") throw err;
+
+      const now = Date.now();
+      if (now - startedAt > timeoutMs) {
+        throw new Error(`timeout acquiring session store lock: ${lockPath}`);
+      }
+
+      // Best-effort stale lock eviction (e.g. crashed process).
+      try {
+        const st = await fs.promises.stat(lockPath);
+        const ageMs = now - st.mtimeMs;
+        if (ageMs > staleMs) {
+          await fs.promises.unlink(lockPath);
+          continue;
+        }
+      } catch {
+        // ignore
+      }
+
+      await new Promise((r) => setTimeout(r, pollIntervalMs));
+    }
+  }
+
+  try {
+    return await fn();
+  } finally {
+    await fs.promises.unlink(lockPath).catch(() => undefined);
+  }
+}
+
+export async function updateSessionStoreEntry(params: {
+  storePath: string;
+  sessionKey: string;
+  update: (entry: SessionEntry) => Promise<Partial<SessionEntry> | null>;
+}): Promise<SessionEntry | null> {
+  const { storePath, sessionKey, update } = params;
+  return await withSessionStoreLock(storePath, async () => {
+    const store = loadSessionStore(storePath);
+    const existing = store[sessionKey];
+    if (!existing) return null;
+    const patch = await update(existing);
+    if (!patch) return existing;
+    const next = mergeSessionEntry(existing, patch);
+    store[sessionKey] = next;
+    await saveSessionStore(storePath, store);
+    return next;
+  });
+}
+
 export async function updateLastRoute(params: {
   storePath: string;
   sessionKey: string;