/** * Reading a DeepSeek Harness session transcript. * * Two of these assertions exist because the obvious implementation was measured * to be wrong against real files: * * - dsh appends ONE ZSTD FRAME PER WRITE, and Node's `zlib` zstd decoder stops * at the first frame end. A 56-line transcript decoded as 1 line / 158 bytes, * which reads as "the worker never answered" rather than as an error. The * multi-frame fixtures below are the guard. * - a fresh worker in a case directory that had been used before answered its * first `last-response` with the PREVIOUS session's reply. Session-to- * transcript pairing is therefore its own describe block. */ import { describe, expect, it, beforeAll, afterAll } from 'vitest'; import { mkdtemp, mkdir, writeFile, rm, utimes } from 'node:fs/promises'; import { join } from 'node:path'; import { tmpdir } from 'node:os'; import * as zlib from 'node:zlib'; import { decodeZstdFrames, findDeepSeekTranscript, parseDeepSeekTranscript, readDeepSeekLastResponse, resetDeepSeekTranscriptMemoForTest, resolveDeepSeekHome, zstdFrameRanges, zstdSupported, } from '../src/deepseek-transcript.js'; const zstdCompressSync = (zlib as unknown as { zstdCompressSync?: (b: Buffer) => Buffer }).zstdCompressSync; /** Compress each line into its own frame — exactly how dsh appends. */ function framed(lines: string[]): Buffer { if (!zstdCompressSync) throw new Error('zstd unavailable'); return Buffer.concat(lines.map((line) => zstdCompressSync(Buffer.from(`${line}\n`, 'utf8')))); } const sessionHeader = (cwd: string, createdAt: number, id = 'aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee') => JSON.stringify({ type: 'session', version: 0, id, createdAt, cwd, delegationDepth: 0 }); const userPrompt = (text: string) => JSON.stringify({ type: 'user/message', data: { content: [{ type: 'text', text }], source: { kind: 'user' }, role: 'user' }, }); const pluginContext = (text: string) => JSON.stringify({ type: 'user/message', data: { content: [{ type: 'text', text }], source: { kind: 'plugin', plugin: '@deepseek-ai/dsh-system-prompt' } }, }); const assistantMessage = (text: string, turn = 1, step = 1, time = 1_700_000_000_000) => JSON.stringify({ type: 'assistant/message', time, data: { turn, step, message: { role: 'assistant', content: [{ type: 'text', text }] } }, }); const turnEnd = (turn: number, reason: Record, time = 1_700_000_000_001) => JSON.stringify({ type: 'turn/end', time, data: { turn, reason } }); describe.skipIf(!zstdSupported())('zstd frame walking', () => { it('decodes every frame, not just the first (the silent-truncation bug)', () => { const lines = Array.from({ length: 40 }, (_, i) => JSON.stringify({ type: 'noise', seq: i })); const buf = framed(lines); // The one-shot decoder is what this module had to replace. const oneShot = (zlib as unknown as { zstdDecompressSync?: (b: Buffer) => Buffer }).zstdDecompressSync!(buf); expect(oneShot.toString('utf8').trim().split('\n')).toHaveLength(1); expect(zstdFrameRanges(buf)).toHaveLength(40); expect(decodeZstdFrames(buf).trim().split('\n')).toHaveLength(40); }); it('round-trips a single-frame file', () => { const buf = framed(['{"type":"session"}']); expect(decodeZstdFrames(buf)).toBe('{"type":"session"}\n'); }); it('passes an uncompressed transcript straight through', () => { const plain = Buffer.from('{"type":"session"}\n{"type":"turn/start"}\n', 'utf8'); expect(decodeZstdFrames(plain)).toBe('{"type":"session"}\n{"type":"turn/start"}\n'); }); it('keeps the whole frames before a torn tail instead of failing the read', () => { const buf = framed(['{"a":1}', '{"b":2}', '{"c":3}']); const torn = buf.subarray(0, buf.length - 4); const decoded = decodeZstdFrames(torn); expect(decoded).toContain('{"a":1}'); expect(decoded).toContain('{"b":2}'); expect(decoded).not.toContain('{"c":3}'); }); it('refuses to walk a buffer that is not zstd', () => { expect(zstdFrameRanges(Buffer.from('not zstd at all', 'utf8'))).toEqual([]); }); }); describe('parseDeepSeekTranscript', () => { it('returns the last turn text and skips plugin-injected context', () => { const raw = [ sessionHeader('/w', 1), userPrompt('what is 2+2?'), pluginContext('Current runtime context. This snapshot supersedes earlier snapshots.'), assistantMessage('4.'), turnEnd(1, { kind: 'completed' }), ].join('\n'); const result = parseDeepSeekTranscript(raw, { blocks: true }); expect(result.text).toBe('4.'); expect(result.cwd).toBe('/w'); expect(result.blocks.filter((b) => b.kind === 'prompt').map((b) => b.text)).toEqual(['what is 2+2?']); expect(result.blocks.some((b) => b.text.includes('runtime context'))).toBe(false); }); it('drops a leaked reasoning prefix at the closing tag', () => { const raw = [ sessionHeader('/w', 1), assistantMessage('I should read the file first.\n\nThe add function is wrong.'), turnEnd(1, { kind: 'completed' }), ].join('\n'); expect(parseDeepSeekTranscript(raw).text).toBe('The add function is wrong.'); }); it('renders tool calls and tool results as tool blocks', () => { const raw = [ sessionHeader('/w', 1), JSON.stringify({ type: 'assistant/message', data: { turn: 1, step: 1, message: { role: 'assistant', content: [ { type: 'text', text: 'Reading it.' }, { type: 'tool-call', id: 'c1', name: 'read', arguments: '{"file_path":"calc.py"}' }, ], }, }, }), JSON.stringify({ type: 'tool/result', data: { turn: 1, step: 1, message: { content: [{ type: 'tool-result', toolCallId: 'c1', content: [{ type: 'text', text: 'def add' }] }], }, }, }), assistantMessage('It subtracts instead of adding.', 1, 2), turnEnd(1, { kind: 'completed' }), ].join('\n'); const result = parseDeepSeekTranscript(raw, { blocks: true }); expect(result.blocks.filter((b) => b.kind === 'tool').map((b) => b.text)).toEqual([ 'read({"file_path":"calc.py"})', 'def add', ]); // Both steps of the turn read back, in order. expect(result.text).toBe('Reading it.\n\nIt subtracts instead of adding.'); }); it('uses streamed deltas only for a step the model never finalized', () => { const raw = [ sessionHeader('/w', 1), JSON.stringify({ type: 'assistant/chunk', data: { turn: 1, step: 1, chunk: { type: 'text-delta', index: 0, text: 'Par' } }, }), JSON.stringify({ type: 'text-chunks', seq: null, data: { turn: 1, step: 1, index: 0, texts: ['is is ', 'the'] }, }), JSON.stringify({ type: 'assistant/chunk', data: { turn: 1, step: 1, chunk: { type: 'text-delta', index: 0, text: ' capital.' } }, }), ].join('\n'); // Still streaming: the partial answer is readable. expect(parseDeepSeekTranscript(raw).text).toBe('Paris is the capital.'); // Once finalized, the deltas must not be appended a second time. const finalized = `${raw}\n${assistantMessage('Paris is the capital.')}\n${turnEnd(1, { kind: 'completed' })}`; expect(parseDeepSeekTranscript(finalized).text).toBe('Paris is the capital.'); }); it('does not resurrect raw deltas for a step whose reply was all reasoning', () => { // Measured on a real conversation: step 1 finalized as reasoning only, so // its text stripped to '' and the (unstripped) deltas took its place, // putting `` and the monologue back in front of the caller. const raw = [ sessionHeader('/w', 1), JSON.stringify({ type: 'assistant/chunk', data: { turn: 1, step: 1, chunk: { type: 'text-delta', index: 0, text: "I'll read the file.\n\n" } }, }), assistantMessage("I'll read the file.\n\n", 1, 1), assistantMessage('The add function is wrong.', 1, 2), turnEnd(1, { kind: 'completed' }), ].join('\n'); expect(parseDeepSeekTranscript(raw).text).toBe('The add function is wrong.'); }); it('answers a failed turn with its error rather than the previous turn text', () => { const raw = [ sessionHeader('/w', 1), assistantMessage('First answer.', 1), turnEnd(1, { kind: 'completed' }), turnEnd(2, { kind: 'error', error: { message: '400: model does not support tools', code: 'INVALID_REQUEST' } }), ].join('\n'); expect(parseDeepSeekTranscript(raw).text).toBe('Turn error: 400: model does not support tools'); }); it('calls an early stop an ending, not an error, and keeps the text it did produce', () => { const raw = [sessionHeader('/w', 1), assistantMessage('Most'), turnEnd(1, { kind: 'max-tokens' })].join('\n'); const result = parseDeepSeekTranscript(raw, { blocks: true }); expect(result.text).toBe('Most'); expect(result.blocks.map((b) => b.text)).toContain('Turn ended: max-tokens'); }); it('survives a torn last line', () => { const raw = [sessionHeader('/w', 1), assistantMessage('Complete.'), '{"type":"turn/e'].join('\n'); expect(parseDeepSeekTranscript(raw).text).toBe('Complete.'); }); it('is empty for a session that has said nothing', () => { expect(parseDeepSeekTranscript(sessionHeader('/w', 1)).text).toBe(''); }); }); describe('resolveDeepSeekHome', () => { it('prefers the session override over the environment', () => { const previous = process.env.DSH_HOME; process.env.DSH_HOME = '/from-env'; try { expect(resolveDeepSeekHome({ deepSeekHomeOverride: '/from-session' })).toBe('/from-session'); expect(resolveDeepSeekHome({})).toBe('/from-env'); } finally { if (previous === undefined) delete process.env.DSH_HOME; else process.env.DSH_HOME = previous; } }); it('falls back to ~/.dsh', () => { const previous = process.env.DSH_HOME; delete process.env.DSH_HOME; try { expect(resolveDeepSeekHome({})).toMatch(/\.dsh$/); } finally { if (previous !== undefined) process.env.DSH_HOME = previous; } }); }); describe.skipIf(!zstdSupported())('session-to-transcript pairing', () => { let dshHome: string; const workspace = '/home/tester/cases/worker-1'; const sessionStart = 1_800_000_000_000; /** Write a transcript for `cwd`, created at `createdAt`, mtime `mtime`. */ async function writeTranscript(name: string, cwd: string, createdAt: number, mtime: number, answer?: string) { const dir = join(dshHome, 'sessions', '--home-tester-cases-worker-1--', name); await mkdir(dir, { recursive: true }); const lines = [sessionHeader(cwd, createdAt, name)]; if (answer) lines.push(assistantMessage(answer), turnEnd(1, { kind: 'completed' })); const path = join(dir, 'session.jsonl.zstd'); await writeFile(path, framed(lines)); await utimes(path, new Date(mtime), new Date(mtime)); return path; } beforeAll(async () => { dshHome = await mkdtemp(join(tmpdir(), 'dsh-home-')); }); afterAll(async () => { await rm(dshHome, { recursive: true, force: true }); }); it("never hands a fresh session its predecessor's answer", async () => { await writeTranscript('older', workspace, sessionStart - 600_000, sessionStart - 590_000, 'stale answer'); const found = await findDeepSeekTranscript({ dshHome, workingDir: workspace, startedAt: sessionStart }); expect(found).toBeNull(); const result = await readDeepSeekLastResponse({ workingDir: workspace, createdAt: sessionStart, deepSeekHomeOverride: dshHome, }); expect(result).not.toBeNull(); expect(result?.text).toBe(''); }); it('pairs on the boot window even when a sibling wrote more recently', async () => { await writeTranscript('mine', workspace, sessionStart + 2_000, sessionStart + 2_000, 'my answer'); await writeTranscript('sibling', workspace, sessionStart + 300_000, sessionStart + 400_000, 'sibling answer'); const found = await findDeepSeekTranscript({ dshHome, workingDir: workspace, startedAt: sessionStart }); expect(found).toContain('/mine/'); const result = await readDeepSeekLastResponse({ workingDir: workspace, createdAt: new Date(sessionStart), deepSeekHomeOverride: dshHome, }); expect(result?.text).toBe('my answer'); }); it('ignores a transcript recorded for another workspace', async () => { const other = await mkdtemp(join(tmpdir(), 'dsh-home-')); try { const dir = join(other, 'sessions', '--home-tester-cases-worker-1--', 'foreign'); await mkdir(dir, { recursive: true }); await writeFile( join(dir, 'session.jsonl.zstd'), framed([sessionHeader('/somewhere/else', sessionStart + 1_000), assistantMessage('not yours')]) ); const found = await findDeepSeekTranscript({ dshHome: other, workingDir: workspace, startedAt: sessionStart }); expect(found).toBeNull(); } finally { await rm(other, { recursive: true, force: true }); } }); it('reads a transcript created later in the session (a /new conversation)', async () => { const later = await mkdtemp(join(tmpdir(), 'dsh-home-')); try { const dir = join(later, 'sessions', '--home-tester-cases-worker-1--', 'after-new'); await mkdir(dir, { recursive: true }); await writeFile( join(dir, 'session.jsonl.zstd'), framed([ sessionHeader(workspace, sessionStart + 1_800_000), assistantMessage('after /new'), turnEnd(1, { kind: 'completed' }), ]) ); const result = await readDeepSeekLastResponse({ workingDir: workspace, createdAt: sessionStart, deepSeekHomeOverride: later, }); expect(result?.text).toBe('after /new'); } finally { await rm(later, { recursive: true, force: true }); } }); it('reports an unreadable home as empty, not as an error', async () => { const result = await readDeepSeekLastResponse({ workingDir: workspace, createdAt: sessionStart, deepSeekHomeOverride: join(tmpdir(), 'dsh-home-that-does-not-exist'), }); expect(result).toEqual({ text: '', timestamp: '', blocks: [] }); }); it('serves an unchanged transcript from the memo and re-reads when the stat moves', async () => { const home = await mkdtemp(join(tmpdir(), 'dsh-home-')); try { const dir = join(home, 'sessions', '--home-tester-cases-worker-1--', 'memo'); await mkdir(dir, { recursive: true }); const path = join(dir, 'session.jsonl'); const stamp = new Date(sessionStart + 1_000); const read = () => readDeepSeekLastResponse({ workingDir: workspace, createdAt: sessionStart, deepSeekHomeOverride: home }); const body = (answer: string) => `${[sessionHeader(workspace, sessionStart + 1_000), assistantMessage(answer), turnEnd(1, { kind: 'completed' })].join('\n')}\n`; resetDeepSeekTranscriptMemoForTest(); await writeFile(path, body('AAAA')); await utimes(path, stamp, stamp); expect((await read())?.text).toBe('AAAA'); // Same byte length, same forced mtime: indistinguishable from unchanged // by stat, and deliberately served from the memo — the 1s/poll skill loop // must not decode an unchanged file, and dsh only ever APPENDS, so a // same-stat rewrite does not exist outside a test. await writeFile(path, body('BBBB')); await utimes(path, stamp, stamp); expect((await read())?.text).toBe('AAAA'); // An append moves mtime (and normally size), which is the invalidation. await utimes(path, new Date(sessionStart + 2_000), new Date(sessionStart + 2_000)); expect((await read())?.text).toBe('BBBB'); } finally { await rm(home, { recursive: true, force: true }); } }); });