diff --git a/src/hooks/session-queue.ts b/src/hooks/session-queue.ts
index f2e17ffcb..869a1ca13 100644
--- a/src/hooks/session-queue.ts
+++ b/src/hooks/session-queue.ts
@@ -2,18 +2,23 @@ import {
   appendFileSync,
   closeSync,
   existsSync,
+  fstatSync,
+  ftruncateSync,
   mkdirSync,
   openSync,
   readFileSync,
+  readSync,
   readdirSync,
   renameSync,
   rmSync,
   statSync,
   writeFileSync,
+  writeSync,
 } from "node:fs";
 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 +72,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,12 +129,175 @@ export function buildQueuedSessionRow(args: {
   };
 }
 
-export function appendQueuedSessionRow(row: QueuedSessionRow, queueDir = DEFAULT_QUEUE_DIR): string {
+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,
+): 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.
+ *
+ * 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[],
+  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);
-  appendFileSync(queuePath, `${JSON.stringify(row)}\n`);
-  return queuePath;
+  const queuePath = getQueuePath(queueDir, extractSessionId(rows[0].path));
+  const rowsPayload = Buffer.from(rows.map(row => `${JSON.stringify(row)}\n`).join(""), "utf-8");
+
+  // "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 };
+    }
+    let written = 0;
+    try {
+      while (written < payload.length) {
+        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 {
+        if (fstatSync(fd).size <= startedAt + payload.length) ftruncateSync(fd, startedAt);
+      } catch {
+        /* rollback failed — the malformed line is skipped at read time */
+      }
+      log("session-queue", `append failed after ${written}/${payload.length} bytes: ${e instanceof Error ? e.message : String(e)}`);
+      return { queuePath, appended: false };
+    }
+  } finally {
+    closeSync(fd);
+  }
+  return { queuePath, appended: true };
+}
+
+/**
+ * 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,
+  onDropped?: (path: string, sizeBytes: number) => void,
+): number {
+  let reclaimed = 0;
+  let names: string[];
+  try {
+    names = readdirSync(queueDir);
+  } catch {
+    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);
+    // 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;
+      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 */
+    }
+  }
+  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)));
+}
+
+/** 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;
+  } catch {
+    return 0;
+  }
 }
 
 export function buildSessionInsertSql(sessionsTable: string, rows: QueuedSessionRow[]): string {
@@ -369,11 +545,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/src/mcp/cowork-ingest.ts b/src/mcp/cowork-ingest.ts
index 6d177e32f..0628c616d 100644
--- a/src/mcp/cowork-ingest.ts
+++ b/src/mcp/cowork-ingest.ts
@@ -22,6 +22,7 @@
 import {
   closeSync,
   existsSync,
+  fstatSync,
   mkdirSync,
   openSync,
   readFileSync,
@@ -30,6 +31,7 @@ import {
   statSync,
   utimesSync,
   writeFileSync,
+  writeSync,
 } from "node:fs";
 import { join } from "node:path";
 import { homedir } from "node:os";
@@ -39,10 +41,13 @@ 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,
+  MAX_SESSION_QUEUE_BYTES,
 } from "../hooks/session-queue.js";
 import { spawnWikiWorker, bundleDirFromImportMeta } from "../hooks/spawn-wiki-worker.js";
 import { forceSessionEndTrigger } from "../skillify/triggers.js";
@@ -59,6 +64,8 @@ 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(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.
@@ -153,6 +160,55 @@ function findTranscripts(root: string): string[] {
   return out;
 }
 
+/** True when the Cowork queue still holds rows the backend has not taken. */
+function hasQueuedRows(): boolean {
+  try {
+    // 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
+  }
+}
+
+/**
+ * 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.
+ */
+/**
+ * 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 });
+    // 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 {
+      // 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;
+      }
+      let written = 0;
+      while (written < record.length) written += writeSync(fd, record, written, record.length - written);
+    } finally {
+      closeSync(fd);
+    }
+  } catch {
+    /* best effort */
+  }
+  log("cowork-ingest", `recorded queue loss: ${JSON.stringify(detail)}`);
+}
+
 function loadState(): IngestState {
   try {
     const raw = JSON.parse(readFileSync(STATE_PATH, "utf-8"));
@@ -363,9 +419,15 @@ 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, undefined, (path, sizeBytes) =>
+      recordLoss({ droppedQueueFile: path, sizeBytes }),
+    );
     const state = loadState();
     const transcripts = findTranscripts(root);
     let appendedAny = false;
+    let queueFull = false;
 
     for (const path of transcripts) {
       let lines: string[];
@@ -377,34 +439,85 @@ 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),
-          });
-          appendQueuedSessionRow(row, COWORK_QUEUE_DIR);
-          appendedAny = true;
-          ingested += 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;
         }
