feat(opencode): add interactive split-footer mode to run (#23557)
This commit is contained in:
Simon Klee
970
packages/opencode/src/cli/cmd/run/session-data.ts
Normal file
970
packages/opencode/src/cli/cmd/run/session-data.ts
Normal file
@@ -0,0 +1,970 @@
|
||||
// Core reducer for direct interactive mode.
|
||||
//
|
||||
// Takes raw SDK events and produces two outputs:
|
||||
// - StreamCommit[]: append-only scrollback entries (text, tool, error, etc.)
|
||||
// - FooterOutput: status bar patches and view transitions (permission, question)
|
||||
//
|
||||
// The reducer mutates SessionData in place for performance but has no
|
||||
// external side effects -- no IO, no footer calls. The caller
|
||||
// (stream.transport.ts) feeds events in and forwards output to the footer
|
||||
// through stream.ts.
|
||||
//
|
||||
// Key design decisions:
|
||||
//
|
||||
// - Text parts buffer in `data.text` until their message role is confirmed as
|
||||
// "assistant". This prevents echoing user-role text parts. The `ready()`
|
||||
// check gates output: if we see a text delta before the message.updated
|
||||
// event that tells us the role, we stash it and flush later via `replay()`.
|
||||
//
|
||||
// - Tool echo stripping: bash tools may echo their own output in the next
|
||||
// assistant text part. `stashEcho()` records completed bash output, and
|
||||
// `stripEcho()` removes it from the start of the next assistant chunk.
|
||||
//
|
||||
// - Permission and question requests queue in `data.permissions` and
|
||||
// `data.questions`. The footer shows whichever is first. When a reply
|
||||
// event arrives, the queue entry is removed and the footer falls back
|
||||
// to the next pending request or to the prompt view.
|
||||
import type { Event, Part, PermissionRequest, QuestionRequest, ToolPart } from "@opencode-ai/sdk/v2"
|
||||
import * as Locale from "@/util/locale"
|
||||
import { toolView } from "./tool"
|
||||
import type { FooterOutput, FooterPatch, FooterView, StreamCommit } from "./types"
|
||||
|
||||
const money = new Intl.NumberFormat("en-US", {
|
||||
style: "currency",
|
||||
currency: "USD",
|
||||
})
|
||||
|
||||
type Tokens = {
|
||||
input?: number
|
||||
output?: number
|
||||
reasoning?: number
|
||||
cache?: {
|
||||
read?: number
|
||||
write?: number
|
||||
}
|
||||
}
|
||||
|
||||
type PartKind = "assistant" | "reasoning" | "user"
|
||||
type MessageRole = "assistant" | "user"
|
||||
type Dict = Record<string, unknown>
|
||||
type SessionCommit = StreamCommit
|
||||
|
||||
// Mutable accumulator for the reducer. Each field tracks a different aspect
|
||||
// of the stream so we can produce correct incremental output:
|
||||
//
|
||||
// - ids: parts and error keys we've already committed (dedup guard)
|
||||
// - tools: tool parts we've emitted a "start" for but not yet completed
|
||||
// - call: tool call inputs, keyed by msg:call, for enriching permission views
|
||||
// - role: message ID → "assistant" | "user", learned from message.updated
|
||||
// - msg: part ID → message ID
|
||||
// - part: part ID → "assistant" | "reasoning" (text parts only)
|
||||
// - text: part ID → full accumulated text so far
|
||||
// - sent: part ID → byte offset of last flushed text (for incremental output)
|
||||
// - end: part IDs whose time.end has arrived (part is finished)
|
||||
// - echo: message ID → bash outputs to strip from the next assistant chunk
|
||||
export type SessionData = {
|
||||
includeUserText: boolean
|
||||
announced: boolean
|
||||
ids: Set<string>
|
||||
tools: Set<string>
|
||||
call: Map<string, Dict>
|
||||
permissions: PermissionRequest[]
|
||||
questions: QuestionRequest[]
|
||||
role: Map<string, MessageRole>
|
||||
msg: Map<string, string>
|
||||
part: Map<string, PartKind>
|
||||
text: Map<string, string>
|
||||
sent: Map<string, number>
|
||||
end: Set<string>
|
||||
echo: Map<string, Set<string>>
|
||||
}
|
||||
|
||||
export type SessionDataInput = {
|
||||
data: SessionData
|
||||
event: Event
|
||||
sessionID: string
|
||||
thinking: boolean
|
||||
limits: Record<string, number>
|
||||
}
|
||||
|
||||
export type SessionDataOutput = {
|
||||
data: SessionData
|
||||
commits: SessionCommit[]
|
||||
footer?: FooterOutput
|
||||
}
|
||||
|
||||
export function createSessionData(
|
||||
input: {
|
||||
includeUserText?: boolean
|
||||
} = {},
|
||||
): SessionData {
|
||||
return {
|
||||
includeUserText: input.includeUserText ?? false,
|
||||
announced: false,
|
||||
ids: new Set(),
|
||||
tools: new Set(),
|
||||
call: new Map(),
|
||||
permissions: [],
|
||||
questions: [],
|
||||
role: new Map(),
|
||||
msg: new Map(),
|
||||
part: new Map(),
|
||||
text: new Map(),
|
||||
sent: new Map(),
|
||||
end: new Set(),
|
||||
echo: new Map(),
|
||||
}
|
||||
}
|
||||
|
||||
function modelKey(provider: string, model: string): string {
|
||||
return `${provider}/${model}`
|
||||
}
|
||||
|
||||
function formatUsage(
|
||||
tokens: Tokens | undefined,
|
||||
limit: number | undefined,
|
||||
cost: number | undefined,
|
||||
): string | undefined {
|
||||
const total =
|
||||
(tokens?.input ?? 0) +
|
||||
(tokens?.output ?? 0) +
|
||||
(tokens?.reasoning ?? 0) +
|
||||
(tokens?.cache?.read ?? 0) +
|
||||
(tokens?.cache?.write ?? 0)
|
||||
|
||||
if (total <= 0) {
|
||||
if (typeof cost === "number" && cost > 0) {
|
||||
return money.format(cost)
|
||||
}
|
||||
return undefined
|
||||
}
|
||||
|
||||
const text =
|
||||
limit && limit > 0 ? `${Locale.number(total)} (${Math.round((total / limit) * 100)}%)` : Locale.number(total)
|
||||
|
||||
if (typeof cost === "number" && cost > 0) {
|
||||
return `${text} · ${money.format(cost)}`
|
||||
}
|
||||
|
||||
return text
|
||||
}
|
||||
|
||||
export function formatError(error: {
|
||||
name?: string
|
||||
message?: string
|
||||
data?: {
|
||||
message?: string
|
||||
}
|
||||
}): string {
|
||||
if (error.data?.message) {
|
||||
return error.data.message
|
||||
}
|
||||
|
||||
if (error.message) {
|
||||
return error.message
|
||||
}
|
||||
|
||||
if (error.name) {
|
||||
return error.name
|
||||
}
|
||||
|
||||
return "unknown error"
|
||||
}
|
||||
|
||||
function isAbort(error: { name?: string } | undefined): boolean {
|
||||
return error?.name === "MessageAbortedError"
|
||||
}
|
||||
|
||||
function msgErr(id: string): string {
|
||||
return `msg:${id}:error`
|
||||
}
|
||||
|
||||
function patch(patch?: FooterPatch, view?: FooterView): FooterOutput | undefined {
|
||||
if (!patch && !view) {
|
||||
return undefined
|
||||
}
|
||||
|
||||
return {
|
||||
patch,
|
||||
view,
|
||||
}
|
||||
}
|
||||
|
||||
function out(data: SessionData, commits: SessionCommit[], footer?: FooterOutput): SessionDataOutput {
|
||||
if (!footer) {
|
||||
return {
|
||||
data,
|
||||
commits,
|
||||
}
|
||||
}
|
||||
|
||||
return {
|
||||
data,
|
||||
commits,
|
||||
footer,
|
||||
}
|
||||
}
|
||||
|
||||
export function pickBlockerView(input: {
|
||||
permission?: PermissionRequest
|
||||
question?: QuestionRequest
|
||||
}): FooterView {
|
||||
if (input.permission) {
|
||||
return { type: "permission", request: input.permission }
|
||||
}
|
||||
|
||||
if (input.question) {
|
||||
return { type: "question", request: input.question }
|
||||
}
|
||||
|
||||
return { type: "prompt" }
|
||||
}
|
||||
|
||||
export function blockerStatus(view: FooterView) {
|
||||
if (view.type === "permission") {
|
||||
return "awaiting permission"
|
||||
}
|
||||
|
||||
if (view.type === "question") {
|
||||
return "awaiting answer"
|
||||
}
|
||||
|
||||
return ""
|
||||
}
|
||||
|
||||
function pickSessionView(data: SessionData): FooterView {
|
||||
return pickBlockerView({
|
||||
permission: data.permissions[0],
|
||||
question: data.questions[0],
|
||||
})
|
||||
}
|
||||
|
||||
function queueFooter(data: SessionData): FooterOutput {
|
||||
const view = pickSessionView(data)
|
||||
|
||||
return {
|
||||
view,
|
||||
patch: { status: blockerStatus(view) },
|
||||
}
|
||||
}
|
||||
|
||||
function queueOut(data: SessionData, commits: SessionCommit[]): SessionDataOutput {
|
||||
return out(data, commits, queueFooter(data))
|
||||
}
|
||||
|
||||
function upsert<T extends { id: string }>(list: T[], item: T) {
|
||||
const idx = list.findIndex((entry) => entry.id === item.id)
|
||||
if (idx === -1) {
|
||||
list.push(item)
|
||||
return
|
||||
}
|
||||
|
||||
list[idx] = item
|
||||
}
|
||||
|
||||
function remove(list: Array<{ id: string }>, id: string): boolean {
|
||||
const idx = list.findIndex((entry) => entry.id === id)
|
||||
if (idx === -1) {
|
||||
return false
|
||||
}
|
||||
|
||||
list.splice(idx, 1)
|
||||
return true
|
||||
}
|
||||
|
||||
export function bootstrapSessionData(input: {
|
||||
data: SessionData
|
||||
messages: Array<{
|
||||
parts: Part[]
|
||||
}>
|
||||
permissions: PermissionRequest[]
|
||||
questions: QuestionRequest[]
|
||||
}) {
|
||||
for (const message of input.messages) {
|
||||
for (const part of message.parts) {
|
||||
if (part.type !== "tool") {
|
||||
continue
|
||||
}
|
||||
|
||||
input.data.call.set(key(part.messageID, part.callID), part.state.input)
|
||||
}
|
||||
}
|
||||
|
||||
for (const request of input.permissions.slice().sort((a, b) => a.id.localeCompare(b.id))) {
|
||||
upsert(input.data.permissions, enrichPermission(input.data, request))
|
||||
}
|
||||
|
||||
for (const request of input.questions.slice().sort((a, b) => a.id.localeCompare(b.id))) {
|
||||
upsert(input.data.questions, request)
|
||||
}
|
||||
}
|
||||
|
||||
function key(msg: string, call: string): string {
|
||||
return `${msg}:${call}`
|
||||
}
|
||||
|
||||
function enrichPermission(data: SessionData, request: PermissionRequest): PermissionRequest {
|
||||
if (!request.tool) {
|
||||
return request
|
||||
}
|
||||
|
||||
const input = data.call.get(key(request.tool.messageID, request.tool.callID))
|
||||
if (!input) {
|
||||
return request
|
||||
}
|
||||
|
||||
const meta = request.metadata ?? {}
|
||||
if (meta.input === input) {
|
||||
return request
|
||||
}
|
||||
|
||||
return {
|
||||
...request,
|
||||
metadata: {
|
||||
...meta,
|
||||
input,
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
// Updates the active permission request when the matching tool part gets
|
||||
// new input (e.g., a diff). This keeps the permission UI in sync with the
|
||||
// tool's evolving state. Only triggers a footer update if the currently
|
||||
// displayed permission was the one that changed.
|
||||
function syncPermission(data: SessionData, part: ToolPart): FooterOutput | undefined {
|
||||
data.call.set(key(part.messageID, part.callID), part.state.input)
|
||||
if (data.permissions.length === 0) {
|
||||
return undefined
|
||||
}
|
||||
|
||||
let changed = false
|
||||
let active = false
|
||||
data.permissions = data.permissions.map((request, index) => {
|
||||
if (!request.tool || request.tool.messageID !== part.messageID || request.tool.callID !== part.callID) {
|
||||
return request
|
||||
}
|
||||
|
||||
const next = enrichPermission(data, request)
|
||||
if (next === request) {
|
||||
return request
|
||||
}
|
||||
|
||||
changed = true
|
||||
active ||= index === 0
|
||||
return next
|
||||
})
|
||||
|
||||
if (!changed || !active) {
|
||||
return undefined
|
||||
}
|
||||
|
||||
return {
|
||||
view: pickSessionView(data),
|
||||
}
|
||||
}
|
||||
|
||||
// Question tool replies can complete without a matching question.replied event.
|
||||
// When that happens, drop the recovered pending request tied to this tool call so
|
||||
// the footer can return to the next blocker or to the prompt.
|
||||
function syncQuestion(data: SessionData, part: ToolPart): FooterOutput | undefined {
|
||||
if (part.tool !== "question") {
|
||||
return undefined
|
||||
}
|
||||
|
||||
if (part.state.status !== "completed" && part.state.status !== "error") {
|
||||
return undefined
|
||||
}
|
||||
|
||||
const next = data.questions.filter(
|
||||
(request) => request.tool?.messageID !== part.messageID || request.tool?.callID !== part.callID,
|
||||
)
|
||||
if (next.length === data.questions.length) {
|
||||
return undefined
|
||||
}
|
||||
|
||||
data.questions = next
|
||||
return queueFooter(data)
|
||||
}
|
||||
|
||||
function toolStatus(part: ToolPart): string {
|
||||
if (part.tool !== "task") {
|
||||
return `running ${part.tool}`
|
||||
}
|
||||
|
||||
const state = part.state as {
|
||||
input?: {
|
||||
description?: unknown
|
||||
subagent_type?: unknown
|
||||
}
|
||||
}
|
||||
const desc = state.input?.description
|
||||
if (typeof desc === "string" && desc.trim()) {
|
||||
return `running ${desc.trim()}`
|
||||
}
|
||||
|
||||
const type = state.input?.subagent_type
|
||||
if (typeof type === "string" && type.trim()) {
|
||||
return `running ${type.trim()}`
|
||||
}
|
||||
|
||||
return "running task"
|
||||
}
|
||||
|
||||
// Returns true if we can flush this part's text to scrollback.
|
||||
//
|
||||
// We gate on the message role being "assistant" because user-role messages
|
||||
// also contain text parts (the user's own input) which we don't want to
|
||||
// echo. If we haven't received the message.updated event yet, we return
|
||||
// false and the text stays buffered until replay() flushes it.
|
||||
function ready(data: SessionData, partID: string): boolean {
|
||||
const msg = data.msg.get(partID)
|
||||
if (!msg) {
|
||||
return true
|
||||
}
|
||||
|
||||
const role = data.role.get(msg)
|
||||
if (!role) {
|
||||
return false
|
||||
}
|
||||
|
||||
if (role === "assistant") {
|
||||
return true
|
||||
}
|
||||
|
||||
return data.includeUserText && role === "user"
|
||||
}
|
||||
|
||||
function syncText(data: SessionData, partID: string, next: string) {
|
||||
const prev = data.text.get(partID) ?? ""
|
||||
if (!next) {
|
||||
return prev
|
||||
}
|
||||
|
||||
if (!prev || next.length >= prev.length) {
|
||||
data.text.set(partID, next)
|
||||
return next
|
||||
}
|
||||
|
||||
return prev
|
||||
}
|
||||
|
||||
// Records bash tool output for echo stripping. Some models echo bash output
|
||||
// verbatim at the start of their next text part. We save both the raw and
|
||||
// trimmed forms so stripEcho() can match either.
|
||||
function stashEcho(data: SessionData, part: ToolPart) {
|
||||
if (part.tool !== "bash") {
|
||||
return
|
||||
}
|
||||
|
||||
if (typeof part.messageID !== "string" || !part.messageID) {
|
||||
return
|
||||
}
|
||||
|
||||
const output = "output" in part.state ? part.state.output : undefined
|
||||
if (typeof output !== "string") {
|
||||
return
|
||||
}
|
||||
|
||||
const text = output.replace(/^\n+/, "")
|
||||
if (!text.trim()) {
|
||||
return
|
||||
}
|
||||
|
||||
const set = data.echo.get(part.messageID) ?? new Set<string>()
|
||||
set.add(text)
|
||||
const trim = text.replace(/\n+$/, "")
|
||||
if (trim && trim !== text) {
|
||||
set.add(trim)
|
||||
}
|
||||
data.echo.set(part.messageID, set)
|
||||
}
|
||||
|
||||
function stripEcho(data: SessionData, msg: string | undefined, chunk: string): string {
|
||||
if (!msg) {
|
||||
return chunk
|
||||
}
|
||||
|
||||
const set = data.echo.get(msg)
|
||||
if (!set || set.size === 0) {
|
||||
return chunk
|
||||
}
|
||||
|
||||
data.echo.delete(msg)
|
||||
const list = [...set].sort((a, b) => b.length - a.length)
|
||||
for (const item of list) {
|
||||
if (!item || !chunk.startsWith(item)) {
|
||||
continue
|
||||
}
|
||||
|
||||
return chunk.slice(item.length).replace(/^\n+/, "")
|
||||
}
|
||||
|
||||
return chunk
|
||||
}
|
||||
|
||||
function flushPart(data: SessionData, commits: SessionCommit[], partID: string, interrupted = false) {
|
||||
const kind = data.part.get(partID)
|
||||
if (!kind) {
|
||||
return
|
||||
}
|
||||
|
||||
const text = data.text.get(partID) ?? ""
|
||||
const sent = data.sent.get(partID) ?? 0
|
||||
let chunk = text.slice(sent)
|
||||
const msg = data.msg.get(partID)
|
||||
|
||||
if (sent === 0) {
|
||||
chunk = chunk.replace(/^\n+/, "")
|
||||
// Some models emit a standalone whitespace token before real content.
|
||||
// Keep buffering until we have visible text so scrollback doesn't get a blank row.
|
||||
if (!chunk.trim()) {
|
||||
return
|
||||
}
|
||||
if (kind === "reasoning" && chunk) {
|
||||
chunk = `Thinking: ${chunk.replace(/\[REDACTED\]/g, "")}`
|
||||
}
|
||||
if (kind === "assistant" && chunk) {
|
||||
chunk = stripEcho(data, msg, chunk)
|
||||
if (!chunk.trim()) {
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if (chunk) {
|
||||
data.sent.set(partID, text.length)
|
||||
commits.push({
|
||||
kind,
|
||||
text: chunk,
|
||||
phase: "progress",
|
||||
source: kind === "user" ? "system" : kind,
|
||||
messageID: msg,
|
||||
partID,
|
||||
})
|
||||
}
|
||||
|
||||
if (!interrupted) {
|
||||
return
|
||||
}
|
||||
|
||||
commits.push({
|
||||
kind,
|
||||
text: "",
|
||||
phase: "final",
|
||||
source: kind === "user" ? "system" : kind,
|
||||
messageID: msg,
|
||||
partID,
|
||||
interrupted: true,
|
||||
})
|
||||
}
|
||||
|
||||
function drop(data: SessionData, partID: string) {
|
||||
data.part.delete(partID)
|
||||
data.text.delete(partID)
|
||||
data.sent.delete(partID)
|
||||
data.msg.delete(partID)
|
||||
data.end.delete(partID)
|
||||
}
|
||||
|
||||
// Called when we learn a message's role (from message.updated). Flushes any
|
||||
// buffered text parts that were waiting on role confirmation. User-role
|
||||
// parts are silently dropped.
|
||||
function replay(data: SessionData, commits: SessionCommit[], messageID: string, role: MessageRole, thinking: boolean) {
|
||||
for (const [partID, msg] of data.msg.entries()) {
|
||||
if (msg !== messageID || data.ids.has(partID)) {
|
||||
continue
|
||||
}
|
||||
|
||||
if (role === "user" && !data.includeUserText) {
|
||||
data.ids.add(partID)
|
||||
drop(data, partID)
|
||||
continue
|
||||
}
|
||||
|
||||
const kind = data.part.get(partID)
|
||||
if (!kind) {
|
||||
continue
|
||||
}
|
||||
|
||||
if (role === "user" && kind === "assistant") {
|
||||
data.part.set(partID, "user")
|
||||
}
|
||||
|
||||
if (kind === "reasoning" && !thinking) {
|
||||
if (data.end.has(partID)) {
|
||||
data.ids.add(partID)
|
||||
}
|
||||
drop(data, partID)
|
||||
continue
|
||||
}
|
||||
|
||||
flushPart(data, commits, partID)
|
||||
|
||||
if (!data.end.has(partID)) {
|
||||
continue
|
||||
}
|
||||
|
||||
data.ids.add(partID)
|
||||
drop(data, partID)
|
||||
}
|
||||
}
|
||||
|
||||
function toolCommit(
|
||||
part: ToolPart,
|
||||
next: Pick<SessionCommit, "text" | "phase" | "toolState"> & { toolError?: string },
|
||||
): SessionCommit {
|
||||
return {
|
||||
kind: "tool",
|
||||
source: "tool",
|
||||
messageID: part.messageID,
|
||||
partID: part.id,
|
||||
tool: part.tool,
|
||||
part,
|
||||
...next,
|
||||
}
|
||||
}
|
||||
|
||||
function startTool(part: ToolPart): SessionCommit {
|
||||
return toolCommit(part, {
|
||||
text: toolStatus(part),
|
||||
phase: "start",
|
||||
toolState: "running",
|
||||
})
|
||||
}
|
||||
|
||||
function doneTool(part: ToolPart): SessionCommit {
|
||||
return toolCommit(part, {
|
||||
text: "",
|
||||
phase: "final",
|
||||
toolState: "completed",
|
||||
})
|
||||
}
|
||||
|
||||
function failTool(part: ToolPart, text: string): SessionCommit {
|
||||
return toolCommit(part, {
|
||||
text,
|
||||
phase: "final",
|
||||
toolState: "error",
|
||||
toolError: text,
|
||||
})
|
||||
}
|
||||
|
||||
// Emits "interrupted" final entries for all in-flight parts. Called when a turn is aborted.
|
||||
export function flushInterrupted(data: SessionData, commits: SessionCommit[]) {
|
||||
for (const partID of data.part.keys()) {
|
||||
if (data.ids.has(partID)) {
|
||||
continue
|
||||
}
|
||||
|
||||
const msg = data.msg.get(partID)
|
||||
if (msg && data.role.get(msg) === "user" && !data.includeUserText) {
|
||||
data.ids.add(partID)
|
||||
drop(data, partID)
|
||||
continue
|
||||
}
|
||||
|
||||
flushPart(data, commits, partID, true)
|
||||
data.ids.add(partID)
|
||||
drop(data, partID)
|
||||
}
|
||||
}
|
||||
|
||||
// The main reducer. Takes one SDK event and returns scrollback commits and
|
||||
// footer updates. Called once per event from the stream transport's watch loop.
|
||||
//
|
||||
// Event handling follows the SDK event types:
|
||||
// message.updated → learn role, flush buffered parts, track usage
|
||||
// message.part.delta → accumulate text, flush if ready
|
||||
// message.part.updated → handle text/reasoning/tool state transitions
|
||||
// permission.* → manage the permission queue, drive footer view
|
||||
// question.* → manage the question queue, drive footer view
|
||||
// session.error → emit error scrollback entry
|
||||
export function reduceSessionData(input: SessionDataInput): SessionDataOutput {
|
||||
const commits: SessionCommit[] = []
|
||||
const data = input.data
|
||||
const event = input.event
|
||||
|
||||
if (event.type === "message.updated") {
|
||||
if (event.properties.sessionID !== input.sessionID) {
|
||||
return out(data, commits)
|
||||
}
|
||||
|
||||
const info = event.properties.info
|
||||
if (typeof info.id === "string") {
|
||||
data.role.set(info.id, info.role)
|
||||
replay(data, commits, info.id, info.role, input.thinking)
|
||||
}
|
||||
|
||||
if (info.role !== "assistant") {
|
||||
return out(data, commits)
|
||||
}
|
||||
|
||||
let next: FooterPatch | undefined
|
||||
if (!data.announced) {
|
||||
data.announced = true
|
||||
next = { status: "assistant responding" }
|
||||
}
|
||||
|
||||
const usage = formatUsage(
|
||||
info.tokens,
|
||||
input.limits[modelKey(info.providerID, info.modelID)],
|
||||
typeof info.cost === "number" ? info.cost : undefined,
|
||||
)
|
||||
if (usage) {
|
||||
next = {
|
||||
...next,
|
||||
usage,
|
||||
}
|
||||
}
|
||||
|
||||
if (typeof info.id === "string" && info.error && !isAbort(info.error) && !data.ids.has(msgErr(info.id))) {
|
||||
data.ids.add(msgErr(info.id))
|
||||
commits.push({
|
||||
kind: "error",
|
||||
text: formatError(info.error),
|
||||
phase: "start",
|
||||
source: "system",
|
||||
messageID: info.id,
|
||||
})
|
||||
}
|
||||
|
||||
return out(data, commits, patch(next))
|
||||
}
|
||||
|
||||
if (event.type === "message.part.delta") {
|
||||
if (event.properties.sessionID !== input.sessionID) {
|
||||
return out(data, commits)
|
||||
}
|
||||
|
||||
if (
|
||||
typeof event.properties.partID !== "string" ||
|
||||
typeof event.properties.field !== "string" ||
|
||||
typeof event.properties.delta !== "string"
|
||||
) {
|
||||
return out(data, commits)
|
||||
}
|
||||
|
||||
if (event.properties.field !== "text") {
|
||||
return out(data, commits)
|
||||
}
|
||||
|
||||
const partID = event.properties.partID
|
||||
if (data.ids.has(partID)) {
|
||||
return out(data, commits)
|
||||
}
|
||||
|
||||
if (typeof event.properties.messageID === "string") {
|
||||
data.msg.set(partID, event.properties.messageID)
|
||||
}
|
||||
|
||||
const text = data.text.get(partID) ?? ""
|
||||
data.text.set(partID, text + event.properties.delta)
|
||||
|
||||
const kind = data.part.get(partID)
|
||||
if (!kind) {
|
||||
return out(data, commits)
|
||||
}
|
||||
|
||||
if (kind === "reasoning" && !input.thinking) {
|
||||
return out(data, commits)
|
||||
}
|
||||
|
||||
if (!ready(data, partID)) {
|
||||
return out(data, commits)
|
||||
}
|
||||
|
||||
flushPart(data, commits, partID)
|
||||
return out(data, commits)
|
||||
}
|
||||
|
||||
if (event.type === "message.part.updated") {
|
||||
const part = event.properties.part
|
||||
if (part.sessionID !== input.sessionID) {
|
||||
return out(data, commits)
|
||||
}
|
||||
|
||||
if (part.type === "tool") {
|
||||
const view = syncPermission(data, part) ?? syncQuestion(data, part)
|
||||
|
||||
if (part.state.status === "running") {
|
||||
if (data.ids.has(part.id)) {
|
||||
return out(data, commits, view)
|
||||
}
|
||||
|
||||
if (!data.tools.has(part.id)) {
|
||||
data.tools.add(part.id)
|
||||
commits.push(startTool(part))
|
||||
}
|
||||
|
||||
return out(data, commits, view ?? patch({ status: toolStatus(part) }))
|
||||
}
|
||||
|
||||
if (part.state.status === "completed") {
|
||||
const seen = data.tools.has(part.id)
|
||||
const mode = toolView(part.tool)
|
||||
data.tools.delete(part.id)
|
||||
if (data.ids.has(part.id)) {
|
||||
return out(data, commits, view)
|
||||
}
|
||||
|
||||
if (!seen) {
|
||||
commits.push(startTool(part))
|
||||
}
|
||||
|
||||
data.ids.add(part.id)
|
||||
stashEcho(data, part)
|
||||
|
||||
const output = part.state.output
|
||||
if (mode.output && typeof output === "string" && output.trim()) {
|
||||
commits.push({
|
||||
kind: "tool",
|
||||
text: output,
|
||||
phase: "progress",
|
||||
source: "tool",
|
||||
messageID: part.messageID,
|
||||
partID: part.id,
|
||||
tool: part.tool,
|
||||
part,
|
||||
toolState: "completed",
|
||||
})
|
||||
}
|
||||
|
||||
if (mode.final) {
|
||||
commits.push(doneTool(part))
|
||||
}
|
||||
|
||||
return out(data, commits, view)
|
||||
}
|
||||
|
||||
if (part.state.status === "error") {
|
||||
const seen = data.tools.has(part.id)
|
||||
data.tools.delete(part.id)
|
||||
if (data.ids.has(part.id)) {
|
||||
return out(data, commits, view)
|
||||
}
|
||||
|
||||
if (!seen) {
|
||||
commits.push(startTool(part))
|
||||
}
|
||||
|
||||
data.ids.add(part.id)
|
||||
const text =
|
||||
typeof part.state.error === "string" && part.state.error.trim() ? part.state.error : "unknown error"
|
||||
commits.push(failTool(part, text))
|
||||
return out(data, commits, view)
|
||||
}
|
||||
}
|
||||
|
||||
if (part.type !== "text" && part.type !== "reasoning") {
|
||||
return out(data, commits)
|
||||
}
|
||||
|
||||
if (data.ids.has(part.id)) {
|
||||
return out(data, commits)
|
||||
}
|
||||
|
||||
const kind = part.type === "text" ? "assistant" : "reasoning"
|
||||
if (typeof part.messageID === "string") {
|
||||
data.msg.set(part.id, part.messageID)
|
||||
}
|
||||
|
||||
const msg = part.messageID
|
||||
const role = msg ? data.role.get(msg) : undefined
|
||||
if (role === "user" && part.type === "text" && !data.includeUserText) {
|
||||
data.ids.add(part.id)
|
||||
drop(data, part.id)
|
||||
return out(data, commits)
|
||||
}
|
||||
|
||||
if (kind === "reasoning" && !input.thinking) {
|
||||
if (part.time?.end) {
|
||||
data.ids.add(part.id)
|
||||
}
|
||||
drop(data, part.id)
|
||||
return out(data, commits)
|
||||
}
|
||||
|
||||
data.part.set(part.id, role === "user" && kind === "assistant" ? "user" : kind)
|
||||
syncText(data, part.id, part.text)
|
||||
|
||||
if (part.time?.end) {
|
||||
data.end.add(part.id)
|
||||
}
|
||||
|
||||
if (msg && !role) {
|
||||
return out(data, commits)
|
||||
}
|
||||
|
||||
if (!ready(data, part.id)) {
|
||||
return out(data, commits)
|
||||
}
|
||||
|
||||
flushPart(data, commits, part.id)
|
||||
|
||||
if (!part.time?.end) {
|
||||
return out(data, commits)
|
||||
}
|
||||
|
||||
data.ids.add(part.id)
|
||||
drop(data, part.id)
|
||||
return out(data, commits)
|
||||
}
|
||||
|
||||
if (event.type === "permission.asked") {
|
||||
if (event.properties.sessionID !== input.sessionID) {
|
||||
return out(data, commits)
|
||||
}
|
||||
|
||||
upsert(data.permissions, enrichPermission(data, event.properties))
|
||||
return queueOut(data, commits)
|
||||
}
|
||||
|
||||
if (event.type === "permission.replied") {
|
||||
if (event.properties.sessionID !== input.sessionID) {
|
||||
return out(data, commits)
|
||||
}
|
||||
|
||||
if (!remove(data.permissions, event.properties.requestID)) {
|
||||
return out(data, commits)
|
||||
}
|
||||
|
||||
return queueOut(data, commits)
|
||||
}
|
||||
|
||||
if (event.type === "question.asked") {
|
||||
if (event.properties.sessionID !== input.sessionID) {
|
||||
return out(data, commits)
|
||||
}
|
||||
|
||||
upsert(data.questions, event.properties)
|
||||
return queueOut(data, commits)
|
||||
}
|
||||
|
||||
if (event.type === "question.replied" || event.type === "question.rejected") {
|
||||
if (event.properties.sessionID !== input.sessionID) {
|
||||
return out(data, commits)
|
||||
}
|
||||
|
||||
if (!remove(data.questions, event.properties.requestID)) {
|
||||
return out(data, commits)
|
||||
}
|
||||
|
||||
return queueOut(data, commits)
|
||||
}
|
||||
|
||||
if (event.type === "session.error") {
|
||||
if (event.properties.sessionID !== input.sessionID || !event.properties.error) {
|
||||
return out(data, commits)
|
||||
}
|
||||
|
||||
commits.push({
|
||||
kind: "error",
|
||||
text: formatError(event.properties.error),
|
||||
phase: "start",
|
||||
source: "system",
|
||||
})
|
||||
return out(data, commits)
|
||||
}
|
||||
|
||||
return out(data, commits)
|
||||
}
|
||||
Reference in New Issue
Block a user