From 007b4e1e6bd5989c4cecdca2e0638ecb9c5a4f8f Mon Sep 17 00:00:00 2001
From: Emanuele Fenocchi <emanuele.fenocchi@activeloop.dev>
Date: Thu, 20 Aug 2026 06:33:58 +0000
Subject: [PATCH 01/12] fix(cowork): stop the queue replay that grew a 40 GB
 local jsonl

ingestCoworkSessions() appended new transcript entries to the local queue
file, uploaded, and only persisted the per-transcript line watermark after
the upload succeeded. Any upload failure (offline host, disconnected
network filesystem, 5xx) lost the watermark, so the next 30s tick re-read
the same transcript lines and appended the entire transcript to the queue
again - forever. A customer running Claude Cowork reached a single 39.9 GB
~/.deeplake/queue-cowork/<session>.jsonl growing ~5 GB/day.

The queue file, not the watermark, is what owes the backend those rows,
and it retries them with the same ids (the INSERT is idempotent). So
persist the watermark as soon as the rows are queued, before the upload,
and let a drain failure leave the rows queued instead of aborting the tick.

Regression test reproduces the leak: on the previous code a second failing
tick doubled the queue file (678 -> 1356 bytes) with no new transcript
content.
---
 src/mcp/cowork-ingest.ts                    |  34 ++++--
 tests/claude-code/cowork-queue-leak.test.ts | 114 ++++++++++++++++++++
 2 files changed, 139 insertions(+), 9 deletions(-)
 create mode 100644 tests/claude-code/cowork-queue-leak.test.ts

diff --git a/src/mcp/cowork-ingest.ts b/src/mcp/cowork-ingest.ts
index 6d177e32f..64fb3b07c 100644
--- a/src/mcp/cowork-ingest.ts
+++ b/src/mcp/cowork-ingest.ts
@@ -43,6 +43,7 @@ import {
   buildQueuedSessionRow,
   buildSessionPath,
   drainSessionQueues,
+  gcOversizedQueueFiles,
 } from "../hooks/session-queue.js";
 import { spawnWikiWorker, bundleDirFromImportMeta } from "../hooks/spawn-wiki-worker.js";
 import { forceSessionEndTrigger } from "../skillify/triggers.js";