+
+        // 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: String(parsed.sessionId), neededBytes: needed });
+            processed += 1;
+            continue;
+          }
+          // 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;
+        }
+
+        appendedAny = true;
+        ingested += rows.length;
+        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
+    // 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 (queueFull) {
+      // 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");
     }
 
-    if (appendedAny) {
+    // 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,
@@ -412,15 +525,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..a4773c23d
--- /dev/null
+++ b/tests/claude-code/cowork-queue-leak.test.ts
@@ -0,0 +1,228 @@
+/**
+ * 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, readFileSync, statSync, appendFileSync, existsSync } from "node:fs";
+import { join } from "node:path";
+import { tmpdir } from "node:os";
+
+const SESSION_ID = "b27efa59-a8bc-4ea3-8b02-18cbc608ae17";
+
+// 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(sql: string): Promise<never[]> {
+      if (offline) throw new Error("fetch failed: ECONNREFUSED");
+      uploads.push(sql);
+      return [];
+    }
+    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;
+  }
+}
+
+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")
+      .split("\n")
+      .filter(Boolean).length;
+  } catch {
+    return 0;
+  }
+}
+
+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"),
+    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();
+    expect(queuedRows()).toBe(1);
+
+    appendFileSync(path, line("two"));
+    await ingestCoworkSessions();
+
+    // 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;
+    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();
+    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);
+  });
+
+  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);
+  });
+
+  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);
+});
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..4bc57761b
--- /dev/null
+++ b/tests/claude-code/session-queue-append-atomicity.test.ts
@@ -0,0 +1,160 @@
+/**
+ * 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, 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", async () => {
+    const queuePath = appendQueuedSessionRows([makeRow("s3", 0)], queueDir).queuePath;
+    const before = readFileSync(queuePath, "utf-8");
+    // 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));
+      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("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.
+    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");
+  });
+});
diff --git a/tests/claude-code/session-queue.test.ts b/tests/claude-code/session-queue.test.ts
index 928347124..86e90cc48 100644
--- a/tests/claude-code/session-queue.test.ts
+++ b/tests/claude-code/session-queue.test.ts
@@ -12,13 +12,17 @@ import { join } from "node:path";
 import { tmpdir } from "node:os";
 import {
   appendQueuedSessionRow,
+  appendQueuedSessionRows,
   buildQueuedSessionRow,
   buildSessionInsertSql,
   buildSessionPath,
   clearSessionWriteDisabled,
   drainSessionQueues,
   flushSessionQueue,
+  gcOversizedQueueFiles,
   isSessionWriteDisabled,
+  MAX_SESSION_QUEUE_BYTES,
+  queuedRowBytes,
   isSessionWriteAuthError,
   markSessionWriteDisabled,
   type QueuedSessionRow,
@@ -87,7 +91,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);
@@ -594,3 +598,83 @@ describe("session queue", () => {
     release?.();
   });
 });
+
+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);
+    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 { queuePath: small } = appendQueuedSessionRow(makeRow("s-small", 0), queueDir);
+    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));
+
+    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,
+    // 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("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);
+  });
+});