fix(connector): make raw dump reads resilient to just-written-file races

src/connectors/io.ts

@@ -1,4 +1,5 @@
-import { chmod, readFile, writeFile } from "node:fs/promises";
+import { randomUUID } from "node:crypto";
+import { chmod, mkdir, open, readFile, rename, rm } from "node:fs/promises";
 import path from "node:path";
 import {
   ensureConnectorHome,
@@ -85,18 +86,64 @@ export function updateStateWithRun(
   };
 }
 
+/**
+ * Writes JSON to a private (0o600) file atomically and durably: the content is
+ * written to a temporary sibling, flushed to disk, and then renamed into
+ * place. The destination path therefore only ever becomes visible with
+ * complete content, so readers racing the write (for example the synthesis
+ * agent reading a raw dump the connector just reported) can never observe a
+ * partial file, and the write is on disk before the path is reported.
+ */
 async function writePrivateJson(
   filePath: string,
   value: unknown,
 ): Promise<void> {
-  await import("node:fs/promises").then(({ mkdir }) =>
-    mkdir(path.dirname(filePath), { recursive: true, mode: 0o700 }),
-  );
-  await writeFile(filePath, `${JSON.stringify(value, null, 2)}\n`, {
-    encoding: "utf8",
-    mode: 0o600,
-  });
+  const directoryPath = path.dirname(filePath);
+  await mkdir(directoryPath, { recursive: true, mode: 0o700 });
+
+  const tempPath = `${filePath}.${randomUUID()}.tmp`;
+  const fileHandle = await open(tempPath, "wx", 0o600);
+
+  try {
+    await fileHandle.writeFile(`${JSON.stringify(value, null, 2)}\n`, "utf8");
+    await fileHandle.sync();
+  } catch (error) {
+    await fileHandle.close();
+    await rm(tempPath, { force: true });
+    throw error;
+  }
+
+  await fileHandle.close();
+
+  try {
+    await rename(tempPath, filePath);
+  } catch (error) {
+    await rm(tempPath, { force: true });
+    throw error;
+  }
+
   await chmod(filePath, 0o600);
+  await syncDirectory(directoryPath);
+}
+
+/**
+ * Flushes a directory so a just-renamed entry survives a crash and is visible
+ * to other readers of the directory. Best effort: directory fsync is not
+ * supported on some platforms (notably Windows), where the rename above is
+ * still atomic.
+ */
+async function syncDirectory(directoryPath: string): Promise<void> {
+  try {
+    const directoryHandle = await open(directoryPath, "r");
+
+    try {
+      await directoryHandle.sync();
+    } finally {
+      await directoryHandle.close();
+    }
+  } catch {
+    // Ignore: opening or fsyncing a directory is platform-dependent.
+  }
 }
 
 function isFileNotFoundError(error: unknown): boolean {

src/connectors/tools.ts

@@ -3,7 +3,7 @@ import {
   type StructuredToolInterface,
 } from "@langchain/core/tools";
 import { constants as fsConstants } from "node:fs";
-import { lstat, open, readdir, stat } from "node:fs/promises";
+import { lstat, open, readdir, stat, type FileHandle } from "node:fs/promises";
 import path from "node:path";
 import {
   getConnectorConfigPath,
@@ -302,15 +302,17 @@ async function listRawItems(connectorId: ConnectorId) {
   };
 }
 
+const RAW_ITEM_READ_ENOENT_MAX_ATTEMPTS = 5;
+const RAW_ITEM_READ_ENOENT_INITIAL_DELAY_MS = 25;
+
 async function readRawItem(
   connectorId: ConnectorId,
   relativePath: string,
   maxBytes: number,
 ) {
   const rawDir = getConnectorRawDir(connectorId);
   const filePath = resolveConnectorRawPath(connectorId, relativePath);
-  await assertRawItemPathHasNoSymlinks(rawDir, filePath);
-  const fileHandle = await open(filePath, getRawItemOpenFlags());
+  const fileHandle = await openRawItemWithEnoentRetry(rawDir, filePath);
 
   try {
     const fileStat = await fileHandle.stat();
@@ -333,6 +335,43 @@ async function readRawItem(
   }
 }
 
+/**
+ * Opens a raw item, retrying briefly on ENOENT with bounded exponential
+ * backoff. Connectors report raw dump paths the moment ingestion finishes and
+ * the synthesis agent reads them immediately afterwards, so a transient ENOENT
+ * on a just-written path is retryable rather than fatal; without the retry the
+ * read either crashes the run or silently drops captures.
+ */
+async function openRawItemWithEnoentRetry(
+  rawDir: string,
+  filePath: string,
+): Promise<FileHandle> {
+  let delayMs = RAW_ITEM_READ_ENOENT_INITIAL_DELAY_MS;
+
+  for (let attempt = 1; ; attempt += 1) {
+    try {
+      await assertRawItemPathHasNoSymlinks(rawDir, filePath);
+      return await open(filePath, getRawItemOpenFlags());
+    } catch (error) {
+      if (
+        !isFileNotFoundError(error) ||
+        attempt >= RAW_ITEM_READ_ENOENT_MAX_ATTEMPTS
+      ) {
+        throw error;
+      }
+    }
+
+    await sleep(delayMs);
+    delayMs *= 2;
+  }
+}
+
+function sleep(durationMs: number): Promise<void> {
+  return new Promise((resolve) => {
+    setTimeout(resolve, durationMs);
+  });
+}
+
 async function listFiles(
   rootDir: string,
   currentDir: string,

test/raw-dump-read-race.test.ts

@@ -0,0 +1,151 @@
+import type { StructuredToolInterface } from "@langchain/core/tools";
+import { mkdtemp, readFile, readdir, rm, stat } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import path from "node:path";
+import { afterEach, describe, expect, test, vi } from "vitest";
+
+const originalHome = process.env.HOME;
+const originalUserProfile = process.env.USERPROFILE;
+const tempHomes: string[] = [];
+
+afterEach(async () => {
+  vi.restoreAllMocks();
+  vi.resetModules();
+  restoreEnv("HOME", originalHome);
+  restoreEnv("USERPROFILE", originalUserProfile);
+
+  await Promise.all(
+    tempHomes
+      .splice(0)
+      .map((home) => rm(home, { force: true, recursive: true })),
+  );
+});
+
+describe("raw dump read race", () => {
+  test("read retries ENOENT until a dump written moments later is visible", async () => {
+    const home = await createTempHome();
+    const { connectorIo, tools } = await loadConnectorModules(home);
+    const runId = "2026-07-27T00-00-00-000Z";
+    const dump = {
+      fetchedAt: "2026-07-27T00:00:00.000Z",
+      pages: [{ data: [{ id: "1" }, { id: "2" }] }],
+      stream: "bookmarks",
+    };
+
+    // Simulate the race from issue #460: synthesis starts reading the
+    // reported path before the connector's write has landed on disk.
+    const delayedWrite = (async () => {
+      await sleep(60);
+      await connectorIo.writeRawJson("x", runId, "bookmarks.json", dump);
+    })();
+    const result = await invokeJson<RawReadResult>(
+      getTool(tools, "openwiki_read_raw_item"),
+      { connectorId: "x", path: `${runId}/bookmarks.json` },
+    );
+    await delayedWrite;
+
+    expect(JSON.parse(result.content)).toEqual(dump);
+    expect(result.truncated).toBe(false);
+  });
+
+  test("read still throws ENOENT after bounded retries when the dump never appears", async () => {
+    const home = await createTempHome();
+    const { tools } = await loadConnectorModules(home);
+
+    await expect(
+      getTool(tools, "openwiki_read_raw_item").invoke({
+        connectorId: "x",
+        path: "2026-07-27T00-00-00-000Z/missing.json",
+      }),
+    ).rejects.toThrow(/ENOENT/u);
+  });
+
+  test("writeRawJson publishes the dump atomically with no temp files left behind", async () => {
+    const home = await createTempHome();
+    const { connectorIo } = await loadConnectorModules(home);
+    const runId = "2026-07-27T00-00-00-000Z";
+    const dump = { pages: [{ data: [{ id: "1" }] }], stream: "bookmarks" };
+
+    const filePath = await connectorIo.writeRawJson(
+      "x",
+      runId,
+      "bookmarks.json",
+      dump,
+    );
+
+    // The path the connector reports must already hold the complete dump,
+    // and no intermediate temp file may remain next to it.
+    expect(JSON.parse(await readFile(filePath, "utf8"))).toEqual(dump);
+    expect(await readdir(path.dirname(filePath))).toEqual(["bookmarks.json"]);
+
+    if (process.platform !== "win32") {
+      expect((await stat(filePath)).mode & 0o777).toBe(0o600);
+    }
+  });
+});
+
+interface RawReadResult {
+  content: string;
+  truncated: boolean;
+}
+
+async function loadConnectorModules(home: string): Promise<{
+  connectorIo: typeof import("../src/connectors/io.ts");
+  tools: StructuredToolInterface[];
+}> {
+  vi.resetModules();
+  process.env.HOME = home;
+  process.env.USERPROFILE = home;
+  const { createOpenWikiConnectorTools } =
+    await import("../src/connectors/tools.ts");
+  const connectorIo = await import("../src/connectors/io.ts");
+
+  return { connectorIo, tools: createOpenWikiConnectorTools() };
+}
+
+function getTool(
+  tools: StructuredToolInterface[],
+  name: string,
+): StructuredToolInterface {
+  const tool = tools.find((candidate) => candidate.name === name);
+
+  if (!tool) {
+    throw new Error(`Missing connector tool: ${name}`);
+  }
+
+  return tool;
+}
+
+async function invokeJson<T>(
+  tool: StructuredToolInterface,
+  input: Record<string, unknown>,
+): Promise<T> {
+  const result: unknown = await tool.invoke(input);
+
+  if (typeof result !== "string") {
+    throw new Error("Expected connector tool to return a JSON string.");
+  }
+
+  return JSON.parse(result) as T;
+}
+
+async function createTempHome(): Promise<string> {
+  const home = await mkdtemp(path.join(tmpdir(), "openwiki-dump-race-"));
+  tempHomes.push(home);
+
+  return home;
+}
+
+function sleep(durationMs: number): Promise<void> {
+  return new Promise((resolve) => {
+    setTimeout(resolve, durationMs);
+  });
+}
+
+function restoreEnv(key: string, value: string | undefined): void {
+  if (value === undefined) {
+    delete process.env[key];
+  } else {
+    process.env[key] = value;
+  }
+}