From 655e40559e6d1822f4a8a1342050bb8ea7424755 Mon Sep 17 00:00:00 2001 From: Divyam Talwar Date: Tue, 22 Sep 2026 04:25:21 +0530 Subject: [PATCH 1/2] Preserve incomplete Cowork transcript records across polling Keep an unparseable final JSONL record below the transcript watermark until the producer finishes writing it, while retaining the existing skip behavior for malformed newline-terminated records. Constraint: Cowork transcripts are appended incrementally and may split UTF-8 records across polls Rejected: Advance every parse failure | would permanently lose a record that is still being written Confidence: high Scope-risk: narrow Reversibility: clean Directive: Keep transcript-tail handling tied to the final unterminated record only Tested: Native Vitest regression 3/3; affected Cowork suite excluding pre-existing macOS fixture failures; npm run typecheck; npm run build Not-tested: Live Cowork, Deeplake cloud, and real agent spawners --- src/mcp/cowork-ingest.ts | 10 +- .../cowork-ingest-partial-record.test.ts | 172 ++++++++++++++++++ 2 files changed, 181 insertions(+), 1 deletion(-) create mode 100644 tests/claude-code/cowork-ingest-partial-record.test.ts diff --git a/src/mcp/cowork-ingest.ts b/src/mcp/cowork-ingest.ts index f774bd12..9091283e 100644 --- a/src/mcp/cowork-ingest.ts +++ b/src/mcp/cowork-ingest.ts @@ -461,8 +461,11 @@ export async function ingestCoworkSessions(): Promise<{ ingested: number } | { s for (const path of transcripts) { let lines: string[]; + let hasUnterminatedTail: boolean; try { - lines = readFileSync(path, "utf-8").split("\n").filter(Boolean); + const transcript = readFileSync(path, "utf-8"); + lines = transcript.split("\n").filter(Boolean); + hasUnterminatedTail = !transcript.endsWith("\n"); } catch { continue; } @@ -479,6 +482,11 @@ export async function ingestCoworkSessions(): Promise<{ ingested: number } | { s try { parsed = JSON.parse(raw); } catch { + // The producer may still be appending this final record. Retrying + // it next tick preserves the watermark until it becomes parseable. + // Complete records without a final newline remain supported; only + // an unparseable, unterminated tail waits for more bytes. + if (hasUnterminatedTail && processed === lines.length - 1) break; processed += 1; continue; } diff --git a/tests/claude-code/cowork-ingest-partial-record.test.ts b/tests/claude-code/cowork-ingest-partial-record.test.ts new file mode 100644 index 00000000..001102c7 --- /dev/null +++ b/tests/claude-code/cowork-ingest-partial-record.test.ts @@ -0,0 +1,172 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { + appendFileSync, + mkdirSync, + mkdtempSync, + readFileSync, + writeFileSync, +} from "node:fs"; +import { join } from "node:path"; +import { tmpdir } from "node:os"; + +const uploads: string[] = []; +let networkAvailable = true; + +vi.mock("../../src/deeplake-api.js", () => ({ + DeeplakeApi: class { + async query(sql: string): Promise { + if (!networkAvailable) throw new Error("offline"); + uploads.push(sql); + return []; + } + async ensureSessionsTable(): Promise {} + }, +})); + +vi.mock("../../src/hooks/spawn-wiki-worker.js", () => ({ + bundleDirFromImportMeta: () => "test-bundle", + spawnWikiWorker: vi.fn(), +})); + +vi.mock("../../src/skillify/triggers.js", () => ({ + forceSessionEndTrigger: vi.fn(), +})); + +const SESSION_A = "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa"; +const SESSION_B = "bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb"; + +let home: string; +let previousHome: string | undefined; + +function transcriptPath(sessionId: string): string { + const desktopDir = process.platform === "darwin" + ? join(home, "Library", "Application Support", "Claude") + : join(home, ".config", "Claude"); + const dir = join(desktopDir, "local-agent-mode-sessions", "run", ".claude", "projects", "proj"); + mkdirSync(dir, { recursive: true }); + return join(dir, `${sessionId}.jsonl`); +} + +function record(sessionId: string, content: string): string { + return JSON.stringify({ + type: "user", + sessionId, + timestamp: "2026-09-22T00:00:00.000Z", + cwd: "/cowork", + message: { role: "user", content }, + }); +} + +function queuePath(sessionId: string): string { + return join(home, ".deeplake", "queue-cowork", `${sessionId}.jsonl`); +} + +function queuedMessages(sessionId: string): string[] { + return readFileSync(queuePath(sessionId), "utf8") + .trim() + .split("\n") + .map(line => (JSON.parse(line) as { message: string }).message) + .map(message => JSON.parse(message) as { content: string }) + .map(message => message.content); +} + +async function loadIngest() { + vi.resetModules(); + return import("../../src/mcp/cowork-ingest.js"); +} + +beforeEach(() => { + previousHome = process.env.HOME; + home = mkdtempSync(join(tmpdir(), "hivemind-cowork-partial-")); + process.env.HOME = home; + mkdirSync(join(home, ".deeplake"), { recursive: true }); + writeFileSync( + join(home, ".deeplake", "credentials.json"), + JSON.stringify({ + token: "test-token", + orgId: "test-org", + orgName: "test-org", + userName: "test-user", + workspaceId: "test-workspace", + apiUrl: "http://127.0.0.1:1", + }), + ); + uploads.length = 0; + networkAvailable = true; +}); + +afterEach(() => { + if (previousHome === undefined) delete process.env.HOME; + else process.env.HOME = previousHome; +}); + +describe("Cowork ingest partial transcript records", () => { + it("holds an unparseable final record, then ingests it after split writes", async () => { + const path = transcriptPath(SESSION_A); + const first = record(SESSION_A, "first"); + const malformed = "{not-json}"; + const partial = record(SESSION_A, "split record"); + const splitAt = Buffer.from(partial, "utf8").indexOf(Buffer.from("split", "utf8")); + writeFileSync(path, `${first}\n\n${malformed}\n` + Buffer.from(partial, "utf8").subarray(0, splitAt)); + + const { ingestCoworkSessions } = await loadIngest(); + await ingestCoworkSessions(); + + // The earlier good line was committed; the malformed newline-terminated + // line was skipped; the unterminated tail remains in the real queue state. + expect(uploads).toHaveLength(1); + expect(uploads[0]).toContain('"content":"first"'); + expect(readFileSync(join(home, ".deeplake", "cowork-ingest-state.json"), "utf8")).toContain('"processedLines"'); + expect(readFileSync(path, "utf8")).not.toContain("split record"); + + appendFileSync(path, Buffer.from(partial, "utf8").subarray(splitAt)); + appendFileSync(path, "\r\n"); + await ingestCoworkSessions(); + await ingestCoworkSessions(); + + expect(uploads).toHaveLength(2); + expect(uploads.filter(sql => sql.includes('"content":"split record"'))).toHaveLength(1); + expect(uploads.filter(sql => sql.includes('"content":"first"'))).toHaveLength(1); + }); + + it("waits for split UTF-8 bytes and accepts the completed CRLF record", async () => { + const path = transcriptPath(SESSION_A); + const full = Buffer.from(record(SESSION_A, "emoji 💾 survives") + "\r\n", "utf8"); + const emojiBytes = Buffer.from("💾", "utf8"); + const splitAt = full.indexOf(emojiBytes) + 1; + writeFileSync(path, full.subarray(0, splitAt)); + + const { ingestCoworkSessions } = await loadIngest(); + await ingestCoworkSessions(); + expect(uploads).toHaveLength(0); + + appendFileSync(path, full.subarray(splitAt)); + await ingestCoworkSessions(); + expect(uploads).toHaveLength(1); + expect(uploads[0]).toContain('"content":"emoji 💾 survives"'); + }); + + it("keeps transcript watermarks independent and does not duplicate repeated polls", async () => { + const pathA = transcriptPath(SESSION_A); + const pathB = transcriptPath(SESSION_B); + const partial = record(SESSION_A, "A partial"); + const splitAt = Buffer.byteLength(partial, "utf8") - 3; + writeFileSync(pathA, Buffer.from(partial, "utf8").subarray(0, splitAt)); + writeFileSync(pathB, `${record(SESSION_B, "B complete")}\n`); + + const { ingestCoworkSessions } = await loadIngest(); + await ingestCoworkSessions(); + await ingestCoworkSessions(); + expect(uploads).toHaveLength(1); + expect(uploads[0]).toContain('"content":"B complete"'); + expect(uploads[0]).not.toContain("A partial"); + + appendFileSync(pathA, Buffer.from(partial, "utf8").subarray(splitAt)); + await ingestCoworkSessions(); + await ingestCoworkSessions(); + + expect(uploads).toHaveLength(2); + expect(uploads.filter(sql => sql.includes('"content":"A partial"'))).toHaveLength(1); + expect(uploads.filter(sql => sql.includes('"content":"B complete"'))).toHaveLength(1); + }); +}); From 4eb32cabe026b6eab177352986bdc4430b5104e8 Mon Sep 17 00:00:00 2001 From: Divyam Talwar Date: Tue, 22 Sep 2026 05:28:16 +0530 Subject: [PATCH 2/2] test(cowork): assert the exact saved partial-record watermark --- tests/claude-code/cowork-ingest-partial-record.test.ts | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/tests/claude-code/cowork-ingest-partial-record.test.ts b/tests/claude-code/cowork-ingest-partial-record.test.ts index 001102c7..ba2560dc 100644 --- a/tests/claude-code/cowork-ingest-partial-record.test.ts +++ b/tests/claude-code/cowork-ingest-partial-record.test.ts @@ -116,7 +116,8 @@ describe("Cowork ingest partial transcript records", () => { // line was skipped; the unterminated tail remains in the real queue state. expect(uploads).toHaveLength(1); expect(uploads[0]).toContain('"content":"first"'); - expect(readFileSync(join(home, ".deeplake", "cowork-ingest-state.json"), "utf8")).toContain('"processedLines"'); + const state = JSON.parse(readFileSync(join(home, ".deeplake", "cowork-ingest-state.json"), "utf8")); + expect(state.processedLines[path]).toBe(2); expect(readFileSync(path, "utf8")).not.toContain("split record"); appendFileSync(path, Buffer.from(partial, "utf8").subarray(splitAt));