Mirror_Repos/anomalyco_opencode
2
0
Fork 0
Code Issues Pull Requests Actions 71 Packages Projects Releases Wiki Activity

refactor(schema): session shell payloads and event prefix restore (#35229)

Browse Source
This commit is contained in:
Kit Langton
2026-07-03 17:30:25 -04:00
committed by GitHub
parent 64e4f6f91b
commit 650d774372
49 changed files with 1521 additions and 1200 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/opencode/src/cli/cmd/run/noninteractive.ts
+15 -15
View File
@@ -158,14 +158,14 @@ export async function runNonInteractivePrompt(input: Input) {
if (!("sessionID" in event.data) || event.data.sessionID !== input.sessionID) continue
const time = toMillis(event.created)
if (event.type === "prompt.promoted") {
if (event.type === "session.prompt.promoted") {
if (event.data.inputID === messageID) {
promoted = true
continue
}
}
if (
event.type === "execution.settled" &&
event.type === "session.execution.settled" &&
event.data.outcome === "interrupted" &&
(interrupted || permissionRejected || questionRejected || formCancelled)
) {
@@ -173,7 +173,7 @@ export async function runNonInteractivePrompt(input: Input) {
}
if (!promoted) continue
if (event.type === "step.started") {
if (event.type === "session.step.started") {
const part: StepStartPart = {
id: partID(event.id),
sessionID: input.sessionID,
@@ -189,11 +189,11 @@ export async function runNonInteractivePrompt(input: Input) {
continue
}
if (event.type === "text.started") {
if (event.type === "session.text.started") {
starts.set(event.data.textID, { id: partID(event.id), timestamp: time })
continue
}
if (event.type === "text.ended") {
if (event.type === "session.text.ended") {
const started = starts.get(event.data.textID)
const part: TextPart = {
id: started?.id ?? partID(event.id),
@@ -207,11 +207,11 @@ export async function runNonInteractivePrompt(input: Input) {
continue
}
if (event.type === "reasoning.started") {
if (event.type === "session.reasoning.started") {
starts.set(event.data.reasoningID, { id: partID(event.id), timestamp: time })
continue
}
if (event.type === "reasoning.ended" && input.thinking) {
if (event.type === "session.reasoning.ended" && input.thinking) {
const started = starts.get(event.data.reasoningID)
const part: ReasoningPart = {
id: started?.id ?? partID(event.id),
@@ -236,7 +236,7 @@ export async function runNonInteractivePrompt(input: Input) {
continue
}
if (event.type === "tool.input.started") {
if (event.type === "session.tool.input.started") {
tools.set(event.data.callID, {
id: partID(event.id),
timestamp: time,
@@ -246,12 +246,12 @@ export async function runNonInteractivePrompt(input: Input) {
})
continue
}
if (event.type === "tool.input.ended") {
if (event.type === "session.tool.input.ended") {
const current = tools.get(event.data.callID)
if (current) current.raw = event.data.text
continue
}
if (event.type === "tool.called") {
if (event.type === "session.tool.called") {
const current = tools.get(event.data.callID)
tools.set(event.data.callID, {
id: current?.id ?? partID(event.id),
@@ -264,7 +264,7 @@ export async function runNonInteractivePrompt(input: Input) {
})
continue
}
if (event.type === "tool.success") {
if (event.type === "session.tool.success") {
const current = tools.get(event.data.callID) ?? fallbackTool(event)
const part: ToolPart = {
id: current.id,
@@ -297,7 +297,7 @@ export async function runNonInteractivePrompt(input: Input) {
if (!emit("tool_use", time, { part })) await input.renderTool(part)
continue
}
if (event.type === "tool.failed") {
if (event.type === "session.tool.failed") {
const current = tools.get(event.data.callID) ?? fallbackTool(event)
const error = event.data.error.message
const part: ToolPart = {
@@ -328,7 +328,7 @@ export async function runNonInteractivePrompt(input: Input) {
continue
}
if (event.type === "step.ended") {
if (event.type === "session.step.ended") {
const part: StepFinishPart = {
id: partID(event.id),
sessionID: input.sessionID,
@@ -342,14 +342,14 @@ export async function runNonInteractivePrompt(input: Input) {
emit("step_finish", time, { part })
continue
}
if (event.type === "step.failed") {
if (event.type === "session.step.failed") {
if (interrupted || permissionRejected || questionRejected || formCancelled) continue
emittedError = true
process.exitCode = 1
if (!emit("error", time, { error: event.data.error })) UI.error(event.data.error.message)
continue
}
if (event.type === "execution.settled") {
if (event.type === "session.execution.settled") {
if (event.data.outcome === "failure" && !emittedError && !questionRejected && !formCancelled) {
emittedError = true
process.exitCode = 1
packages/opencode/src/cli/cmd/run/session-data.ts
+15 -15
View File
@@ -62,7 +62,7 @@ type SessionCommit = StreamCommit
// - sent: part ID → byte offset of last flushed text (for incremental output)
// - visible: part ID → rendered text for an active part after display transforms
// - end: part IDs whose time.end has arrived (part is finished)
// - shell: shell call ID → chosen transcript source for direct shell calls
// - shell: shell ID → chosen transcript source for direct shell calls
// - echo: message ID → bash outputs to strip from the next assistant chunk
type ShellCall = {
source: "shell" | "tool"
@@ -607,12 +607,12 @@ function toolCommit(
}
}
function shellPartID(callID: string): string {
return `shell:${callID}`
function shellPartID(shellID: string): string {
return `shell:${shellID}`
}
function claimShell(data: SessionData, callID: string, source: ShellCall["source"], command?: string): ShellCall {
const current = data.shell.get(callID)
function claimShell(data: SessionData, shellID: string, source: ShellCall["source"], command?: string): ShellCall {
const current = data.shell.get(shellID)
if (current) {
if (command && !current.command) {
current.command = command
@@ -625,7 +625,7 @@ function claimShell(data: SessionData, callID: string, source: ShellCall["source
source,
...(command ? { command } : {}),
} satisfies ShellCall
data.shell.set(callID, next)
data.shell.set(shellID, next)
return next
}
@@ -728,37 +728,37 @@ export function reduceSessionData(input: SessionDataInput): SessionDataOutput {
const data = input.data
const event = input.event
if (event.type === "shell.started") {
if (event.type === "session.shell.started") {
if (event.properties.sessionID !== input.sessionID) {
return out(data, commits)
}
const shell = claimShell(data, event.properties.callID, "shell", event.properties.command)
const shell = claimShell(data, event.properties.shell.id, "shell", event.properties.shell.command)
if (shell.source !== "shell") {
return out(data, commits)
}
const partID = shellPartID(event.properties.callID)
const partID = shellPartID(event.properties.shell.id)
if (data.ids.has(partID) || data.tools.has(partID)) {
return out(data, commits, patch({ status: "running shell" }))
}
data.tools.add(partID)
commits.push(startShell(event.properties.callID, shell.command ?? event.properties.command))
commits.push(startShell(event.properties.shell.id, shell.command ?? event.properties.shell.command))
return out(data, commits, patch({ status: "running shell" }))
}
if (event.type === "shell.ended") {
if (event.type === "session.shell.ended") {
if (event.properties.sessionID !== input.sessionID) {
return out(data, commits)
}
const shell = claimShell(data, event.properties.callID, "shell")
const shell = claimShell(data, event.properties.shell.id, "shell")
if (shell.source !== "shell") {
return out(data, commits)
}
const partID = shellPartID(event.properties.callID)
const partID = shellPartID(event.properties.shell.id)
const seen = data.tools.has(partID)
const command = shell.command ?? ""
data.tools.delete(partID)
@@ -767,11 +767,11 @@ export function reduceSessionData(input: SessionDataInput): SessionDataOutput {
}
if (!seen && command) {
commits.push(startShell(event.properties.callID, command))
commits.push(startShell(event.properties.shell.id, command))
}
data.ids.add(partID)
commits.push(doneShell(event.properties.callID, command, event.properties.output))
commits.push(doneShell(event.properties.shell.id, command, event.properties.output.output))
return out(data, commits)
}
packages/opencode/src/cli/cmd/run/stream-v2.subagent.ts
+15 -15
View File
@@ -424,21 +424,21 @@ export function createSubagentTracker(input: SubagentTrackerInput): SubagentTrac
}
const reduce = (child: ChildState, event: V2Event) => {
if (event.type === "prompt.promoted") {
if (event.type === "session.prompt.promoted") {
if (userFrame(child, event.data.inputID, "")) {
touch(child, event.created)
notifyDetail(child)
}
return
}
if (event.type === "step.started") {
if (event.type === "session.step.started") {
touch(child, event.created)
if (child.label === FALLBACK_LABEL && event.data.agent) child.label = Locale.titlecase(event.data.agent)
if (child.status !== "running") child.status = "running"
input.emit()
return
}
if (event.type === "text.delta") {
if (event.type === "session.text.delta") {
const projected = child.projectedText.get(event.data.textID)
const covered = projected?.indexOf(event.data.delta) ?? -1
if (projected && covered >= 0) {
@@ -459,7 +459,7 @@ export function createSubagentTracker(input: SubagentTrackerInput): SubagentTrac
notifyDetail(child)
return
}
if (event.type === "text.ended") {
if (event.type === "session.text.ended") {
child.text.set(event.data.textID, event.data.text)
child.projectedText.delete(event.data.textID)
setFrame(child, `text:${event.data.textID}`, {
@@ -474,7 +474,7 @@ export function createSubagentTracker(input: SubagentTrackerInput): SubagentTrac
notifyDetail(child)
return
}
if (event.type === "reasoning.delta") {
if (event.type === "session.reasoning.delta") {
const projected = child.projectedReasoning.get(event.data.reasoningID)
const covered = projected?.indexOf(event.data.delta) ?? -1
if (projected && covered >= 0) {
@@ -495,7 +495,7 @@ export function createSubagentTracker(input: SubagentTrackerInput): SubagentTrac
notifyDetail(child)
return
}
if (event.type === "reasoning.ended") {
if (event.type === "session.reasoning.ended") {
child.reasoning.set(event.data.reasoningID, event.data.text)
child.projectedReasoning.delete(event.data.reasoningID)
if (!input.thinking) return
@@ -510,11 +510,11 @@ export function createSubagentTracker(input: SubagentTrackerInput): SubagentTrac
notifyDetail(child)
return
}
if (event.type === "tool.input.started") {
if (event.type === "session.tool.input.started") {
child.tools.set(event.data.callID, { name: event.data.name, input: {}, started: event.created })
return
}
if (event.type === "tool.called") {
if (event.type === "session.tool.called") {
const current = child.tools.get(event.data.callID)
child.tools.set(event.data.callID, {
name: event.data.tool,
@@ -537,10 +537,10 @@ export function createSubagentTracker(input: SubagentTrackerInput): SubagentTrac
notifyDetail(child)
return
}
if (event.type === "tool.success" || event.type === "tool.failed") {
if (event.type === "session.tool.success" || event.type === "session.tool.failed") {
if (child.finishedTools.has(event.data.callID)) return
const current = child.tools.get(event.data.callID)
const failed = event.type === "tool.failed"
const failed = event.type === "session.tool.failed"
childTool(
child,
{
@@ -577,7 +577,7 @@ export function createSubagentTracker(input: SubagentTrackerInput): SubagentTrac
notifyDetail(child)
return
}
if (event.type === "step.failed") {
if (event.type === "session.step.failed") {
setFrame(child, `error:step:${event.data.assistantMessageID}`, {
kind: "error",
source: "system",
@@ -589,7 +589,7 @@ export function createSubagentTracker(input: SubagentTrackerInput): SubagentTrac
notifyDetail(child)
return
}
if (event.type === "execution.settled") {
if (event.type === "session.execution.settled") {
child.status =
event.data.outcome === "success" ? "completed" : event.data.outcome === "interrupted" ? "cancelled" : "error"
touch(child, event.created)
@@ -613,15 +613,15 @@ export function createSubagentTracker(input: SubagentTrackerInput): SubagentTrac
return {
main(event) {
if (event.type === "tool.called") {
if (event.type === "session.tool.called") {
if (event.data.tool === "subagent") pendingCalls.set(event.data.callID, event.data.input)
return
}
if (event.type === "tool.failed") {
if (event.type === "session.tool.failed") {
pendingCalls.delete(event.data.callID)
return
}
if (event.type !== "tool.success") return
if (event.type !== "session.tool.success") return
const pending = pendingCalls.get(event.data.callID)
pendingCalls.delete(event.data.callID)
const found = childSessionID(record(event.data.structured))
packages/opencode/src/cli/cmd/run/stream-v2.transport.ts
+77 -48
View File
@@ -213,7 +213,9 @@ function promptAgents(next: SessionTurnInput) {
? [
{
name: part.name,
source: part.source ? { start: part.source.start, end: part.source.end, text: part.source.value } : undefined,
source: part.source
? { start: part.source.start, end: part.source.end, text: part.source.value }
: undefined,
},
]
: [],
@@ -404,27 +406,39 @@ export async function createSessionTransport(input: StreamInput): Promise<Sessio
return
}
if (message.type === "shell") {
state.shellCommands.set(message.callID, message.command)
state.shellCommands.set(message.shell.id, message.shell.command)
const completed = message.time.completed !== undefined
if (!render) {
// Suppressed history: mark settled shells rendered so live redelivery
// stays silent. A still-running shell stays unmarked and renders in
// full when its live shell.ended event arrives.
if (completed) {
state.shellStarted.add(message.callID)
state.shellEnded.add(message.callID)
state.shellStarted.add(message.shell.id)
state.shellEnded.add(message.shell.id)
}
return
}
if (!state.shellStarted.has(message.callID)) {
state.shellStarted.add(message.callID)
write([shellCommit(message.callID, message.command, { text: "running shell", phase: "start", toolState: "running" })])
if (!state.shellStarted.has(message.shell.id)) {
state.shellStarted.add(message.shell.id)
write([
shellCommit(message.shell.id, message.shell.command, {
text: "running shell",
phase: "start",
toolState: "running",
}),
])
}
if (completed && !state.shellEnded.has(message.callID)) {
state.shellEnded.add(message.callID)
write([shellCommit(message.callID, message.command, { text: message.output, phase: "progress", toolState: "completed" })])
if (completed && message.output && !state.shellEnded.has(message.shell.id)) {
state.shellEnded.add(message.shell.id)
write([
shellCommit(message.shell.id, message.shell.command, {
text: message.output.output,
phase: "progress",
toolState: "completed",
}),
])
}
if (completed && state.shellWait?.callID === message.callID) state.shellWait.resolve()
if (completed && state.shellWait?.callID === message.shell.id) state.shellWait.resolve()
return
}
if (message.type !== "assistant") return
@@ -515,17 +529,17 @@ export async function createSessionTransport(input: StreamInput): Promise<Sessio
}
input.trace?.write("recv.event", event)
subagents.main(event)
if (event.type === "prompt.promoted") {
if (event.type === "session.prompt.promoted") {
if (state.wait?.messageID === event.data.inputID) state.wait.promoted = true
state.messageIDs.add(event.data.inputID)
write([], { phase: "running", status: "waiting for assistant" })
return
}
if (event.type === "step.started") {
if (event.type === "session.step.started") {
write([], { phase: "running", status: "assistant responding" })
return
}
if (event.type === "skill.activated") {
if (event.type === "session.skill.activated") {
const messageID = event.id.replace(/^evt_/, "msg_")
if (state.wait) state.wait.promoted = true
if (state.skillMessages.has(messageID)) return
@@ -533,38 +547,56 @@ export async function createSessionTransport(input: StreamInput): Promise<Sessio
write([skillCommit(messageID, event.data.name)])
return
}
if (event.type === "shell.started") {
state.shellCommands.set(event.data.callID, event.data.command)
if (event.type === "session.shell.started") {
state.shellCommands.set(event.data.shell.id, event.data.shell.command)
const wait = state.shellWait
if (wait && wait.callID === undefined) wait.callID = event.data.callID
if (state.shellStarted.has(event.data.callID)) return
state.shellStarted.add(event.data.callID)
write([shellCommit(event.data.callID, event.data.command, { text: "running shell", phase: "start", toolState: "running" })], {
phase: "running",
status: "running shell",
})
if (wait && wait.callID === undefined) wait.callID = event.data.shell.id
if (state.shellStarted.has(event.data.shell.id)) return
state.shellStarted.add(event.data.shell.id)
write(
[
shellCommit(event.data.shell.id, event.data.shell.command, {
text: "running shell",
phase: "start",
toolState: "running",
}),
],
{
phase: "running",
status: "running shell",
},
)
return
}
if (event.type === "shell.ended") {
const command = state.shellCommands.get(event.data.callID) ?? ""
if (event.type === "session.shell.ended") {
const command = state.shellCommands.get(event.data.shell.id) ?? event.data.shell.command
const commits: StreamCommit[] = []
if (!state.shellStarted.has(event.data.callID)) {
state.shellStarted.add(event.data.callID)
if (command) commits.push(shellCommit(event.data.callID, command, { text: "running shell", phase: "start", toolState: "running" }))
if (!state.shellStarted.has(event.data.shell.id)) {
state.shellStarted.add(event.data.shell.id)
if (command)
commits.push(
shellCommit(event.data.shell.id, command, { text: "running shell", phase: "start", toolState: "running" }),
)
}
if (!state.shellEnded.has(event.data.callID)) {
state.shellEnded.add(event.data.callID)
commits.push(shellCommit(event.data.callID, command, { text: event.data.output, phase: "progress", toolState: "completed" }))
if (!state.shellEnded.has(event.data.shell.id)) {
state.shellEnded.add(event.data.shell.id)
commits.push(
shellCommit(event.data.shell.id, command, {
text: event.data.output.output,
phase: "progress",
toolState: "completed",
}),
)
}
const wait = state.shellWait
// An unset callID means shell.started has not been observed yet (event
// delivery lag); mini serializes its own shells, so adopt this ended.
const owned = wait !== undefined && (wait.callID === undefined || wait.callID === event.data.callID)
const owned = wait !== undefined && (wait.callID === undefined || wait.callID === event.data.shell.id)
write(commits, owned || state.wait ? undefined : { phase: "idle", status: "" })
if (owned) wait.resolve()
return
}
if (event.type === "text.delta") {
if (event.type === "session.text.delta") {
const key = streamPartKey(event.data.assistantMessageID, event.data.textID)
const projected = state.projectedText.get(key)
const covered = projected?.indexOf(event.data.delta) ?? -1
@@ -586,7 +618,7 @@ export async function createSessionTransport(input: StreamInput): Promise<Sessio
])
return
}
if (event.type === "text.ended") {
if (event.type === "session.text.ended") {
const key = streamPartKey(event.data.assistantMessageID, event.data.textID)
const previous = state.text.get(key) ?? ""
if (event.data.text.length > previous.length)
@@ -604,7 +636,7 @@ export async function createSessionTransport(input: StreamInput): Promise<Sessio
state.projectedText.delete(key)
return
}
if (event.type === "reasoning.delta") {
if (event.type === "session.reasoning.delta") {
const key = streamPartKey(event.data.assistantMessageID, event.data.reasoningID)
const projected = state.projectedReasoning.get(key)
const covered = projected?.indexOf(event.data.delta) ?? -1
@@ -627,7 +659,7 @@ export async function createSessionTransport(input: StreamInput): Promise<Sessio
])
return
}
if (event.type === "reasoning.ended") {
if (event.type === "session.reasoning.ended") {
const key = streamPartKey(event.data.assistantMessageID, event.data.reasoningID)
const previous = state.reasoning.get(key) ?? ""
if (input.thinking && event.data.text.length > previous.length)
@@ -645,7 +677,7 @@ export async function createSessionTransport(input: StreamInput): Promise<Sessio
state.projectedReasoning.delete(key)
return
}
if (event.type === "tool.input.started") {
if (event.type === "session.tool.input.started") {
state.tools.set(event.data.callID, {
messageID: event.data.assistantMessageID,
name: event.data.name,
@@ -655,7 +687,7 @@ export async function createSessionTransport(input: StreamInput): Promise<Sessio
})
return
}
if (event.type === "tool.called") {
if (event.type === "session.tool.called") {
if (state.finishedTools.has(event.data.callID)) return
const current = state.tools.get(event.data.callID)
const item: SessionMessageAssistantTool = {
@@ -669,10 +701,10 @@ export async function createSessionTransport(input: StreamInput): Promise<Sessio
renderTool(event.data.assistantMessageID, item)
return
}
if (event.type === "tool.progress") return
if (event.type === "tool.success" || event.type === "tool.failed") {
if (event.type === "session.tool.progress") return
if (event.type === "session.tool.success" || event.type === "session.tool.failed") {
const current = state.tools.get(event.data.callID)
const failed = event.type === "tool.failed"
const failed = event.type === "session.tool.failed"
const item: SessionMessageAssistantTool = {
type: "tool",
id: event.data.callID,
@@ -720,7 +752,7 @@ export async function createSessionTransport(input: StreamInput): Promise<Sessio
syncBlockers()
return
}
if (event.type === "step.ended") {
if (event.type === "session.step.ended") {
const total =
event.data.tokens.input +
event.data.tokens.output +
@@ -734,13 +766,13 @@ export async function createSessionTransport(input: StreamInput): Promise<Sessio
})
return
}
if (event.type === "step.failed") {
if (event.type === "session.step.failed") {
state.errors.add(event.data.assistantMessageID)
if (state.wait) state.wait.failureRendered = true
write([{ kind: "error", source: "system", text: errorMessage(event.data.error), phase: "start" }])
return
}
if (event.type === "execution.settled") {
if (event.type === "session.execution.settled") {
write([], { phase: "idle", status: "" })
const current = state.wait
if (!current || (!current.promoted && !current.interrupted)) return
@@ -977,10 +1009,7 @@ export async function createSessionTransport(input: StreamInput): Promise<Sessio
sessionID: input.sessionID,
id: messageID,
prompt: {
text: [
next.prompt.text,
...prepared.flatMap((file) => (file.text ? [file.text] : [])),
].join("\n\n"),
text: [next.prompt.text, ...prepared.flatMap((file) => (file.text ? [file.text] : []))].join("\n\n"),
files: attachments.length ? attachments : undefined,
agents: agents.length ? agents : undefined,
},