@@ -363,6 +364,9 @@ export async function ingestCoworkSessions(): Promise<{ ingested: number } | { s
 
   let ingested = 0;
   try {
+    // Reclaim any queue file left oversized by an earlier upload outage before
+    // writing more rows — such a file can no longer be flushed.
+    gcOversizedQueueFiles(COWORK_QUEUE_DIR);
     const state = loadState();
     const transcripts = findTranscripts(root);
     let appendedAny = false;
@@ -404,6 +408,17 @@ export async function ingestCoworkSessions(): Promise<{ ingested: number } | { s
       state.processedLines[path] = lines.length;
     }
 
+    // Persist the watermark as soon as the rows are in the on-disk queue, and
+    // BEFORE the upload. The queue file — not this watermark — is what owes the
+    // backend those rows, and it retries them on the next tick with the same
+    // row ids (the INSERT is idempotent). Saving after the upload instead meant
+    // that any upload failure lost the watermark, so the next tick re-read the
+    // same transcript lines and appended the whole transcript to the queue
+    // again, every 30s, without bound: a customer running Cowork with a
+    // flapping network filesystem reached a single 39.9 GB queue file growing
+    // ~5 GB/day (2026-08-19).
+    if (appendedAny) saveState(state);
+
     if (appendedAny) {
       const api = new DeeplakeApi(
         config.token,
@@ -412,15 +427,16 @@ export async function ingestCoworkSessions(): Promise<{ ingested: number } | { s
         config.workspaceId,
         config.sessionsTableName,
       );
-      await drainSessionQueues(api, {
-        sessionsTable: config.sessionsTableName,
-        queueDir: COWORK_QUEUE_DIR,
-      });
-      // Persist the line watermark immediately after the upload, before the
-      // slow summarize step below. Rows carry random ids, so a crash between
-      // the insert and a later saveState would replay these lines under fresh
-      // ids and duplicate them. Saving here shrinks that window to this write.
-      saveState(state);
+      try {
+        await drainSessionQueues(api, {
+          sessionsTable: config.sessionsTableName,
+          queueDir: COWORK_QUEUE_DIR,
+        });
+      } catch (e: unknown) {
+        // Upload failure is expected while offline. The rows stay queued and
+        // the next tick retries them; do not let it abort the summarize pass.
+        log("cowork-ingest", `queue drain failed, rows stay queued: ${e instanceof Error ? e.message : String(e)}`);
+      }
     }
 
     // Summarize sessions that have gone idle — Cowork has no SessionEnd hook,
diff --git a/tests/claude-code/cowork-queue-leak.test.ts b/tests/claude-code/cowork-queue-leak.test.ts
new file mode 100644
index 000000000..462ecde58
--- /dev/null
+++ b/tests/claude-code/cowork-queue-leak.test.ts
@@ -0,0 +1,114 @@
+/**
+ * Regression test for the Cowork local-queue disk leak.
+ *
+ * Reported 2026-08-19 by a customer running Claude Cowork: a single
+ * ~/.deeplake/queue-cowork/<session>.jsonl file had grown to 39.9 GB at
+ * ~1 MB / 20 s while their network filesystem was flapping.
+ *
+ * Cause: ingestCoworkSessions() appended every new transcript entry to the
+ * queue file, then uploaded, and only persisted the per-transcript line
+ * watermark AFTER a successful upload. When the upload threw (offline, DNS,
+ * 5xx) the watermark was lost, so the next 30 s tick re-read the same
+ * transcript lines from the old watermark and appended the whole transcript
+ * to the queue again — forever, growing without bound.
+ */
+import { describe, it, expect, beforeEach, afterEach, vi } from "vitest";
+import { mkdtempSync, mkdirSync, writeFileSync, statSync, appendFileSync } from "node:fs";
+import { join } from "node:path";
+import { tmpdir } from "node:os";
+
+const SESSION_ID = "b27efa59-a8bc-4ea3-8b02-18cbc608ae17";
+
+// Every upload fails, the way it does on a disconnected network filesystem.
+vi.mock("../../src/deeplake-api.js", () => ({
+  DeeplakeApi: class {
+    async query(): Promise<never> {
+      throw new Error("fetch failed: ECONNREFUSED");
+    }
+    async ensureSessionsTable(): Promise<void> {}
+  },
+}));
+
+let home: string;
+let prevHome: string | undefined;
+
+function transcriptPath(): string {
+  const dir = join(home, ".config", "Claude", "local-agent-mode-sessions", "s1", ".claude", "projects", "proj");
+  mkdirSync(dir, { recursive: true });
+  return join(dir, `${SESSION_ID}.jsonl`);
+}
+
+function line(text: string): string {
+  return `${JSON.stringify({
+    type: "user",
+    sessionId: SESSION_ID,
+    timestamp: new Date().toISOString(),
+    cwd: "/cowork",
+    message: { role: "user", content: text },
+  })}\n`;
+}
+
+function queueBytes(): number {
+  try {
+    return statSync(join(home, ".deeplake", "queue-cowork", `${SESSION_ID}.jsonl`)).size;
+  } catch {
+    return 0;
+  }
+}
+
+beforeEach(() => {
+  prevHome = process.env.HOME;
+  home = mkdtempSync(join(tmpdir(), "cowork-leak-"));
+  process.env.HOME = home;
+  mkdirSync(join(home, ".deeplake"), { recursive: true });
+  writeFileSync(
+    join(home, ".deeplake", "credentials.json"),
+    JSON.stringify({ token: "t", orgId: "org", orgName: "org", userName: "u", workspaceId: "default", apiUrl: "http://127.0.0.1:1" }),
+  );
+  vi.resetModules();
+});
+
+afterEach(() => {
+  if (prevHome === undefined) delete process.env.HOME;
+  else process.env.HOME = prevHome;
+});
+
+describe("cowork queue growth when uploads fail", () => {
+  it("does not re-append the same transcript lines on every failed upload", async () => {
+    const path = transcriptPath();
+    writeFileSync(path, line("first prompt"));
+
+    const { ingestCoworkSessions } = await import("../../src/mcp/cowork-ingest.js");
+
+    await ingestCoworkSessions();
+    const afterFirst = queueBytes();
+    expect(afterFirst).toBeGreaterThan(0);
+
+    // Nothing new was written to the transcript — a second tick must queue
+    // nothing, even though the first upload failed and the rows are still
+    // sitting in the queue file waiting to be retried.
+    await ingestCoworkSessions();
+    expect(queueBytes()).toBe(afterFirst);
+
+    // A third tick, still offline, still no transcript growth.
+    await ingestCoworkSessions();
+    expect(queueBytes()).toBe(afterFirst);
+  });
+
+  it("queues each new transcript line exactly once across failing ticks", async () => {
+    const path = transcriptPath();
+    writeFileSync(path, line("one"));
+
+    const { ingestCoworkSessions } = await import("../../src/mcp/cowork-ingest.js");
+    await ingestCoworkSessions();
+    const afterOne = queueBytes();
+
+    appendFileSync(path, line("two"));
+    await ingestCoworkSessions();
+    const afterTwo = queueBytes();
+
+    // Growth for the second tick is one message, not the whole transcript.
+    expect(afterTwo - afterOne).toBeLessThan(afterOne * 1.5);
+    expect(afterTwo).toBeGreaterThan(afterOne);
+  });
+});

From 9b659f4b663971e091cb31897a3ed9f5fa60e89e Mon Sep 17 00:00:00 2001
From: Emanuele Fenocchi <emanuele.fenocchi@activeloop.dev>
Date: Thu, 20 Aug 2026 06:33:58 +0000
Subject: [PATCH 02/12] fix(session-queue): cap a session queue file and drop
 oversized ones

Backstop for the Cowork queue leak: nothing bounded a session queue file,
and a file past a few hundred MB is unflushable anyway because
readQueuedRows() reads it into a single string.

Stop appending at 256 MB per session file, and gcOversizedQueueFiles()
deletes queue/inflight files past the ceiling so a host that already has
one reclaims the disk on the next ingest tick.
---
 src/hooks/session-queue.ts              | 57 ++++++++++++++++++++++++-
 tests/claude-code/session-queue.test.ts | 40 +++++++++++++++++
 2 files changed, 96 insertions(+), 1 deletion(-)

diff --git a/src/hooks/session-queue.ts b/src/hooks/session-queue.ts
index f2e17ffcb..d290d23b2 100644
--- a/src/hooks/session-queue.ts
+++ b/src/hooks/session-queue.ts
@@ -14,6 +14,7 @@ import {
 import { dirname, join } from "node:path";
 import { homedir } from "node:os";
 import { sqlIdent, sqlStr } from "../utils/sql.js";
+import { log } from "../utils/debug.js";
 
 export interface SessionQueueApi {
   query(sql: string): Promise<Record<string, unknown>[]>;
@@ -67,6 +68,14 @@ export interface DrainSessionQueueResult {
 }
 
 const DEFAULT_QUEUE_DIR = join(homedir(), ".deeplake", "queue");
+// Hard ceiling for a single session's queue file. The queue is a retry buffer
+// for rows the backend has not accepted yet, so it only reaches this size when
+// uploads have been failing for a very long time (offline host, disconnected
+// network filesystem). Past the ceiling we stop appending instead of filling
+// the user's disk: a customer running Cowork hit a single 39.9 GB queue file
+// growing ~5 GB/day (2026-08-19). A file this large is also unflushable —
+// readQueuedRows() reads it whole — so gcOversizedQueueFiles() drops it.
+export const MAX_SESSION_QUEUE_BYTES = 256 * 1024 * 1024;
 const DEFAULT_MAX_BATCH_ROWS = 50;
 const DEFAULT_STALE_INFLIGHT_MS = 60_000;
 const DEFAULT_AUTH_FAILURE_TTL_MS = 5 * 60_000;
@@ -116,14 +125,60 @@ export function buildQueuedSessionRow(args: {
   };
 }
 
-export function appendQueuedSessionRow(row: QueuedSessionRow, queueDir = DEFAULT_QUEUE_DIR): string {
+export function appendQueuedSessionRow(
+  row: QueuedSessionRow,
+  queueDir = DEFAULT_QUEUE_DIR,
+  maxQueueBytes = MAX_SESSION_QUEUE_BYTES,
+): string {
   mkdirSync(queueDir, { recursive: true });
   const sessionId = extractSessionId(row.path);
   const queuePath = getQueuePath(queueDir, sessionId);
+  if (fileSize(queuePath) >= maxQueueBytes) {
+    log("session-queue", `queue file at the ${maxQueueBytes}-byte ceiling, dropping row: ${queuePath}`);
+    return queuePath;
+  }
   appendFileSync(queuePath, `${JSON.stringify(row)}\n`);
   return queuePath;
 }
 
+/**
+ * Delete queue/inflight files that have grown past the ceiling. Such a file
+ * cannot be flushed anyway (readQueuedRows reads it into a single string) and
+ * would otherwise sit on disk forever — it is the residue of an upload outage,
+ * not data the backend is still waiting for. Returns the bytes reclaimed.
+ */
+export function gcOversizedQueueFiles(queueDir = DEFAULT_QUEUE_DIR, maxQueueBytes = MAX_SESSION_QUEUE_BYTES): number {
+  let reclaimed = 0;
+  let names: string[];
+  try {
+    names = readdirSync(queueDir);
+  } catch {
+    return 0;
+  }
+  for (const name of names) {
+    if (!name.endsWith(".jsonl") && !name.endsWith(".inflight")) continue;
+    const path = join(queueDir, name);
+    const size = fileSize(path);
+    if (size < maxQueueBytes) continue;
+    try {
+      rmSync(path, { force: true });
+      reclaimed += size;
+      log("session-queue", `dropped oversized queue file (${size} bytes): ${path}`);
+    } catch {
+      /* best effort */
+    }
+  }
+  return reclaimed;
+}
+
+function fileSize(path: string): number {
+  try {
+    return statSync(path).size;
+  } catch {
+    return 0;
+  }
+}
+
 export function buildSessionInsertSql(sessionsTable: string, rows: QueuedSessionRow[]): string {
   if (rows.length === 0) throw new Error("buildSessionInsertSql: rows must not be empty");
   const table = sqlIdent(sessionsTable);
diff --git a/tests/claude-code/session-queue.test.ts b/tests/claude-code/session-queue.test.ts
index 928347124..3d38b7292 100644
--- a/tests/claude-code/session-queue.test.ts
+++ b/tests/claude-code/session-queue.test.ts
@@ -18,7 +18,9 @@ import {
   clearSessionWriteDisabled,
   drainSessionQueues,
   flushSessionQueue,
+  gcOversizedQueueFiles,
   isSessionWriteDisabled,
+  MAX_SESSION_QUEUE_BYTES,
   isSessionWriteAuthError,
   markSessionWriteDisabled,
   type QueuedSessionRow,
@@ -594,3 +596,41 @@ describe("session queue", () => {
     release?.();
   });
 });
+
+describe("oversized queue files", () => {
+  it("stops appending once the queue file reaches the size ceiling", () => {
+    const queueDir = makeQueueDir();
+    const queuePath = appendQueuedSessionRow(makeRow("s-cap", 0), queueDir, 10_000);
+    const afterFirst = readFileSync(queuePath, "utf-8");
+
+    // Simulate a queue that has already reached the ceiling.
+    writeFileSync(queuePath, "x".repeat(10_000));
+    appendQueuedSessionRow(makeRow("s-cap", 1), queueDir, 10_000);
+    expect(readFileSync(queuePath, "utf-8").length).toBe(10_000);
+
+    // Below the ceiling it appends as usual.
+    writeFileSync(queuePath, afterFirst);
+    appendQueuedSessionRow(makeRow("s-cap", 2), queueDir, 10_000);
+    expect(readFileSync(queuePath, "utf-8").split("\n").filter(Boolean)).toHaveLength(2);
+  });
+
+  it("drops queue and inflight files past the ceiling and leaves the rest alone", () => {
+    const queueDir = makeQueueDir();
+    const small = appendQueuedSessionRow(makeRow("s-small", 0), queueDir);
+    const big = join(queueDir, "s-big.jsonl");
+    const bigInflight = join(queueDir, "s-big-2.inflight");
+    writeFileSync(big, "x".repeat(4096));
+    writeFileSync(bigInflight, "x".repeat(4096));
+
+    const reclaimed = gcOversizedQueueFiles(queueDir, 4096);
+
+    expect(reclaimed).toBe(8192);
+    expect(existsSync(big)).toBe(false);
+    expect(existsSync(bigInflight)).toBe(false);
+    expect(existsSync(small)).toBe(true);
+  });
+
+  it("defaults the ceiling to 256 MB", () => {
+    expect(MAX_SESSION_QUEUE_BYTES).toBe(256 * 1024 * 1024);
+  });
+});

From 8ff8411b9eb675222cadbd9680f189130145e87f Mon Sep 17 00:00:00 2001
From: Emanuele Fenocchi <emanuele.fenocchi@activeloop.dev>
Date: Thu, 20 Aug 2026 06:39:36 +0000
Subject: [PATCH 03/12] test(session-queue): assert exact row counts and a hard
 ceiling

CodeRabbit review on #343: project the serialized row size before appending so
the ceiling cannot be overshot by one row, and assert queued-row counts instead
of relative byte growth.
---
 src/hooks/session-queue.ts                  |  7 +++++--
 tests/claude-code/cowork-queue-leak.test.ts | 20 ++++++++++++++------
 tests/claude-code/session-queue.test.ts     | 11 +++++++++++
 3 files changed, 30 insertions(+), 8 deletions(-)

diff --git a/src/hooks/session-queue.ts b/src/hooks/session-queue.ts
index d290d23b2..73fff9ec7 100644
--- a/src/hooks/session-queue.ts
+++ b/src/hooks/session-queue.ts
@@ -133,11 +133,14 @@ export function appendQueuedSessionRow(
   mkdirSync(queueDir, { recursive: true });
   const sessionId = extractSessionId(row.path);
   const queuePath = getQueuePath(queueDir, sessionId);
-  if (fileSize(queuePath) >= maxQueueBytes) {
+  const payload = `${JSON.stringify(row)}\n`;
+  // Project the post-append size, so the ceiling is a real ceiling rather than
+  // "the last row may overshoot it by however large that row happened to be".
+  if (fileSize(queuePath) + Buffer.byteLength(payload, "utf-8") > maxQueueBytes) {
     log("session-queue", `queue file at the ${maxQueueBytes}-byte ceiling, dropping row: ${queuePath}`);
     return queuePath;
   }
-  appendFileSync(queuePath, `${JSON.stringify(row)}\n`);
+  appendFileSync(queuePath, payload);
   return queuePath;
 }
 
diff --git a/tests/claude-code/cowork-queue-leak.test.ts b/tests/claude-code/cowork-queue-leak.test.ts
index 462ecde58..7450a5803 100644
--- a/tests/claude-code/cowork-queue-leak.test.ts
+++ b/tests/claude-code/cowork-queue-leak.test.ts
@@ -13,7 +13,7 @@
  * to the queue again — forever, growing without bound.
  */
 import { describe, it, expect, beforeEach, afterEach, vi } from "vitest";
-import { mkdtempSync, mkdirSync, writeFileSync, statSync, appendFileSync } from "node:fs";
+import { mkdtempSync, mkdirSync, writeFileSync, readFileSync, statSync, appendFileSync } from "node:fs";
 import { join } from "node:path";
 import { tmpdir } from "node:os";
 
@@ -56,6 +56,16 @@ function queueBytes(): number {
   }
 }
 
+function queuedRows(): number {
+  try {
+    return readFileSync(join(home, ".deeplake", "queue-cowork", `${SESSION_ID}.jsonl`), "utf-8")
+      .split("\n")
+      .filter(Boolean).length;
+  } catch {
+    return 0;
+  }
+}
+
 beforeEach(() => {
   prevHome = process.env.HOME;
   home = mkdtempSync(join(tmpdir(), "cowork-leak-"));
@@ -101,14 +111,12 @@ describe("cowork queue growth when uploads fail", () => {
 
     const { ingestCoworkSessions } = await import("../../src/mcp/cowork-ingest.js");
     await ingestCoworkSessions();
-    const afterOne = queueBytes();
+    expect(queuedRows()).toBe(1);
 
     appendFileSync(path, line("two"));
     await ingestCoworkSessions();
-    const afterTwo = queueBytes();
 
-    // Growth for the second tick is one message, not the whole transcript.
-    expect(afterTwo - afterOne).toBeLessThan(afterOne * 1.5);
-    expect(afterTwo).toBeGreaterThan(afterOne);
+    // The second tick queues the one new message, not the whole transcript.
+    expect(queuedRows()).toBe(2);
   });
 });
diff --git a/tests/claude-code/session-queue.test.ts b/tests/claude-code/session-queue.test.ts
index 3d38b7292..f88fb97cf 100644
--- a/tests/claude-code/session-queue.test.ts
+++ b/tests/claude-code/session-queue.test.ts
@@ -598,6 +598,17 @@ describe("session queue", () => {
 });
 
 describe("oversized queue files", () => {
+  it("rejects an append that would push the file past the ceiling", () => {
+    const queueDir = makeQueueDir();
+    const queuePath = appendQueuedSessionRow(makeRow("s-edge", 0), queueDir);
+    const existing = readFileSync(queuePath, "utf-8");
+
+    // Ceiling one byte above the current size: the file is below the limit,
+    // but the next row crosses it, so the file must be left byte-identical.
+    appendQueuedSessionRow(makeRow("s-edge", 1), queueDir, existing.length + 1);
+    expect(readFileSync(queuePath, "utf-8")).toBe(existing);
+  });
+
   it("stops appending once the queue file reaches the size ceiling", () => {
     const queueDir = makeQueueDir();
     const queuePath = appendQueuedSessionRow(makeRow("s-cap", 0), queueDir, 10_000);

From 35615ce52242bfae5eb37a4e4cf9f459623f643f Mon Sep 17 00:00:00 2001
From: Emanuele Fenocchi <emanuele.fenocchi@activeloop.dev>
Date: Thu, 20 Aug 2026 06:56:31 +0000
Subject: [PATCH 04/12] fix(cowork): drain a queue left by an outage, and stop
 silent drops

Codex adversarial review on #343 blocked on three items, all real:

1. The drain only ran when the tick appended something (`if (appendedAny)`).
   Before the watermark fix the replay itself kept that flag true, so the leak
   was also what retried the upload. Without it, a queue left behind by an
   outage was never uploaded if the Cowork session had gone quiet meanwhile.
   Now the drain also runs whenever the queue still holds rows. Regression
   test: recovery tick with an unchanged transcript must upload and clear.

2. A row dropped at the size ceiling still advanced the watermark, losing the
   message with only a debug log nobody has enabled. appendQueuedSessionRow
   now reports whether it appended, and the ingest records drops to
   ~/.deeplake/queue-cowork/.dropped-rows.jsonl.

3. gcOversizedQueueFiles deleted files at exactly the ceiling, which appends
   allow and which still hold unsent rows. It now only drops files strictly
   above the ceiling - unreachable for the fixed code, so it only ever
   collects residue from a build that predates the ceiling.
---
 src/hooks/session-queue.ts                  | 18 +++++--
 src/mcp/cowork-ingest.ts                    | 53 +++++++++++++++++++--
 tests/claude-code/cowork-queue-leak.test.ts | 33 +++++++++++--
 tests/claude-code/session-queue.test.ts     | 19 +++++---
 4 files changed, 105 insertions(+), 18 deletions(-)

diff --git a/src/hooks/session-queue.ts b/src/hooks/session-queue.ts
index 73fff9ec7..cd9cfa9b6 100644
--- a/src/hooks/session-queue.ts
+++ b/src/hooks/session-queue.ts
@@ -125,11 +125,17 @@ export function buildQueuedSessionRow(args: {
   };
 }
 
+export interface AppendQueuedRowResult {
+  queuePath: string;
+  /** false when the row was dropped because the file sits at its ceiling. */
+  appended: boolean;
+}
+
 export function appendQueuedSessionRow(
   row: QueuedSessionRow,
   queueDir = DEFAULT_QUEUE_DIR,
   maxQueueBytes = MAX_SESSION_QUEUE_BYTES,
-): string {
+): AppendQueuedRowResult {
   mkdirSync(queueDir, { recursive: true });
   const sessionId = extractSessionId(row.path);
   const queuePath = getQueuePath(queueDir, sessionId);
@@ -138,10 +144,10 @@ export function appendQueuedSessionRow(
   // "the last row may overshoot it by however large that row happened to be".
   if (fileSize(queuePath) + Buffer.byteLength(payload, "utf-8") > maxQueueBytes) {
     log("session-queue", `queue file at the ${maxQueueBytes}-byte ceiling, dropping row: ${queuePath}`);
-    return queuePath;
+    return { queuePath, appended: false };
   }
   appendFileSync(queuePath, payload);
-  return queuePath;
+  return { queuePath, appended: true };
 }
 
 /**
@@ -162,7 +168,11 @@ export function gcOversizedQueueFiles(queueDir = DEFAULT_QUEUE_DIR, maxQueueByte
     if (!name.endsWith(".jsonl") && !name.endsWith(".inflight")) continue;
     const path = join(queueDir, name);
     const size = fileSize(path);
-    if (size < maxQueueBytes) continue;
+    // Strictly ABOVE the ceiling. appendQueuedSessionRow never lets a file
+    // exceed it, so anything caught here is residue from a build that predates
+    // the ceiling — not rows the backend is still waiting for. A file sitting
+    // exactly at the ceiling is legitimate and is left alone to be flushed.
+    if (size <= maxQueueBytes) continue;
     try {
       rmSync(path, { force: true });
       reclaimed += size;
diff --git a/src/mcp/cowork-ingest.ts b/src/mcp/cowork-ingest.ts
index 64fb3b07c..f52fcf5c8 100644
--- a/src/mcp/cowork-ingest.ts
+++ b/src/mcp/cowork-ingest.ts
@@ -20,6 +20,7 @@
  * several concurrent MCP processes Cowork spawns from double-inserting.
  */
 import {
+  appendFileSync,
   closeSync,
   existsSync,
   mkdirSync,
@@ -60,6 +61,7 @@ const STATE_PATH = join(DEEPLAKE_DIR, "cowork-ingest-state.json");
 const LOCK_PATH = join(DEEPLAKE_DIR, ".cowork-ingest.lock");
 const COWORK_QUEUE_DIR = join(DEEPLAKE_DIR, "queue-cowork");
 const NOTICE_MARKER = join(DEEPLAKE_DIR, ".cowork-data-notice-shown");
+const DROPPED_MARKER = join(COWORK_QUEUE_DIR, ".dropped-rows.jsonl");
 const LOCK_STALE_MS = 60_000;
 // Refresh the held lock's mtime well inside LOCK_STALE_MS so a long ingest is
 // never mistaken for a dead run and stolen mid-flight by a second process.
@@ -154,6 +156,33 @@ function findTranscripts(root: string): string[] {
   return out;
 }
 
+/** True when the Cowork queue still holds rows the backend has not taken. */
+function hasQueuedRows(): boolean {
+  try {
+    return readdirSync(COWORK_QUEUE_DIR).some(n => n.endsWith(".jsonl") || n.endsWith(".inflight"));
+  } catch {
+    return false; // no queue dir yet
+  }
+}
+
+/**
+ * Note rows the queue ceiling refused. The transcript watermark has already
+ * moved past them, so they will never be uploaded; this file is the only
+ * durable trace of that loss.
+ */
+function recordDroppedRows(count: number): void {
+  try {
+    mkdirSync(COWORK_QUEUE_DIR, { recursive: true });
+    appendFileSync(
+      DROPPED_MARKER,
+      `${JSON.stringify({ at: new Date().toISOString(), droppedRows: count })}\n`,
+    );
+  } catch {
+    /* best effort */
+  }
+  log("cowork-ingest", `dropped ${count} row(s): the session queue is at its size ceiling`);
+}
+
 function loadState(): IngestState {
   try {
     const raw = JSON.parse(readFileSync(STATE_PATH, "utf-8"));
@@ -370,6 +399,7 @@ export async function ingestCoworkSessions(): Promise<{ ingested: number } | { s
     const state = loadState();
     const transcripts = findTranscripts(root);
     let appendedAny = false;
+    let dropped = 0;
 
     for (const path of transcripts) {
       let lines: string[];
@@ -400,9 +430,13 @@ export async function ingestCoworkSessions(): Promise<{ ingested: number } | { s
             pluginVersion: getVersion(),
             timestamp: String(entry.timestamp),
           });
-          appendQueuedSessionRow(row, COWORK_QUEUE_DIR);
-          appendedAny = true;
-          ingested += 1;
+          const { appended } = appendQueuedSessionRow(row, COWORK_QUEUE_DIR);
+          if (appended) {
+            appendedAny = true;
+            ingested += 1;
+          } else {
+            dropped += 1;
+          }
         }
       }
       state.processedLines[path] = lines.length;
@@ -419,7 +453,18 @@ export async function ingestCoworkSessions(): Promise<{ ingested: number } | { s
     // ~5 GB/day (2026-08-19).
     if (appendedAny) saveState(state);
 
-    if (appendedAny) {
+    if (dropped > 0) {
+      // The watermark advanced past these lines, so they are gone for good.
+      // Record it where support can find it — a debug log nobody has enabled
+      // is not a record. Only reachable after a very long upload outage.
+      recordDroppedRows(dropped);
+    }
+
+    // Drain whenever anything is queued — not only when this tick appended.
+    // Before the watermark fix, the replay itself kept appendedAny true and so
+    // kept retrying; without it, a queue left behind by an outage would never
+    // be uploaded if the Cowork session had meanwhile gone quiet.
+    if (appendedAny || hasQueuedRows()) {
       const api = new DeeplakeApi(
         config.token,
         config.apiUrl,
diff --git a/tests/claude-code/cowork-queue-leak.test.ts b/tests/claude-code/cowork-queue-leak.test.ts
index 7450a5803..280d97c67 100644
--- a/tests/claude-code/cowork-queue-leak.test.ts
+++ b/tests/claude-code/cowork-queue-leak.test.ts
@@ -19,11 +19,16 @@ import { tmpdir } from "node:os";
 
 const SESSION_ID = "b27efa59-a8bc-4ea3-8b02-18cbc608ae17";
 
-// Every upload fails, the way it does on a disconnected network filesystem.
+// The upload fails while `offline` is true, the way it does on a disconnected
+// network filesystem, and succeeds once it flips back.
+const uploads: string[] = [];
+let offline = true;
 vi.mock("../../src/deeplake-api.js", () => ({
   DeeplakeApi: class {
-    async query(): Promise<never> {
-      throw new Error("fetch failed: ECONNREFUSED");
+    async query(sql: string): Promise<never[]> {
+      if (offline) throw new Error("fetch failed: ECONNREFUSED");
+      uploads.push(sql);
+      return [];
     }
     async ensureSessionsTable(): Promise<void> {}
   },
@@ -70,6 +75,8 @@ beforeEach(() => {
   prevHome = process.env.HOME;
   home = mkdtempSync(join(tmpdir(), "cowork-leak-"));
   process.env.HOME = home;
+  offline = true;
+  uploads.length = 0;
   mkdirSync(join(home, ".deeplake"), { recursive: true });
   writeFileSync(
     join(home, ".deeplake", "credentials.json"),
@@ -119,4 +126,24 @@ describe("cowork queue growth when uploads fail", () => {
     // The second tick queues the one new message, not the whole transcript.
     expect(queuedRows()).toBe(2);
   });
+
+  it("uploads the queue left by an outage on a later tick, with no new transcript content", async () => {
+    const path = transcriptPath();
+    writeFileSync(path, line("queued while offline"));
+
+    const { ingestCoworkSessions } = await import("../../src/mcp/cowork-ingest.js");
+    await ingestCoworkSessions();
+    expect(queuedRows()).toBe(1);
+    expect(uploads).toHaveLength(0);
+
+    // Network is back. The transcript has NOT changed, so this tick appends
+    // nothing — the queued row must still be uploaded and the file cleared.
+    offline = false;
+    await ingestCoworkSessions();
+
+    expect(uploads).toHaveLength(1);
+    expect(uploads[0]).toContain("queued while offline");
+    expect(queuedRows()).toBe(0);
+    expect(queueBytes()).toBe(0);
+  });
 });
diff --git a/tests/claude-code/session-queue.test.ts b/tests/claude-code/session-queue.test.ts
index f88fb97cf..9e1d6e00a 100644
--- a/tests/claude-code/session-queue.test.ts
+++ b/tests/claude-code/session-queue.test.ts
@@ -89,7 +89,7 @@ describe("session queue", () => {
     const queueDir = makeQueueDir();
     const row = makeRow("session-append", 1);
 
-    const queuePath = appendQueuedSessionRow(row, queueDir);
+    const { queuePath } = appendQueuedSessionRow(row, queueDir);
     const lines = readFileSync(queuePath, "utf-8").trim().split("\n");
 
     expect(lines).toHaveLength(1);
@@ -600,7 +600,7 @@ describe("session queue", () => {
 describe("oversized queue files", () => {
   it("rejects an append that would push the file past the ceiling", () => {
     const queueDir = makeQueueDir();
-    const queuePath = appendQueuedSessionRow(makeRow("s-edge", 0), queueDir);
+    const { queuePath } = appendQueuedSessionRow(makeRow("s-edge", 0), queueDir);
     const existing = readFileSync(queuePath, "utf-8");
 
     // Ceiling one byte above the current size: the file is below the limit,
@@ -611,7 +611,7 @@ describe("oversized queue files", () => {
 
   it("stops appending once the queue file reaches the size ceiling", () => {
     const queueDir = makeQueueDir();
-    const queuePath = appendQueuedSessionRow(makeRow("s-cap", 0), queueDir, 10_000);
+    const { queuePath } = appendQueuedSessionRow(makeRow("s-cap", 0), queueDir, 10_000);
     const afterFirst = readFileSync(queuePath, "utf-8");
 
     // Simulate a queue that has already reached the ceiling.
@@ -627,17 +627,22 @@ describe("oversized queue files", () => {
 
   it("drops queue and inflight files past the ceiling and leaves the rest alone", () => {
     const queueDir = makeQueueDir();
-    const small = appendQueuedSessionRow(makeRow("s-small", 0), queueDir);
+    const { queuePath: small } = appendQueuedSessionRow(makeRow("s-small", 0), queueDir);
     const big = join(queueDir, "s-big.jsonl");
     const bigInflight = join(queueDir, "s-big-2.inflight");
-    writeFileSync(big, "x".repeat(4096));
-    writeFileSync(bigInflight, "x".repeat(4096));
+    const atCeiling = join(queueDir, "s-exact.jsonl");
+    writeFileSync(big, "x".repeat(4097));
+    writeFileSync(bigInflight, "x".repeat(5000));
+    writeFileSync(atCeiling, "x".repeat(4096));
 
     const reclaimed = gcOversizedQueueFiles(queueDir, 4096);
 
-    expect(reclaimed).toBe(8192);
+    expect(reclaimed).toBe(9097);
     expect(existsSync(big)).toBe(false);
     expect(existsSync(bigInflight)).toBe(false);
+    // A file sitting exactly at the ceiling is legitimate — appends stop there,
+    // so it still holds rows the backend has not taken. It must survive.
+    expect(existsSync(atCeiling)).toBe(true);
     expect(existsSync(small)).toBe(true);
   });
 

From 2fea4fe3e0f2d29e9bb599b1783165041d2fba48 Mon Sep 17 00:00:00 2001
From: Emanuele Fenocchi <emanuele.fenocchi@activeloop.dev>
Date: Thu, 20 Aug 2026 07:02:38 +0000
Subject: [PATCH 05/12] fix(cowork): keep the dropped-row journal out of queue
 discovery

CodeRabbit on #343: the journal lived in the queue dir and ended in .jsonl, so
hasQueuedRows() reported pending work forever after a single drop, and the GC
could delete the only record of that loss. It now lives at
~/.deeplake/cowork-dropped-rows.jsonl, and both queue discovery and the GC skip
dot-prefixed metadata (drain lock, disabled marker) outright.

Also asserts the exact uploaded row (parsed from the jsonb literal) instead of
a substring of the INSERT statement.
---
 src/hooks/session-queue.ts                  |  3 +++
 src/mcp/cowork-ingest.ts                    |  9 ++++++---
 tests/claude-code/cowork-queue-leak.test.ts | 11 ++++++++++-
 tests/claude-code/session-queue.test.ts     |  5 +++++
 4 files changed, 24 insertions(+), 4 deletions(-)

diff --git a/src/hooks/session-queue.ts b/src/hooks/session-queue.ts
index cd9cfa9b6..89f054d1f 100644
--- a/src/hooks/session-queue.ts
+++ b/src/hooks/session-queue.ts
@@ -165,6 +165,9 @@ export function gcOversizedQueueFiles(queueDir = DEFAULT_QUEUE_DIR, maxQueueByte
     return 0;
   }
   for (const name of names) {
+    // Skip queue metadata (drain lock, disabled marker, any journal a caller
+    // parks here): it is bookkeeping, never rows the backend is owed.
+    if (name.startsWith(".")) continue;
     if (!name.endsWith(".jsonl") && !name.endsWith(".inflight")) continue;
     const path = join(queueDir, name);
     const size = fileSize(path);
diff --git a/src/mcp/cowork-ingest.ts b/src/mcp/cowork-ingest.ts
index f52fcf5c8..6f7238560 100644
--- a/src/mcp/cowork-ingest.ts
+++ b/src/mcp/cowork-ingest.ts
@@ -61,7 +61,7 @@ const STATE_PATH = join(DEEPLAKE_DIR, "cowork-ingest-state.json");
 const LOCK_PATH = join(DEEPLAKE_DIR, ".cowork-ingest.lock");
 const COWORK_QUEUE_DIR = join(DEEPLAKE_DIR, "queue-cowork");
 const NOTICE_MARKER = join(DEEPLAKE_DIR, ".cowork-data-notice-shown");
-const DROPPED_MARKER = join(COWORK_QUEUE_DIR, ".dropped-rows.jsonl");
+const DROPPED_MARKER = join(DEEPLAKE_DIR, "cowork-dropped-rows.jsonl");
 const LOCK_STALE_MS = 60_000;
 // Refresh the held lock's mtime well inside LOCK_STALE_MS so a long ingest is
 // never mistaken for a dead run and stolen mid-flight by a second process.
@@ -159,7 +159,10 @@ function findTranscripts(root: string): string[] {
 /** True when the Cowork queue still holds rows the backend has not taken. */
 function hasQueuedRows(): boolean {
   try {
-    return readdirSync(COWORK_QUEUE_DIR).some(n => n.endsWith(".jsonl") || n.endsWith(".inflight"));
+    // Dot-prefixed entries are queue metadata (drain lock, disabled marker),
+    // never rows owed to the backend.
+    return readdirSync(COWORK_QUEUE_DIR)
+      .some(n => !n.startsWith(".") && (n.endsWith(".jsonl") || n.endsWith(".inflight")));
   } catch {
     return false; // no queue dir yet
   }
@@ -172,7 +175,7 @@ function hasQueuedRows(): boolean {
  */
 function recordDroppedRows(count: number): void {
   try {
-    mkdirSync(COWORK_QUEUE_DIR, { recursive: true });
+    mkdirSync(DEEPLAKE_DIR, { recursive: true });
     appendFileSync(
       DROPPED_MARKER,
       `${JSON.stringify({ at: new Date().toISOString(), droppedRows: count })}\n`,
diff --git a/tests/claude-code/cowork-queue-leak.test.ts b/tests/claude-code/cowork-queue-leak.test.ts
index 280d97c67..e6482d963 100644
--- a/tests/claude-code/cowork-queue-leak.test.ts
+++ b/tests/claude-code/cowork-queue-leak.test.ts
@@ -142,7 +142,16 @@ describe("cowork queue growth when uploads fail", () => {
     await ingestCoworkSessions();
 
     expect(uploads).toHaveLength(1);
-    expect(uploads[0]).toContain("queued while offline");
+    // Assert the row that actually went up, not a substring of the statement.
+    const jsonb = uploads[0].match(/'(\{.*?\})'::jsonb/)?.[1];
+    expect(jsonb).toBeDefined();
+    expect(JSON.parse(jsonb!.replace(/''/g, "'"))).toMatchObject({
+      session_id: SESSION_ID,
+      type: "user_message",
+      content: "queued while offline",
+      agent: "claude_cowork",
+      cwd: "/cowork",
+    });
     expect(queuedRows()).toBe(0);
     expect(queueBytes()).toBe(0);
   });
diff --git a/tests/claude-code/session-queue.test.ts b/tests/claude-code/session-queue.test.ts
index 9e1d6e00a..d6b8f2fd8 100644
--- a/tests/claude-code/session-queue.test.ts
+++ b/tests/claude-code/session-queue.test.ts
@@ -631,6 +631,8 @@ describe("oversized queue files", () => {
     const big = join(queueDir, "s-big.jsonl");
     const bigInflight = join(queueDir, "s-big-2.inflight");
     const atCeiling = join(queueDir, "s-exact.jsonl");
+    const journal = join(queueDir, ".dropped-rows.jsonl");
+    writeFileSync(journal, "x".repeat(9000));
     writeFileSync(big, "x".repeat(4097));
     writeFileSync(bigInflight, "x".repeat(5000));
     writeFileSync(atCeiling, "x".repeat(4096));
@@ -644,6 +646,9 @@ describe("oversized queue files", () => {
     // so it still holds rows the backend has not taken. It must survive.
     expect(existsSync(atCeiling)).toBe(true);
     expect(existsSync(small)).toBe(true);
+    // Dot-prefixed entries are queue metadata, not rows owed to the backend —
+    // GC must never collect them, however large they get.
+    expect(existsSync(journal)).toBe(true);
   });
 
   it("defaults the ceiling to 256 MB", () => {

From 55a9718bb2f81e15bee9ba3ac7dd65c5761fc018 Mon Sep 17 00:00:00 2001
From: Emanuele Fenocchi <emanuele.fenocchi@activeloop.dev>
Date: Thu, 20 Aug 2026 07:08:28 +0000
Subject: [PATCH 06/12] fix(cowork): journal what the GC throws away, and
 tighten the recovery test

Codex re-review on #343:

- gcOversizedQueueFiles() now reports each dropped file to an optional callback;
  the Cowork ingest records it to ~/.deeplake/cowork-dropped-rows.jsonl, so a
  deletion of never-acknowledged rows leaves a durable trace instead of only a
  debug log. Deleting is still the deliberate tradeoff: a file above the ceiling
  cannot be flushed (readQueuedRows reads it whole) and can only come from a
  build that predates the ceiling.
- The stale-queue recovery test now asserts ingested === 0 and exactly one
  VALUES tuple in the uploaded statement, so it cannot be satisfied by a tick
  that re-appended the transcript first - the origin/main behaviour.
---
 src/hooks/session-queue.ts                  | 10 +++++++++-
 src/mcp/cowork-ingest.ts                    | 17 ++++++++++-------
 tests/claude-code/cowork-queue-leak.test.ts |  8 +++++++-
 tests/claude-code/session-queue.test.ts     |  6 +++++-
 4 files changed, 31 insertions(+), 10 deletions(-)

diff --git a/src/hooks/session-queue.ts b/src/hooks/session-queue.ts
index 89f054d1f..397a731ea 100644
--- a/src/hooks/session-queue.ts
+++ b/src/hooks/session-queue.ts
@@ -156,7 +156,11 @@ export function appendQueuedSessionRow(
  * would otherwise sit on disk forever — it is the residue of an upload outage,
  * not data the backend is still waiting for. Returns the bytes reclaimed.
  */
-export function gcOversizedQueueFiles(queueDir = DEFAULT_QUEUE_DIR, maxQueueBytes = MAX_SESSION_QUEUE_BYTES): number {
+export function gcOversizedQueueFiles(
+  queueDir = DEFAULT_QUEUE_DIR,
+  maxQueueBytes = MAX_SESSION_QUEUE_BYTES,
+  onDropped?: (path: string, sizeBytes: number) => void,
+): number {
   let reclaimed = 0;
   let names: string[];
   try {
@@ -180,6 +184,10 @@ export function gcOversizedQueueFiles(queueDir = DEFAULT_QUEUE_DIR, maxQueueByte
       rmSync(path, { force: true });
       reclaimed += size;
       log("session-queue", `dropped oversized queue file (${size} bytes): ${path}`);
+      // The rows in it were never acknowledged by the backend. Hand the caller
+      // the loss so it can be recorded somewhere durable rather than only in a
+      // debug log nobody has enabled.
+      onDropped?.(path, size);
     } catch {
       /* best effort */
     }
diff --git a/src/mcp/cowork-ingest.ts b/src/mcp/cowork-ingest.ts
index 6f7238560..08d6291bd 100644
--- a/src/mcp/cowork-ingest.ts
+++ b/src/mcp/cowork-ingest.ts
@@ -173,17 +173,18 @@ function hasQueuedRows(): boolean {
  * moved past them, so they will never be uploaded; this file is the only
  * durable trace of that loss.
  */
-function recordDroppedRows(count: number): void {
+function recordLoss(detail: Record<string, unknown>): void {
   try {
     mkdirSync(DEEPLAKE_DIR, { recursive: true });
-    appendFileSync(
-      DROPPED_MARKER,
-      `${JSON.stringify({ at: new Date().toISOString(), droppedRows: count })}\n`,
-    );
+    appendFileSync(DROPPED_MARKER, `${JSON.stringify({ at: new Date().toISOString(), ...detail })}\n`);
   } catch {
     /* best effort */
   }
-  log("cowork-ingest", `dropped ${count} row(s): the session queue is at its size ceiling`);
+  log("cowork-ingest", `recorded queue loss: ${JSON.stringify(detail)}`);
+}
+
+function recordDroppedRows(count: number): void {
+  recordLoss({ droppedRows: count, reason: "session queue at its size ceiling" });
 }
 
 function loadState(): IngestState {
@@ -398,7 +399,9 @@ export async function ingestCoworkSessions(): Promise<{ ingested: number } | { s
   try {
     // Reclaim any queue file left oversized by an earlier upload outage before
     // writing more rows — such a file can no longer be flushed.
-    gcOversizedQueueFiles(COWORK_QUEUE_DIR);
+    gcOversizedQueueFiles(COWORK_QUEUE_DIR, undefined, (path, sizeBytes) =>
+      recordLoss({ droppedQueueFile: path, sizeBytes }),
+    );
     const state = loadState();
     const transcripts = findTranscripts(root);
     let appendedAny = false;
diff --git a/tests/claude-code/cowork-queue-leak.test.ts b/tests/claude-code/cowork-queue-leak.test.ts
index e6482d963..4b14b972f 100644
--- a/tests/claude-code/cowork-queue-leak.test.ts
+++ b/tests/claude-code/cowork-queue-leak.test.ts
@@ -139,9 +139,15 @@ describe("cowork queue growth when uploads fail", () => {
     // Network is back. The transcript has NOT changed, so this tick appends
     // nothing — the queued row must still be uploaded and the file cleared.
     offline = false;
-    await ingestCoworkSessions();
+    const result = await ingestCoworkSessions();
 
+    // ingested === 0 is what separates this from the old behaviour: on
+    // origin/main the tick only uploaded because it had re-appended the
+    // transcript first, which would show up here as ingested > 0.
+    expect(result).toEqual({ ingested: 0 });
     expect(uploads).toHaveLength(1);
+    // Exactly one row in that statement — one VALUES tuple, not a replayed batch.
+    expect(uploads[0].match(/::jsonb/g)).toHaveLength(1);
     // Assert the row that actually went up, not a substring of the statement.
     const jsonb = uploads[0].match(/'(\{.*?\})'::jsonb/)?.[1];
     expect(jsonb).toBeDefined();
diff --git a/tests/claude-code/session-queue.test.ts b/tests/claude-code/session-queue.test.ts
index d6b8f2fd8..5ea48c9db 100644
--- a/tests/claude-code/session-queue.test.ts
+++ b/tests/claude-code/session-queue.test.ts
@@ -637,9 +637,13 @@ describe("oversized queue files", () => {
     writeFileSync(bigInflight, "x".repeat(5000));
     writeFileSync(atCeiling, "x".repeat(4096));
 
-    const reclaimed = gcOversizedQueueFiles(queueDir, 4096);
+    const losses: Array<[string, number]> = [];
+    const reclaimed = gcOversizedQueueFiles(queueDir, 4096, (path, size) => losses.push([path, size]));
 
     expect(reclaimed).toBe(9097);
+    // The caller is told exactly what was thrown away, so the loss can be
+    // recorded durably instead of only in a debug log.
+    expect(losses.sort()).toEqual([[big, 4097], [bigInflight, 5000]].sort());
     expect(existsSync(big)).toBe(false);
     expect(existsSync(bigInflight)).toBe(false);
     // A file sitting exactly at the ceiling is legitimate — appends stop there,

From 700309ae9725c19584c82b896a665629e4f5e321 Mon Sep 17 00:00:00 2001
From: Emanuele Fenocchi <emanuele.fenocchi@activeloop.dev>
Date: Thu, 20 Aug 2026 07:24:50 +0000
Subject: [PATCH 07/12] fix(cowork): never drop a transcript line at the queue
 ceiling

CodeRabbit (critical) and codex both flagged the same thing: a row rejected at
the ceiling still advanced the watermark, so that message was lost for good.

The watermark now advances one transcript line at a time, and a line's rows are
queued all-or-nothing: if they do not all fit, the watermark stays put and the
line is retried after the drain makes room. Nothing is dropped and nothing is
half-queued. This cannot bring the leak back, because the ceiling refuses the
appends that would grow the file.

The one case that still skips a line is a single line larger than the entire
ceiling, which can never be queued and would otherwise freeze that transcript
forever; it is journalled to ~/.deeplake/cowork-dropped-rows.jsonl.

Regression test: with the queue filled to the ceiling the tick returns
ingested 0 and loses nothing, and after room is freed the held line uploads
exactly once while the earlier ones are not re-queued. Against the previous
drop-and-advance behaviour it fails with 'expected { ingested: 0 } to deeply
equal { ingested: 1 }'.
---
 src/hooks/session-queue.ts                  | 18 +++++
 src/mcp/cowork-ingest.ts                    | 79 +++++++++++++--------
 tests/claude-code/cowork-queue-leak.test.ts | 42 +++++++++++
 3 files changed, 110 insertions(+), 29 deletions(-)

diff --git a/src/hooks/session-queue.ts b/src/hooks/session-queue.ts
index 397a731ea..5c032e9f4 100644
--- a/src/hooks/session-queue.ts
+++ b/src/hooks/session-queue.ts
@@ -195,6 +195,24 @@ export function gcOversizedQueueFiles(
   return reclaimed;
 }
 
+/** Bytes one row occupies in a queue file, newline included. */
+export function queuedRowBytes(row: QueuedSessionRow): number {
+  return Buffer.byteLength(`${JSON.stringify(row)}\n`, "utf-8");
+}
+
+/**
+ * Bytes still available in a session's queue file before it hits the ceiling.
+ * Lets a caller check that a whole group of rows fits BEFORE appending any of
+ * them, so it never leaves half a transcript line queued.
+ */
+export function sessionQueueRoomBytes(
+  sessionId: string,
+  queueDir = DEFAULT_QUEUE_DIR,
+  maxQueueBytes = MAX_SESSION_QUEUE_BYTES,
+): number {
+  return Math.max(0, maxQueueBytes - fileSize(getQueuePath(queueDir, sessionId)));
+}
+
 function fileSize(path: string): number {
   try {
     return statSync(path).size;
diff --git a/src/mcp/cowork-ingest.ts b/src/mcp/cowork-ingest.ts
index 08d6291bd..a3334b878 100644
--- a/src/mcp/cowork-ingest.ts
+++ b/src/mcp/cowork-ingest.ts
@@ -45,6 +45,9 @@ import {
   buildSessionPath,
   drainSessionQueues,
   gcOversizedQueueFiles,
+  queuedRowBytes,
+  sessionQueueRoomBytes,
+  MAX_SESSION_QUEUE_BYTES,
 } from "../hooks/session-queue.js";
 import { spawnWikiWorker, bundleDirFromImportMeta } from "../hooks/spawn-wiki-worker.js";
 import { forceSessionEndTrigger } from "../skillify/triggers.js";
@@ -183,10 +186,6 @@ function recordLoss(detail: Record<string, unknown>): void {
   log("cowork-ingest", `recorded queue loss: ${JSON.stringify(detail)}`);
 }
 
-function recordDroppedRows(count: number): void {
-  recordLoss({ droppedRows: count, reason: "session queue at its size ceiling" });
-}
-
 function loadState(): IngestState {
   try {
     const raw = JSON.parse(readFileSync(STATE_PATH, "utf-8"));
@@ -405,7 +404,7 @@ export async function ingestCoworkSessions(): Promise<{ ingested: number } | { s
     const state = loadState();
     const transcripts = findTranscripts(root);
     let appendedAny = false;
-    let dropped = 0;
+    let queueFull = false;
 
     for (const path of transcripts) {
       let lines: string[];
@@ -417,35 +416,60 @@ export async function ingestCoworkSessions(): Promise<{ ingested: number } | { s
       const already = state.processedLines[path] ?? 0;
       if (lines.length <= already) continue;
 
+      // Advance the watermark one transcript line at a time. A line's rows are
+      // queued all-or-nothing: if they do not all fit under the queue ceiling,
+      // the watermark stays put and the line is retried once the queue drains,
+      // rather than being half-queued or silently dropped.
+      let processed = already;
       for (const raw of lines.slice(already)) {
         let parsed: TranscriptLine;
         try {
           parsed = JSON.parse(raw);
         } catch {
+          processed += 1;
           continue;
         }
-        for (const entry of entriesForLine(parsed)) {
-          const serialized = JSON.stringify(entry);
-          const row = buildQueuedSessionRow({
-            sessionPath: buildSessionPath(config, String(entry.session_id)),
-            line: serialized,
-            userName: config.userName,
-            projectName: COWORK_PROJECT,
-            description: String(entry.type ?? ""),
-            agent: COWORK_AGENT,
-            pluginVersion: getVersion(),
-            timestamp: String(entry.timestamp),
-          });
-          const { appended } = appendQueuedSessionRow(row, COWORK_QUEUE_DIR);
-          if (appended) {
-            appendedAny = true;
-            ingested += 1;
-          } else {
-            dropped += 1;
+
+        const rows = entriesForLine(parsed).map(entry => buildQueuedSessionRow({
+          sessionPath: buildSessionPath(config, String(entry.session_id)),
+          line: JSON.stringify(entry),
+          userName: config.userName,
+          projectName: COWORK_PROJECT,
+          description: String(entry.type ?? ""),
+          agent: COWORK_AGENT,
+          pluginVersion: getVersion(),
+          timestamp: String(entry.timestamp),
+        }));
+        if (rows.length === 0) {
+          processed += 1;
+          continue;
+        }
+
+        const sessionId = String(parsed.sessionId);
+        const needed = rows.reduce((n, row) => n + queuedRowBytes(row), 0);
+        if (needed > sessionQueueRoomBytes(sessionId, COWORK_QUEUE_DIR)) {
+          if (needed > MAX_SESSION_QUEUE_BYTES) {
+            // Bigger than the whole ceiling — it can never be queued, and
+            // stalling here would freeze this transcript forever. Skip it, on
+            // the record.
+            recordLoss({ skippedTranscriptLine: path, sessionId, neededBytes: needed });
+            processed += 1;
+            continue;
           }
+          // The queue is full. Stop here; the watermark keeps this line for the
+          // next tick, once the drain below has made room.
+          queueFull = true;
+          break;
+        }
+
+        for (const row of rows) {
+          appendQueuedSessionRow(row, COWORK_QUEUE_DIR);
+          appendedAny = true;
+          ingested += 1;
         }
+        processed += 1;
       }
-      state.processedLines[path] = lines.length;
+      state.processedLines[path] = processed;
     }
 
     // Persist the watermark as soon as the rows are in the on-disk queue, and
@@ -459,11 +483,8 @@ export async function ingestCoworkSessions(): Promise<{ ingested: number } | { s
     // ~5 GB/day (2026-08-19).
     if (appendedAny) saveState(state);
 
-    if (dropped > 0) {
-      // The watermark advanced past these lines, so they are gone for good.
-      // Record it where support can find it — a debug log nobody has enabled
-      // is not a record. Only reachable after a very long upload outage.
-      recordDroppedRows(dropped);
+    if (queueFull) {
+      recordLoss({ queueFull: true, note: "ingestion paused at the queue ceiling; no rows dropped" });
     }
 
     // Drain whenever anything is queued — not only when this tick appended.
diff --git a/tests/claude-code/cowork-queue-leak.test.ts b/tests/claude-code/cowork-queue-leak.test.ts
index 4b14b972f..7a35fd782 100644
--- a/tests/claude-code/cowork-queue-leak.test.ts
+++ b/tests/claude-code/cowork-queue-leak.test.ts
@@ -61,6 +61,10 @@ function queueBytes(): number {
   }
 }
 
+function queuePath(): string {
+  return join(home, ".deeplake", "queue-cowork", `${SESSION_ID}.jsonl`);
+}
+
 function queuedRows(): number {
   try {
     return readFileSync(join(home, ".deeplake", "queue-cowork", `${SESSION_ID}.jsonl`), "utf-8")
@@ -161,4 +165,42 @@ describe("cowork queue growth when uploads fail", () => {
     expect(queuedRows()).toBe(0);
     expect(queueBytes()).toBe(0);
   });
+
+  it("holds the watermark when the queue is full, then ingests those lines exactly once", async () => {
+    const path = transcriptPath();
+    writeFileSync(path, line("first") + line("second"));
+
+    const { ingestCoworkSessions } = await import("../../src/mcp/cowork-ingest.js");
+    const { MAX_SESSION_QUEUE_BYTES } = await import("../../src/hooks/session-queue.js");
+
+    // First tick queues both lines while offline.
+    await ingestCoworkSessions();
+    expect(queuedRows()).toBe(2);
+
+    // Fill the queue file to its ceiling: the next tick cannot append anything.
+    const queued = readFileSync(queuePath(), "utf-8");
+    writeFileSync(queuePath(), queued + "x".repeat(MAX_SESSION_QUEUE_BYTES - queued.length));
+    appendFileSync(path, line("third while full"));
+    const full = await ingestCoworkSessions();
+
+    // Nothing queued, and — the point of the test — the line is NOT lost.
+    expect(full).toEqual({ ingested: 0 });
+
+    // Make room and come back online: the held line is ingested exactly once,
+    // and the two earlier ones are not queued a second time.
+    writeFileSync(queuePath(), queued);
+    offline = false;
+    const recovered = await ingestCoworkSessions();
+
+    expect(recovered).toEqual({ ingested: 1 });
+    const jsonbs = uploads.join(" ").match(/::jsonb/g) ?? [];
+    expect(jsonbs).toHaveLength(3);
+    const contents = uploads.join(" ").match(/"content":"[^"]*"/g) ?? [];
+    expect(contents.sort()).toEqual([
+      '"content":"first"',
+      '"content":"second"',
+      '"content":"third while full"',
+    ]);
+    expect(queuedRows()).toBe(0);
+  });
 });

From c320684ceac8b3e829e287ad06891a6fa8d6d2ed Mon Sep 17 00:00:00 2001
From: Emanuele Fenocchi <emanuele.fenocchi@activeloop.dev>
Date: Thu, 20 Aug 2026 07:30:59 +0000
Subject: [PATCH 08/12] fix(cowork): stop the loss journal from becoming the
 next unbounded file

Codex pass 4 on #343: the journal appended a line on every 30s tick for as long
as the queue stayed full, so a fix for an unbounded file introduced a smaller
unbounded file.

A full queue loses nothing now - ingestion is simply paused until the drain
frees room - so that state is a debug log, not a journal entry. The journal is
left for real, irreversible losses (a queue file dropped by the GC, a transcript
line larger than the whole ceiling), both rare and one-shot, and it carries its
own 1 MB ceiling.

Test: three consecutive full-queue ticks must leave no journal file at all.
---
 src/mcp/cowork-ingest.ts                    | 20 ++++++++++++++++-
 tests/claude-code/cowork-queue-leak.test.ts | 24 ++++++++++++++++++++-
 2 files changed, 42 insertions(+), 2 deletions(-)

diff --git a/src/mcp/cowork-ingest.ts b/src/mcp/cowork-ingest.ts
index a3334b878..b941a440c 100644
--- a/src/mcp/cowork-ingest.ts
+++ b/src/mcp/cowork-ingest.ts
@@ -65,6 +65,7 @@ const LOCK_PATH = join(DEEPLAKE_DIR, ".cowork-ingest.lock");
 const COWORK_QUEUE_DIR = join(DEEPLAKE_DIR, "queue-cowork");
 const NOTICE_MARKER = join(DEEPLAKE_DIR, ".cowork-data-notice-shown");
 const DROPPED_MARKER = join(DEEPLAKE_DIR, "cowork-dropped-rows.jsonl");
+const MAX_LOSS_JOURNAL_BYTES = 1024 * 1024;
 const LOCK_STALE_MS = 60_000;
 // Refresh the held lock's mtime well inside LOCK_STALE_MS so a long ingest is
 // never mistaken for a dead run and stolen mid-flight by a second process.
@@ -176,9 +177,23 @@ function hasQueuedRows(): boolean {
  * moved past them, so they will never be uploaded; this file is the only
  * durable trace of that loss.
  */
+/**
+ * Record a real, irreversible loss. Only reachable for a queue file dropped by
+ * the GC or a transcript line larger than the whole ceiling — both rare and
+ * one-shot. The journal carries its own ceiling so it can never become the next
+ * unbounded file.
+ */
 function recordLoss(detail: Record<string, unknown>): void {
   try {
     mkdirSync(DEEPLAKE_DIR, { recursive: true });
+    try {
+      if (statSync(DROPPED_MARKER).size >= MAX_LOSS_JOURNAL_BYTES) {
+        log("cowork-ingest", "loss journal is at its ceiling, not recording further entries");
+        return;
+      }
+    } catch {
+      /* no journal yet */
+    }
     appendFileSync(DROPPED_MARKER, `${JSON.stringify({ at: new Date().toISOString(), ...detail })}\n`);
   } catch {
     /* best effort */
@@ -484,7 +499,10 @@ export async function ingestCoworkSessions(): Promise<{ ingested: number } | { s
     if (appendedAny) saveState(state);
 
     if (queueFull) {
-      recordLoss({ queueFull: true, note: "ingestion paused at the queue ceiling; no rows dropped" });
+      // Debug log only, deliberately: nothing is lost here, ingestion is just
+      // paused until the drain frees room, and this state repeats on every
+      // 30s tick — journalling it would itself grow without bound.
+      log("cowork-ingest", "ingestion paused at the queue ceiling; no rows dropped");
     }
 
     // Drain whenever anything is queued — not only when this tick appended.
diff --git a/tests/claude-code/cowork-queue-leak.test.ts b/tests/claude-code/cowork-queue-leak.test.ts
index 7a35fd782..a4773c23d 100644
--- a/tests/claude-code/cowork-queue-leak.test.ts
+++ b/tests/claude-code/cowork-queue-leak.test.ts
@@ -13,7 +13,7 @@
  * to the queue again — forever, growing without bound.
  */
 import { describe, it, expect, beforeEach, afterEach, vi } from "vitest";
-import { mkdtempSync, mkdirSync, writeFileSync, readFileSync, statSync, appendFileSync } from "node:fs";
+import { mkdtempSync, mkdirSync, writeFileSync, readFileSync, statSync, appendFileSync, existsSync } from "node:fs";
 import { join } from "node:path";
 import { tmpdir } from "node:os";
 
@@ -203,4 +203,26 @@ describe("cowork queue growth when uploads fail", () => {
     ]);
     expect(queuedRows()).toBe(0);
   });
+
+  it("does not journal anything while the queue is merely full", async () => {
+    const path = transcriptPath();
+    writeFileSync(path, line("first"));
+
+    const { ingestCoworkSessions } = await import("../../src/mcp/cowork-ingest.js");
+    const { MAX_SESSION_QUEUE_BYTES } = await import("../../src/hooks/session-queue.js");
+    await ingestCoworkSessions();
+
+    const queued = readFileSync(queuePath(), "utf-8");
+    writeFileSync(queuePath(), queued + "x".repeat(MAX_SESSION_QUEUE_BYTES - queued.length));
+    appendFileSync(path, line("held back"));
+
+    // Three full-queue ticks in a row: nothing is lost, so nothing is written
+    // to the loss journal — otherwise it would grow on every 30s tick forever.
+    await ingestCoworkSessions();
+    await ingestCoworkSessions();
+    await ingestCoworkSessions();
+
+    expect(existsSync(join(home, ".deeplake", "cowork-dropped-rows.jsonl"))).toBe(false);
+    // Each tick re-reads a 256 MB queue file, so this is slower than the rest.
+  }, 60_000);
 });

From 0b5f7c2bf07eb3e3d2f737e941cb25ab08be977b Mon Sep 17 00:00:00 2001
From: Emanuele Fenocchi <emanuele.fenocchi@activeloop.dev>
Date: Thu, 20 Aug 2026 07:35:11 +0000
Subject: [PATCH 09/12] fix(queue): close the file-system race CodeQL flagged
 on the new size checks

CodeQL (high) on #343: both new ceiling checks were stat-then-append on a path,
a js/file-system-race, and several Cowork MCP processes write these files
concurrently. Both now open once and use fstat + write on that descriptor, so
the size the check sees is the size the write extends.
---
 src/hooks/session-queue.ts | 18 ++++++++++++++----
 src/mcp/cowork-ingest.ts   | 15 ++++++++++-----
 2 files changed, 24 insertions(+), 9 deletions(-)

diff --git a/src/hooks/session-queue.ts b/src/hooks/session-queue.ts
index 5c032e9f4..e194b2b16 100644
--- a/src/hooks/session-queue.ts
+++ b/src/hooks/session-queue.ts
@@ -2,6 +2,7 @@ import {
   appendFileSync,
   closeSync,
   existsSync,
+  fstatSync,
   mkdirSync,
   openSync,
   readFileSync,
@@ -10,6 +11,7 @@ import {
   rmSync,
   statSync,
   writeFileSync,
+  writeSync,
 } from "node:fs";
 import { dirname, join } from "node:path";
 import { homedir } from "node:os";
@@ -142,11 +144,19 @@ export function appendQueuedSessionRow(
   const payload = `${JSON.stringify(row)}\n`;
   // Project the post-append size, so the ceiling is a real ceiling rather than
   // "the last row may overshoot it by however large that row happened to be".
-  if (fileSize(queuePath) + Buffer.byteLength(payload, "utf-8") > maxQueueBytes) {
-    log("session-queue", `queue file at the ${maxQueueBytes}-byte ceiling, dropping row: ${queuePath}`);
-    return { queuePath, appended: false };
+  // Checked and written through ONE descriptor: a stat-then-append on the path
+  // is a file-system race (CodeQL js/file-system-race), and several Cowork MCP
+  // processes write this file concurrently.
+  const fd = openSync(queuePath, "a");
+  try {
+    if (fstatSync(fd).size + Buffer.byteLength(payload, "utf-8") > maxQueueBytes) {
+      log("session-queue", `queue file at the ${maxQueueBytes}-byte ceiling, dropping row: ${queuePath}`);
+      return { queuePath, appended: false };
+    }
+    writeSync(fd, payload);
+  } finally {
+    closeSync(fd);
   }
-  appendFileSync(queuePath, payload);
   return { queuePath, appended: true };
 }
 
diff --git a/src/mcp/cowork-ingest.ts b/src/mcp/cowork-ingest.ts
index b941a440c..2bb1c9a9b 100644
--- a/src/mcp/cowork-ingest.ts
+++ b/src/mcp/cowork-ingest.ts
@@ -20,9 +20,9 @@
  * several concurrent MCP processes Cowork spawns from double-inserting.
  */
 import {
-  appendFileSync,
   closeSync,
   existsSync,
+  fstatSync,
   mkdirSync,
   openSync,
   readFileSync,
@@ -31,6 +31,7 @@ import {
   statSync,
   utimesSync,
   writeFileSync,
+  writeSync,
 } from "node:fs";
 import { join } from "node:path";
 import { homedir } from "node:os";
@@ -186,15 +187,19 @@ function hasQueuedRows(): boolean {
 function recordLoss(detail: Record<string, unknown>): void {
   try {
     mkdirSync(DEEPLAKE_DIR, { recursive: true });
+    // Size-check and write through ONE descriptor: a stat-then-append on the
+    // path is a file-system race (CodeQL js/file-system-race), and this file is
+    // written by several concurrent Cowork MCP processes.
+    const fd = openSync(DROPPED_MARKER, "a");
     try {
-      if (statSync(DROPPED_MARKER).size >= MAX_LOSS_JOURNAL_BYTES) {
+      if (fstatSync(fd).size >= MAX_LOSS_JOURNAL_BYTES) {
         log("cowork-ingest", "loss journal is at its ceiling, not recording further entries");
         return;
       }
-    } catch {
-      /* no journal yet */
+      writeSync(fd, `${JSON.stringify({ at: new Date().toISOString(), ...detail })}\n`);
+    } finally {
+      closeSync(fd);
     }
-    appendFileSync(DROPPED_MARKER, `${JSON.stringify({ at: new Date().toISOString(), ...detail })}\n`);
   } catch {
     /* best effort */
   }

From d90db899bda537d9c93c93a6800c4abbae78f91a Mon Sep 17 00:00:00 2001
From: Emanuele Fenocchi <emanuele.fenocchi@activeloop.dev>
Date: Thu, 20 Aug 2026 07:48:12 +0000
Subject: [PATCH 10/12] fix(queue): append a transcript line's rows atomically,
 with rollback

CodeRabbit on #343:

- writeSync may write fewer bytes than asked for; both new call sites now loop
  until the payload is fully written.
- A write that throws part-way left half a JSON line in the queue, which would
  fail every later drain. appendQueuedSessionRows() records the file size before
  writing and truncates back to it on failure, so the group is all-or-nothing
  and the caller keeps the watermark on that line.
- The loss journal checked its size before serializing the record, so the last
  write could step over the 1 MB ceiling. It now projects the record's size.

The Cowork ingest queues a line through the new group append instead of doing
its own room math, so 'the whole line or nothing' is enforced in one place.
---
 src/hooks/session-queue.ts              | 57 ++++++++++++++++++++-----
 src/mcp/cowork-ingest.ts                | 33 +++++++-------
 tests/claude-code/session-queue.test.ts | 19 +++++++++
 3 files changed, 83 insertions(+), 26 deletions(-)

diff --git a/src/hooks/session-queue.ts b/src/hooks/session-queue.ts
index e194b2b16..a45d328b7 100644
--- a/src/hooks/session-queue.ts
+++ b/src/hooks/session-queue.ts
@@ -3,6 +3,7 @@ import {
   closeSync,
   existsSync,
   fstatSync,
+  ftruncateSync,
   mkdirSync,
   openSync,
   readFileSync,
@@ -138,22 +139,56 @@ export function appendQueuedSessionRow(
   queueDir = DEFAULT_QUEUE_DIR,
   maxQueueBytes = MAX_SESSION_QUEUE_BYTES,
 ): AppendQueuedRowResult {
+  return appendQueuedSessionRows([row], queueDir, maxQueueBytes);
+}
+
+/**
+ * Append a group of rows to one session's queue file, all-or-nothing.
+ *
+ * The caller uses the group to keep a transcript line atomic: either every row
+ * for that line is queued, or none is and the caller can retry the whole line
+ * later. Three things make that hold:
+ *   - the size check and the write share ONE descriptor, so the size the check
+ *     saw is the size the write extends (a stat-then-append on the path is
+ *     CodeQL's js/file-system-race, and several Cowork MCP processes write
+ *     these files);
+ *   - writeSync is looped, because it is allowed to write fewer bytes than
+ *     asked for;
+ *   - a write that throws part-way is truncated back to where it started, so
+ *     the queue never holds half a JSON line (which would fail every later
+ *     drain).
+ */
+export function appendQueuedSessionRows(
+  rows: QueuedSessionRow[],
+  queueDir = DEFAULT_QUEUE_DIR,
+  maxQueueBytes = MAX_SESSION_QUEUE_BYTES,
+): AppendQueuedRowResult {
+  if (rows.length === 0) throw new Error("appendQueuedSessionRows: rows must not be empty");
   mkdirSync(queueDir, { recursive: true });
-  const sessionId = extractSessionId(row.path);
-  const queuePath = getQueuePath(queueDir, sessionId);
-  const payload = `${JSON.stringify(row)}\n`;
-  // Project the post-append size, so the ceiling is a real ceiling rather than
-  // "the last row may overshoot it by however large that row happened to be".
-  // Checked and written through ONE descriptor: a stat-then-append on the path
-  // is a file-system race (CodeQL js/file-system-race), and several Cowork MCP
-  // processes write this file concurrently.
+  const queuePath = getQueuePath(queueDir, extractSessionId(rows[0].path));
+  const payload = Buffer.from(rows.map(row => `${JSON.stringify(row)}\n`).join(""), "utf-8");
+
   const fd = openSync(queuePath, "a");
   try {
-    if (fstatSync(fd).size + Buffer.byteLength(payload, "utf-8") > maxQueueBytes) {
-      log("session-queue", `queue file at the ${maxQueueBytes}-byte ceiling, dropping row: ${queuePath}`);
+    const startedAt = fstatSync(fd).size;
+    if (startedAt + payload.length > maxQueueBytes) {
+      log("session-queue", `queue file at the ${maxQueueBytes}-byte ceiling, refusing ${rows.length} row(s): ${queuePath}`);
+      return { queuePath, appended: false };
+    }
+    let written = 0;
+    try {
+      while (written < payload.length) {
+        written += writeSync(fd, payload, written, payload.length - written);
+      }
+    } catch (e: unknown) {
+      try {
+        ftruncateSync(fd, startedAt);
+      } catch {
+        /* nothing better to do — the drain will report the bad line */
+      }
+      log("session-queue", `append failed, rolled back ${rows.length} row(s): ${e instanceof Error ? e.message : String(e)}`);
       return { queuePath, appended: false };
     }
-    writeSync(fd, payload);
   } finally {
     closeSync(fd);
   }
diff --git a/src/mcp/cowork-ingest.ts b/src/mcp/cowork-ingest.ts
index 2bb1c9a9b..0628c616d 100644
--- a/src/mcp/cowork-ingest.ts
+++ b/src/mcp/cowork-ingest.ts
@@ -41,13 +41,12 @@ import { DeeplakeApi } from "../deeplake-api.js";
 import { claudeDesktopConfigDir } from "../cli/util.js";
 import { getVersion } from "../cli/version.js";
 import {
-  appendQueuedSessionRow,
+  appendQueuedSessionRows,
   buildQueuedSessionRow,
   buildSessionPath,
   drainSessionQueues,
   gcOversizedQueueFiles,
   queuedRowBytes,
-  sessionQueueRoomBytes,
   MAX_SESSION_QUEUE_BYTES,
 } from "../hooks/session-queue.js";
 import { spawnWikiWorker, bundleDirFromImportMeta } from "../hooks/spawn-wiki-worker.js";
@@ -192,11 +191,15 @@ function recordLoss(detail: Record<string, unknown>): void {
     // written by several concurrent Cowork MCP processes.
     const fd = openSync(DROPPED_MARKER, "a");
     try {
-      if (fstatSync(fd).size >= MAX_LOSS_JOURNAL_BYTES) {
+      // Project the record's own size, so the journal cannot step over its
+      // ceiling on the last write.
+      const record = Buffer.from(`${JSON.stringify({ at: new Date().toISOString(), ...detail })}\n`, "utf-8");
+      if (fstatSync(fd).size + record.length > MAX_LOSS_JOURNAL_BYTES) {
         log("cowork-ingest", "loss journal is at its ceiling, not recording further entries");
         return;
       }
-      writeSync(fd, `${JSON.stringify({ at: new Date().toISOString(), ...detail })}\n`);
+      let written = 0;
+      while (written < record.length) written += writeSync(fd, record, written, record.length - written);
     } finally {
       closeSync(fd);
     }
@@ -465,28 +468,28 @@ export async function ingestCoworkSessions(): Promise<{ ingested: number } | { s
           continue;
         }
 
-        const sessionId = String(parsed.sessionId);
-        const needed = rows.reduce((n, row) => n + queuedRowBytes(row), 0);
-        if (needed > sessionQueueRoomBytes(sessionId, COWORK_QUEUE_DIR)) {
+        // All-or-nothing: the group either lands whole or not at all, so the
+        // watermark below is never left pointing past a half-queued line.
+        const { appended } = appendQueuedSessionRows(rows, COWORK_QUEUE_DIR);
+        if (!appended) {
+          const needed = rows.reduce((n, row) => n + queuedRowBytes(row), 0);
           if (needed > MAX_SESSION_QUEUE_BYTES) {
             // Bigger than the whole ceiling — it can never be queued, and
             // stalling here would freeze this transcript forever. Skip it, on
             // the record.
-            recordLoss({ skippedTranscriptLine: path, sessionId, neededBytes: needed });
+            recordLoss({ skippedTranscriptLine: path, sessionId: String(parsed.sessionId), neededBytes: needed });
             processed += 1;
             continue;
           }
-          // The queue is full. Stop here; the watermark keeps this line for the
-          // next tick, once the drain below has made room.
+          // Queue full, or the write failed and was rolled back. Stop here; the
+          // watermark keeps this line for the next tick, once the drain below
+          // has made room.
           queueFull = true;
           break;
         }
 
-        for (const row of rows) {
-          appendQueuedSessionRow(row, COWORK_QUEUE_DIR);
-          appendedAny = true;
-          ingested += 1;
-        }
+        appendedAny = true;
+        ingested += rows.length;
         processed += 1;
       }
       state.processedLines[path] = processed;
diff --git a/tests/claude-code/session-queue.test.ts b/tests/claude-code/session-queue.test.ts
index 5ea48c9db..86e90cc48 100644
--- a/tests/claude-code/session-queue.test.ts
+++ b/tests/claude-code/session-queue.test.ts
@@ -12,6 +12,7 @@ import { join } from "node:path";
 import { tmpdir } from "node:os";
 import {
   appendQueuedSessionRow,
+  appendQueuedSessionRows,
   buildQueuedSessionRow,
   buildSessionInsertSql,
   buildSessionPath,
@@ -21,6 +22,7 @@ import {
   gcOversizedQueueFiles,
   isSessionWriteDisabled,
   MAX_SESSION_QUEUE_BYTES,
+  queuedRowBytes,
   isSessionWriteAuthError,
   markSessionWriteDisabled,
   type QueuedSessionRow,
@@ -655,6 +657,23 @@ describe("oversized queue files", () => {
     expect(existsSync(journal)).toBe(true);
   });
 
+  it("appends a group of rows all-or-nothing", () => {
+    const queueDir = makeQueueDir();
+    const rows = [makeRow("s-group", 0), makeRow("s-group", 1), makeRow("s-group", 2)];
+    const bytes = rows.reduce((n, r) => n + queuedRowBytes(r), 0);
+
+    // One byte short of what the group needs: nothing at all may be written.
+    const refused = appendQueuedSessionRows(rows, queueDir, bytes - 1);
+    expect(refused.appended).toBe(false);
+    expect(existsSync(refused.queuePath)).toBe(true);
+    expect(readFileSync(refused.queuePath, "utf-8")).toBe("");
+
+    // Exactly enough room: all three land together.
+    const accepted = appendQueuedSessionRows(rows, queueDir, bytes);
+    expect(accepted.appended).toBe(true);
+    expect(readFileSync(accepted.queuePath, "utf-8").split("\n").filter(Boolean)).toHaveLength(3);
+  });
+
   it("defaults the ceiling to 256 MB", () => {
     expect(MAX_SESSION_QUEUE_BYTES).toBe(256 * 1024 * 1024);
   });

From 1ebca127c39d1d8d9c4f2cce51d7b315950de6fb Mon Sep 17 00:00:00 2001
From: Emanuele Fenocchi <emanuele.fenocchi@activeloop.dev>
Date: Thu, 20 Aug 2026 07:56:02 +0000
Subject: [PATCH 11/12] fix(queue): a malformed record no longer strands the
 whole queue

Codex pass 6 on #343 blocked on the new append path. Two real problems and the
tests it asked for:

- readQueuedRows threw on any unparseable line, so ONE truncated record - from
  a crash mid-append, a rollback that could not run, an older build - failed
  that flush and every flush after it, stranding the queue permanently. Bad
  lines are now skipped and counted in the debug log.
- The rollback truncated to this call's start offset unconditionally, which
  would destroy a concurrent appender's rows. It now rolls back only while the
  file has not grown past what this call could have written; a throwing
  writeSync does not report its progress, so that bound is what can be checked.
  The single-writer contract (the Cowork ingest holds an exclusive lock for the
  whole tick) is documented on the function.

New tests/claude-code/session-queue-append-atomicity.test.ts drives writeSync
directly: a write that throws half-way leaves the file byte-identical, a short
write is completed rather than losing its tail, a concurrent appender's row
survives a rollback, and a queue with a truncated last record still flushes its
good rows.
---
 src/hooks/session-queue.ts                    |  43 ++++--
 .../session-queue-append-atomicity.test.ts    | 130 ++++++++++++++++++
 2 files changed, 163 insertions(+), 10 deletions(-)
 create mode 100644 tests/claude-code/session-queue-append-atomicity.test.ts

diff --git a/src/hooks/session-queue.ts b/src/hooks/session-queue.ts
index a45d328b7..88bea27da 100644
--- a/src/hooks/session-queue.ts
+++ b/src/hooks/session-queue.ts
@@ -155,8 +155,15 @@ export function appendQueuedSessionRow(
  *   - writeSync is looped, because it is allowed to write fewer bytes than
  *     asked for;
  *   - a write that throws part-way is truncated back to where it started, so
- *     the queue never holds half a JSON line (which would fail every later
- *     drain).
+ *     the queue never holds half a JSON line.
+ *
+ * Rollback is what makes this single-writer: truncating while another process
+ * appends would destroy its rows, so the rollback is skipped unless the file is
+ * exactly as this call left it. The Cowork ingest — the only caller — holds an
+ * exclusive lock (~/.deeplake/.cowork-ingest.lock) for the whole tick, so two
+ * appenders do not overlap in practice. A malformed tail that survives anyway
+ * (failed rollback, a crash mid-write, an older build) is skipped by
+ * readQueuedRows rather than wedging the drain forever.
  */
 export function appendQueuedSessionRows(
   rows: QueuedSessionRow[],
@@ -181,12 +188,17 @@ export function appendQueuedSessionRows(
         written += writeSync(fd, payload, written, payload.length - written);
       }
     } catch (e: unknown) {
+      // A throwing writeSync does not report how much it wrote, so bound the
+      // rollback by what this call could possibly have written: if the file has
+      // grown past startedAt + payload.length, another writer appended in the
+      // meantime and truncating would destroy its rows. Leave it alone then —
+      // readQueuedRows skips the malformed tail instead of wedging the drain.
       try {
-        ftruncateSync(fd, startedAt);
+        if (fstatSync(fd).size <= startedAt + payload.length) ftruncateSync(fd, startedAt);
       } catch {
-        /* nothing better to do — the drain will report the bad line */
+        /* rollback failed — the malformed line is skipped at read time */
       }
-      log("session-queue", `append failed, rolled back ${rows.length} row(s): ${e instanceof Error ? e.message : String(e)}`);
+      log("session-queue", `append failed after ${written}/${payload.length} bytes: ${e instanceof Error ? e.message : String(e)}`);
       return { queuePath, appended: false };
     }
   } finally {
@@ -511,11 +523,22 @@ async function flushInflightFile(
 
 function readQueuedRows(path: string): QueuedSessionRow[] {
   const raw = readFileSync(path, "utf-8");
-  return raw
-    .split("\n")
-    .map(line => line.trim())
-    .filter(Boolean)
-    .map((line) => JSON.parse(line) as QueuedSessionRow);
+  const rows: QueuedSessionRow[] = [];
+  let malformed = 0;
+  for (const line of raw.split("\n")) {
+    const trimmed = line.trim();
+    if (!trimmed) continue;
+    try {
+      rows.push(JSON.parse(trimmed) as QueuedSessionRow);
+    } catch {
+      // A half-written record — a crash mid-append, a rollback that could not
+      // run. Skipping it costs one message; throwing would fail this flush and
+      // every flush after it, stranding the whole queue permanently.
+      malformed += 1;
+    }
+  }
+  if (malformed > 0) log("session-queue", `skipped ${malformed} malformed row(s) in ${path}`);
+  return rows;
 }
 
 function requeueInflight(queuePath: string, inflightPath: string): void {
diff --git a/tests/claude-code/session-queue-append-atomicity.test.ts b/tests/claude-code/session-queue-append-atomicity.test.ts
new file mode 100644
index 000000000..edb2842bb
--- /dev/null
+++ b/tests/claude-code/session-queue-append-atomicity.test.ts
@@ -0,0 +1,130 @@
+/**
+ * Failure-path tests for the atomic queue append.
+ *
+ * These cover what the happy-path tests cannot: a write that throws part-way,
+ * a second writer that appends during that failure, and a malformed record
+ * that survives on disk anyway. The last one is the important one — before it
+ * was skipped at read time, one truncated line failed every future flush and
+ * stranded the whole queue.
+ */
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+import { mkdtempSync, readFileSync, writeFileSync, appendFileSync } from "node:fs";
+import { join } from "node:path";
+import { tmpdir } from "node:os";
+
+const realFs = await vi.importActual<typeof import("node:fs")>("node:fs");
+
+/** Installed by each test to hijack writeSync; null = pass through. */
+let writeHook: ((fd: number, buf: Buffer, off: number, len: number) => number) | null = null;
+
+vi.mock("node:fs", async () => {
+  const actual = await vi.importActual<typeof import("node:fs")>("node:fs");
+  return {
+    ...actual,
+    writeSync: (fd: number, buf: Buffer, off: number, len: number) =>
+      writeHook ? writeHook(fd, buf, off, len) : (actual.writeSync as any)(fd, buf, off, len),
+  };
+});
+
+const {
+  appendQueuedSessionRows,
+  buildQueuedSessionRow,
+  buildSessionPath,
+  flushSessionQueue,
+} = await import("../../src/hooks/session-queue.js");
+
+function makeRow(sessionId: string, seq: number) {
+  return buildQueuedSessionRow({
+    sessionPath: buildSessionPath({ userName: "alice", orgName: "acme", workspaceId: "default" }, sessionId),
+    line: JSON.stringify({ type: "user_message", content: `msg-${seq}` }),
+    userName: "alice",
+    projectName: "p",
+    description: "user_message",
+    agent: "claude_cowork",
+    timestamp: "2026-08-20T00:00:00.000Z",
+  });
+}
+
+let queueDir: string;
+
+beforeEach(() => {
+  queueDir = mkdtempSync(join(tmpdir(), "queue-atomicity-"));
+  writeHook = null;
+});
+
+afterEach(() => {
+  writeHook = null;
+});
+
+describe("appendQueuedSessionRows failure paths", () => {
+  it("rolls back a write that throws part-way, leaving the file as it was", () => {
+    const queuePath = appendQueuedSessionRows([makeRow("s1", 0)], queueDir).queuePath;
+    const before = readFileSync(queuePath, "utf-8");
+
+    // Write half the payload, then fail — the classic partial write.
+    writeHook = (fd, buf, off, len) => {
+      (realFs.writeSync as any)(fd, buf, off, Math.floor(len / 2));
+      throw new Error("ENOSPC: no space left on device");
+    };
+    const result = appendQueuedSessionRows([makeRow("s1", 1), makeRow("s1", 2)], queueDir);
+
+    expect(result.appended).toBe(false);
+    expect(readFileSync(queuePath, "utf-8")).toBe(before);
+  });
+
+  it("completes a short write instead of losing its tail", () => {
+    let calls = 0;
+    // First call writes one byte — writeSync is allowed to do that.
+    writeHook = (fd, buf, off, len) => (realFs.writeSync as any)(fd, buf, off, calls++ === 0 ? 1 : len);
+
+    const { queuePath, appended } = appendQueuedSessionRows([makeRow("s2", 0)], queueDir);
+
+    expect(appended).toBe(true);
+    expect(calls).toBeGreaterThan(1);
+    const lines = readFileSync(queuePath, "utf-8").split("\n").filter(Boolean);
+    expect(lines).toHaveLength(1);
+    expect(JSON.parse(lines[0]).message).toContain("msg-0");
+  });
+
+  it("does not truncate over a concurrent appender's rows when rolling back", () => {
+    const queuePath = appendQueuedSessionRows([makeRow("s3", 0)], queueDir).queuePath;
+    const before = readFileSync(queuePath, "utf-8");
+    const otherWriter = `${JSON.stringify(makeRow("s3", 99))}\n`;
+
+    // Fail, and have somebody else append while we are down: rolling back to
+    // our start offset would destroy their row, so it must be skipped.
+    writeHook = (fd, buf, off, len) => {
+      (realFs.writeSync as any)(fd, buf, off, Math.floor(len / 2));
+      appendFileSync(queuePath, otherWriter);
+      throw new Error("EIO");
+    };
+    expect(appendQueuedSessionRows([makeRow("s3", 1)], queueDir).appended).toBe(false);
+
+    const after = readFileSync(queuePath, "utf-8");
+    expect(after.startsWith(before)).toBe(true);
+    expect(after).toContain(otherWriter.trim());
+  });
+});
+
+describe("a malformed record does not strand the queue", () => {
+  it("skips the bad line and flushes the good ones", async () => {
+    const good = appendQueuedSessionRows([makeRow("s4", 0), makeRow("s4", 1)], queueDir);
+    // A half-written record, the way a crash mid-append leaves one.
+    appendFileSync(good.queuePath, '{"id":"truncated","path":"/sessions/ali');
+
+    const sent: string[] = [];
+    const api = {
+      query: async (sql: string) => { sent.push(sql); return []; },
+      ensureSessionsTable: async () => {},
+    };
+
+    const result = await flushSessionQueue(api, { sessionId: "s4", sessionsTable: "sessions", queueDir });
+
+    expect(result.status).toBe("flushed");
+    expect(result.rows).toBe(2);
+    expect(sent).toHaveLength(1);
+    expect(sent[0]).toContain("msg-0");
+    expect(sent[0]).toContain("msg-1");
+    expect(sent[0]).not.toContain("truncated");
+  });
+});

From 93e3c797608045125c125c7ddb62f3033cf9b04c Mon Sep 17 00:00:00 2001
From: Emanuele Fenocchi <emanuele.fenocchi@activeloop.dev>
Date: Thu, 20 Aug 2026 08:02:10 +0000
Subject: [PATCH 12/12] fix(queue): start a new line when the queue file ends
 mid-record

Codex pass 7 approved but noted the concurrency test only proved the other
writer's BYTES survived, not that its row was still parseable. Strengthening the
test proved it was not: a row appended after a half-written one is glued onto
the fragment and skipped with it at read time.

appendQueuedSessionRows() now checks the file's last byte and prefixes a newline
when a previous append died mid-record, so the incoming rows stay parseable and
only the fragment is skipped. The descriptor is opened 'a+' rather than 'a' -
with 'a' it is write-only and that check failed with EBADF, silently doing
nothing, which is how this survived the first round.

Tests: the concurrent writer's row now has to flush and reach the backend, and a
file ending mid-record must yield three lines with the fragment isolated.
---
 src/hooks/session-queue.ts                    | 26 ++++++++++-
 .../session-queue-append-atomicity.test.ts    | 46 +++++++++++++++----
 2 files changed, 62 insertions(+), 10 deletions(-)

diff --git a/src/hooks/session-queue.ts b/src/hooks/session-queue.ts
index 88bea27da..869a1ca13 100644
--- a/src/hooks/session-queue.ts
+++ b/src/hooks/session-queue.ts
@@ -7,6 +7,7 @@ import {
   mkdirSync,
   openSync,
   readFileSync,
+  readSync,
   readdirSync,
   renameSync,
   rmSync,
@@ -173,11 +174,20 @@ export function appendQueuedSessionRows(
   if (rows.length === 0) throw new Error("appendQueuedSessionRows: rows must not be empty");
   mkdirSync(queueDir, { recursive: true });
   const queuePath = getQueuePath(queueDir, extractSessionId(rows[0].path));
-  const payload = Buffer.from(rows.map(row => `${JSON.stringify(row)}\n`).join(""), "utf-8");
+  const rowsPayload = Buffer.from(rows.map(row => `${JSON.stringify(row)}\n`).join(""), "utf-8");
 
-  const fd = openSync(queuePath, "a");
+  // "a+" (not "a"): appends, but is also readable, so endsWithNewline() below
+  // can inspect the last byte. With "a" the descriptor is write-only and that
+  // read fails with EBADF.
+  const fd = openSync(queuePath, "a+");
   try {
     const startedAt = fstatSync(fd).size;
+    // If the file does not end in a newline, a previous append died half-way.
+    // Start on a fresh line so THESE rows stay parseable — otherwise they are
+    // glued onto the broken fragment and skipped with it at read time.
+    const payload = endsWithNewline(fd, startedAt)
+      ? rowsPayload
+      : Buffer.concat([Buffer.from("\n", "utf-8"), rowsPayload]);
     if (startedAt + payload.length > maxQueueBytes) {
       log("session-queue", `queue file at the ${maxQueueBytes}-byte ceiling, refusing ${rows.length} row(s): ${queuePath}`);
       return { queuePath, appended: false };
@@ -270,6 +280,18 @@ export function sessionQueueRoomBytes(
   return Math.max(0, maxQueueBytes - fileSize(getQueuePath(queueDir, sessionId)));
 }
 
+/** True when the file is empty or its last byte is a newline. */
+function endsWithNewline(fd: number, size: number): boolean {
+  if (size === 0) return true;
+  try {
+    const tail = Buffer.alloc(1);
+    readSync(fd, tail, 0, 1, size - 1);
+    return tail[0] === 0x0a;
+  } catch {
+    return true; // cannot tell — do not inject a stray newline
+  }
+}
+
 function fileSize(path: string): number {
   try {
     return statSync(path).size;
diff --git a/tests/claude-code/session-queue-append-atomicity.test.ts b/tests/claude-code/session-queue-append-atomicity.test.ts
index edb2842bb..4bc57761b 100644
--- a/tests/claude-code/session-queue-append-atomicity.test.ts
+++ b/tests/claude-code/session-queue-append-atomicity.test.ts
@@ -8,7 +8,7 @@
  * stranded the whole queue.
  */
 import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
-import { mkdtempSync, readFileSync, writeFileSync, appendFileSync } from "node:fs";
+import { mkdtempSync, readFileSync, appendFileSync } from "node:fs";
 import { join } from "node:path";
 import { tmpdir } from "node:os";
 
@@ -86,27 +86,57 @@ describe("appendQueuedSessionRows failure paths", () => {
     expect(JSON.parse(lines[0]).message).toContain("msg-0");
   });
 
-  it("does not truncate over a concurrent appender's rows when rolling back", () => {
+  it("does not truncate over a concurrent appender's rows when rolling back", async () => {
     const queuePath = appendQueuedSessionRows([makeRow("s3", 0)], queueDir).queuePath;
     const before = readFileSync(queuePath, "utf-8");
-    const otherWriter = `${JSON.stringify(makeRow("s3", 99))}\n`;
-
-    // Fail, and have somebody else append while we are down: rolling back to
-    // our start offset would destroy their row, so it must be skipped.
+    // Fail, and have a real second writer append through the same API while we
+    // are down: rolling back to our start offset would destroy its row.
     writeHook = (fd, buf, off, len) => {
       (realFs.writeSync as any)(fd, buf, off, Math.floor(len / 2));
-      appendFileSync(queuePath, otherWriter);
+      const hook = writeHook;
+      writeHook = null;
+      appendQueuedSessionRows([makeRow("s3", 99)], queueDir);
+      writeHook = hook;
       throw new Error("EIO");
     };
     expect(appendQueuedSessionRows([makeRow("s3", 1)], queueDir).appended).toBe(false);
 
     const after = readFileSync(queuePath, "utf-8");
     expect(after.startsWith(before)).toBe(true);
-    expect(after).toContain(otherWriter.trim());
+    expect(after).toContain("msg-99");
+
+    // Surviving the truncate is not enough: the other writer's row must still
+    // be parseable and actually reach the backend, with the half-written bytes
+    // dropped along the way.
+    writeHook = null;
+    const sent: string[] = [];
+    const result = await flushSessionQueue(
+      { query: async (sql: string) => { sent.push(sql); return []; }, ensureSessionsTable: async () => {} },
+      { sessionId: "s3", sessionsTable: "sessions", queueDir },
+    );
+
+    expect(result.status).toBe("flushed");
+    expect(result.rows).toBe(2); // the pre-existing row + the concurrent one
+    expect(sent.join(" ")).toContain("msg-99");
+    expect(sent.join(" ")).toContain("msg-0");
   });
 });
 
 describe("a malformed record does not strand the queue", () => {
+  it("starts a new line when the file ends mid-record", () => {
+    const { queuePath } = appendQueuedSessionRows([makeRow("s5", 0)], queueDir);
+    appendFileSync(queuePath, '{"id":"half-written","path":"/sess');
+
+    // Without the healing newline this row would be glued onto the fragment
+    // and skipped along with it.
+    appendQueuedSessionRows([makeRow("s5", 1)], queueDir);
+
+    const lines = readFileSync(queuePath, "utf-8").split("\n").filter(Boolean);
+    expect(lines).toHaveLength(3);
+    expect(lines[1]).toBe('{"id":"half-written","path":"/sess');
+    expect(JSON.parse(lines[2]).message).toContain("msg-1");
+  });
+
   it("skips the bad line and flushes the good ones", async () => {
     const good = appendQueuedSessionRows([makeRow("s4", 0), makeRow("s4", 1)], queueDir);
     // A half-written record, the way a crash mid-append leaves one.