diff --git a/src/providers/dsh.ts b/src/providers/dsh.ts index 7595e106..4fa8bd2d 100644 --- a/src/providers/dsh.ts +++ b/src/providers/dsh.ts @@ -3,7 +3,7 @@ import { join } from 'path' import { homedir } from 'os' import zlib from 'zlib' -import { readSessionFile } from '../fs-utils.js' +import { MAX_SESSION_FILE_BYTES, readSessionFile } from '../fs-utils.js' import { calculateCost, getShortModelName } from '../models.js' import { extractBashCommands } from '../bash-utils.js' import type { ProbeRoot, Provider, SessionSource, SessionParser, ParsedProviderCall } from './types.js' @@ -23,6 +23,23 @@ const zstdDecompress = (zlib as { zstdDecompressSync?: (buf: Buffer) => Buffer } const ZSTD_MAGIC = 0xfd2fb528 +// SESSION_FORMAT_VERSION in @deepseek-ai/dsh-session. DSH refuses to load a log +// stamped with any other version, and a bump means an event's meaning changed, +// so a foreign version is skipped rather than read with today's assumptions. +const SESSION_FORMAT_VERSION = 0 + +const MIN_REASONABLE_TIMESTAMP_MS = 1_000_000_000_000 + +// Discovery walks every session, so a per-file notice would repeat once per +// log; each distinct message is worth saying exactly once. +const noticed = new Set() + +function notice(message: string): void { + if (noticed.has(message)) return + noticed.add(message) + process.stderr.write(message) +} + type ZstdFrame = { start: number; end: number } // Locate complete frames without decompressing their blocks. An EOF inside the @@ -87,13 +104,21 @@ type DshEvent = { seq?: number time?: number // Session header fields live at the top level of the first event. + version?: number id?: string cwd?: string + createdAt?: number + parentSession?: string + seedLength?: number data?: { turn?: number step?: number content?: Array<{ type?: string; text?: string }> + // `user/message` carries the message author: a real prompt is + // `{ kind: 'user' }`, agent-injected context is `{ kind: 'plugin' }`. + source?: { kind?: string } header?: { config?: { model?: string; provider?: string } } + message?: { source?: { kind?: string; model?: string; provider?: string } } chunk?: { type?: string; usage?: DshUsage } usage?: DshUsage name?: string @@ -109,8 +134,9 @@ type StepBucket = { // projection). Time follows the winning report. final: boolean time?: number - // Model in force when this step's usage was reported (the most recent - // request/header config at that point in the log). + // Model that produced this step: the reporting assistant/message's own + // `message.source` when it names one, else the most recent request/header + // config (a header can change the model mid-turn between steps). model: string tools: string[] skills: string[] @@ -138,6 +164,25 @@ function mapToolName(raw: string): string { return toolNameMap[raw] ?? raw } +// A log stamped with a version this parser was not written against is skipped +// whole: a bump means an event's meaning changed, so reading it with today's +// assumptions would report confident wrong numbers. +function isReadableVersion(header: DshEvent, filePath: string): boolean { + if (header.version === SESSION_FORMAT_VERSION) return true + notice(`codeburn: skipping DSH session ${filePath}: unsupported session format version ${String(header.version)}; upgrade codeburn.\n`) + return false +} + +// DSH writes epoch milliseconds; promote a seconds-resolution value and reject +// what stays implausible, matching the guard cline-cli.ts uses on the hazard. +function isoTimestamp(value: number | undefined, fallback: string): string { + if (typeof value !== 'number' || !Number.isFinite(value) || value <= 0) return fallback + const ms = value < MIN_REASONABLE_TIMESTAMP_MS ? value * 1000 : value + const date = new Date(ms) + if (Number.isNaN(date.getTime()) || date.getTime() < MIN_REASONABLE_TIMESTAMP_MS) return fallback + return date.toISOString() +} + function getDshHome(override?: string): string { // An empty-string DSH_HOME is treated as unset. return override ?? (process.env['DSH_HOME'] || undefined) ?? join(homedir(), '.dsh') @@ -165,11 +210,18 @@ function* readZstdLines(buffer: Buffer, maxFrames = Number.POSITIVE_INFINITY): G async function readEventLines(filePath: string): Promise { if (filePath.endsWith('.zstd')) { if (!zstdDecompress) { - process.stderr.write('codeburn: DSH sessions need Node >= 22.15 (zstd support); skipping DSH usage.\n') + notice('codeburn: DSH sessions need Node >= 22.15 (zstd support); skipping DSH usage.\n') return null } let buffer: Buffer try { + // The whole log is buffered to scan its frames, so it needs the same + // oversize guard readSessionFile applies to the uncompressed variant. + const size = (await stat(filePath)).size + if (size > MAX_SESSION_FILE_BYTES) { + notice(`codeburn: skipped oversize DSH session log ${filePath} (${size} bytes)\n`) + return null + } buffer = await readFile(filePath) } catch { return null @@ -177,7 +229,7 @@ async function readEventLines(filePath: string): Promise { try { return [...readZstdLines(buffer)] } catch (err) { - process.stderr.write(`codeburn: skipped corrupt DSH session log ${filePath}: ${err instanceof Error ? err.message : err}\n`) + notice(`codeburn: skipped corrupt DSH session log ${filePath}: ${err instanceof Error ? err.message : err}\n`) return null } } @@ -230,7 +282,8 @@ async function readSessionHeader(filePath: string): Promise { const line = await firstLine() if (!line) return null const event = JSON.parse(line) as DshEvent - return event.type === 'session' ? event : null + if (event.type !== 'session') return null + return isReadableVersion(event, filePath) ? event : null } catch { return null } @@ -307,6 +360,11 @@ function createParser(source: SessionSource, seenKeys: Set): SessionPars let cwd = '' let model = 'unknown' let currentTurn = 0 + let sessionStart = '' + // Events a forked session inherited from its parent. They are a verbatim + // copy of the parent's log, which codeburn parses as its own session, so + // counting them here would bill the same calls twice. + let seedLength = 0 const userMessageByTurn = new Map() const buckets = new Map() @@ -319,11 +377,18 @@ function createParser(source: SessionSource, seenKeys: Set): SessionPars } if (event.type === 'session') { + if (!isReadableVersion(event, source.path)) return sessionId = event.id ?? sessionId cwd = event.cwd ?? cwd + sessionStart = isoTimestamp(event.createdAt, sessionStart) + if (typeof event.parentSession === 'string' && event.parentSession && typeof event.seedLength === 'number') { + seedLength = event.seedLength + } continue } + if (typeof event.seq === 'number' && event.seq < seedLength) continue + if (event.type === 'turn/start') { currentTurn = event.data?.turn ?? currentTurn continue @@ -338,10 +403,15 @@ function createParser(source: SessionSource, seenKeys: Set): SessionPars } if (event.type === 'user/message') { + // Plugin-injected context (runtime snapshots, skill bodies, file-change + // notices) rides the same event type as a typed prompt; only the latter + // is a useful preview. + if (event.data?.source?.kind !== 'user') continue + if (userMessageByTurn.has(currentTurn)) continue const texts = (event.data?.content ?? []) .filter(c => c.type === 'text' && typeof c.text === 'string' && c.text) .map(c => c.text!) - if (texts.length > 0) userMessageByTurn.set(currentTurn, texts.join(' ')) + if (texts.length > 0) userMessageByTurn.set(currentTurn, texts.join(' ').slice(0, 500)) continue } @@ -369,11 +439,16 @@ function createParser(source: SessionSource, seenKeys: Set): SessionPars let usage: DshUsage | undefined let isFinal = false + // The model that actually served the call, when the message records it. + // request/header only describes the request codeburn is about to see. + let reportedModel = model if (event.type === 'assistant/chunk' && event.data?.chunk?.type === 'usage') { usage = event.data.chunk.usage } else if (event.type === 'assistant/message' && event.data?.usage) { usage = event.data.usage isFinal = true + const messageModel = event.data.message?.source?.model + if (typeof messageModel === 'string' && messageModel) reportedModel = messageModel } else { continue } @@ -394,7 +469,7 @@ function createParser(source: SessionSource, seenKeys: Set): SessionPars bucket.usage = usage bucket.final = isFinal bucket.time = event.time - bucket.model = model + bucket.model = reportedModel } } @@ -435,13 +510,14 @@ function createParser(source: SessionSource, seenKeys: Set): SessionPars tools: [...new Set(bucket.tools)], bashCommands: bucket.bashCommands, skills: bucket.skills.length > 0 ? [...new Set(bucket.skills)] : undefined, - timestamp: typeof bucket.time === 'number' ? new Date(bucket.time).toISOString() : '', + timestamp: isoTimestamp(bucket.time, sessionStart), speed: 'standard', deduplicationKey: dedupKey, userMessage: userMessageByTurn.get(turn!) ?? '', sessionId: sessionId || source.path, project: cwd ? projectFromCwd(cwd, source.project) : source.project, projectPath: cwd || undefined, + workingDirectory: cwd || undefined, } } }, diff --git a/src/session-cache.ts b/src/session-cache.ts index 6ded7829..a1305ae1 100644 --- a/src/session-cache.ts +++ b/src/session-cache.ts @@ -285,7 +285,10 @@ export const PROVIDER_PARSE_VERSIONS: Record = { // input/cache rollup; this bump re-parses them so the missing tokens land. copilot: 'cli-shutdown-cost-v1-skills-source-provenance-v1', grok: 'estimated-cost-v1', - dsh: 'v1', + // seed-aware-v1: the parser now skips the parent events a forked session + // replays (double-counted before), takes the model from the reporting + // assistant/message, and keeps agent-injected context out of the preview. + dsh: 'seed-aware-v1', hermes: 'reasoning-output-accounting-v1-est-cost', 'lingtai-tui': 'token-ledger-registry-activity-v3', 'ibm-bob': 'worktree-project-grouping-v1', diff --git a/tests/fixtures/dsh/bash-tool-turn.jsonl b/tests/fixtures/dsh/bash-tool-turn.jsonl new file mode 100644 index 00000000..add85175 --- /dev/null +++ b/tests/fixtures/dsh/bash-tool-turn.jsonl @@ -0,0 +1,35 @@ +{"type":"session","version":0,"id":"e128dda9-ed11-4868-8266-0ef90d03c3d6","createdAt":1783352050748,"cwd":"/home/u/proj","delegationDepth":0} +{"type":"agent/inbox/spliced","seq":0,"time":1785498771334,"data":{"target":"next-turn","start":0,"inserted":[{"content":[{"type":"text","text":"Use the bash tool to run exactly: echo TERMINAL_OK. Then reply with the single word DONE and stop."}],"source":{"kind":"user"},"role":"user","id":"38694db6-921d-41fd-b1fb-3b0c40caf67c"}]}} +{"type":"turn/start","seq":1,"time":1785821375023,"data":{"turn":1}} +{"type":"agent/inbox/spliced","seq":2,"time":1785821375023,"data":{"target":"next-turn","start":0,"removedCount":1,"inserted":[]}} +{"type":"step/start","seq":3,"time":1783352050755,"data":{"turn":1,"step":1}} +{"type":"user/message","seq":4,"time":1785498771360,"data":{"content":[{"type":"text","text":"Use the bash tool to run exactly: echo TERMINAL_OK. Then reply with the single word DONE and stop."}],"source":{"kind":"user"},"role":"user","id":"38694db6-921d-41fd-b1fb-3b0c40caf67c"},"surfaceOp":"append"} +{"type":"user/message","seq":5,"time":1785730424635,"data":{"content":[{"type":"text","text":"Current runtime context. This snapshot supersedes earlier runtime-context snapshots.\n\nCurrent DSH file policy: danger-full-access. The DSH file sandbox does not restrict file modifications by available operations.\n\nApproval prompts are disabled in this session: actions that require approval are rejected automatically — do not request sandbox escalation (do not set `sandbox_permissions`)."}],"source":{"kind":"plugin","plugin":"@deepseek-ai/dsh-system-prompt","form":"snapshot","sections":[{"name":"sandbox:policy","text":"Current DSH file policy: danger-full-access. The DSH file sandbox does not restrict file modifications by available operations."},{"name":"approval:policy","text":"Approval prompts are disabled in this session: actions that require approval are rejected automatically — do not request sandbox escalation (do not set `sandbox_permissions`)."}]},"role":"user","id":"80474489-442a-4e98-beef-df6cd1e85870"},"surfaceOp":"append"} +{"type":"session/title","seq":6,"time":1785730424635,"data":{"title":"Use the bash tool to","messageSeqs":[4],"source":{"kind":"fallback"}}} +{"type":"request/header","seq":7,"time":1785498771361,"data":{"header":{"config":{"provider":"deepseek-official","model":"deepseek-v4-flash"},"system":"you are dsh","tools":[]},"reason":"initial"}} +{"type":"request/context","seq":8,"time":1785730424636,"data":{"provider":"deepseek-official","model":"deepseek-v4-flash"}} +{"type":"assistant/chunk","seq":9,"time":1783352051590,"data":{"turn":1,"step":1,"chunk":{"type":"block-start","index":0,"blockType":"reasoning"}}} +{"type":"reasoning-chunks","seq0":10,"time0":1783352051618,"data":{"turn":1,"step":1,"index":0,"dt":[0,1,0,0,26,30,0,0,1,0,27,1,0,0,0,86,1],"texts":["The"," user"," wants"," me"," to"," run"," a"," simple"," bash"," command"," and"," then"," reply"," with"," \"","D","ONE","\"."]}} +{"type":"assistant/chunk","seq":28,"time":1783352051820,"data":{"turn":1,"step":1,"chunk":{"type":"block-start","index":1,"blockType":"tool-call"}}} +{"type":"tool-call-chunks","seq0":29,"time0":1783352051820,"data":{"turn":1,"step":1,"index":1,"dt":[0,0,0,28,0,0,0,29,0,0,28,1,0,29,0,0,0,32,0,0,0,0,0,74,0,0,13,0,63,1],"id":"call_00_fkbBRJsUrGKd1pWVc4Gn8233","name":"bash","args":["","{","\"","command","\"",": ","\"","echo"," TER","MIN","AL","_OK","\"",", ","\"","description","\"",": ","\"","E","cho"," TER","MIN","AL","_OK"," to"," verify"," terminal"," access","\"","}"]}} +{"type":"assistant/chunk","seq":60,"time":1783352052118,"data":{"turn":1,"step":1,"chunk":{"type":"block-end","index":0,"block":{"type":"reasoning","text":"The user wants me to run a simple bash command and then reply with \"DONE\"."}}}} +{"type":"assistant/chunk","seq":61,"time":1783352052118,"data":{"turn":1,"step":1,"chunk":{"type":"block-end","index":1,"block":{"type":"tool-call","id":"call_00_fkbBRJsUrGKd1pWVc4Gn8233","name":"bash","arguments":"{\"command\": \"echo TERMINAL_OK\", \"description\": \"Echo TERMINAL_OK to verify terminal access\"}"}}}} +{"type":"assistant/chunk","seq":62,"time":1785498771373,"data":{"turn":1,"step":1,"chunk":{"type":"usage","usage":{"inputTokens":2877,"outputTokens":90,"cacheReadTokens":0,"reasoningTokens":18}}}} +{"type":"assistant/chunk","seq":63,"time":1785730424645,"data":{"turn":1,"step":1,"chunk":{"type":"finish","reason":{"kind":"tool-calls"}}}} +{"type":"assistant/message","seq":64,"time":1785730424645,"data":{"turn":1,"step":1,"message":{"role":"assistant","content":[{"type":"reasoning","text":"The user wants me to run a simple bash command and then reply with \"DONE\"."},{"type":"tool-call","id":"call_00_fkbBRJsUrGKd1pWVc4Gn8233","name":"bash","arguments":"{\"command\": \"echo TERMINAL_OK\", \"description\": \"Echo TERMINAL_OK to verify terminal access\"}"}],"source":{"kind":"model","provider":"deepseek-official","model":"deepseek-v4-flash"},"id":"0a855246-fbf6-4f91-87b4-c6f1889effe7"},"usage":{"inputTokens":2877,"outputTokens":90,"cacheReadTokens":0,"reasoningTokens":18}},"sourceEventSeqs":[9,10,11,12,13,14,15,16,17,18,19,20,21,22,23,24,25,26,27,28,29,30,31,32,33,34,35,36,37,38,39,40,41,42,43,44,45,46,47,48,49,50,51,52,53,54,55,56,57,58,59,60,61,62,63],"surfaceOp":"append"} +{"type":"tool/call","seq":65,"time":1785730424646,"data":{"turn":1,"step":1,"callId":"call_00_fkbBRJsUrGKd1pWVc4Gn8233","name":"bash","arguments":"{\"command\": \"echo TERMINAL_OK\", \"description\": \"Echo TERMINAL_OK to verify terminal access\"}"}} +{"type":"tool/result","seq":66,"time":1785730424665,"data":{"turn":1,"step":1,"message":{"source":{"kind":"tool","callId":"call_00_fkbBRJsUrGKd1pWVc4Gn8233"},"content":[{"type":"tool-result","toolCallId":"call_00_fkbBRJsUrGKd1pWVc4Gn8233","content":[{"type":"text","text":"TERMINAL_OK\n"}],"isError":false}],"role":"user","id":"908ca4f5-efbb-443b-9b07-acbf25edf954"}},"sourceEventSeqs":[65],"surfaceOp":"append"} +{"type":"step/end","seq":67,"time":1785730424665,"data":{"turn":1,"step":1}} +{"type":"step/start","seq":68,"time":1785730424676,"data":{"turn":1,"step":2}} +{"type":"assistant/chunk","seq":69,"time":1783352052780,"data":{"turn":1,"step":2,"chunk":{"type":"block-start","index":0,"blockType":"reasoning"}}} +{"type":"reasoning-chunks","seq0":70,"time0":1783352052809,"data":{"turn":1,"step":2,"index":0,"dt":[29,0,0,29,0,0,0,0,0,28,1,28,1,0,0,32,0,0,0,0,0],"texts":["The"," command"," ran"," successfully"," and"," output"," \"","TER","MIN","AL","_OK","\"."," I"," should"," now"," reply"," with"," just"," \"","D","ONE","\"."]}} +{"type":"assistant/chunk","seq":92,"time":1783352052986,"data":{"turn":1,"step":2,"chunk":{"type":"block-start","index":1,"blockType":"text"}}} +{"type":"assistant/chunk","seq":93,"time":1783352052986,"data":{"turn":1,"step":2,"chunk":{"type":"text-delta","index":1,"text":"D"}}} +{"type":"assistant/chunk","seq":94,"time":1783352052986,"data":{"turn":1,"step":2,"chunk":{"type":"text-delta","index":1,"text":"ONE"}}} +{"type":"assistant/chunk","seq":95,"time":1783352052986,"data":{"turn":1,"step":2,"chunk":{"type":"block-end","index":0,"block":{"type":"reasoning","text":"The command ran successfully and output \"TERMINAL_OK\". I should now reply with just \"DONE\"."}}}} +{"type":"assistant/chunk","seq":96,"time":1783352052986,"data":{"turn":1,"step":2,"chunk":{"type":"block-end","index":1,"block":{"type":"text","text":"DONE"}}}} +{"type":"assistant/chunk","seq":97,"time":1785498771406,"data":{"turn":1,"step":2,"chunk":{"type":"usage","usage":{"inputTokens":168,"outputTokens":25,"cacheReadTokens":2816,"reasoningTokens":22}}}} +{"type":"assistant/chunk","seq":98,"time":1785730424681,"data":{"turn":1,"step":2,"chunk":{"type":"finish","reason":{"kind":"stop"}}}} +{"type":"assistant/message","seq":99,"time":1785730424681,"data":{"turn":1,"step":2,"message":{"role":"assistant","content":[{"type":"reasoning","text":"The command ran successfully and output \"TERMINAL_OK\". I should now reply with just \"DONE\"."},{"type":"text","text":"DONE"}],"source":{"kind":"model","provider":"deepseek-official","model":"deepseek-v4-flash"},"id":"aa705bf0-9b5b-4af3-9763-dbf93c98e4c4"},"usage":{"inputTokens":168,"outputTokens":25,"cacheReadTokens":2816,"reasoningTokens":22}},"sourceEventSeqs":[69,70,71,72,73,74,75,76,77,78,79,80,81,82,83,84,85,86,87,88,89,90,91,92,93,94,95,96,97,98],"surfaceOp":"append"} +{"type":"step/end","seq":100,"time":1785730424682,"data":{"turn":1,"step":2}} +{"type":"turn/end","seq":101,"time":1785730424682,"data":{"turn":1,"reason":{"kind":"completed"}}} diff --git a/tests/providers/dsh.test.ts b/tests/providers/dsh.test.ts index b5fbfcb0..05994566 100644 --- a/tests/providers/dsh.test.ts +++ b/tests/providers/dsh.test.ts @@ -1,5 +1,5 @@ import { describe, it, expect, beforeEach, afterEach } from 'vitest' -import { mkdtemp, mkdir, writeFile, rm } from 'fs/promises' +import { mkdtemp, mkdir, writeFile, readFile, rm } from 'fs/promises' import { join } from 'path' import { homedir, tmpdir } from 'os' import zlib from 'zlib' @@ -404,3 +404,136 @@ describe('dsh provider - display names', () => { expect(provider.toolDisplayName('cordis_run')).toBe('cordis_run') }) }) + +describe('dsh provider - real log fidelity', () => { + // The upstream snapshot from deepseek-ai/deepseek-harness + // (examples/acp-agent/tests/snapshots/bash-tool-turn/session.jsonl), with its + // template placeholders filled in. It is the reference for every shape the + // parser reads: packed `reasoning-chunks`/`tool-call-chunks` storage rows, a + // plugin-injected user/message beside the typed one, and both the streamed + // usage chunk and the final assistant/message usage for the same step. + async function writeRealSession(): Promise { + const lines = (await readFile(join(import.meta.dirname, '../fixtures/dsh/bash-tool-turn.jsonl'), 'utf-8')) + .split('\n').filter(l => l.trim()) + return writePlainSession('--home-u-proj--', 'e128dda9-ed11-4868-8266-0ef90d03c3d6', lines) + } + + it('parses the upstream snapshot: two steps, exact usage, model from the message source', async () => { + const calls = await parseAll(createDshProvider(tmpDir), await writeRealSession()) + + expect(calls).toHaveLength(2) + expect(calls.map(c => c.model)).toEqual(['deepseek-v4-flash', 'deepseek-v4-flash']) + expect(calls[0]).toMatchObject({ + inputTokens: 2877, + outputTokens: 90, + cacheReadInputTokens: 0, + reasoningTokens: 18, + sessionId: 'e128dda9-ed11-4868-8266-0ef90d03c3d6', + project: 'proj', + projectPath: '/home/u/proj', + workingDirectory: '/home/u/proj', + }) + expect(calls[1]).toMatchObject({ inputTokens: 168, outputTokens: 25, cacheReadInputTokens: 2816, reasoningTokens: 22 }) + // Reasoning bills at the output rate, so it must not appear as input. + expect(calls[0]!.costUSD).toBe(calculateCost('deepseek-v4-flash', 2877, 90 + 18, 0, 0, 0)) + expect(calls[0]!.costUSD).toBeGreaterThan(0) + }) + + it('takes the typed prompt as the preview, not the plugin-injected context', async () => { + const calls = await parseAll(createDshProvider(tmpDir), await writeRealSession()) + + expect(calls[0]!.userMessage).toBe('Use the bash tool to run exactly: echo TERMINAL_OK. Then reply with the single word DONE and stop.') + expect(calls[0]!.userMessage).not.toContain('Current runtime context') + }) + + it('reads the tool call through the packed chunk rows around it', async () => { + const calls = await parseAll(createDshProvider(tmpDir), await writeRealSession()) + + expect(calls[0]!.tools).toEqual(['Bash']) + expect(calls[0]!.bashCommands).toEqual(['echo']) + }) +}) + +describe('dsh provider - defensive reads', () => { + it('skips a log stamped with an unsupported session format version', async () => { + const filePath = await writePlainSession('--home-u-proj--', 'session-future', [ + JSON.stringify({ type: 'session', version: 1, id: 'session-future', createdAt: 1786707336131, cwd: '/home/u/proj', delegationDepth: 0 }), + chunkUsage(1, 1, { inputTokens: 100, outputTokens: 10 }, 1786707340000), + ]) + + expect(await createDshProvider(tmpDir).discoverSessions()).toEqual([]) + expect(await parseAll(createDshProvider(tmpDir), filePath)).toEqual([]) + }) + + it('does not bill a forked session for the events it inherited from its parent', async () => { + const filePath = await writePlainSession('--home-u-proj--', 'session-fork', [ + JSON.stringify({ + type: 'session', version: 0, id: 'session-fork', createdAt: 1786707336131, + cwd: '/home/u/proj', parentSession: 'session-parent', seedLength: 3, delegationDepth: 0, + }), + // seq 0..2 are a verbatim copy of the parent's log, which codeburn parses + // as its own session; only seq >= 3 is this session's own work. + JSON.stringify({ type: 'turn/start', seq: 0, time: 1786707337000, data: { turn: 1 } }), + JSON.stringify({ type: 'assistant/message', seq: 1, time: 1786707337100, data: { turn: 1, step: 1, message: { role: 'assistant', content: [] }, usage: { inputTokens: 9999, outputTokens: 999 } } }), + JSON.stringify({ type: 'session/end-seed', seq: 2, time: 1786707337200, data: {} }), + JSON.stringify({ type: 'turn/start', seq: 3, time: 1786707338000, data: { turn: 2 } }), + JSON.stringify({ type: 'assistant/message', seq: 4, time: 1786707338100, data: { turn: 2, step: 1, message: { role: 'assistant', content: [] }, usage: { inputTokens: 100, outputTokens: 10 } } }), + ]) + + const calls = await parseAll(createDshProvider(tmpDir), filePath) + expect(calls).toHaveLength(1) + expect(calls[0]!.inputTokens).toBe(100) + }) + + it('ignores unknown event types, packed chunk rows, and unparsable lines', async () => { + const filePath = await writePlainSession('--home-u-proj--', 'session-noise', [ + sessionHeader({ id: 'session-noise', cwd: '/home/u/proj' }), + JSON.stringify({ type: 'agent/inbox/spliced', seq: 0, time: 1786707337000, data: { target: 'next-turn' } }), + JSON.stringify({ type: 'reasoning-chunks', seq0: 1, time0: 1786707337100, data: { turn: 1, step: 1, index: 0, dt: [0], texts: ['a', 'b'] } }), + '{ not json at all', + ' ', + chunkUsage(1, 1, { inputTokens: 100, outputTokens: 10 }, 1786707340000), + ]) + + const calls = await parseAll(createDshProvider(tmpDir), filePath) + expect(calls).toHaveLength(1) + expect(calls[0]!.inputTokens).toBe(100) + }) + + it('falls back to the header createdAt when a usage event carries no usable time', async () => { + const filePath = await writePlainSession('--home-u-proj--', 'session-notime', [ + JSON.stringify({ type: 'session', version: 0, id: 'session-notime', createdAt: 1786707336131, cwd: '/home/u/proj', delegationDepth: 0 }), + JSON.stringify({ type: 'assistant/message', seq: 1, data: { turn: 1, step: 1, message: { role: 'assistant', content: [] }, usage: { inputTokens: 100, outputTokens: 10 } } }), + ]) + + const calls = await parseAll(createDshProvider(tmpDir), filePath) + expect(calls).toHaveLength(1) + expect(calls[0]!.timestamp).toBe(new Date(1786707336131).toISOString()) + }) +}) + +describe('dsh provider - real log, real container', () => { + it('reads the upstream snapshot out of multi-frame zstd with a torn tail identically to plain jsonl', async () => { + const lines = (await readFile(join(import.meta.dirname, '../fixtures/dsh/bash-tool-turn.jsonl'), 'utf-8')) + .split('\n').filter(l => l.trim()) + const plain = await parseAll( + createDshProvider(tmpDir), + await writePlainSession('--home-u-proj--', 'plain', lines), + ) + + // Header batch, then three append batches — the layout DSH writes. + const dir = join(tmpDir, 'sessions', '--home-u-proj--', 'framed') + await mkdir(dir, { recursive: true }) + const filePath = join(dir, 'session.jsonl.zstd') + const frames = [[lines[0]!], lines.slice(1, 10), lines.slice(10, 25), lines.slice(25)] + .map(batch => zstdCompress!(Buffer.from(batch.join('\n') + '\n', 'utf-8'))) + // A crashed writer's half-written final batch, carrying usage that must not count. + const torn = zstdCompress!(Buffer.from(assistantMessage(9, 9, { inputTokens: 123456, outputTokens: 1 }, 1785730424999) + '\n', 'utf-8')) + await writeFile(filePath, Buffer.concat([...frames, torn.subarray(0, Math.floor(torn.length / 2))])) + + const framed = await parseAll(createDshProvider(tmpDir), filePath) + expect(framed.map(c => [c.inputTokens, c.outputTokens, c.reasoningTokens, c.model])) + .toEqual(plain.map(c => [c.inputTokens, c.outputTokens, c.reasoningTokens, c.model])) + expect(framed).toHaveLength(2) + }) +})