diff --git a/packages/client/src/solid/data.ts b/packages/client/src/solid/data.ts index d8285c21f86..e061ecbca93 100644 --- a/packages/client/src/solid/data.ts +++ b/packages/client/src/solid/data.ts @@ -401,15 +401,44 @@ export function createData(config: CreateDataInput) { index.set(item.id, messages.length) messages.push(item) }, + insert(sessionID: string, item: SessionMessageInfo) { + message.update(sessionID, (draft, index) => message.append(draft, index, item)) + }, + // Streaming events target one assistant message and, within it, the latest part of a kind. + // A missing target means the row was never loaded or was evicted; the event is dropped. + editAssistant(sessionID: string, messageID: string, fn: (assistant: SessionMessageAssistant) => void) { + message.update(sessionID, (draft, index) => { + const position = index.get(messageID) + const item = position === undefined ? undefined : draft[position] + if (item?.type === "assistant") fn(item) + }) + }, + editTool(sessionID: string, messageID: string, toolID: string, fn: (tool: SessionMessageAssistantTool) => void) { + message.editAssistant(sessionID, messageID, (assistant) => { + const tool = assistant.content.findLast( + (item): item is SessionMessageAssistantTool => item.type === "tool" && item.id === toolID, + ) + if (tool) fn(tool) + }) + }, + editText(sessionID: string, messageID: string, fn: (text: SessionMessageAssistantText) => void) { + message.editAssistant(sessionID, messageID, (assistant) => { + const text = assistant.content.findLast((item): item is SessionMessageAssistantText => item.type === "text") + if (text) fn(text) + }) + }, + editReasoning(sessionID: string, messageID: string, fn: (reasoning: SessionMessageAssistantReasoning) => void) { + message.editAssistant(sessionID, messageID, (assistant) => { + const reasoning = assistant.content.findLast( + (item): item is SessionMessageAssistantReasoning => item.type === "reasoning" && !item.time?.completed, + ) + if (reasoning) fn(reasoning) + }) + }, activeAssistant(messages: SessionMessageInfo[]) { const item = messages.findLast((item) => item.type === "assistant" && !item.time.completed) return item?.type === "assistant" ? item : undefined }, - assistant(messages: SessionMessageInfo[], index: Map, messageID: string) { - const position = index.get(messageID) - const item = position === undefined ? undefined : messages[position] - return item?.type === "assistant" ? item : undefined - }, shell(messages: SessionMessageInfo[], shellID: string) { const item = messages.findLast((item) => item.type === "shell" && item.shellID === shellID) return item?.type === "shell" ? item : undefined @@ -418,19 +447,6 @@ export function createData(config: CreateDataInput) { const item = messages.findLast((item) => item.type === "compaction" && item.status === "running") return item?.type === "compaction" ? item : undefined }, - latestTool(assistant: SessionMessageAssistant | undefined, id?: string) { - return assistant?.content.findLast( - (item): item is SessionMessageAssistantTool => item.type === "tool" && (id === undefined || item.id === id), - ) - }, - latestText(assistant: SessionMessageAssistant | undefined) { - return assistant?.content.findLast((item): item is SessionMessageAssistantText => item.type === "text") - }, - latestReasoning(assistant: SessionMessageAssistant | undefined) { - return assistant?.content.findLast( - (item): item is SessionMessageAssistantReasoning => item.type === "reasoning" && !item.time?.completed, - ) - }, reindex(messages: SessionMessageInfo[], index: Map, start: number) { for (let position = start; position < messages.length; position++) { const item = messages[position] @@ -609,14 +625,12 @@ export function createData(config: CreateDataInput) { const previous = store.session.info[event.data.sessionID]?.agent if (store.session.info[event.data.sessionID]) setStore("session", "info", event.data.sessionID, "agent", event.data.agent) - message.update(event.data.sessionID, (draft, index) => { - message.append(draft, index, { - id: messageIDFromEvent(event.id), - type: "agent-switched", - agent: event.data.agent, - previous, - time: { created: event.created }, - }) + message.insert(event.data.sessionID, { + id: messageIDFromEvent(event.id), + type: "agent-switched", + agent: event.data.agent, + previous, + time: { created: event.created }, }) return } @@ -624,13 +638,11 @@ export function createData(config: CreateDataInput) { if (store.session.info[event.data.sessionID]) setStore("session", "info", event.data.sessionID, "model", event.data.model) if (!store.session.message[event.data.sessionID]) return - message.update(event.data.sessionID, (draft, index) => { - message.append(draft, index, { - id: messageIDFromEvent(event.id), - type: "model-switched", - model: event.data.model, - time: { created: event.created }, - }) + message.insert(event.data.sessionID, { + id: messageIDFromEvent(event.id), + type: "model-switched", + model: event.data.model, + time: { created: event.created }, }) void api() .session.message({ sessionID: event.data.sessionID, messageID: messageIDFromEvent(event.id) }) @@ -665,16 +677,14 @@ export function createData(config: CreateDataInput) { setStore("session", "info", event.data.sessionID, "location", event.data.location) if (event.data.projectID) setStore("session", "info", event.data.sessionID, "projectID", event.data.projectID) setStore("session", "info", event.data.sessionID, "subpath", event.data.subpath) - message.update(event.data.sessionID, (draft, index) => { - message.append(draft, index, { - id: messageIDFromEvent(event.id), - type: "location-switched", - location: event.data.location, - projectID: event.data.projectID, - subpath: event.data.subpath, - previous, - time: { created: event.created }, - }) + message.insert(event.data.sessionID, { + id: messageIDFromEvent(event.id), + type: "location-switched", + location: event.data.location, + projectID: event.data.projectID, + subpath: event.data.subpath, + previous, + time: { created: event.created }, }) } return @@ -748,42 +758,36 @@ export function createData(config: CreateDataInput) { // and produce no transcript message. const updateText = event.data.text if (updateText === undefined) return - message.update(event.data.sessionID, (draft, index) => { - message.append(draft, index, { - id: messageIDFromEvent(event.id), - type: "system", - text: updateText, - description: `Instructions updated: ${Object.keys(event.data.delta).join(", ")}`, - metadata: event.metadata, - time: { created: event.created }, - }) + message.insert(event.data.sessionID, { + id: messageIDFromEvent(event.id), + type: "system", + text: updateText, + description: `Instructions updated: ${Object.keys(event.data.delta).join(", ")}`, + metadata: event.metadata, + time: { created: event.created }, }) return case "session.synthetic": - message.update(event.data.sessionID, (draft, index) => { - message.append(draft, index, { - id: messageIDFromEvent(event.id), - type: "synthetic", - text: event.data.text, - description: event.data.description, - metadata: event.data.metadata, - time: { created: event.created }, - }) + message.insert(event.data.sessionID, { + id: messageIDFromEvent(event.id), + type: "synthetic", + text: event.data.text, + description: event.data.description, + metadata: event.data.metadata, + time: { created: event.created }, }) return case "session.shell.started": - message.update(event.data.sessionID, (draft, index) => { - message.append(draft, index, { - id: messageIDFromEvent(event.id), - type: "shell", - shellID: event.data.shell.id, - command: event.data.shell.command, - status: event.data.shell.status, - exit: event.data.shell.exit, - metadata: - event.data.shell.metadata.background === true ? { ...event.metadata, background: true } : event.metadata, - time: { created: event.created }, - }) + message.insert(event.data.sessionID, { + id: messageIDFromEvent(event.id), + type: "shell", + shellID: event.data.shell.id, + command: event.data.shell.command, + status: event.data.shell.status, + exit: event.data.shell.exit, + metadata: + event.data.shell.metadata.background === true ? { ...event.metadata, background: true } : event.metadata, + time: { created: event.created }, }) return case "session.shell.ended": @@ -798,9 +802,8 @@ export function createData(config: CreateDataInput) { return case "session.message.content.updated": { if (store.session.message[event.data.sessionID]) - message.update(event.data.sessionID, (draft, index) => { - const assistant = message.assistant(draft, index, event.data.messageID) - if (assistant) assistant.content = [...event.data.content] + message.editAssistant(event.data.sessionID, event.data.messageID, (assistant) => { + assistant.content = [...event.data.content] }) if (!sync.pending(`session.message:${event.data.sessionID}`)) return result.session.message.invalidate(event.data.sessionID) @@ -842,65 +845,54 @@ export function createData(config: CreateDataInput) { }) return case "session.step.streamed": - message.update(event.data.sessionID, (draft, index) => { - const currentAssistant = message.assistant(draft, index, event.data.assistantMessageID) - if (currentAssistant) currentAssistant.time.streamed = event.created + message.editAssistant(event.data.sessionID, event.data.assistantMessageID, (assistant) => { + assistant.time.streamed = event.created }) return case "session.step.ended": { - message.update(event.data.sessionID, (draft, index) => { - const currentAssistant = message.assistant(draft, index, event.data.assistantMessageID) - if (!currentAssistant) return - currentAssistant.time.completed = event.created - currentAssistant.finish = event.data.finish - currentAssistant.rawFinish = event.data.rawFinish - currentAssistant.providerState = event.data.providerState - currentAssistant.cost = event.data.cost - currentAssistant.tokens = event.data.tokens - if (event.data.snapshot) - currentAssistant.snapshot = { ...currentAssistant.snapshot, end: event.data.snapshot } + message.editAssistant(event.data.sessionID, event.data.assistantMessageID, (assistant) => { + assistant.time.completed = event.created + assistant.finish = event.data.finish + assistant.rawFinish = event.data.rawFinish + assistant.providerState = event.data.providerState + assistant.cost = event.data.cost + assistant.tokens = event.data.tokens + if (event.data.snapshot) assistant.snapshot = { ...assistant.snapshot, end: event.data.snapshot } }) return } case "session.step.failed": - message.update(event.data.sessionID, (draft, index) => { - const currentAssistant = message.assistant(draft, index, event.data.assistantMessageID) - if (!currentAssistant) return - currentAssistant.time.completed = event.created - currentAssistant.finish = event.data.finish ?? "error" - currentAssistant.rawFinish = event.data.rawFinish - currentAssistant.providerState = event.data.providerState - currentAssistant.error = event.data.error - currentAssistant.retry = undefined + message.editAssistant(event.data.sessionID, event.data.assistantMessageID, (assistant) => { + assistant.time.completed = event.created + assistant.finish = event.data.finish ?? "error" + assistant.rawFinish = event.data.rawFinish + assistant.providerState = event.data.providerState + assistant.error = event.data.error + assistant.retry = undefined if (event.data.cost !== undefined && event.data.tokens !== undefined) { - currentAssistant.cost = event.data.cost - currentAssistant.tokens = event.data.tokens + assistant.cost = event.data.cost + assistant.tokens = event.data.tokens } }) return case "session.text.started": - message.update(event.data.sessionID, (draft, index) => { - message.assistant(draft, index, event.data.assistantMessageID)?.content.push({ - type: "text", - text: "", - }) + message.editAssistant(event.data.sessionID, event.data.assistantMessageID, (assistant) => { + assistant.content.push({ type: "text", text: "" }) }) return case "session.text.delta": - message.update(event.data.sessionID, (draft, index) => { - const match = message.latestText(message.assistant(draft, index, event.data.assistantMessageID)) - if (match) match.text += event.data.delta + message.editText(event.data.sessionID, event.data.assistantMessageID, (text) => { + text.text += event.data.delta }) return case "session.text.ended": - message.update(event.data.sessionID, (draft, index) => { - const match = message.latestText(message.assistant(draft, index, event.data.assistantMessageID)) - if (match) match.text = event.data.text + message.editText(event.data.sessionID, event.data.assistantMessageID, (text) => { + text.text = event.data.text }) return case "session.tool.input.started": - message.update(event.data.sessionID, (draft, index) => { - message.assistant(draft, index, event.data.assistantMessageID)?.content.push({ + message.editAssistant(event.data.sessionID, event.data.assistantMessageID, (assistant) => { + assistant.content.push({ type: "tool", id: event.data.id, name: event.data.name, @@ -910,86 +902,60 @@ export function createData(config: CreateDataInput) { }) return case "session.tool.input.delta": - message.update(event.data.sessionID, (draft, index) => { - const match = message.latestTool( - message.assistant(draft, index, event.data.assistantMessageID), - event.data.id, - ) - if (match?.state.status === "streaming") match.state.input += event.data.delta + message.editTool(event.data.sessionID, event.data.assistantMessageID, event.data.id, (tool) => { + if (tool.state.status === "streaming") tool.state.input += event.data.delta }) return case "session.tool.input.ended": - message.update(event.data.sessionID, (draft, index) => { - const match = message.latestTool( - message.assistant(draft, index, event.data.assistantMessageID), - event.data.id, - ) - if (match?.state.status === "streaming") match.state.input = event.data.text + message.editTool(event.data.sessionID, event.data.assistantMessageID, event.data.id, (tool) => { + if (tool.state.status === "streaming") tool.state.input = event.data.text }) return case "session.tool.called": - message.update(event.data.sessionID, (draft, index) => { - const match = message.latestTool( - message.assistant(draft, index, event.data.assistantMessageID), - event.data.id, - ) - if (!match) return - match.time.ran = event.created - match.executed = event.data.executed - match.providerState = event.data.state - match.state = { status: "running", input: event.data.input, metadata: {} } + message.editTool(event.data.sessionID, event.data.assistantMessageID, event.data.id, (tool) => { + tool.time.ran = event.created + tool.executed = event.data.executed + tool.providerState = event.data.state + tool.state = { status: "running", input: event.data.input, metadata: {} } }) return case "session.tool.progress": - message.update(event.data.sessionID, (draft, index) => { - const match = message.latestTool( - message.assistant(draft, index, event.data.assistantMessageID), - event.data.id, - ) - if (match?.state.status !== "running") return - match.state.metadata = event.data.metadata + message.editTool(event.data.sessionID, event.data.assistantMessageID, event.data.id, (tool) => { + if (tool.state.status === "running") tool.state.metadata = event.data.metadata }) return case "session.tool.success": - message.update(event.data.sessionID, (draft, index) => { - const match = message.latestTool( - message.assistant(draft, index, event.data.assistantMessageID), - event.data.id, - ) - if (match?.state.status !== "running") return - match.state = { + message.editTool(event.data.sessionID, event.data.assistantMessageID, event.data.id, (tool) => { + if (tool.state.status !== "running") return + tool.state = { status: "completed", - input: match.state.input, + input: tool.state.input, metadata: event.data.metadata, content: [...event.data.content], } - match.executed = event.data.executed || match.executed === true - match.providerResultState = event.data.resultState - match.time.completed = event.created + tool.executed = event.data.executed || tool.executed === true + tool.providerResultState = event.data.resultState + tool.time.completed = event.created }) return case "session.tool.failed": - message.update(event.data.sessionID, (draft, index) => { - const match = message.latestTool( - message.assistant(draft, index, event.data.assistantMessageID), - event.data.id, - ) - if (!match || (match.state.status !== "streaming" && match.state.status !== "running")) return - match.state = { + message.editTool(event.data.sessionID, event.data.assistantMessageID, event.data.id, (tool) => { + if (tool.state.status !== "streaming" && tool.state.status !== "running") return + tool.state = { status: "error", error: event.data.error, - input: typeof match.state.input === "string" ? {} : match.state.input, + input: typeof tool.state.input === "string" ? {} : tool.state.input, metadata: event.data.metadata, content: event.data.content, } - match.executed = event.data.executed || match.executed === true - match.providerResultState = event.data.resultState - match.time.completed = event.created + tool.executed = event.data.executed || tool.executed === true + tool.providerResultState = event.data.resultState + tool.time.completed = event.created }) return case "session.reasoning.started": - message.update(event.data.sessionID, (draft, index) => { - message.assistant(draft, index, event.data.assistantMessageID)?.content.push({ + message.editAssistant(event.data.sessionID, event.data.assistantMessageID, (assistant) => { + assistant.content.push({ type: "reasoning", text: "", state: event.data.state, @@ -998,30 +964,20 @@ export function createData(config: CreateDataInput) { }) return case "session.reasoning.delta": - message.update(event.data.sessionID, (draft, index) => { - const match = message.latestReasoning(message.assistant(draft, index, event.data.assistantMessageID)) - if (match) match.text += event.data.delta + message.editReasoning(event.data.sessionID, event.data.assistantMessageID, (reasoning) => { + reasoning.text += event.data.delta }) return case "session.reasoning.ended": - message.update(event.data.sessionID, (draft, index) => { - const match = message.latestReasoning(message.assistant(draft, index, event.data.assistantMessageID)) - if (match) { - match.text = event.data.text - match.time = { created: match.time?.created ?? event.created, completed: event.created } - if (event.data.state !== undefined) match.state = event.data.state - } + message.editReasoning(event.data.sessionID, event.data.assistantMessageID, (reasoning) => { + reasoning.text = event.data.text + reasoning.time = { created: reasoning.time?.created ?? event.created, completed: event.created } + if (event.data.state !== undefined) reasoning.state = event.data.state }) return case "session.retry.scheduled": - message.update(event.data.sessionID, (draft, index) => { - const currentAssistant = message.assistant(draft, index, event.data.assistantMessageID) - if (!currentAssistant) return - currentAssistant.retry = { - attempt: event.data.attempt, - at: event.data.at, - error: event.data.error, - } + message.editAssistant(event.data.sessionID, event.data.assistantMessageID, (assistant) => { + assistant.retry = { attempt: event.data.attempt, at: event.data.at, error: event.data.error } }) return case "session.execution.started": @@ -1029,16 +985,14 @@ export function createData(config: CreateDataInput) { return case "session.compaction.started": if (event.data.inputID) removePending(event.data.sessionID, event.data.inputID) - message.update(event.data.sessionID, (draft, index) => { - message.append(draft, index, { - id: event.data.inputID ?? messageIDFromEvent(event.id), - type: "compaction", - status: "running", - reason: event.data.reason, - summary: "", - recent: event.data.recent ?? "", - time: { created: event.created }, - }) + message.insert(event.data.sessionID, { + id: event.data.inputID ?? messageIDFromEvent(event.id), + type: "compaction", + status: "running", + reason: event.data.reason, + summary: "", + recent: event.data.recent ?? "", + time: { created: event.created }, }) if (event.data.inputID) compacting.get(event.data.sessionID)?.observed.add(event.data.inputID) return