fix: normalize tool names and remove metadata.sessionId from running/completed events
- Apply normalizeToolName consistently in all tool_result paths (wasStreamed=true and wasStreamed=false) - Remove metadata.sessionId from running state in streamStateTracker content_block_stop - Remove metadata.sessionId from emitTaskProgress and emitTaskCompleted - Add streamStateTracker.ts with stream-to-opencode event conversion - Add subagent task lifecycle events (started/progress/completed) - Handle tombstone, attachment, progress message types
This commit is contained in:
DoSun
parent
da1985708f
commit
07f6ebe004
449
src/server/__tests__/sessionMessageRouter.test.ts
Normal file
449
src/server/__tests__/sessionMessageRouter.test.ts
Normal file
|
|
@ -0,0 +1,449 @@
|
|||
import { describe, expect, test, vi, afterEach } from 'bun:test'
|
||||
import { routeMessage, type StdoutMessage, type MessageRouterCtx, type SubagentInfo } from '../sessionMessageRouter.js'
|
||||
import { resetAllState } from '../streamStateTracker.js'
|
||||
import type { ControlChannel } from '../sessionControlChannel.js'
|
||||
import type { SessionState, SessionBusyStatus, InitData, PendingPermission, PendingQuestion } from '../types.js'
|
||||
import type { SessionMessage } from '../transcriptReader.js'
|
||||
|
||||
type OCEvent = { event: string; properties: Record<string, unknown> }
|
||||
|
||||
type TestCtx = MessageRouterCtx & {
|
||||
oc: OCEvent[]
|
||||
ev: Array<{ event: string; data: Record<string, unknown> }>
|
||||
msgs: StdoutMessage[]
|
||||
buf: SessionMessage[]
|
||||
}
|
||||
|
||||
afterEach(() => {
|
||||
resetAllState('test-session-1')
|
||||
})
|
||||
|
||||
function makeCtx(overrides?: Partial<MessageRouterCtx>): TestCtx {
|
||||
const oc: OCEvent[] = []
|
||||
const ev: Array<{ event: string; data: Record<string, unknown> }> = []
|
||||
const msgs: StdoutMessage[] = []
|
||||
const buf: SessionMessage[] = []
|
||||
let _model: string | undefined
|
||||
let _providerId: string | undefined
|
||||
const _activeSubagents = new Map<string, SubagentInfo>()
|
||||
const base: MessageRouterCtx = {
|
||||
sessionId: 'test-session-1',
|
||||
silent: false,
|
||||
getStatus: () => 'running' as SessionState,
|
||||
getBusyStatus: () => ({ type: 'idle' }) as SessionBusyStatus,
|
||||
getModel: () => _model,
|
||||
getProviderId: () => _providerId,
|
||||
getInitRequestId: () => null,
|
||||
getLastMessageUuid: () => null,
|
||||
setStatus: vi.fn(),
|
||||
setBusyStatus: vi.fn(),
|
||||
setModel: (m: string) => { _model = m },
|
||||
setProviderId: (id: string) => { _providerId = id },
|
||||
setInitData: vi.fn(),
|
||||
setLastMessageUuid: vi.fn(),
|
||||
setLastActiveAt: vi.fn(),
|
||||
getActiveSubagents: () => _activeSubagents,
|
||||
registerSubagent: (info: SubagentInfo) => { _activeSubagents.set(info.agentId, info) },
|
||||
unregisterSubagent: (agentId: string) => { _activeSubagents.delete(agentId) },
|
||||
getControlChannel: () => ({ tryResolve: vi.fn() }) as unknown as ControlChannel,
|
||||
getPendingPermissions: () => new Map<string, PendingPermission>(),
|
||||
getPendingQuestions: () => new Map<string, PendingQuestion>(),
|
||||
resolveInit: vi.fn(),
|
||||
resolvePrompt: vi.fn(),
|
||||
addCost: vi.fn(),
|
||||
addInputTokens: vi.fn(),
|
||||
addOutputTokens: vi.fn(),
|
||||
emitEvent: (event, data) => { ev.push({ event, data }) },
|
||||
emitOpencodeEvent: (event, properties) => { oc.push({ event, properties }) },
|
||||
emitBusyStatus: vi.fn(),
|
||||
emitMessage: (msg) => { msgs.push(msg) },
|
||||
writeStdin: vi.fn(),
|
||||
pushBufferMessage: (msg) => { buf.push(msg) },
|
||||
addTombstonedUuid: vi.fn(),
|
||||
}
|
||||
const merged = { ...base, ...overrides }
|
||||
return { ...merged, oc, ev, msgs, buf } as TestCtx
|
||||
}
|
||||
|
||||
describe('routeMessage — stream_event', () => {
|
||||
test('message_start emits message.updated + step-start', () => {
|
||||
const ctx = makeCtx()
|
||||
routeMessage({
|
||||
type: 'stream_event',
|
||||
event: { type: 'message_start', message: { model: 'claude-sonnet-4', usage: { input_tokens: 50 } } },
|
||||
}, ctx)
|
||||
const types = ctx.oc.map(e => e.event)
|
||||
expect(types).toContain('message.updated')
|
||||
expect(types).toContain('message.part.updated')
|
||||
})
|
||||
|
||||
test('stream_event without event field is silently dropped', () => {
|
||||
const ctx = makeCtx()
|
||||
routeMessage({ type: 'stream_event' }, ctx)
|
||||
expect(ctx.oc.length).toBe(0)
|
||||
expect(ctx.ev.length).toBe(0)
|
||||
})
|
||||
})
|
||||
|
||||
describe('routeMessage — system subtype dispatch', () => {
|
||||
test('task_started → task.started + virtual session events + assistant message', () => {
|
||||
const ctx = makeCtx()
|
||||
routeMessage({ type: 'system', subtype: 'task_started', task_id: 'task-1', description: 'Running agent', task_type: 'local_agent' }, ctx)
|
||||
expect(ctx.oc.length).toBe(4)
|
||||
expect(ctx.oc[0].event).toBe('task.started')
|
||||
expect(ctx.oc[0].properties).toMatchObject({ sessionID: 'test-session-1', taskID: 'task-1', taskType: 'local_agent' })
|
||||
expect(ctx.oc[1].event).toBe('session.created')
|
||||
expect(ctx.oc[1].properties).toMatchObject({ sessionID: 'task-1', info: { parentID: 'test-session-1', agent: 'local_agent' } })
|
||||
expect(ctx.oc[2].event).toBe('message.updated')
|
||||
expect(ctx.oc[2].properties).toMatchObject({ sessionID: 'task-1', info: { role: 'assistant', modelID: 'subagent' } })
|
||||
expect(ctx.oc[3].event).toBe('session.status')
|
||||
expect(ctx.oc[3].properties).toMatchObject({ sessionID: 'task-1', status: { type: 'busy' } })
|
||||
expect(ctx.getActiveSubagents().has('task-1')).toBe(true)
|
||||
})
|
||||
|
||||
test('task_progress → task.progress', () => {
|
||||
const ctx = makeCtx()
|
||||
routeMessage({ type: 'system', subtype: 'task_progress', task_id: 'task-1', usage: { total_tokens: 100 } }, ctx)
|
||||
expect(ctx.oc[0].event).toBe('task.progress')
|
||||
})
|
||||
|
||||
test('task_progress with tool_uses emits tool parts on subagent session', () => {
|
||||
const ctx = makeCtx()
|
||||
routeMessage({ type: 'system', subtype: 'task_started', task_id: 'task-1', description: 'Test agent' }, ctx)
|
||||
const assistantMsgID = (ctx.oc.find(e => e.event === 'message.updated' && e.properties.sessionID === 'task-1')?.properties as Record<string, unknown>)?.info as Record<string, unknown>
|
||||
expect(assistantMsgID).toBeDefined()
|
||||
expect(assistantMsgID.role).toBe('assistant')
|
||||
ctx.oc.length = 0
|
||||
|
||||
routeMessage({
|
||||
type: 'system',
|
||||
subtype: 'task_progress',
|
||||
task_id: 'task-1',
|
||||
last_tool_name: 'Bash',
|
||||
description: 'Running ls',
|
||||
usage: { duration_ms: 1000, tool_uses: 1, total_tokens: 0 },
|
||||
}, ctx)
|
||||
|
||||
const progress = ctx.oc.find(e => e.event === 'task.progress')
|
||||
expect(progress).toBeDefined()
|
||||
|
||||
const toolParts = ctx.oc.filter(e => e.event === 'message.part.updated')
|
||||
expect(toolParts.length).toBe(1)
|
||||
const part = toolParts[0].properties.part as Record<string, unknown>
|
||||
expect(part.type).toBe('tool')
|
||||
expect(part.tool).toBe('bash')
|
||||
expect((part.state as Record<string, unknown>).status).toBe('completed')
|
||||
expect(part.messageID).toBe(assistantMsgID.id)
|
||||
expect(part.sessionID).toBe('task-1')
|
||||
})
|
||||
|
||||
test('task_progress emits incremental tool parts as tool_uses increases', () => {
|
||||
const ctx = makeCtx()
|
||||
routeMessage({ type: 'system', subtype: 'task_started', task_id: 'task-1', description: 'Test agent' }, ctx)
|
||||
ctx.oc.length = 0
|
||||
|
||||
routeMessage({
|
||||
type: 'system', subtype: 'task_progress', task_id: 'task-1',
|
||||
last_tool_name: 'Bash', description: 'Running ls', usage: { tool_uses: 1 },
|
||||
}, ctx)
|
||||
expect(ctx.oc.filter(e => e.event === 'message.part.updated').length).toBe(1)
|
||||
|
||||
routeMessage({
|
||||
type: 'system', subtype: 'task_progress', task_id: 'task-1',
|
||||
last_tool_name: 'Read', description: 'Reading file', usage: { tool_uses: 2 },
|
||||
}, ctx)
|
||||
expect(ctx.oc.filter(e => e.event === 'message.part.updated').length).toBe(2)
|
||||
|
||||
const parts = ctx.oc.filter(e => e.event === 'message.part.updated')
|
||||
const tool1 = parts[0].properties.part as Record<string, unknown>
|
||||
const tool2 = parts[1].properties.part as Record<string, unknown>
|
||||
expect(tool1.tool).toBe('bash')
|
||||
expect(tool2.tool).toBe('read')
|
||||
expect(tool1.id).not.toBe(tool2.id)
|
||||
})
|
||||
|
||||
test('task_progress without prior task_started is a no-op for tool parts', () => {
|
||||
const ctx = makeCtx()
|
||||
routeMessage({
|
||||
type: 'system', subtype: 'task_progress', task_id: 'unknown-task',
|
||||
last_tool_name: 'Bash', usage: { tool_uses: 1 },
|
||||
}, ctx)
|
||||
expect(ctx.oc.filter(e => e.event === 'message.part.updated').length).toBe(0)
|
||||
})
|
||||
|
||||
test('task_notification → task.completed + assistant message finalize + virtual session idle', () => {
|
||||
const ctx = makeCtx()
|
||||
routeMessage({ type: 'system', subtype: 'task_started', task_id: 'task-1', agent_id: 'task-1', description: 'Test' }, ctx)
|
||||
ctx.oc.length = 0
|
||||
routeMessage({ type: 'system', subtype: 'task_notification', task_id: 'task-1', agent_id: 'task-1', status: 'completed', summary: 'Done', output_file: '/tmp/out' }, ctx)
|
||||
expect(ctx.oc[0].event).toBe('task.completed')
|
||||
expect(ctx.oc[0].properties).toMatchObject({ taskID: 'task-1', status: 'completed' })
|
||||
const msgUpdated = ctx.oc.filter(e => e.event === 'message.updated')
|
||||
expect(msgUpdated.length).toBe(1)
|
||||
expect(msgUpdated[0].properties).toMatchObject({ sessionID: 'task-1', info: { role: 'assistant', modelID: 'subagent' } })
|
||||
const info = msgUpdated[0].properties.info as Record<string, unknown>
|
||||
expect((info.time as Record<string, unknown>).completed).toBeDefined()
|
||||
const sessionUpdated = ctx.oc.filter(e => e.event === 'session.updated')
|
||||
expect(sessionUpdated.length).toBe(1)
|
||||
const sessionStatus = ctx.oc.filter(e => e.event === 'session.status')
|
||||
expect(sessionStatus.length).toBe(1)
|
||||
expect(sessionStatus[0].properties).toMatchObject({ sessionID: 'task-1', status: { type: 'idle' } })
|
||||
const idle = ctx.oc.filter(e => e.event === 'session.idle')
|
||||
expect(idle.length).toBe(1)
|
||||
expect(idle[0].properties).toMatchObject({ sessionID: 'task-1' })
|
||||
expect(ctx.getActiveSubagents().has('task-1')).toBe(false)
|
||||
})
|
||||
|
||||
test('api_error → session.error', () => {
|
||||
const ctx = makeCtx()
|
||||
routeMessage({ type: 'system', subtype: 'api_error', content: 'Rate limited', retry_in_ms: 5000, retry_attempt: 2, max_retries: 3 }, ctx)
|
||||
expect(ctx.oc[0].event).toBe('session.error')
|
||||
const err = ctx.oc[0].properties.error as Record<string, unknown>
|
||||
expect(err.subtype).toBe('api_error')
|
||||
expect(err.retryInMs).toBe(5000)
|
||||
})
|
||||
|
||||
test('compact_boundary → compaction part', () => {
|
||||
const ctx = makeCtx()
|
||||
routeMessage({ type: 'system', subtype: 'compact_boundary', uuid: 'uuid-1' }, ctx)
|
||||
expect(ctx.oc[0].event).toBe('message.part.updated')
|
||||
const part = ctx.oc[0].properties.part as Record<string, unknown>
|
||||
expect(part.type).toBe('compaction')
|
||||
expect(part.auto).toBe(false)
|
||||
})
|
||||
|
||||
test('microcompact_boundary → compaction with auto=true', () => {
|
||||
const ctx = makeCtx()
|
||||
routeMessage({ type: 'system', subtype: 'microcompact_boundary' }, ctx)
|
||||
const part = ctx.oc[0].properties.part as Record<string, unknown>
|
||||
expect(part.auto).toBe(true)
|
||||
})
|
||||
|
||||
test('stop_hook_summary → session.hook_summary', () => {
|
||||
const ctx = makeCtx()
|
||||
routeMessage({ type: 'system', subtype: 'stop_hook_summary', hook_count: 2, total_duration_ms: 100 }, ctx)
|
||||
expect(ctx.oc[0].event).toBe('session.hook_summary')
|
||||
})
|
||||
|
||||
test('turn_duration → session.metrics', () => {
|
||||
const ctx = makeCtx()
|
||||
routeMessage({ type: 'system', subtype: 'turn_duration', duration: 5000 }, ctx)
|
||||
expect(ctx.oc[0].event).toBe('session.metrics')
|
||||
})
|
||||
|
||||
test('cache_warning → session.warning', () => {
|
||||
const ctx = makeCtx()
|
||||
routeMessage({ type: 'system', subtype: 'cache_warning', content: 'Miss' }, ctx)
|
||||
expect(ctx.oc[0].event).toBe('session.warning')
|
||||
})
|
||||
|
||||
test('informational → session.info', () => {
|
||||
const ctx = makeCtx()
|
||||
routeMessage({ type: 'system', subtype: 'informational', content: 'Info' }, ctx)
|
||||
expect(ctx.oc[0].event).toBe('session.info')
|
||||
})
|
||||
|
||||
test('unknown subtype → session.info', () => {
|
||||
const ctx = makeCtx()
|
||||
routeMessage({ type: 'system', subtype: 'custom_subtype', content: 'data' }, ctx)
|
||||
expect(ctx.oc[0].event).toBe('session.info')
|
||||
expect(ctx.oc[0].properties.subtype).toBe('custom_subtype')
|
||||
})
|
||||
|
||||
test('session_state_changed → session.status', () => {
|
||||
const ctx = makeCtx()
|
||||
routeMessage({ type: 'system', subtype: 'session_state_changed', status: 'compacting' }, ctx)
|
||||
expect(ctx.oc[0].event).toBe('session.status')
|
||||
})
|
||||
})
|
||||
|
||||
describe('routeMessage — attachment', () => {
|
||||
test('attachment → message.attachment', () => {
|
||||
const ctx = makeCtx()
|
||||
routeMessage({ type: 'attachment', attachment: { type: 'hook_success', data: 'ok' } }, ctx)
|
||||
expect(ctx.oc.length).toBe(1)
|
||||
expect(ctx.oc[0].event).toBe('message.attachment')
|
||||
expect(ctx.oc[0].properties.attachmentType).toBe('hook_success')
|
||||
})
|
||||
})
|
||||
|
||||
describe('routeMessage — progress', () => {
|
||||
test('progress → tool.progress', () => {
|
||||
const ctx = makeCtx()
|
||||
routeMessage({ type: 'progress', toolUseID: 'tool-1', data: { type: 'bash_progress' } }, ctx)
|
||||
expect(ctx.oc.length).toBe(1)
|
||||
expect(ctx.oc[0].event).toBe('tool.progress')
|
||||
expect(ctx.oc[0].properties.toolUseID).toBe('tool-1')
|
||||
})
|
||||
})
|
||||
|
||||
describe('routeMessage — tombstone', () => {
|
||||
test('tombstone → message.removed', () => {
|
||||
const ctx = makeCtx()
|
||||
routeMessage({ type: 'tombstone', message: { uuid: 'dead-msg-uuid' } }, ctx)
|
||||
expect(ctx.oc.length).toBe(1)
|
||||
expect(ctx.oc[0].event).toBe('message.removed')
|
||||
expect(ctx.oc[0].properties.messageID).toBe('dead-msg-uuid')
|
||||
})
|
||||
|
||||
test('tombstone without uuid is dropped', () => {
|
||||
const ctx = makeCtx()
|
||||
routeMessage({ type: 'tombstone', message: {} }, ctx)
|
||||
expect(ctx.oc.length).toBe(0)
|
||||
})
|
||||
})
|
||||
|
||||
describe('routeMessage — assistant (non-streaming)', () => {
|
||||
test('emits message.updated + parts for content blocks', () => {
|
||||
const ctx = makeCtx()
|
||||
routeMessage({
|
||||
type: 'assistant',
|
||||
uuid: 'msg-1',
|
||||
model: 'claude-sonnet-4',
|
||||
provider_id: 'anthropic',
|
||||
timestamp: '2026-05-18T00:00:00Z',
|
||||
parentUuid: 'parent-1',
|
||||
message: {
|
||||
content: [
|
||||
{ type: 'text', text: 'Hello' },
|
||||
{ type: 'thinking', thinking: 'Let me think...' },
|
||||
{ type: 'tool_use', id: 'tool-1', name: 'Bash', input: { command: 'ls' } },
|
||||
],
|
||||
},
|
||||
}, ctx)
|
||||
|
||||
const updated = ctx.oc.filter(e => e.event === 'message.updated')
|
||||
const parts = ctx.oc.filter(e => e.event === 'message.part.updated')
|
||||
expect(updated.length).toBe(1)
|
||||
expect(updated[0].properties.info).toMatchObject({ role: 'assistant', modelID: 'claude-sonnet-4', parentID: 'parent-1' })
|
||||
expect(parts.length).toBe(3)
|
||||
const partTypes = parts.map(e => (e.properties.part as Record<string, unknown>)?.type)
|
||||
expect(partTypes).toEqual(['text', 'reasoning', 'tool'])
|
||||
})
|
||||
|
||||
test('redacted_thinking emits reasoning part with redacted=true', () => {
|
||||
const ctx = makeCtx()
|
||||
routeMessage({ type: 'assistant', uuid: 'msg-1', message: { content: [{ type: 'redacted_thinking' }] } }, ctx)
|
||||
const part = ctx.oc.find(e => e.event === 'message.part.updated')
|
||||
const p = part?.properties.part as Record<string, unknown>
|
||||
expect(p.type).toBe('reasoning')
|
||||
expect(p.redacted).toBe(true)
|
||||
})
|
||||
|
||||
test('pushes to buffer when uuid present', () => {
|
||||
const ctx = makeCtx()
|
||||
routeMessage({ type: 'assistant', uuid: 'msg-1', message: { content: [] } }, ctx)
|
||||
expect(ctx.buf.length).toBe(1)
|
||||
expect(ctx.buf[0].uuid).toBe('msg-1')
|
||||
})
|
||||
|
||||
test('API error content emits message.updated with role=error + error object', () => {
|
||||
const ctx = makeCtx()
|
||||
routeMessage({
|
||||
type: 'assistant',
|
||||
uuid: 'err-1',
|
||||
model: 'claude-sonnet-4',
|
||||
provider_id: 'anthropic',
|
||||
message: {
|
||||
content: [{ type: 'text', text: 'API Error: 429 {"error":{"code":"1305","message":"rate limited"}}' }],
|
||||
},
|
||||
}, ctx)
|
||||
const updated = ctx.oc.filter(e => e.event === 'message.updated')
|
||||
expect(updated.length).toBe(1)
|
||||
const info = updated[0].properties.info as Record<string, unknown>
|
||||
expect(info.role).toBe('assistant')
|
||||
expect(info.error).toMatchObject({ name: 'APIError', data: { statusCode: 429, isRetryable: true } })
|
||||
expect((info.error as any).data.message).toContain('rate limited')
|
||||
expect(ctx.buf[0].role).toBe('assistant')
|
||||
expect(ctx.buf[0].error).toBeDefined()
|
||||
})
|
||||
|
||||
test('CoStrict API error content also detected', () => {
|
||||
const ctx = makeCtx()
|
||||
routeMessage({
|
||||
type: 'assistant',
|
||||
uuid: 'err-2',
|
||||
message: {
|
||||
content: [{ type: 'text', text: 'CoStrict API Error: something broke' }],
|
||||
},
|
||||
}, ctx)
|
||||
const info = (ctx.oc.find(e => e.event === 'message.updated')?.properties as Record<string, unknown>)?.info as Record<string, unknown>
|
||||
expect(info.role).toBe('assistant')
|
||||
expect((info.error as any).data.message).toBe('something broke')
|
||||
})
|
||||
})
|
||||
|
||||
describe('routeMessage — result', () => {
|
||||
test('success emits session.result only', () => {
|
||||
const ctx = makeCtx()
|
||||
routeMessage({ type: 'result', subtype: 'success', cost_usd: 0.05, usage: { input_tokens: 100, output_tokens: 50 } }, ctx)
|
||||
expect(ctx.oc.filter(e => e.event === 'session.result').length).toBe(1)
|
||||
expect(ctx.oc.filter(e => e.event === 'session.error').length).toBe(0)
|
||||
})
|
||||
|
||||
test('error subtype emits session.error', () => {
|
||||
const ctx = makeCtx()
|
||||
routeMessage({ type: 'result', subtype: 'error_max_turns', is_error: true, errors: [{ message: 'Max turns reached' }] }, ctx)
|
||||
const errors = ctx.oc.filter(e => e.event === 'session.error')
|
||||
expect(errors.length).toBe(1)
|
||||
expect(errors[0].properties.error).toMatchObject({ subtype: 'error_max_turns', level: 'error', message: 'Max turns reached' })
|
||||
})
|
||||
|
||||
test('accumulates cost and tokens', () => {
|
||||
const ctx = makeCtx()
|
||||
routeMessage({ type: 'result', subtype: 'success', cost_usd: 0.05, usage: { input_tokens: 100, output_tokens: 50 } }, ctx)
|
||||
expect(ctx.addCost).toHaveBeenCalledWith(0.05)
|
||||
expect(ctx.addInputTokens).toHaveBeenCalledWith(100)
|
||||
expect(ctx.addOutputTokens).toHaveBeenCalledWith(50)
|
||||
})
|
||||
})
|
||||
|
||||
describe('routeMessage — init', () => {
|
||||
test('init response emits session.updated', () => {
|
||||
const ctx = makeCtx({
|
||||
getStatus: () => 'starting',
|
||||
getInitRequestId: () => 'init-req-1',
|
||||
})
|
||||
routeMessage({
|
||||
type: 'control_response',
|
||||
response: {
|
||||
subtype: 'success',
|
||||
request_id: 'init-req-1',
|
||||
response: { models: [{ value: 'claude-sonnet-4' }], account: { apiProvider: 'anthropic' } },
|
||||
},
|
||||
}, ctx)
|
||||
const updated = ctx.oc.filter(e => e.event === 'session.updated')
|
||||
expect(updated.length).toBe(1)
|
||||
expect(updated[0].properties).toMatchObject({ status: 'running', model: 'claude-sonnet-4', providerID: 'anthropic' })
|
||||
})
|
||||
})
|
||||
|
||||
describe('routeMessage — user', () => {
|
||||
test('emits message.updated', () => {
|
||||
const ctx = makeCtx()
|
||||
routeMessage({ type: 'user', uuid: 'user-1', content: 'Hello' }, ctx)
|
||||
expect(ctx.oc.length).toBe(1)
|
||||
expect(ctx.oc[0].event).toBe('message.updated')
|
||||
expect(ctx.oc[0].properties.info).toMatchObject({ role: 'user' })
|
||||
})
|
||||
|
||||
test('replay is dropped', () => {
|
||||
const ctx = makeCtx()
|
||||
routeMessage({ type: 'user', content: 'replay', isReplay: true }, ctx)
|
||||
expect(ctx.oc.length).toBe(0)
|
||||
})
|
||||
|
||||
test('local-command-stdout is dropped', () => {
|
||||
const ctx = makeCtx()
|
||||
routeMessage({ type: 'user', content: '<local-command-stdout>out</local-command-stdout>' }, ctx)
|
||||
expect(ctx.oc.length).toBe(0)
|
||||
})
|
||||
})
|
||||
|
||||
describe('routeMessage — unknown type', () => {
|
||||
test('unknown message type is silently dropped', () => {
|
||||
const ctx = makeCtx()
|
||||
routeMessage({ type: 'unknown_type', data: 'test' }, ctx)
|
||||
expect(ctx.oc.length).toBe(0)
|
||||
})
|
||||
})
|
||||
341
src/server/__tests__/streamStateTracker.test.ts
Normal file
341
src/server/__tests__/streamStateTracker.test.ts
Normal file
|
|
@ -0,0 +1,341 @@
|
|||
import { describe, expect, test, beforeEach } from 'bun:test'
|
||||
import { processStreamEvent, resetAllState } from '../streamStateTracker.js'
|
||||
|
||||
describe('processStreamEvent', () => {
|
||||
const sessionID = 'test-session-1'
|
||||
|
||||
beforeEach(() => {
|
||||
resetAllState(sessionID)
|
||||
})
|
||||
|
||||
test('message_start emits message.updated + step-start', () => {
|
||||
const events = processStreamEvent(sessionID, {
|
||||
type: 'message_start',
|
||||
message: {
|
||||
id: 'msg-1',
|
||||
model: 'claude-sonnet-4-20250514',
|
||||
usage: { input_tokens: 100, cache_read_input_tokens: 50, cache_creation_input_tokens: 10 },
|
||||
},
|
||||
})
|
||||
|
||||
expect(events.length).toBe(2)
|
||||
expect(events[0].type).toBe('message.updated')
|
||||
expect(events[0].properties).toMatchObject({
|
||||
sessionID,
|
||||
info: { role: 'assistant', modelID: 'claude-sonnet-4-20250514' },
|
||||
})
|
||||
expect(events[1].type).toBe('message.part.updated')
|
||||
expect(events[1].properties).toMatchObject({
|
||||
sessionID,
|
||||
part: { type: 'step-start' },
|
||||
})
|
||||
})
|
||||
|
||||
test('content_block_start (text) emits text part', () => {
|
||||
const events = processStreamEvent(sessionID, {
|
||||
type: 'content_block_start',
|
||||
index: 0,
|
||||
content_block: { type: 'text', text: '' },
|
||||
})
|
||||
|
||||
expect(events.length).toBe(1)
|
||||
expect(events[0].type).toBe('message.part.updated')
|
||||
expect(events[0].properties.part).toMatchObject({ type: 'text', text: '' })
|
||||
})
|
||||
|
||||
test('content_block_start (thinking) emits reasoning part', () => {
|
||||
const events = processStreamEvent(sessionID, {
|
||||
type: 'content_block_start',
|
||||
index: 0,
|
||||
content_block: { type: 'thinking', thinking: '' },
|
||||
})
|
||||
|
||||
expect(events.length).toBe(1)
|
||||
expect(events[0].properties.part).toMatchObject({ type: 'reasoning', redacted: false })
|
||||
})
|
||||
|
||||
test('content_block_start (redacted_thinking) emits reasoning part with redacted=true', () => {
|
||||
const events = processStreamEvent(sessionID, {
|
||||
type: 'content_block_start',
|
||||
index: 0,
|
||||
content_block: { type: 'redacted_thinking' },
|
||||
})
|
||||
|
||||
expect(events.length).toBe(1)
|
||||
expect(events[0].properties.part).toMatchObject({ type: 'reasoning', redacted: true })
|
||||
})
|
||||
|
||||
test('content_block_start (tool_use) emits pending tool part', () => {
|
||||
const events = processStreamEvent(sessionID, {
|
||||
type: 'content_block_start',
|
||||
index: 0,
|
||||
content_block: { type: 'tool_use', id: 'tool-1', name: 'Bash' },
|
||||
})
|
||||
|
||||
expect(events.length).toBe(1)
|
||||
expect(events[0].properties.part).toMatchObject({
|
||||
type: 'tool',
|
||||
callID: 'tool-1',
|
||||
tool: 'bash',
|
||||
state: { status: 'pending', input: {} },
|
||||
})
|
||||
})
|
||||
|
||||
test('content_block_delta (text_delta) emits part delta', () => {
|
||||
processStreamEvent(sessionID, {
|
||||
type: 'content_block_start',
|
||||
index: 0,
|
||||
content_block: { type: 'text', text: '' },
|
||||
})
|
||||
|
||||
const events = processStreamEvent(sessionID, {
|
||||
type: 'content_block_delta',
|
||||
index: 0,
|
||||
delta: { type: 'text_delta', text: 'Hello' },
|
||||
})
|
||||
|
||||
expect(events.length).toBe(1)
|
||||
expect(events[0].type).toBe('message.part.delta')
|
||||
expect(events[0].properties).toMatchObject({ field: 'text', delta: 'Hello' })
|
||||
})
|
||||
|
||||
test('content_block_delta (thinking_delta) emits part delta', () => {
|
||||
processStreamEvent(sessionID, {
|
||||
type: 'content_block_start',
|
||||
index: 0,
|
||||
content_block: { type: 'thinking', thinking: '' },
|
||||
})
|
||||
|
||||
const events = processStreamEvent(sessionID, {
|
||||
type: 'content_block_delta',
|
||||
index: 0,
|
||||
delta: { type: 'thinking_delta', thinking: 'Let me think...' },
|
||||
})
|
||||
|
||||
expect(events[0].properties).toMatchObject({ field: 'text', delta: 'Let me think...' })
|
||||
})
|
||||
|
||||
test('content_block_delta (input_json_delta) accumulates and emits', () => {
|
||||
processStreamEvent(sessionID, {
|
||||
type: 'content_block_start',
|
||||
index: 0,
|
||||
content_block: { type: 'tool_use', id: 'tool-1', name: 'Bash' },
|
||||
})
|
||||
|
||||
const events1 = processStreamEvent(sessionID, {
|
||||
type: 'content_block_delta',
|
||||
index: 0,
|
||||
delta: { type: 'input_json_delta', partial_json: '{"command":"ls' },
|
||||
})
|
||||
expect(events1[0].properties.delta).toBe('{"command":"ls')
|
||||
|
||||
const events2 = processStreamEvent(sessionID, {
|
||||
type: 'content_block_delta',
|
||||
index: 0,
|
||||
delta: { type: 'input_json_delta', partial_json: '"}' },
|
||||
})
|
||||
expect(events2[0].properties.delta).toBe('"}')
|
||||
})
|
||||
|
||||
test('content_block_stop (tool_use) emits running state with parsed input', () => {
|
||||
processStreamEvent(sessionID, {
|
||||
type: 'content_block_start',
|
||||
index: 0,
|
||||
content_block: { type: 'tool_use', id: 'tool-1', name: 'Read' },
|
||||
})
|
||||
processStreamEvent(sessionID, {
|
||||
type: 'content_block_delta',
|
||||
index: 0,
|
||||
delta: { type: 'input_json_delta', partial_json: '{"file_path":"/test.ts"}' },
|
||||
})
|
||||
|
||||
const events = processStreamEvent(sessionID, {
|
||||
type: 'content_block_stop',
|
||||
index: 0,
|
||||
})
|
||||
|
||||
expect(events.length).toBe(1)
|
||||
expect((events[0].properties.part as Record<string, unknown>)?.state).toMatchObject({
|
||||
status: 'running',
|
||||
input: { filePath: '/test.ts' },
|
||||
})
|
||||
})
|
||||
|
||||
test('content_block_stop (tool_use) handles corrupted JSON gracefully', () => {
|
||||
processStreamEvent(sessionID, {
|
||||
type: 'content_block_start',
|
||||
index: 0,
|
||||
content_block: { type: 'tool_use', id: 'tool-1', name: 'Bash' },
|
||||
})
|
||||
processStreamEvent(sessionID, {
|
||||
type: 'content_block_delta',
|
||||
index: 0,
|
||||
delta: { type: 'input_json_delta', partial_json: '{broken' },
|
||||
})
|
||||
|
||||
const events = processStreamEvent(sessionID, {
|
||||
type: 'content_block_stop',
|
||||
index: 0,
|
||||
})
|
||||
|
||||
expect((events[0].properties.part as Record<string, unknown>)?.state).toMatchObject({ input: {} })
|
||||
})
|
||||
|
||||
test('content_block_stop (text) emits part with text and time.end', () => {
|
||||
processStreamEvent(sessionID, {
|
||||
type: 'content_block_start',
|
||||
index: 0,
|
||||
content_block: { type: 'text', text: '' },
|
||||
})
|
||||
processStreamEvent(sessionID, {
|
||||
type: 'content_block_delta',
|
||||
index: 0,
|
||||
delta: { type: 'text_delta', text: 'hello' },
|
||||
})
|
||||
|
||||
const events = processStreamEvent(sessionID, {
|
||||
type: 'content_block_stop',
|
||||
index: 0,
|
||||
})
|
||||
|
||||
expect(events.length).toBe(1)
|
||||
expect(events[0].type).toBe('message.part.updated')
|
||||
const part = events[0].properties.part as Record<string, unknown>
|
||||
expect(part.type).toBe('text')
|
||||
expect(part.text).toBe('hello')
|
||||
expect((part.time as Record<string, unknown>).end).toBeDefined()
|
||||
})
|
||||
|
||||
test('message_delta extracts usage and stopReason (no event emitted)', () => {
|
||||
const events = processStreamEvent(sessionID, {
|
||||
type: 'message_delta',
|
||||
delta: { stop_reason: 'end_turn' },
|
||||
usage: { output_tokens: 42 },
|
||||
})
|
||||
|
||||
expect(events.length).toBe(0)
|
||||
})
|
||||
|
||||
test('message_stop emits step-finish with tokens', () => {
|
||||
processStreamEvent(sessionID, {
|
||||
type: 'message_start',
|
||||
message: {
|
||||
model: 'claude-sonnet-4-20250514',
|
||||
usage: { input_tokens: 100, cache_read_input_tokens: 50, cache_creation_input_tokens: 10 },
|
||||
},
|
||||
})
|
||||
processStreamEvent(sessionID, {
|
||||
type: 'message_delta',
|
||||
delta: { stop_reason: 'end_turn' },
|
||||
usage: { output_tokens: 42 },
|
||||
})
|
||||
|
||||
const events = processStreamEvent(sessionID, { type: 'message_stop' })
|
||||
|
||||
expect(events.length).toBe(2)
|
||||
expect(events[0].type).toBe('message.part.updated')
|
||||
expect(events[0].properties.part).toMatchObject({
|
||||
type: 'step-finish',
|
||||
reason: 'end_turn',
|
||||
tokens: {
|
||||
input: 100,
|
||||
output: 42,
|
||||
reasoning: 0,
|
||||
cache: { read: 50, write: 10 },
|
||||
},
|
||||
})
|
||||
expect(events[1].type).toBe('message.updated')
|
||||
expect((events[1].properties.info as Record<string, unknown>).time).toMatchObject({ completed: expect.any(Number) })
|
||||
})
|
||||
|
||||
test('full text flow: start → delta → stop → message_delta → message_stop', () => {
|
||||
const allEvents: Array<{ type: string; properties: Record<string, unknown> }> = []
|
||||
|
||||
const collect = (evts: Array<{ type: string; properties: Record<string, unknown> }>) => {
|
||||
allEvents.push(...evts)
|
||||
}
|
||||
|
||||
collect(processStreamEvent(sessionID, {
|
||||
type: 'message_start',
|
||||
message: { model: 'test-model', usage: { input_tokens: 10 } },
|
||||
}))
|
||||
collect(processStreamEvent(sessionID, {
|
||||
type: 'content_block_start',
|
||||
index: 0,
|
||||
content_block: { type: 'text', text: '' },
|
||||
}))
|
||||
collect(processStreamEvent(sessionID, {
|
||||
type: 'content_block_delta',
|
||||
index: 0,
|
||||
delta: { type: 'text_delta', text: 'Hello ' },
|
||||
}))
|
||||
collect(processStreamEvent(sessionID, {
|
||||
type: 'content_block_delta',
|
||||
index: 0,
|
||||
delta: { type: 'text_delta', text: 'world' },
|
||||
}))
|
||||
collect(processStreamEvent(sessionID, {
|
||||
type: 'content_block_stop',
|
||||
index: 0,
|
||||
}))
|
||||
collect(processStreamEvent(sessionID, {
|
||||
type: 'message_delta',
|
||||
delta: { stop_reason: 'end_turn' },
|
||||
usage: { output_tokens: 5 },
|
||||
}))
|
||||
collect(processStreamEvent(sessionID, { type: 'message_stop' }))
|
||||
|
||||
const types = allEvents.map(e => e.type)
|
||||
expect(types).toEqual([
|
||||
'message.updated',
|
||||
'message.part.updated',
|
||||
'message.part.updated',
|
||||
'message.part.delta',
|
||||
'message.part.delta',
|
||||
'message.part.updated',
|
||||
'message.part.updated',
|
||||
'message.updated',
|
||||
])
|
||||
|
||||
const deltas = allEvents.filter(e => e.type === 'message.part.delta') as Array<{ type: string; properties: Record<string, unknown> }>
|
||||
expect(deltas[0].properties.delta).toBe('Hello ')
|
||||
expect(deltas[1].properties.delta).toBe('world')
|
||||
})
|
||||
|
||||
test('multi-step (tool loop): second message_start creates new step', () => {
|
||||
processStreamEvent(sessionID, { type: 'message_stop' })
|
||||
|
||||
const events = processStreamEvent(sessionID, {
|
||||
type: 'message_start',
|
||||
message: { model: 'test', usage: { input_tokens: 50 } },
|
||||
})
|
||||
|
||||
const stepStarts = events.filter(e =>
|
||||
e.type === 'message.part.updated' &&
|
||||
(e.properties.part as Record<string, unknown>)?.type === 'step-start',
|
||||
)
|
||||
expect(stepStarts.length).toBe(1)
|
||||
})
|
||||
|
||||
test('unknown event type returns empty array', () => {
|
||||
const events = processStreamEvent(sessionID, { type: 'ping' })
|
||||
expect(events).toEqual([])
|
||||
})
|
||||
|
||||
test('content_block_delta with missing block state returns empty', () => {
|
||||
const events = processStreamEvent(sessionID, {
|
||||
type: 'content_block_delta',
|
||||
index: 99,
|
||||
delta: { type: 'text_delta', text: 'orphan' },
|
||||
})
|
||||
expect(events).toEqual([])
|
||||
})
|
||||
|
||||
test('content_block_start with no content_block returns empty', () => {
|
||||
const events = processStreamEvent(sessionID, {
|
||||
type: 'content_block_start',
|
||||
index: 0,
|
||||
})
|
||||
expect(events).toEqual([])
|
||||
})
|
||||
})
|
||||
|
|
@ -1,5 +1,5 @@
|
|||
import { describe, expect, test, beforeEach } from 'bun:test'
|
||||
import { readSessionMessages, readSessionTodos, clearPathCache } from '../transcriptReader.js'
|
||||
import { readSessionMessages, readSessionTodos, readSessionTasks, clearPathCache } from '../transcriptReader.js'
|
||||
|
||||
describe('transcriptReader', () => {
|
||||
beforeEach(() => {
|
||||
|
|
@ -20,6 +20,13 @@ describe('transcriptReader', () => {
|
|||
expect(result).toEqual([])
|
||||
})
|
||||
|
||||
test('readSessionTasks returns empty for non-existent session', async () => {
|
||||
const result = await readSessionTasks({
|
||||
sessionId: 'nonexistent-session-id-12345',
|
||||
})
|
||||
expect(result).toEqual([])
|
||||
})
|
||||
|
||||
test('readSessionMessages respects limit', async () => {
|
||||
const result = await readSessionMessages({
|
||||
sessionId: 'nonexistent-session-id-12345',
|
||||
|
|
|
|||
127
src/server/__tests__/transcriptReaderParts.test.ts
Normal file
127
src/server/__tests__/transcriptReaderParts.test.ts
Normal file
|
|
@ -0,0 +1,127 @@
|
|||
import { describe, expect, test } from 'bun:test'
|
||||
import { decomposeMessageToParts, type SessionMessage } from '../transcriptReader.js'
|
||||
|
||||
describe('decomposeMessageToParts', () => {
|
||||
test('assistant with content blocks produces parts', () => {
|
||||
const msg: SessionMessage = {
|
||||
uuid: 'msg-1',
|
||||
type: 'assistant',
|
||||
role: 'assistant',
|
||||
content: [
|
||||
{ type: 'text', text: 'Hello' },
|
||||
{ type: 'thinking', thinking: 'Let me think...' },
|
||||
{ type: 'tool_use', id: 'tool-1', name: 'Bash', input: { command: 'ls' } },
|
||||
],
|
||||
timestamp: Date.now(),
|
||||
parent_uuid: null,
|
||||
}
|
||||
|
||||
const result = decomposeMessageToParts(msg)
|
||||
expect(result.parts).toBeDefined()
|
||||
expect(result.parts!.length).toBe(3)
|
||||
expect(result.parts![0].type).toBe('text')
|
||||
expect(result.parts![1].type).toBe('reasoning')
|
||||
expect(result.parts![2].type).toBe('tool')
|
||||
const toolPart = result.parts![2] as { tool: string; callID: string }
|
||||
expect(toolPart.tool).toBe('bash')
|
||||
expect(toolPart.callID).toBe('tool-1')
|
||||
})
|
||||
|
||||
test('assistant with redacted_thinking produces reasoning part', () => {
|
||||
const msg: SessionMessage = {
|
||||
uuid: 'msg-1',
|
||||
type: 'assistant',
|
||||
role: 'assistant',
|
||||
content: [{ type: 'redacted_thinking' }],
|
||||
timestamp: Date.now(),
|
||||
parent_uuid: null,
|
||||
}
|
||||
|
||||
const result = decomposeMessageToParts(msg)
|
||||
expect(result.parts!.length).toBe(1)
|
||||
const p = result.parts![0] as { redacted?: boolean }
|
||||
expect(p.redacted).toBe(true)
|
||||
})
|
||||
|
||||
test('user with tool_result content produces tool-result part', () => {
|
||||
const msg: SessionMessage = {
|
||||
uuid: 'msg-1',
|
||||
type: 'user',
|
||||
role: 'user',
|
||||
content: [
|
||||
{ type: 'text', text: 'Here is the result' },
|
||||
{ type: 'tool_result', tool_use_id: 'tool-1', content: 'file contents' },
|
||||
],
|
||||
timestamp: Date.now(),
|
||||
parent_uuid: null,
|
||||
}
|
||||
|
||||
const result = decomposeMessageToParts(msg)
|
||||
expect(result.parts!.length).toBe(2)
|
||||
expect(result.parts![0].type).toBe('text')
|
||||
expect(result.parts![1].type).toBe('tool-result')
|
||||
})
|
||||
|
||||
test('message without array content returns unchanged', () => {
|
||||
const msg: SessionMessage = {
|
||||
uuid: 'msg-1',
|
||||
type: 'assistant',
|
||||
role: 'assistant',
|
||||
content: 'plain text',
|
||||
timestamp: Date.now(),
|
||||
parent_uuid: null,
|
||||
}
|
||||
|
||||
const result = decomposeMessageToParts(msg)
|
||||
expect(result.parts).toBeUndefined()
|
||||
})
|
||||
|
||||
test('message with existing parts returns unchanged', () => {
|
||||
const msg: SessionMessage = {
|
||||
uuid: 'msg-1',
|
||||
type: 'assistant',
|
||||
role: 'assistant',
|
||||
content: [],
|
||||
timestamp: Date.now(),
|
||||
parent_uuid: null,
|
||||
parts: [{ type: 'text', id: 'p-1', text: 'existing' }],
|
||||
}
|
||||
|
||||
const result = decomposeMessageToParts(msg)
|
||||
expect(result.parts!.length).toBe(1)
|
||||
})
|
||||
|
||||
test('system message returns unchanged', () => {
|
||||
const msg: SessionMessage = {
|
||||
uuid: 'msg-1',
|
||||
type: 'system',
|
||||
role: 'system',
|
||||
content: 'some info',
|
||||
timestamp: Date.now(),
|
||||
parent_uuid: null,
|
||||
}
|
||||
|
||||
const result = decomposeMessageToParts(msg)
|
||||
expect(result.parts).toBeUndefined()
|
||||
})
|
||||
|
||||
test('tool name normalization: Write → edit, Agent → task', () => {
|
||||
const msg: SessionMessage = {
|
||||
uuid: 'msg-1',
|
||||
type: 'assistant',
|
||||
role: 'assistant',
|
||||
content: [
|
||||
{ type: 'tool_use', id: 't1', name: 'Write', input: { file_path: '/test.ts' } },
|
||||
{ type: 'tool_use', id: 't2', name: 'Agent', input: { prompt: 'do stuff' } },
|
||||
],
|
||||
timestamp: Date.now(),
|
||||
parent_uuid: null,
|
||||
}
|
||||
|
||||
const result = decomposeMessageToParts(msg)
|
||||
const tool1 = result.parts![0] as { tool: string }
|
||||
const tool2 = result.parts![1] as { tool: string }
|
||||
expect(tool1.tool).toBe('edit')
|
||||
expect(tool2.tool).toBe('task')
|
||||
})
|
||||
})
|
||||
|
|
@ -13,6 +13,7 @@ import {
|
|||
routeMessage,
|
||||
type StdoutMessage,
|
||||
type MessageRouterCtx,
|
||||
type SubagentInfo,
|
||||
} from './sessionMessageRouter.js'
|
||||
import type {
|
||||
SessionBusyStatus,
|
||||
|
|
@ -85,6 +86,16 @@ export class SessionHandle implements DisposableChildProcess {
|
|||
private _titleGenerationAttempted = false
|
||||
private _firstPromptContent?: string
|
||||
private _messageBuffer: SessionMessage[] = []
|
||||
private _tombstonedUuids = new Set<string>()
|
||||
private _activeSubagents = new Map<string, SubagentInfo>()
|
||||
|
||||
get tombstonedUuids(): ReadonlySet<string> {
|
||||
return this._tombstonedUuids
|
||||
}
|
||||
|
||||
get activeSubagents(): ReadonlyMap<string, SubagentInfo> {
|
||||
return this._activeSubagents
|
||||
}
|
||||
|
||||
get status(): SessionState {
|
||||
return this._status
|
||||
|
|
@ -265,7 +276,8 @@ export class SessionHandle implements DisposableChildProcess {
|
|||
this._status = 'stopped'
|
||||
this._busyStatus = { type: 'idle' }
|
||||
this.emitBusyStatus()
|
||||
this.emitEvent('deleted', {
|
||||
this.emitOpencodeEvent('session.deleted', {
|
||||
sessionID: this.sessionId,
|
||||
status: 'stopped',
|
||||
exit_code: code,
|
||||
signal: signal ?? null,
|
||||
|
|
@ -310,7 +322,8 @@ export class SessionHandle implements DisposableChildProcess {
|
|||
this.initResolve = null
|
||||
this.initReject = null
|
||||
this._status = 'stopped'
|
||||
this.emitEvent('deleted', {
|
||||
this.emitOpencodeEvent('session.deleted', {
|
||||
sessionID: this.sessionId,
|
||||
status: 'stopped',
|
||||
reason: 'init_timeout',
|
||||
})
|
||||
|
|
@ -401,11 +414,8 @@ export class SessionHandle implements DisposableChildProcess {
|
|||
properties: Record<string, unknown>,
|
||||
): void {
|
||||
if (this.opts.silent) return
|
||||
if (properties.type === event && 'properties' in properties) {
|
||||
this.eventBus.publish(event, properties as Record<string, unknown>)
|
||||
return
|
||||
}
|
||||
this.eventBus.publish(event, {
|
||||
_native_opencode: true,
|
||||
session_id: properties.sessionID,
|
||||
...properties,
|
||||
})
|
||||
|
|
@ -520,12 +530,22 @@ export class SessionHandle implements DisposableChildProcess {
|
|||
emitMessage: msg => this.emitMessage(msg),
|
||||
writeStdin: data => this.writeStdin(data),
|
||||
pushBufferMessage: msg => this.pushMessage(msg),
|
||||
addTombstonedUuid: uuid => this._tombstonedUuids.add(uuid),
|
||||
getActiveSubagents: () => this._activeSubagents,
|
||||
registerSubagent: info => this._activeSubagents.set(info.agentId, info),
|
||||
unregisterSubagent: agentId => this._activeSubagents.delete(agentId),
|
||||
}
|
||||
}
|
||||
|
||||
setTitle(title: string): void {
|
||||
this._title = title
|
||||
this.emitEvent('ready', this.getInfo())
|
||||
this.emitOpencodeEvent('session.updated', {
|
||||
sessionID: this.sessionId,
|
||||
status: this._status,
|
||||
model: this._model,
|
||||
title: this._title,
|
||||
providerID: this._providerId,
|
||||
})
|
||||
}
|
||||
|
||||
async sendControlRequest(
|
||||
|
|
@ -596,13 +616,16 @@ export class SessionHandle implements DisposableChildProcess {
|
|||
parent_tool_use_id: null,
|
||||
})
|
||||
|
||||
this.emitEvent('message', {
|
||||
type: 'user',
|
||||
content,
|
||||
uuid,
|
||||
session_id: this.sessionId,
|
||||
model: this._model ?? '',
|
||||
provider_id: this._providerId ?? '',
|
||||
this.emitOpencodeEvent('message.updated', {
|
||||
sessionID: this.sessionId,
|
||||
info: {
|
||||
id: uuid,
|
||||
role: 'user',
|
||||
modelID: this._model ?? '',
|
||||
providerID: this._providerId ?? '',
|
||||
time: { created: Date.now() },
|
||||
parentID: null,
|
||||
},
|
||||
})
|
||||
|
||||
this.pushMessage({
|
||||
|
|
@ -645,6 +668,18 @@ export class SessionHandle implements DisposableChildProcess {
|
|||
this._busyStatus = { type: 'idle' }
|
||||
this.emitBusyStatus()
|
||||
|
||||
// Emit a result event with is_interrupted so the adapter layer can
|
||||
// generate the full session.idle event chain that external clients
|
||||
// expect. Without this, if the child is killed before it yields its
|
||||
// own result, the adapter never calls adaptResultEvent and the
|
||||
// client never receives session.idle.
|
||||
this.emitOpencodeEvent('session.result', {
|
||||
sessionID: this.sessionId,
|
||||
subtype: 'error_during_execution',
|
||||
is_error: true,
|
||||
is_interrupted: true,
|
||||
})
|
||||
|
||||
if (!this.promptResolve) return
|
||||
|
||||
await new Promise<void>(resolve => {
|
||||
|
|
|
|||
|
|
@ -1,8 +1,106 @@
|
|||
import { randomUUID } from 'crypto'
|
||||
import { jsonStringify } from '../utils/slowOperations.js'
|
||||
import { processStreamEvent, hasActiveStream, consumeStreamedTurn, getCompletedToolInfo, registerAgentSession } from './streamStateTracker.js'
|
||||
import type { ControlChannel } from './sessionControlChannel.js'
|
||||
import type { SessionBusyStatus, SessionState, InitData, PendingPermission, PendingQuestion } from './types.js'
|
||||
import type { SessionMessage } from './transcriptReader.js'
|
||||
|
||||
const API_ERROR_PREFIX_RE = /^API Error:\s*/
|
||||
const COSTRICT_API_ERROR_PREFIX_RE = /^CoStrict API Error:\s*/
|
||||
|
||||
type SubagentToolState = {
|
||||
emittedToolCount: number
|
||||
assistantMessageID: string
|
||||
toolPartIDs: Map<number, string>
|
||||
mainPartID: string
|
||||
mainMessageID: string
|
||||
mainToolUseID: string
|
||||
progressLines: string[]
|
||||
}
|
||||
|
||||
const subagentToolState = new Map<string, SubagentToolState>()
|
||||
|
||||
function isApiErrorContent(content: unknown): boolean {
|
||||
if (typeof content === 'string') {
|
||||
return API_ERROR_PREFIX_RE.test(content) || COSTRICT_API_ERROR_PREFIX_RE.test(content)
|
||||
}
|
||||
if (!Array.isArray(content)) return false
|
||||
for (const block of content) {
|
||||
if (!block || typeof block !== 'object') continue
|
||||
const b = block as Record<string, unknown>
|
||||
if (b.type === 'text' && typeof b.text === 'string') {
|
||||
const text = b.text as string
|
||||
if (API_ERROR_PREFIX_RE.test(text) || COSTRICT_API_ERROR_PREFIX_RE.test(text)) return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
function buildErrorFromContent(content: unknown): { name: string; data: { message: string; statusCode?: number; isRetryable?: boolean } } {
|
||||
let raw = ''
|
||||
if (typeof content === 'string') {
|
||||
raw = content.replace(API_ERROR_PREFIX_RE, '').replace(COSTRICT_API_ERROR_PREFIX_RE, '')
|
||||
} else if (Array.isArray(content)) {
|
||||
for (const block of content) {
|
||||
if (!block || typeof block !== 'object') continue
|
||||
const b = block as Record<string, unknown>
|
||||
if (b.type === 'text' && typeof b.text === 'string') {
|
||||
raw = (b.text as string).replace(API_ERROR_PREFIX_RE, '').replace(COSTRICT_API_ERROR_PREFIX_RE, '')
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
const statusMatch = raw.match(/^(\d{3})\s+/)
|
||||
const statusCode = statusMatch ? parseInt(statusMatch[1]!, 10) : undefined
|
||||
const isRetryable = statusCode === 429 || statusCode === 503 || statusCode === 529
|
||||
let message = raw
|
||||
try {
|
||||
const jsonStart = raw.indexOf('{')
|
||||
if (jsonStart !== -1) {
|
||||
const parsed = JSON.parse(raw.slice(jsonStart))
|
||||
if (parsed?.error?.message) {
|
||||
message = parsed.error.message
|
||||
} else if (typeof parsed?.message === 'string') {
|
||||
message = parsed.message
|
||||
}
|
||||
}
|
||||
} catch {}
|
||||
return { name: 'APIError', data: { message, statusCode, isRetryable } }
|
||||
}
|
||||
|
||||
function extractToolResultContent(content: unknown): string {
|
||||
if (typeof content === 'string') return content
|
||||
if (!Array.isArray(content)) return ''
|
||||
const parts: string[] = []
|
||||
for (const block of content) {
|
||||
if (!block || typeof block !== 'object') continue
|
||||
const b = block as Record<string, unknown>
|
||||
if (b.type === 'text' && typeof b.text === 'string') {
|
||||
parts.push(b.text)
|
||||
} else if (b.type === 'image') {
|
||||
const src = b.source as Record<string, unknown> | undefined
|
||||
if (typeof src?.data === 'string') {
|
||||
parts.push(`[image: ${src.type}]`)
|
||||
}
|
||||
}
|
||||
}
|
||||
return parts.join('\n')
|
||||
}
|
||||
|
||||
function getActiveSubagentId(ctx: MessageRouterCtx): string | undefined {
|
||||
const subs = ctx.getActiveSubagents()
|
||||
if (subs.size === 0) return undefined
|
||||
return subs.keys().next().value as string | undefined
|
||||
}
|
||||
|
||||
export type SubagentInfo = {
|
||||
agentId: string
|
||||
agentType?: string
|
||||
description?: string
|
||||
prompt?: string
|
||||
toolUseId?: string
|
||||
}
|
||||
|
||||
export type StdoutMessage = {
|
||||
type: string
|
||||
[key: string]: unknown
|
||||
|
|
@ -27,6 +125,10 @@ export type MessageRouterCtx = {
|
|||
setLastMessageUuid(uuid: string): void
|
||||
setLastActiveAt(ts: number): void
|
||||
|
||||
getActiveSubagents(): Map<string, SubagentInfo>
|
||||
registerSubagent(info: SubagentInfo): void
|
||||
unregisterSubagent(agentId: string): void
|
||||
|
||||
getControlChannel(): ControlChannel
|
||||
getPendingPermissions(): Map<string, PendingPermission>
|
||||
getPendingQuestions(): Map<string, PendingQuestion>
|
||||
|
|
@ -44,6 +146,7 @@ export type MessageRouterCtx = {
|
|||
emitMessage(msg: StdoutMessage): void
|
||||
writeStdin(data: string): void
|
||||
pushBufferMessage(msg: SessionMessage): void
|
||||
addTombstonedUuid(uuid: string): void
|
||||
}
|
||||
|
||||
export function routeMessage(msg: StdoutMessage, ctx: MessageRouterCtx): void {
|
||||
|
|
@ -86,12 +189,23 @@ export function routeMessage(msg: StdoutMessage, ctx: MessageRouterCtx): void {
|
|||
break
|
||||
}
|
||||
case 'stream_event': {
|
||||
ctx.setLastActiveAt(Date.now())
|
||||
ctx.emitEvent('stream_event', msg)
|
||||
handleStreamEvent(msg, ctx)
|
||||
break
|
||||
}
|
||||
case 'system': {
|
||||
ctx.emitEvent('message', msg)
|
||||
handleSystemMessage(msg, ctx)
|
||||
break
|
||||
}
|
||||
case 'attachment': {
|
||||
handleAttachment(msg, ctx)
|
||||
break
|
||||
}
|
||||
case 'progress': {
|
||||
handleProgress(msg, ctx)
|
||||
break
|
||||
}
|
||||
case 'tombstone': {
|
||||
handleTombstone(msg, ctx)
|
||||
break
|
||||
}
|
||||
default: {
|
||||
|
|
@ -113,10 +227,11 @@ function handleInitResponse(response: Record<string, unknown>, ctx: MessageRoute
|
|||
ctx.setStatus('running')
|
||||
ctx.setLastActiveAt(Date.now())
|
||||
ctx.resolveInit(initData)
|
||||
ctx.emitEvent('ready', {
|
||||
ctx.emitOpencodeEvent('session.updated', {
|
||||
sessionID: ctx.sessionId,
|
||||
status: 'running',
|
||||
model: ctx.getModel(),
|
||||
provider_id: ctx.getProviderId(),
|
||||
providerID: ctx.getProviderId(),
|
||||
})
|
||||
}
|
||||
|
||||
|
|
@ -126,6 +241,57 @@ function handleAssistantMessage(msg: StdoutMessage, ctx: MessageRouterCtx): void
|
|||
if (msg.uuid && typeof msg.uuid === 'string') {
|
||||
ctx.setLastMessageUuid(msg.uuid)
|
||||
}
|
||||
const rawContent = (msg.message as Record<string, unknown>)?.content ?? msg.content
|
||||
const isApiError = isApiErrorContent(rawContent)
|
||||
const agentId = (msg.agent_id as string | undefined) || getActiveSubagentId(ctx)
|
||||
const emitSessionID = agentId && ctx.getActiveSubagents().has(agentId) ? agentId : ctx.sessionId
|
||||
const wasStreamed = hasActiveStream(emitSessionID) || consumeStreamedTurn(emitSessionID)
|
||||
|
||||
if (msg.uuid) {
|
||||
ctx.pushBufferMessage({
|
||||
uuid: msg.uuid as string,
|
||||
type: 'assistant',
|
||||
role: 'assistant',
|
||||
content: isApiError ? [] : (msg.message ?? msg.content ?? ''),
|
||||
timestamp: msg.timestamp ? new Date(msg.timestamp as string).getTime() : Date.now(),
|
||||
parent_uuid: (msg.parentUuid as string) ?? null,
|
||||
...(isApiError ? { error: buildErrorFromContent(rawContent) } : {}),
|
||||
})
|
||||
}
|
||||
|
||||
if (wasStreamed) {
|
||||
if (Array.isArray(rawContent)) {
|
||||
for (const block of rawContent) {
|
||||
const b = block as Record<string, unknown>
|
||||
if (b.type !== 'tool_result') continue
|
||||
const toolUseID = b.tool_use_id as string
|
||||
if (!toolUseID) continue
|
||||
const toolInfo = getCompletedToolInfo(emitSessionID, toolUseID)
|
||||
const output = extractToolResultContent(b.content)
|
||||
const isError = b.is_error === true
|
||||
ctx.emitOpencodeEvent('message.part.updated', {
|
||||
sessionID: emitSessionID,
|
||||
part: {
|
||||
type: 'tool',
|
||||
id: toolInfo?.partID ?? randomUUID(),
|
||||
callID: toolUseID,
|
||||
tool: toolInfo?.toolName ? normalizeToolName(toolInfo.toolName) : '',
|
||||
messageID: ctx.getLastMessageUuid() ?? '',
|
||||
sessionID: emitSessionID,
|
||||
state: {
|
||||
status: isError ? 'error' : 'completed',
|
||||
input: toolInfo?.input ?? {},
|
||||
title: toolInfo?.title ?? toolInfo?.toolName ?? '',
|
||||
...(isError ? { error: output } : { output }),
|
||||
time: { start: toolInfo?.startTime ?? Date.now(), end: Date.now() },
|
||||
},
|
||||
},
|
||||
})
|
||||
}
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
let enriched = msg
|
||||
if (!msg.model && ctx.getModel()) {
|
||||
enriched = { ...enriched, model: ctx.getModel() }
|
||||
|
|
@ -133,24 +299,91 @@ function handleAssistantMessage(msg: StdoutMessage, ctx: MessageRouterCtx): void
|
|||
if (!msg.provider_id && ctx.getProviderId()) {
|
||||
enriched = { ...enriched, provider_id: ctx.getProviderId() }
|
||||
}
|
||||
ctx.emitEvent('message', enriched)
|
||||
if (msg.uuid) {
|
||||
ctx.pushBufferMessage({
|
||||
uuid: msg.uuid as string,
|
||||
type: 'assistant',
|
||||
const assistantMsgID = (msg.uuid as string) ?? randomUUID()
|
||||
ctx.emitOpencodeEvent('message.updated', {
|
||||
sessionID: emitSessionID,
|
||||
info: {
|
||||
id: assistantMsgID,
|
||||
role: 'assistant',
|
||||
content: msg.message ?? msg.content ?? '',
|
||||
timestamp: msg.timestamp ? new Date(msg.timestamp as string).getTime() : Date.now(),
|
||||
parent_uuid: (msg.parentUuid as string) ?? null,
|
||||
})
|
||||
modelID: enriched.model,
|
||||
providerID: enriched.provider_id,
|
||||
cost: 0,
|
||||
tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } },
|
||||
time: {
|
||||
created: msg.timestamp
|
||||
? new Date(msg.timestamp as string).getTime()
|
||||
: Date.now(),
|
||||
},
|
||||
parentID: (msg.parentUuid as string) ?? null,
|
||||
...(isApiError ? { error: buildErrorFromContent(rawContent) } : {}),
|
||||
},
|
||||
})
|
||||
|
||||
if (Array.isArray(rawContent)) {
|
||||
for (const block of rawContent) {
|
||||
const part = buildPartFromContentBlock(block as Record<string, unknown>, assistantMsgID, emitSessionID)
|
||||
if (part) {
|
||||
if (isApiError && part.type === 'text') continue
|
||||
ctx.emitOpencodeEvent('message.part.updated', {
|
||||
sessionID: emitSessionID,
|
||||
part,
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function handleUserMessage(msg: StdoutMessage, ctx: MessageRouterCtx): void {
|
||||
if (msg.isReplay) return
|
||||
|
||||
const agentId = (msg.agent_id as string | undefined) || getActiveSubagentId(ctx)
|
||||
const emitSessionID = agentId && ctx.getActiveSubagents().has(agentId) ? agentId : ctx.sessionId
|
||||
|
||||
let rawContent = msg.message ?? msg.content
|
||||
if (rawContent && typeof rawContent === 'object' && !Array.isArray(rawContent) && 'content' in (rawContent as Record<string, unknown>)) {
|
||||
rawContent = (rawContent as Record<string, unknown>).content
|
||||
}
|
||||
if (Array.isArray(rawContent)) {
|
||||
for (const block of rawContent) {
|
||||
const b = block as Record<string, unknown>
|
||||
if (b.type !== 'tool_result') continue
|
||||
const toolUseID = b.tool_use_id as string
|
||||
if (!toolUseID) continue
|
||||
const output = extractToolResultContent(b.content)
|
||||
const isError = b.is_error === true
|
||||
const toolInfo = getCompletedToolInfo(emitSessionID, toolUseID)
|
||||
ctx.emitOpencodeEvent('message.part.updated', {
|
||||
sessionID: emitSessionID,
|
||||
part: {
|
||||
type: 'tool',
|
||||
id: toolInfo?.partID ?? randomUUID(),
|
||||
callID: toolUseID,
|
||||
tool: toolInfo?.toolName ? normalizeToolName(toolInfo.toolName) : '',
|
||||
messageID: toolInfo?.messageID,
|
||||
sessionID: emitSessionID,
|
||||
state: {
|
||||
status: isError ? 'error' : 'completed',
|
||||
input: toolInfo?.input ?? {},
|
||||
title: toolInfo?.title ?? toolInfo?.toolName ?? '',
|
||||
...(isError ? { error: output } : { output }),
|
||||
time: { start: toolInfo?.startTime, end: Date.now() },
|
||||
},
|
||||
},
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
const content = typeof msg.content === 'string' ? msg.content : ''
|
||||
if (content.includes('<local-command-stdout>')) return
|
||||
ctx.emitEvent('message', msg)
|
||||
ctx.emitOpencodeEvent('message.updated', {
|
||||
sessionID: emitSessionID,
|
||||
info: {
|
||||
id: (msg.uuid as string) ?? randomUUID(),
|
||||
role: 'user',
|
||||
time: { created: Date.now() },
|
||||
parentID: null,
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
function handleResultMessage(msg: StdoutMessage, ctx: MessageRouterCtx): void {
|
||||
|
|
@ -165,10 +398,119 @@ function handleResultMessage(msg: StdoutMessage, ctx: MessageRouterCtx): void {
|
|||
if (usage?.output_tokens) ctx.addOutputTokens(usage.output_tokens)
|
||||
ctx.setBusyStatus({ type: 'idle' })
|
||||
ctx.emitBusyStatus()
|
||||
ctx.emitEvent('result', msg)
|
||||
ctx.emitOpencodeEvent('session.result', {
|
||||
sessionID: ctx.sessionId,
|
||||
subtype: msg.subtype ?? 'success',
|
||||
costUsd: msg.cost_usd,
|
||||
usage: msg.usage,
|
||||
stopReason: msg.stop_reason,
|
||||
})
|
||||
|
||||
const subtype = msg.subtype as string | undefined
|
||||
if (subtype && subtype !== 'success') {
|
||||
const errorData = msg.errors as Array<Record<string, unknown>> | undefined
|
||||
const errorMessage = errorData?.[0]?.message ?? msg.subtype ?? 'Unknown error'
|
||||
ctx.emitOpencodeEvent('session.error', {
|
||||
sessionID: ctx.sessionId,
|
||||
error: {
|
||||
subtype,
|
||||
level: 'error',
|
||||
message: typeof errorMessage === 'string' ? errorMessage : JSON.stringify(errorMessage),
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
ctx.resolvePrompt({ done: true })
|
||||
}
|
||||
|
||||
function handleStreamEvent(msg: StdoutMessage, ctx: MessageRouterCtx): void {
|
||||
ctx.setLastActiveAt(Date.now())
|
||||
const event = msg.event as Record<string, unknown> | undefined
|
||||
if (!event) return
|
||||
|
||||
const agentId = msg.agent_id as string | undefined
|
||||
const subagentActive = agentId && ctx.getActiveSubagents().has(agentId)
|
||||
const emitSessionID = subagentActive ? agentId! : ctx.sessionId
|
||||
const canonicalEvents = processStreamEvent(emitSessionID, event)
|
||||
for (const ce of canonicalEvents) {
|
||||
ctx.emitOpencodeEvent(ce.type, ce.properties)
|
||||
}
|
||||
}
|
||||
|
||||
function handleSystemMessage(msg: StdoutMessage, ctx: MessageRouterCtx): void {
|
||||
const subtype = msg.subtype as string
|
||||
switch (subtype) {
|
||||
case 'task_notification':
|
||||
emitTaskCompleted(msg, ctx)
|
||||
break
|
||||
case 'task_started':
|
||||
emitTaskStarted(msg, ctx)
|
||||
break
|
||||
case 'task_progress':
|
||||
emitTaskProgress(msg, ctx)
|
||||
break
|
||||
case 'api_error':
|
||||
case 'api_retry':
|
||||
emitSessionError(msg, ctx)
|
||||
break
|
||||
case 'compact_boundary':
|
||||
case 'microcompact_boundary':
|
||||
emitCompactionEvent(msg, ctx)
|
||||
break
|
||||
case 'stop_hook_summary':
|
||||
emitHookSummary(msg, ctx)
|
||||
break
|
||||
case 'turn_duration':
|
||||
emitSessionMetrics(msg, ctx)
|
||||
break
|
||||
case 'cache_warning':
|
||||
emitSessionWarning(msg, ctx)
|
||||
break
|
||||
case 'informational':
|
||||
case 'post_turn_summary':
|
||||
emitSessionInfo(msg, ctx)
|
||||
break
|
||||
case 'session_state_changed':
|
||||
case 'status':
|
||||
emitSessionStatus(msg, ctx)
|
||||
break
|
||||
default:
|
||||
if (subtype) {
|
||||
emitSessionInfo(msg, ctx)
|
||||
}
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
function handleAttachment(msg: StdoutMessage, ctx: MessageRouterCtx): void {
|
||||
ctx.emitOpencodeEvent('message.attachment', {
|
||||
sessionID: ctx.sessionId,
|
||||
attachmentType: (msg.attachment as Record<string, unknown>)?.type,
|
||||
attachment: msg.attachment,
|
||||
})
|
||||
}
|
||||
|
||||
function handleProgress(msg: StdoutMessage, ctx: MessageRouterCtx): void {
|
||||
ctx.emitOpencodeEvent('tool.progress', {
|
||||
sessionID: ctx.sessionId,
|
||||
toolUseID: msg.toolUseID,
|
||||
parentToolUseID: msg.parentToolUseID,
|
||||
data: msg.data,
|
||||
})
|
||||
}
|
||||
|
||||
function handleTombstone(msg: StdoutMessage, ctx: MessageRouterCtx): void {
|
||||
const messageObj = msg.message as Record<string, unknown> | undefined
|
||||
const targetUuid = messageObj?.uuid as string | undefined
|
||||
if (targetUuid) {
|
||||
ctx.addTombstonedUuid(targetUuid)
|
||||
ctx.emitOpencodeEvent('message.removed', {
|
||||
sessionID: ctx.sessionId,
|
||||
messageID: targetUuid,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
function handleControlRequest(msg: StdoutMessage, ctx: MessageRouterCtx): void {
|
||||
const request = msg.request as Record<string, unknown> | undefined
|
||||
const requestId = msg.request_id as string
|
||||
|
|
@ -305,6 +647,341 @@ function handleCancelRequest(msg: StdoutMessage, ctx: MessageRouterCtx): void {
|
|||
}
|
||||
}
|
||||
|
||||
function buildPartFromContentBlock(block: Record<string, unknown>, messageID: string, sessionID: string): Record<string, unknown> | null {
|
||||
switch (block.type) {
|
||||
case 'text':
|
||||
return { type: 'text', id: randomUUID(), text: block.text as string, messageID, sessionID }
|
||||
case 'thinking':
|
||||
return { type: 'reasoning', id: randomUUID(), text: block.thinking as string, messageID, sessionID }
|
||||
case 'redacted_thinking':
|
||||
return { type: 'reasoning', id: randomUUID(), text: '', redacted: true, messageID, sessionID }
|
||||
case 'tool_use': {
|
||||
const toolInput = normalizeToolInput(block.input as Record<string, unknown>)
|
||||
return {
|
||||
type: 'tool',
|
||||
id: randomUUID(),
|
||||
callID: block.id as string,
|
||||
tool: normalizeToolName(block.name as string),
|
||||
state: {
|
||||
status: 'running',
|
||||
input: toolInput,
|
||||
title: (toolInput?.description as string) ?? (block.name as string),
|
||||
time: { start: Date.now() },
|
||||
},
|
||||
messageID,
|
||||
sessionID,
|
||||
}
|
||||
}
|
||||
default:
|
||||
return null
|
||||
}
|
||||
}
|
||||
|
||||
function normalizeToolName(name: string): string {
|
||||
return toPermissionKey(name)
|
||||
}
|
||||
|
||||
function emitTaskStarted(msg: StdoutMessage, ctx: MessageRouterCtx): void {
|
||||
const agentId = (msg.agent_id ?? msg.task_id) as string
|
||||
const taskType = msg.task_type as string | undefined
|
||||
const description = msg.description as string
|
||||
const prompt = msg.prompt as string | undefined
|
||||
const toolUseId = msg.tool_use_id as string | undefined
|
||||
const assistantMessageID = randomUUID()
|
||||
|
||||
ctx.registerSubagent({
|
||||
agentId,
|
||||
agentType: taskType,
|
||||
description,
|
||||
prompt,
|
||||
toolUseId,
|
||||
})
|
||||
|
||||
const toolUseIdForMain = toolUseId
|
||||
subagentToolState.set(agentId, {
|
||||
emittedToolCount: 0,
|
||||
assistantMessageID,
|
||||
toolPartIDs: new Map(),
|
||||
mainPartID: '',
|
||||
mainMessageID: '',
|
||||
mainToolUseID: toolUseIdForMain ?? '',
|
||||
progressLines: [],
|
||||
})
|
||||
|
||||
ctx.emitOpencodeEvent('task.started', {
|
||||
sessionID: ctx.sessionId,
|
||||
taskID: msg.task_id as string,
|
||||
toolUseID: toolUseId,
|
||||
description,
|
||||
taskType,
|
||||
workflowName: msg.workflow_name as string | undefined,
|
||||
prompt,
|
||||
})
|
||||
|
||||
ctx.emitOpencodeEvent('session.created', {
|
||||
sessionID: agentId,
|
||||
info: {
|
||||
id: agentId,
|
||||
parentID: ctx.sessionId,
|
||||
title: description || `Subagent ${agentId.slice(0, 8)}`,
|
||||
agent: taskType ?? 'general-purpose',
|
||||
createdAt: Date.now(),
|
||||
status: 'running',
|
||||
},
|
||||
})
|
||||
|
||||
ctx.emitOpencodeEvent('message.updated', {
|
||||
sessionID: agentId,
|
||||
info: {
|
||||
id: assistantMessageID,
|
||||
role: 'assistant',
|
||||
modelID: 'subagent',
|
||||
time: { created: Date.now() },
|
||||
parentID: null,
|
||||
},
|
||||
})
|
||||
|
||||
ctx.emitOpencodeEvent('session.status', {
|
||||
sessionID: agentId,
|
||||
status: { type: 'busy' },
|
||||
})
|
||||
|
||||
if (toolUseId) {
|
||||
registerAgentSession(ctx.sessionId, toolUseId, agentId)
|
||||
}
|
||||
}
|
||||
|
||||
function emitTaskProgress(msg: StdoutMessage, ctx: MessageRouterCtx): void {
|
||||
const agentId = (msg.agent_id ?? msg.task_id) as string
|
||||
const usage = msg.usage as { duration_ms?: number; tool_uses?: number; total_tokens?: number } | undefined
|
||||
const toolUses = usage?.tool_uses ?? 0
|
||||
const lastToolName = (msg.last_tool_name as string | undefined) ?? 'Tool'
|
||||
const description = msg.description as string | undefined
|
||||
|
||||
ctx.emitOpencodeEvent('task.progress', {
|
||||
sessionID: ctx.sessionId,
|
||||
taskID: msg.task_id as string,
|
||||
description,
|
||||
usage,
|
||||
lastToolName,
|
||||
summary: msg.summary as string | undefined,
|
||||
workflowProgress: msg.workflow_progress,
|
||||
})
|
||||
|
||||
const toolState = subagentToolState.get(agentId)
|
||||
if (!toolState) return
|
||||
|
||||
if (!toolState.mainPartID && toolState.mainToolUseID) {
|
||||
const info = getCompletedToolInfo(ctx.sessionId, toolState.mainToolUseID)
|
||||
if (info) {
|
||||
toolState.mainPartID = info.partID
|
||||
toolState.mainMessageID = info.messageID
|
||||
}
|
||||
}
|
||||
|
||||
if (description && toolState.mainPartID) {
|
||||
const lines = toolState.progressLines
|
||||
lines.push(description)
|
||||
if (lines.length > 3) lines.splice(0, lines.length - 3)
|
||||
toolState.progressLines = lines
|
||||
|
||||
const mainToolInfo = getCompletedToolInfo(ctx.sessionId, toolState.mainToolUseID)
|
||||
ctx.emitOpencodeEvent('message.part.updated', {
|
||||
sessionID: ctx.sessionId,
|
||||
part: {
|
||||
type: 'tool',
|
||||
id: toolState.mainPartID,
|
||||
callID: toolState.mainToolUseID,
|
||||
tool: mainToolInfo?.toolName ? normalizeToolName(mainToolInfo.toolName) : 'task',
|
||||
state: {
|
||||
status: 'running',
|
||||
input: mainToolInfo?.input ?? {},
|
||||
title: mainToolInfo?.title ?? '',
|
||||
time: { start: mainToolInfo?.startTime ?? Date.now() },
|
||||
progress: toolState.progressLines,
|
||||
},
|
||||
messageID: toolState.mainMessageID,
|
||||
sessionID: ctx.sessionId,
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
while (toolState.emittedToolCount < toolUses) {
|
||||
toolState.emittedToolCount++
|
||||
const toolIndex = toolState.emittedToolCount
|
||||
const isLast = toolIndex === toolUses
|
||||
const toolName = toolIndex === toolUses ? lastToolName : 'Tool'
|
||||
|
||||
let partID = toolState.toolPartIDs.get(toolIndex)
|
||||
if (!partID) {
|
||||
partID = randomUUID()
|
||||
toolState.toolPartIDs.set(toolIndex, partID)
|
||||
}
|
||||
|
||||
ctx.emitOpencodeEvent('message.part.updated', {
|
||||
sessionID: agentId,
|
||||
part: {
|
||||
type: 'tool',
|
||||
id: partID,
|
||||
callID: `subagent-tool-${toolIndex}`,
|
||||
tool: normalizeToolName(toolName),
|
||||
state: {
|
||||
status: 'completed',
|
||||
output: isLast && description ? description.replace(/^Running\s*/i, '') : '',
|
||||
title: `${toolName} #${toolIndex}`,
|
||||
time: { start: Date.now() - (usage?.duration_ms ?? 0), end: Date.now() },
|
||||
},
|
||||
messageID: toolState.assistantMessageID,
|
||||
sessionID: agentId,
|
||||
},
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
function emitTaskCompleted(msg: StdoutMessage, ctx: MessageRouterCtx): void {
|
||||
const agentId = (msg.agent_id ?? msg.task_id) as string
|
||||
const status = msg.status as 'completed' | 'failed' | 'stopped'
|
||||
const summary = msg.summary as string
|
||||
const toolUseId = msg.tool_use_id as string | undefined
|
||||
const toolState = subagentToolState.get(agentId)
|
||||
|
||||
ctx.emitOpencodeEvent('task.completed', {
|
||||
sessionID: ctx.sessionId,
|
||||
taskID: msg.task_id as string,
|
||||
toolUseID: toolUseId,
|
||||
status,
|
||||
summary,
|
||||
outputFile: msg.output_file as string,
|
||||
usage: msg.usage,
|
||||
})
|
||||
|
||||
if (toolUseId) {
|
||||
const agentToolInfo = getCompletedToolInfo(ctx.sessionId, toolUseId)
|
||||
ctx.emitOpencodeEvent('message.part.updated', {
|
||||
sessionID: ctx.sessionId,
|
||||
part: {
|
||||
type: 'tool',
|
||||
id: agentToolInfo?.partID ?? randomUUID(),
|
||||
callID: toolUseId,
|
||||
tool: agentToolInfo?.toolName ? normalizeToolName(agentToolInfo.toolName) : undefined,
|
||||
messageID: agentToolInfo?.messageID,
|
||||
sessionID: ctx.sessionId,
|
||||
state: {
|
||||
status: status === 'failed' ? 'error' : 'completed',
|
||||
input: agentToolInfo?.input ?? {},
|
||||
...(status === 'failed' ? { error: summary || 'Task failed' } : { output: summary || '' }),
|
||||
title: summary || '',
|
||||
time: { start: agentToolInfo?.startTime, end: Date.now() },
|
||||
},
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
if (toolState) {
|
||||
ctx.emitOpencodeEvent('message.updated', {
|
||||
sessionID: agentId,
|
||||
info: {
|
||||
id: toolState.assistantMessageID,
|
||||
role: 'assistant',
|
||||
modelID: 'subagent',
|
||||
time: { completed: Date.now() },
|
||||
},
|
||||
})
|
||||
subagentToolState.delete(agentId)
|
||||
}
|
||||
|
||||
const subagent = ctx.getActiveSubagents().get(agentId)
|
||||
if (subagent || agentId) {
|
||||
ctx.emitOpencodeEvent('session.updated', {
|
||||
sessionID: agentId,
|
||||
info: {
|
||||
id: agentId,
|
||||
status,
|
||||
summary,
|
||||
completedAt: Date.now(),
|
||||
},
|
||||
})
|
||||
|
||||
ctx.emitOpencodeEvent('session.status', {
|
||||
sessionID: agentId,
|
||||
status: { type: 'idle' },
|
||||
})
|
||||
|
||||
ctx.emitOpencodeEvent('session.idle', {
|
||||
sessionID: agentId,
|
||||
})
|
||||
|
||||
ctx.unregisterSubagent(agentId)
|
||||
}
|
||||
}
|
||||
|
||||
function emitSessionError(msg: StdoutMessage, ctx: MessageRouterCtx): void {
|
||||
ctx.emitOpencodeEvent('session.error', {
|
||||
sessionID: ctx.sessionId,
|
||||
error: {
|
||||
subtype: msg.subtype,
|
||||
level: msg.level ?? 'error',
|
||||
message: msg.content ?? (msg.error as Record<string, unknown>)?.message,
|
||||
retryInMs: msg.retry_in_ms ?? msg.retryInMs,
|
||||
retryAttempt: msg.retry_attempt ?? msg.retryAttempt,
|
||||
maxRetries: msg.max_retries ?? msg.maxRetries,
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
function emitCompactionEvent(msg: StdoutMessage, ctx: MessageRouterCtx): void {
|
||||
ctx.emitOpencodeEvent('message.part.updated', {
|
||||
sessionID: ctx.sessionId,
|
||||
part: {
|
||||
type: 'compaction',
|
||||
id: (msg.uuid as string) ?? randomUUID(),
|
||||
auto:
|
||||
msg.subtype === 'microcompact_boundary' ||
|
||||
((msg.compact_metadata as Record<string, unknown>)?.trigger === 'auto'),
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
function emitHookSummary(msg: StdoutMessage, ctx: MessageRouterCtx): void {
|
||||
ctx.emitOpencodeEvent('session.hook_summary', {
|
||||
sessionID: ctx.sessionId,
|
||||
hookLabel: msg.hook_label ?? msg.hookLabel,
|
||||
hookCount: msg.hook_count ?? msg.hookCount,
|
||||
hookErrors: msg.hook_errors ?? msg.hookErrors,
|
||||
preventedContinuation: msg.prevented_continuation ?? msg.preventedContinuation,
|
||||
totalDurationMs: msg.total_duration_ms ?? msg.totalDurationMs,
|
||||
})
|
||||
}
|
||||
|
||||
function emitSessionMetrics(msg: StdoutMessage, ctx: MessageRouterCtx): void {
|
||||
ctx.emitOpencodeEvent('session.metrics', {
|
||||
sessionID: ctx.sessionId,
|
||||
turnDuration: msg.duration ?? msg.turn_duration,
|
||||
})
|
||||
}
|
||||
|
||||
function emitSessionWarning(msg: StdoutMessage, ctx: MessageRouterCtx): void {
|
||||
ctx.emitOpencodeEvent('session.warning', {
|
||||
sessionID: ctx.sessionId,
|
||||
message: msg.content,
|
||||
})
|
||||
}
|
||||
|
||||
function emitSessionInfo(msg: StdoutMessage, ctx: MessageRouterCtx): void {
|
||||
ctx.emitOpencodeEvent('session.info', {
|
||||
sessionID: ctx.sessionId,
|
||||
subtype: msg.subtype,
|
||||
content: msg.content,
|
||||
})
|
||||
}
|
||||
|
||||
function emitSessionStatus(msg: StdoutMessage, ctx: MessageRouterCtx): void {
|
||||
ctx.emitOpencodeEvent('session.status', {
|
||||
sessionID: ctx.sessionId,
|
||||
status: msg.status,
|
||||
})
|
||||
}
|
||||
|
||||
function toPermissionKey(toolName: string): string {
|
||||
const map: Record<string, string> = {
|
||||
Read: 'read',
|
||||
|
|
@ -334,3 +1011,19 @@ function extractPatterns(input: Record<string, unknown>): string[] {
|
|||
if (cmd) patterns.push(cmd)
|
||||
return patterns
|
||||
}
|
||||
|
||||
function normalizeToolInput(input: Record<string, unknown>): Record<string, unknown> {
|
||||
const renames: Record<string, string> = {
|
||||
file_path: 'filePath',
|
||||
old_string: 'oldString',
|
||||
new_string: 'newString',
|
||||
replace_all: 'replaceAll',
|
||||
}
|
||||
for (const [oldKey, newKey] of Object.entries(renames)) {
|
||||
if (oldKey in input) {
|
||||
input[newKey] = input[oldKey]
|
||||
delete input[oldKey]
|
||||
}
|
||||
}
|
||||
return input
|
||||
}
|
||||
|
|
|
|||
424
src/server/streamStateTracker.ts
Normal file
424
src/server/streamStateTracker.ts
Normal file
|
|
@ -0,0 +1,424 @@
|
|||
import { randomUUID } from 'crypto'
|
||||
|
||||
type BlockState = {
|
||||
type: string
|
||||
partID: string
|
||||
toolUseID?: string
|
||||
toolName?: string
|
||||
inputJson: string
|
||||
startTime: number
|
||||
text: string
|
||||
}
|
||||
|
||||
type SessionStreamState = {
|
||||
messageID: string
|
||||
parentID: string
|
||||
modelID: string
|
||||
activeBlocks: Map<number, BlockState>
|
||||
stepStartPartID: string
|
||||
assistantPartEmitted: boolean
|
||||
usage: {
|
||||
inputTokens: number
|
||||
outputTokens: number
|
||||
cacheReadTokens: number
|
||||
cacheWriteTokens: number
|
||||
}
|
||||
stopReason: string
|
||||
}
|
||||
|
||||
export type CanonicalEvent = {
|
||||
type: string
|
||||
properties: Record<string, unknown>
|
||||
}
|
||||
|
||||
const states = new Map<string, SessionStreamState>()
|
||||
const streamedSessions = new Set<string>()
|
||||
|
||||
type CompletedToolInfo = {
|
||||
toolName: string
|
||||
partID: string
|
||||
messageID: string
|
||||
startTime: number
|
||||
input: Record<string, unknown>
|
||||
title: string
|
||||
}
|
||||
|
||||
const completedTools = new Map<string, Map<string, CompletedToolInfo>>()
|
||||
|
||||
function getOrCreate(sessionID: string): SessionStreamState {
|
||||
let state = states.get(sessionID)
|
||||
if (!state) {
|
||||
state = {
|
||||
messageID: randomUUID(),
|
||||
parentID: '',
|
||||
modelID: '',
|
||||
activeBlocks: new Map(),
|
||||
stepStartPartID: randomUUID(),
|
||||
assistantPartEmitted: false,
|
||||
usage: { inputTokens: 0, outputTokens: 0, cacheReadTokens: 0, cacheWriteTokens: 0 },
|
||||
stopReason: '',
|
||||
}
|
||||
states.set(sessionID, state)
|
||||
}
|
||||
return state
|
||||
}
|
||||
|
||||
export function resetStreamState(sessionID: string): void {
|
||||
states.delete(sessionID)
|
||||
}
|
||||
|
||||
export function resetAllState(sessionID: string): void {
|
||||
states.delete(sessionID)
|
||||
streamedSessions.delete(sessionID)
|
||||
completedTools.delete(sessionID)
|
||||
pendingAgentSessions.delete(sessionID)
|
||||
}
|
||||
|
||||
export function hasActiveStream(sessionID: string): boolean {
|
||||
return states.has(sessionID)
|
||||
}
|
||||
|
||||
export function consumeStreamedTurn(sessionID: string): boolean {
|
||||
if (streamedSessions.has(sessionID)) {
|
||||
streamedSessions.delete(sessionID)
|
||||
return true
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
const pendingAgentSessions = new Map<string, Map<string, string>>()
|
||||
|
||||
export function registerAgentSession(sessionID: string, toolUseID: string, agentSessionID: string): void {
|
||||
let sessionMap = pendingAgentSessions.get(sessionID)
|
||||
if (!sessionMap) {
|
||||
sessionMap = new Map()
|
||||
pendingAgentSessions.set(sessionID, sessionMap)
|
||||
}
|
||||
sessionMap.set(toolUseID, agentSessionID)
|
||||
}
|
||||
|
||||
export function consumeAgentSession(sessionID: string, toolUseID: string): string | undefined {
|
||||
return pendingAgentSessions.get(sessionID)?.get(toolUseID)
|
||||
}
|
||||
|
||||
export function getCompletedToolInfo(sessionID: string, toolUseID: string): CompletedToolInfo | undefined {
|
||||
return completedTools.get(sessionID)?.get(toolUseID)
|
||||
}
|
||||
|
||||
export function processStreamEvent(
|
||||
sessionID: string,
|
||||
event: Record<string, unknown>,
|
||||
): CanonicalEvent[] {
|
||||
const results: CanonicalEvent[] = []
|
||||
const state = getOrCreate(sessionID)
|
||||
const eventType = event.type as string
|
||||
|
||||
switch (eventType) {
|
||||
case 'message_start': {
|
||||
const msg = (event as { message?: Record<string, unknown> }).message
|
||||
state.messageID = randomUUID()
|
||||
state.modelID = (msg?.model as string) ?? ''
|
||||
const msgUsage = msg?.usage as Record<string, number> | undefined
|
||||
state.usage.inputTokens = msgUsage?.input_tokens ?? 0
|
||||
state.usage.cacheReadTokens = msgUsage?.cache_read_input_tokens ?? 0
|
||||
state.usage.cacheWriteTokens = msgUsage?.cache_creation_input_tokens ?? 0
|
||||
state.activeBlocks.clear()
|
||||
|
||||
if (!state.assistantPartEmitted) {
|
||||
results.push({
|
||||
type: 'message.updated',
|
||||
properties: {
|
||||
sessionID,
|
||||
info: {
|
||||
id: state.messageID,
|
||||
role: 'assistant',
|
||||
modelID: state.modelID,
|
||||
time: { created: Date.now() },
|
||||
},
|
||||
},
|
||||
})
|
||||
state.assistantPartEmitted = true
|
||||
}
|
||||
|
||||
state.stepStartPartID = randomUUID()
|
||||
results.push({
|
||||
type: 'message.part.updated',
|
||||
properties: {
|
||||
sessionID,
|
||||
part: { type: 'step-start', id: state.stepStartPartID, messageID: state.messageID, sessionID },
|
||||
},
|
||||
})
|
||||
break
|
||||
}
|
||||
|
||||
case 'content_block_start': {
|
||||
const evt = event as { content_block?: Record<string, unknown>; index?: number }
|
||||
const block = evt.content_block
|
||||
const index = evt.index as number
|
||||
if (!block) break
|
||||
const partID = randomUUID()
|
||||
const blockState: BlockState = {
|
||||
type: block.type as string,
|
||||
partID,
|
||||
inputJson: '',
|
||||
startTime: Date.now(),
|
||||
text: '',
|
||||
}
|
||||
|
||||
state.activeBlocks.set(index, blockState)
|
||||
|
||||
if (block.type === 'text') {
|
||||
results.push({
|
||||
type: 'message.part.updated',
|
||||
properties: {
|
||||
sessionID,
|
||||
part: { type: 'text', id: partID, text: '', messageID: state.messageID, sessionID },
|
||||
},
|
||||
})
|
||||
} else if (block.type === 'thinking' || block.type === 'redacted_thinking') {
|
||||
results.push({
|
||||
type: 'message.part.updated',
|
||||
properties: {
|
||||
sessionID,
|
||||
part: {
|
||||
type: 'reasoning',
|
||||
id: partID,
|
||||
text: '',
|
||||
redacted: block.type === 'redacted_thinking',
|
||||
messageID: state.messageID, sessionID,
|
||||
},
|
||||
},
|
||||
})
|
||||
} else if (block.type === 'tool_use') {
|
||||
blockState.toolUseID = block.id as string
|
||||
blockState.toolName = block.name as string
|
||||
results.push({
|
||||
type: 'message.part.updated',
|
||||
properties: {
|
||||
sessionID,
|
||||
part: {
|
||||
type: 'tool',
|
||||
id: partID,
|
||||
callID: block.id,
|
||||
tool: normalizeToolName(block.name as string),
|
||||
state: { status: 'pending', input: {} },
|
||||
messageID: state.messageID, sessionID,
|
||||
},
|
||||
},
|
||||
})
|
||||
}
|
||||
break
|
||||
}
|
||||
|
||||
case 'content_block_delta': {
|
||||
const evt = event as { delta?: Record<string, unknown>; index?: number }
|
||||
const delta = evt.delta
|
||||
const index = evt.index as number
|
||||
const blockState = state.activeBlocks.get(index)
|
||||
|
||||
if (!blockState || !delta) break
|
||||
|
||||
if (delta.type === 'text_delta') {
|
||||
blockState.text += (delta.text as string) ?? ''
|
||||
results.push({
|
||||
type: 'message.part.delta',
|
||||
properties: {
|
||||
sessionID,
|
||||
messageID: state.messageID,
|
||||
partID: blockState.partID,
|
||||
field: 'text',
|
||||
delta: delta.text,
|
||||
},
|
||||
})
|
||||
} else if (delta.type === 'thinking_delta') {
|
||||
blockState.text += (delta.thinking as string) ?? ''
|
||||
results.push({
|
||||
type: 'message.part.delta',
|
||||
properties: {
|
||||
sessionID,
|
||||
messageID: state.messageID,
|
||||
partID: blockState.partID,
|
||||
field: 'text',
|
||||
delta: delta.thinking,
|
||||
},
|
||||
})
|
||||
} else if (delta.type === 'input_json_delta') {
|
||||
blockState.inputJson += (delta.partial_json as string) ?? ''
|
||||
results.push({
|
||||
type: 'message.part.delta',
|
||||
properties: {
|
||||
sessionID,
|
||||
messageID: state.messageID,
|
||||
partID: blockState.partID,
|
||||
field: 'input',
|
||||
delta: delta.partial_json,
|
||||
},
|
||||
})
|
||||
}
|
||||
break
|
||||
}
|
||||
|
||||
case 'content_block_stop': {
|
||||
const index = (event as { index?: number }).index as number
|
||||
const blockState = state.activeBlocks.get(index)
|
||||
if (!blockState) break
|
||||
const now = Date.now()
|
||||
|
||||
if (blockState.type === 'tool_use') {
|
||||
let parsedInput: Record<string, unknown> = {}
|
||||
try {
|
||||
parsedInput = JSON.parse(blockState.inputJson || '{}')
|
||||
} catch {
|
||||
parsedInput = {}
|
||||
}
|
||||
parsedInput = normalizeToolInput(parsedInput)
|
||||
const title = (parsedInput.description as string) ?? blockState.toolName ?? ''
|
||||
const toolState: Record<string, unknown> = {
|
||||
status: 'running',
|
||||
input: parsedInput,
|
||||
title,
|
||||
time: { start: blockState.startTime },
|
||||
}
|
||||
results.push({
|
||||
type: 'message.part.updated',
|
||||
properties: {
|
||||
sessionID,
|
||||
part: {
|
||||
type: 'tool',
|
||||
id: blockState.partID,
|
||||
callID: blockState.toolUseID,
|
||||
tool: normalizeToolName(blockState.toolName ?? ''),
|
||||
state: toolState,
|
||||
messageID: state.messageID, sessionID,
|
||||
},
|
||||
},
|
||||
})
|
||||
if (blockState.toolUseID) {
|
||||
let sessionTools = completedTools.get(sessionID)
|
||||
if (!sessionTools) {
|
||||
sessionTools = new Map()
|
||||
completedTools.set(sessionID, sessionTools)
|
||||
}
|
||||
sessionTools.set(blockState.toolUseID, {
|
||||
toolName: blockState.toolName ?? '',
|
||||
partID: blockState.partID,
|
||||
messageID: state.messageID,
|
||||
startTime: blockState.startTime,
|
||||
input: parsedInput,
|
||||
title,
|
||||
})
|
||||
}
|
||||
} else if (blockState.type === 'text' || blockState.type === 'thinking' || blockState.type === 'redacted_thinking') {
|
||||
const partType = blockState.type === 'text' ? 'text' : 'reasoning'
|
||||
results.push({
|
||||
type: 'message.part.updated',
|
||||
properties: {
|
||||
sessionID,
|
||||
part: {
|
||||
type: partType,
|
||||
id: blockState.partID,
|
||||
text: blockState.text.trimEnd(),
|
||||
messageID: state.messageID, sessionID,
|
||||
time: { start: blockState.startTime, end: now },
|
||||
...(blockState.type === 'redacted_thinking' ? { redacted: true } : {}),
|
||||
},
|
||||
},
|
||||
})
|
||||
}
|
||||
state.activeBlocks.delete(index)
|
||||
break
|
||||
}
|
||||
|
||||
case 'message_delta': {
|
||||
const evt = event as {
|
||||
delta?: Record<string, unknown>
|
||||
usage?: Record<string, number>
|
||||
}
|
||||
if (evt.delta?.stop_reason) {
|
||||
state.stopReason = evt.delta.stop_reason as string
|
||||
}
|
||||
if (evt.usage?.output_tokens) {
|
||||
state.usage.outputTokens = evt.usage.output_tokens
|
||||
}
|
||||
break
|
||||
}
|
||||
|
||||
case 'message_stop': {
|
||||
results.push({
|
||||
type: 'message.part.updated',
|
||||
properties: {
|
||||
sessionID,
|
||||
part: {
|
||||
type: 'step-finish',
|
||||
id: randomUUID(),
|
||||
reason: state.stopReason || 'stop',
|
||||
cost: 0,
|
||||
tokens: {
|
||||
input: state.usage.inputTokens,
|
||||
output: state.usage.outputTokens,
|
||||
reasoning: 0,
|
||||
cache: {
|
||||
read: state.usage.cacheReadTokens,
|
||||
write: state.usage.cacheWriteTokens,
|
||||
},
|
||||
},
|
||||
messageID: state.messageID, sessionID,
|
||||
},
|
||||
},
|
||||
})
|
||||
results.push({
|
||||
type: 'message.updated',
|
||||
properties: {
|
||||
sessionID,
|
||||
info: {
|
||||
id: state.messageID,
|
||||
role: 'assistant',
|
||||
modelID: state.modelID,
|
||||
time: { completed: Date.now() },
|
||||
},
|
||||
},
|
||||
})
|
||||
resetStreamState(sessionID)
|
||||
streamedSessions.add(sessionID)
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
return results
|
||||
}
|
||||
|
||||
function normalizeToolInput(input: Record<string, unknown>): Record<string, unknown> {
|
||||
const renames: Record<string, string> = {
|
||||
file_path: 'filePath',
|
||||
old_string: 'oldString',
|
||||
new_string: 'newString',
|
||||
replace_all: 'replaceAll',
|
||||
}
|
||||
for (const [oldKey, newKey] of Object.entries(renames)) {
|
||||
if (oldKey in input) {
|
||||
input[newKey] = input[oldKey]
|
||||
delete input[oldKey]
|
||||
}
|
||||
}
|
||||
return input
|
||||
}
|
||||
|
||||
function normalizeToolName(name: string): string {
|
||||
const map: Record<string, string> = {
|
||||
Read: 'read',
|
||||
Edit: 'edit',
|
||||
Write: 'edit',
|
||||
Glob: 'glob',
|
||||
Grep: 'grep',
|
||||
LS: 'list',
|
||||
Bash: 'bash',
|
||||
PowerShell: 'bash',
|
||||
Agent: 'task',
|
||||
Task: 'task',
|
||||
WebFetch: 'webfetch',
|
||||
WebSearch: 'websearch',
|
||||
TodoRead: 'todoread',
|
||||
TodoWrite: 'todowrite',
|
||||
}
|
||||
return map[name] ?? name.toLowerCase()
|
||||
}
|
||||
|
|
@ -28,6 +28,25 @@ type TranscriptEntry = {
|
|||
[key: string]: unknown
|
||||
}
|
||||
|
||||
export type MessagePart =
|
||||
| { type: 'text'; id: string; text: string }
|
||||
| { type: 'reasoning'; id: string; text: string; redacted?: boolean }
|
||||
| {
|
||||
type: 'tool'
|
||||
id: string
|
||||
callID: string
|
||||
tool: string
|
||||
state:
|
||||
| { status: 'pending'; input: Record<string, unknown> }
|
||||
| { status: 'running'; input: Record<string, unknown>; title?: string; time: { start: number } }
|
||||
| { status: 'completed'; input: Record<string, unknown>; output: string; title: string; time: { start: number; end: number } }
|
||||
| { status: 'error'; input: Record<string, unknown>; error: string; time: { start: number; end: number } }
|
||||
}
|
||||
| { type: 'tool-result'; id: string; toolUseID: string; content: unknown }
|
||||
| { type: 'step-start'; id: string }
|
||||
| { type: 'step-finish'; id: string; reason: string; cost: number; tokens: { input: number; output: number; reasoning: number; cache: { read: number; write: number } } }
|
||||
| { type: 'compaction'; id: string; auto: boolean }
|
||||
|
||||
export type SessionMessage = {
|
||||
uuid: string
|
||||
type: string
|
||||
|
|
@ -36,6 +55,15 @@ export type SessionMessage = {
|
|||
timestamp: number
|
||||
parent_uuid: string | null
|
||||
usage?: { input_tokens: number; output_tokens: number }
|
||||
parts?: MessagePart[]
|
||||
error?: {
|
||||
name: string
|
||||
data: {
|
||||
message: string
|
||||
statusCode?: number
|
||||
isRetryable?: boolean
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
export type TodoItem = {
|
||||
|
|
@ -45,6 +73,91 @@ export type TodoItem = {
|
|||
priority: string
|
||||
}
|
||||
|
||||
const TOOL_NAME_MAP: Record<string, string> = {
|
||||
Read: 'read',
|
||||
Edit: 'edit',
|
||||
Write: 'edit',
|
||||
Glob: 'glob',
|
||||
Grep: 'grep',
|
||||
LS: 'list',
|
||||
Bash: 'bash',
|
||||
PowerShell: 'bash',
|
||||
Agent: 'task',
|
||||
WebFetch: 'webfetch',
|
||||
WebSearch: 'websearch',
|
||||
TodoRead: 'todoread',
|
||||
TodoWrite: 'todowrite',
|
||||
}
|
||||
|
||||
function normalizeToolName(name: string): string {
|
||||
return TOOL_NAME_MAP[name] ?? name.toLowerCase()
|
||||
}
|
||||
|
||||
function contentToParts(
|
||||
content: unknown,
|
||||
role: string,
|
||||
): MessagePart[] {
|
||||
const parts: MessagePart[] = []
|
||||
if (!Array.isArray(content)) return parts
|
||||
|
||||
for (const block of content) {
|
||||
if (!block || typeof block !== 'object') continue
|
||||
const b = block as Record<string, unknown>
|
||||
switch (b.type) {
|
||||
case 'text':
|
||||
parts.push({ type: 'text', id: crypto.randomUUID(), text: (b.text as string) ?? '' })
|
||||
break
|
||||
case 'thinking':
|
||||
parts.push({ type: 'reasoning', id: crypto.randomUUID(), text: (b.thinking as string) ?? '' })
|
||||
break
|
||||
case 'redacted_thinking':
|
||||
parts.push({ type: 'reasoning', id: crypto.randomUUID(), text: '', redacted: true })
|
||||
break
|
||||
case 'tool_use':
|
||||
parts.push({
|
||||
type: 'tool',
|
||||
id: crypto.randomUUID(),
|
||||
callID: (b.id as string) ?? '',
|
||||
tool: normalizeToolName((b.name as string) ?? ''),
|
||||
state: {
|
||||
status: 'running',
|
||||
input: (b.input as Record<string, unknown>) ?? {},
|
||||
time: { start: Date.now() },
|
||||
},
|
||||
})
|
||||
break
|
||||
case 'tool_result': {
|
||||
parts.push({
|
||||
type: 'tool-result',
|
||||
id: crypto.randomUUID(),
|
||||
toolUseID: (b.tool_use_id as string) ?? '',
|
||||
content: b.content,
|
||||
})
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
return parts
|
||||
}
|
||||
|
||||
export function decomposeMessageToParts(msg: SessionMessage): SessionMessage {
|
||||
if (msg.parts && msg.parts.length > 0) return msg
|
||||
const content = msg.content
|
||||
const role = msg.role
|
||||
if (!content || !role) return msg
|
||||
|
||||
if (role === 'assistant' || role === 'user') {
|
||||
const parts = contentToParts(content, role)
|
||||
const filtered = msg.error
|
||||
? parts.filter(p => p.type !== 'text')
|
||||
: parts
|
||||
if (filtered.length > 0) {
|
||||
return { ...msg, parts: filtered }
|
||||
}
|
||||
}
|
||||
return msg
|
||||
}
|
||||
|
||||
const MESSAGE_TYPES = new Set(['user', 'assistant'])
|
||||
const SKIP_TYPES = new Set([
|
||||
'summary',
|
||||
|
|
@ -128,20 +241,30 @@ function entryToSessionMessage(
|
|||
|
||||
const role = entry.role ?? entry.message?.role ?? entry.type
|
||||
const content = entry.message?.content ?? entry.content ?? ''
|
||||
|
||||
if (role === 'user' && isTaskNotificationContent(content)) return null
|
||||
|
||||
const timestamp = entry.timestamp
|
||||
? new Date(entry.timestamp).getTime()
|
||||
: 0
|
||||
|
||||
const errorRaw = extractApiErrorText(content)
|
||||
const detail = errorRaw ? parseApiErrorDetail(errorRaw) : undefined
|
||||
const error = detail
|
||||
? { name: 'APIError', data: detail }
|
||||
: undefined
|
||||
|
||||
return {
|
||||
uuid: entry.uuid,
|
||||
type: entry.type,
|
||||
role: role ?? entry.type,
|
||||
content,
|
||||
content: error ? [] : content,
|
||||
timestamp,
|
||||
parent_uuid: entry.parentUuid ?? null,
|
||||
usage: entry.usage as
|
||||
| { input_tokens: number; output_tokens: number }
|
||||
| undefined,
|
||||
error,
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -149,11 +272,23 @@ function readMessagesFromLines(
|
|||
lines: string[],
|
||||
includeSystem: boolean,
|
||||
): SessionMessage[] {
|
||||
const tombstoned = new Set<string>()
|
||||
for (const line of lines) {
|
||||
if (!line) continue
|
||||
const entry = parseEntry(line)
|
||||
if (!entry) continue
|
||||
if (entry.type === 'tombstone') {
|
||||
const targetUuid = (entry.message as Record<string, unknown> | undefined)?.uuid
|
||||
if (typeof targetUuid === 'string') tombstoned.add(targetUuid)
|
||||
}
|
||||
}
|
||||
|
||||
const messages: SessionMessage[] = []
|
||||
for (const line of lines) {
|
||||
if (!line) continue
|
||||
const entry = parseEntry(line)
|
||||
if (!entry) continue
|
||||
if (entry.uuid && tombstoned.has(entry.uuid)) continue
|
||||
const msg = entryToSessionMessage(entry, includeSystem)
|
||||
if (msg) messages.push(msg)
|
||||
}
|
||||
|
|
@ -425,6 +560,147 @@ export async function readSessionDiff(opts: {
|
|||
}
|
||||
}
|
||||
|
||||
export type TaskInfo = {
|
||||
taskID: string
|
||||
status: 'running' | 'completed' | 'failed' | 'stopped'
|
||||
description: string
|
||||
taskType?: string
|
||||
summary?: string
|
||||
usage?: { total_tokens: number; tool_uses: number; duration_ms: number }
|
||||
toolUseID?: string
|
||||
startTime: number
|
||||
endTime?: number
|
||||
}
|
||||
|
||||
const TASK_NOTIFICATION_RE = /^<task-notification>[\s\S]*<\/task-notification>\s*$/
|
||||
const TASK_ID_RE = /<task-id>([^<]+)<\/task-id>/
|
||||
const API_ERROR_RE = /^API Error:\s*/
|
||||
const COSTRICT_API_ERROR_RE = /^CoStrict API Error:\s*/
|
||||
|
||||
function extractApiErrorText(content: unknown): string | null {
|
||||
if (typeof content === 'string') {
|
||||
if (API_ERROR_RE.test(content)) return content.replace(API_ERROR_RE, '')
|
||||
if (COSTRICT_API_ERROR_RE.test(content)) return content.replace(COSTRICT_API_ERROR_RE, '')
|
||||
}
|
||||
if (!Array.isArray(content)) return null
|
||||
for (const block of content) {
|
||||
if (!block || typeof block !== 'object') continue
|
||||
const b = block as Record<string, unknown>
|
||||
if (b.type === 'text' && typeof b.text === 'string') {
|
||||
const text = b.text as string
|
||||
if (API_ERROR_RE.test(text)) return text.replace(API_ERROR_RE, '')
|
||||
if (COSTRICT_API_ERROR_RE.test(text)) return text.replace(COSTRICT_API_ERROR_RE, '')
|
||||
}
|
||||
}
|
||||
return null
|
||||
}
|
||||
|
||||
function parseApiErrorDetail(raw: string): {
|
||||
message: string
|
||||
statusCode?: number
|
||||
isRetryable?: boolean
|
||||
} {
|
||||
const statusMatch = raw.match(/^(\d{3})\s+/)
|
||||
const statusCode = statusMatch ? parseInt(statusMatch[1]!, 10) : undefined
|
||||
const isRetryable = statusCode === 429 || statusCode === 503 || statusCode === 529
|
||||
let message = raw
|
||||
try {
|
||||
const jsonStart = raw.indexOf('{')
|
||||
if (jsonStart !== -1) {
|
||||
const parsed = JSON.parse(raw.slice(jsonStart))
|
||||
if (parsed?.error?.message) {
|
||||
message = parsed.error.message
|
||||
} else if (typeof parsed?.message === 'string') {
|
||||
message = parsed.message
|
||||
}
|
||||
}
|
||||
} catch {}
|
||||
return { message, statusCode, isRetryable }
|
||||
}
|
||||
const TOOL_USE_ID_RE = /<tool-use-id>([^<]+)<\/tool-use-id>/
|
||||
const TASK_STATUS_RE = /<status>([^<]+)<\/status>/
|
||||
const TASK_SUMMARY_RE = /<summary>([^<]*)<\/summary>/
|
||||
|
||||
function isTaskNotificationContent(content: unknown): boolean {
|
||||
return typeof content === 'string' && TASK_NOTIFICATION_RE.test(content)
|
||||
}
|
||||
|
||||
function parseTaskNotificationXml(xml: string): {
|
||||
taskID: string
|
||||
toolUseID?: string
|
||||
status: 'completed' | 'failed' | 'stopped'
|
||||
summary?: string
|
||||
} | null {
|
||||
const taskID = TASK_ID_RE.exec(xml)?.[1]
|
||||
if (!taskID) return null
|
||||
const rawStatus = TASK_STATUS_RE.exec(xml)?.[1] ?? 'completed'
|
||||
const status: 'completed' | 'failed' | 'stopped' =
|
||||
rawStatus === 'completed' || rawStatus === 'failed' || rawStatus === 'stopped'
|
||||
? rawStatus
|
||||
: 'completed'
|
||||
const toolUseID = TOOL_USE_ID_RE.exec(xml)?.[1]
|
||||
const summary = TASK_SUMMARY_RE.exec(xml)?.[1]
|
||||
return { taskID, toolUseID, status, summary }
|
||||
}
|
||||
|
||||
export async function readSessionTasks(opts: {
|
||||
sessionId: string
|
||||
cwd?: string
|
||||
}): Promise<TaskInfo[]> {
|
||||
const path = await resolveTranscriptPath(opts.sessionId, opts.cwd)
|
||||
if (!path || !existsSync(path)) return []
|
||||
|
||||
const raw = await readFile(path, 'utf-8')
|
||||
const lines = raw.split('\n').filter(Boolean)
|
||||
|
||||
const tasks = new Map<string, TaskInfo>()
|
||||
const descriptions = new Map<string, string>()
|
||||
|
||||
for (const line of lines) {
|
||||
const entry = parseEntry(line)
|
||||
if (!entry) continue
|
||||
|
||||
if (entry.type === 'assistant' && entry.message?.content) {
|
||||
const content = entry.message.content
|
||||
if (!Array.isArray(content)) continue
|
||||
for (const block of content) {
|
||||
if (!block || typeof block !== 'object') continue
|
||||
const b = block as Record<string, unknown>
|
||||
if (b.type === 'tool_use' && (b.name === 'Agent' || b.name === 'LocalMainSessionTask')) {
|
||||
const input = b.input as Record<string, unknown> | undefined
|
||||
const callID = b.id as string | undefined
|
||||
if (input?.description && callID) {
|
||||
descriptions.set(callID, input.description as string)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if (entry.type === 'user' && entry.message?.content) {
|
||||
const content = entry.message.content
|
||||
if (!isTaskNotificationContent(content)) continue
|
||||
const parsed = parseTaskNotificationXml(content as string)
|
||||
if (!parsed) continue
|
||||
|
||||
const existing = tasks.get(parsed.taskID)
|
||||
const endTime = entry.timestamp ? new Date(entry.timestamp).getTime() : Date.now()
|
||||
const desc = existing?.description ?? descriptions.get(parsed.toolUseID ?? '') ?? ''
|
||||
tasks.set(parsed.taskID, {
|
||||
taskID: parsed.taskID,
|
||||
status: parsed.status,
|
||||
description: desc,
|
||||
toolUseID: parsed.toolUseID ?? existing?.toolUseID,
|
||||
summary: parsed.summary ?? existing?.summary,
|
||||
usage: existing?.usage,
|
||||
startTime: existing?.startTime ?? endTime,
|
||||
endTime,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
return [...tasks.values()]
|
||||
}
|
||||
|
||||
export function clearPathCache(): void {
|
||||
pathCache.clear()
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue
Block a user