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

run: render direct shell calls from stream events

Browse Source 
Consume the new session.next.shell.started/ended events instead so
direct shell calls render live, in order with the rest of the
stream, and without SDK fallback polling.

Track each call's source (shell event vs legacy bash tool part) so
whichever arrives first wins; the other is ignored to avoid double
commits when both representations reach the client.
This commit is contained in:
Simon Klee
2026-05-19 09:41:01 +02:00
parent cf3433334a
commit a838ff4c48
6 changed files with 426 additions and 106 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/opencode/src/cli/cmd/run/session-data.ts
+141
View File
@@ -61,13 +61,20 @@ type SessionCommit = StreamCommit
// - 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)
// - shell: shell call 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"
command?: string
}
export type SessionData = {
includeUserText: boolean
announced: boolean
ids: Set<string>
tools: Set<string>
call: Map<string, Dict>
shell: Map<string, ShellCall>
permissions: PermissionRequest[]
questions: QuestionRequest[]
role: Map<string, MessageRole>
@@ -104,6 +111,7 @@ export function createSessionData(
ids: new Set(),
tools: new Set(),
call: new Map(),
shell: new Map(),
permissions: [],
questions: [],
role: new Map(),
@@ -621,6 +629,87 @@ function toolCommit(
}
}
function shellPartID(callID: string): string {
return `shell:${callID}`
}
function claimShell(data: SessionData, callID: string, source: ShellCall["source"], command?: string): ShellCall {
const current = data.shell.get(callID)
if (current) {
if (command && !current.command) {
current.command = command
}
return current
}
const next = {
source,
...(command ? { command } : {}),
} satisfies ShellCall
data.shell.set(callID, next)
return next
}
function bashCommand(part: ToolPart): string | undefined {
if (part.tool !== "bash") {
return undefined
}
const input = part.state.input
if (!input || typeof input !== "object" || Array.isArray(input)) {
return undefined
}
const command = Reflect.get(input, "command")
return typeof command === "string" ? command : undefined
}
function shellCommit(
input: {
callID: string
command: string
},
next: Pick<SessionCommit, "text" | "phase" | "toolState">,
): SessionCommit {
return {
kind: "tool",
source: "tool",
partID: shellPartID(input.callID),
tool: "bash",
shell: input,
...next,
}
}
function startShell(callID: string, command: string): SessionCommit {
return shellCommit(
{
callID,
command,
},
{
text: "running shell",
phase: "start",
toolState: "running",
},
)
}
function doneShell(callID: string, command: string, output: string): SessionCommit {
return shellCommit(
{
callID,
command,
},
{
text: output,
phase: "progress",
toolState: "completed",
},
)
}
function startTool(part: ToolPart): SessionCommit {
return toolCommit(part, {
text: toolStatus(part),
@@ -681,6 +770,53 @@ export function reduceSessionData(input: SessionDataInput): SessionDataOutput {
const data = input.data
const event = input.event
if (event.type === "session.next.shell.started") {
if (event.properties.sessionID !== input.sessionID) {
return out(data, commits)
}
const shell = claimShell(data, event.properties.callID, "shell", event.properties.command)
if (shell.source !== "shell") {
return out(data, commits)
}
const partID = shellPartID(event.properties.callID)
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))
return out(data, commits, patch({ status: "running shell" }))
}
if (event.type === "session.next.shell.ended") {
if (event.properties.sessionID !== input.sessionID) {
return out(data, commits)
}
const shell = claimShell(data, event.properties.callID, "shell")
if (shell.source !== "shell") {
return out(data, commits)
}
const partID = shellPartID(event.properties.callID)
const seen = data.tools.has(partID)
const command = shell.command ?? ""
data.tools.delete(partID)
if (data.ids.has(partID)) {
return out(data, commits)
}
if (!seen && command) {
commits.push(startShell(event.properties.callID, command))
}
data.ids.add(partID)
commits.push(doneShell(event.properties.callID, command, event.properties.output))
return out(data, commits)
}
if (event.type === "message.updated") {
if (event.properties.sessionID !== input.sessionID) {
return out(data, commits)
@@ -782,6 +918,11 @@ export function reduceSessionData(input: SessionDataInput): SessionDataOutput {
if (part.type === "tool") {
const view = syncPermission(data, part) ?? syncQuestion(data, part)
if (part.tool === "bash" && part.callID) {
if (claimShell(data, part.callID, "tool", bashCommand(part)).source === "shell") {
return out(data, commits, view)
}
}
if (part.state.status === "running") {
if (data.ids.has(part.id)) {
packages/opencode/src/cli/cmd/run/stream.transport.ts
+2 -103
View File
@@ -75,23 +75,6 @@ type StreamInput = {
signal?: AbortSignal
}
type ShellMessage = NonNullable<Awaited<ReturnType<OpencodeClient["session"]["shell"]>>["data"]>
type SessionMessage = NonNullable<Awaited<ReturnType<OpencodeClient["session"]["messages"]>>["data"]>[number]
function isShellMessage(value: unknown): value is ShellMessage {
if (!value || typeof value !== "object" || Array.isArray(value)) {
return false
}
const info = Reflect.get(value, "info")
const parts = Reflect.get(value, "parts")
if (!info || typeof info !== "object" || !Array.isArray(parts)) {
return false
}
return Reflect.get(info, "role") === "assistant" && typeof Reflect.get(info, "sessionID") === "string"
}
type Wait = {
tick: number
armed: boolean
@@ -149,6 +132,8 @@ function sid(event: Event): string | undefined {
}
if (
event.type === "session.next.shell.started" ||
event.type === "session.next.shell.ended" ||
event.type === "permission.asked" ||
event.type === "permission.replied" ||
event.type === "question.asked" ||
@@ -530,81 +515,6 @@ function createLayer(input: StreamInput) {
state.footerView = current
}
const applyMessage = (event: Event) => {
const next = reduceSessionData({
data: state.data,
event,
sessionID: input.sessionID,
thinking: input.thinking,
limits: input.limits(),
})
state.data = next.data
syncFooter(next.commits, next.footer?.patch)
}
const applyShellResponse = (message: ShellMessage | SessionMessage | undefined) => {
if (!message || message.info.role !== "assistant" || message.info.sessionID !== input.sessionID) {
return
}
input.trace?.write("recv.shell", {
messageID: message.info.id,
parts: message.parts.length,
})
applyMessage({
type: "message.updated",
properties: {
sessionID: message.info.sessionID,
info: message.info,
},
} as Event)
for (const part of message.parts) {
if (part.type !== "tool") {
continue
}
applyMessage({
type: "message.part.updated",
properties: {
part,
},
} as Event)
}
}
const resolveShellMessage = Effect.fn("RunStreamTransport.resolveShellMessage")(function* (result: unknown) {
if (result && typeof result === "object") {
const data = Reflect.get(result, "data")
if (isShellMessage(data)) {
return data
}
if (isShellMessage(result)) {
return result
}
}
for (let attempt = 0; attempt < 5; attempt += 1) {
const list = yield* Effect.promise(() =>
input.sdk.session.messages({
sessionID: input.sessionID,
limit: 1,
}),
).pipe(Effect.map((item) => item.data ?? []), Effect.orElseSucceed(() => []))
const message = list.find(isShellMessage)
if (message) {
return message
}
if (attempt < 4) {
yield* Effect.sleep("50 millis")
}
}
return undefined
})
const resolveShellAgent = Effect.fn("RunStreamTransport.resolveShellAgent")(function* (agent: string | undefined) {
if (agent) {
return agent
@@ -1145,17 +1055,6 @@ function createLayer(input: StreamInput) {
item.live = true
}),
),
Effect.tap((result) =>
Effect.gen(function* () {
const message = yield* resolveShellMessage(result)
if (!message) {
input.trace?.write("recv.shell.miss")
return
}
applyShellResponse(message)
}),
),
Effect.flatMap(() => Deferred.succeed(item.done, undefined).pipe(Effect.ignore)),
Effect.catch((error) => Deferred.fail(item.done, error).pipe(Effect.ignore)),
Effect.forkIn(scope, { startImmediately: true }),
packages/opencode/src/cli/cmd/run/tool.ts
+28 -3
View File
@@ -35,7 +35,7 @@ import { webSearchProviderLabel, type WebSearchTool } from "@/tool/websearch"
import type { WriteTool } from "@/tool/write"
import { LANGUAGE_EXTENSIONS } from "@/lsp/language"
import * as Locale from "@/util/locale"
import type { RunDiffStyle, RunEntryBody, StreamCommit, ToolSnapshot } from "./types"
import type { RunEntryBody, StreamCommit, ToolSnapshot } from "./types"
export type ToolView = {
output: boolean
@@ -1252,7 +1252,7 @@ function frame(part: ToolPart): ToolFrame {
raw: "",
name: part.tool,
input: dict(state.input),
meta: dict(state.metadata),
meta: "metadata" in part.state ? dict(part.state.metadata) : {},
state,
status: text(state.status),
error: text(state.error),
@@ -1265,7 +1265,7 @@ export function toolFrame(commit: StreamCommit, raw: string): ToolFrame {
raw,
name: commit.tool || commit.part?.tool || "tool",
input: dict(state.input),
meta: dict(state.metadata),
meta: commit.part?.state && "metadata" in commit.part.state ? dict(commit.part.state.metadata) : {},
state,
status: commit.toolState ?? text(state.status),
error: (commit.toolError ?? "").trim(),
@@ -1407,7 +1407,32 @@ function structuredBody(commit: StreamCommit, raw: string): RunEntryBody | undef
}
}
function shellOutput(command: string, raw: string): string | undefined {
const body = stripAnsi(raw).replace(/^\n+/, "").replace(/\n+$/, "")
if (!body) {
return undefined
}
if (!command) {
return body
}
return `\n${body}`
}
export function toolEntryBody(commit: StreamCommit, raw: string): RunEntryBody | undefined {
if (commit.shell) {
if (commit.phase === "start") {
return textBody(`$ ${commit.shell.command}`)
}
if (commit.phase === "progress") {
return textBody(shellOutput(commit.shell.command, raw) ?? "")
}
return undefined
}
const ctx = toolFrame(commit, raw)
const view = toolView(ctx.name)
packages/opencode/src/cli/cmd/run/types.ts
+4
View File
@@ -303,6 +303,10 @@ export type StreamCommit = {
interrupted?: boolean
toolState?: StreamToolState
toolError?: string
shell?: {
callID: string
command: string
}
}
// The public contract between the stream transport / prompt queue and
packages/opencode/test/cli/run/entry.body.test.ts
+67
View File
@@ -359,6 +359,73 @@ describe("run entry body", () => {
})
})
test("renders command-only bash starts without the shell header", () => {
expect(
entryBody(
toolCommit({
tool: "bash",
phase: "start",
toolState: "running",
text: "running shell",
state: {
status: "running",
input: {
command: "ls",
},
time: { start: 1 },
},
}),
),
).toEqual({
type: "text",
content: "$ ls",
})
})
test("renders direct shell commits without a synthetic shell header", () => {
expect(
entryBody(
commit({
kind: "tool",
text: "running shell",
phase: "start",
source: "tool",
tool: "bash",
partID: "shell:call-1",
toolState: "running",
shell: {
callID: "call-1",
command: "pwd",
},
}),
),
).toEqual({
type: "text",
content: "$ pwd",
})
expect(
entryBody(
commit({
kind: "tool",
text: "/tmp/demo\n",
phase: "progress",
source: "tool",
tool: "bash",
partID: "shell:call-1",
toolState: "completed",
shell: {
callID: "call-1",
command: "pwd",
},
}),
),
).toEqual({
type: "text",
content: "\n/tmp/demo",
})
})
test("falls back to patch summary when apply_patch has no visible diff items", () => {
expect(
entryBody(
packages/opencode/test/cli/run/session-data.test.ts
+184
View File
@@ -326,6 +326,190 @@ describe("run session data", () => {
])
})
test("renders direct shell mode from first-class shell events", () => {
let data = createSessionData()
const started = reduce(data, {
type: "session.next.shell.started",
properties: {
sessionID: "session-1",
timestamp: 1,
callID: "call-1",
command: "pwd",
},
})
expect(started.commits).toEqual([
expect.objectContaining({
kind: "tool",
phase: "start",
partID: "shell:call-1",
tool: "bash",
shell: {
callID: "call-1",
command: "pwd",
},
}),
])
data = started.data
const ended = reduce(data, {
type: "session.next.shell.ended",
properties: {
sessionID: "session-1",
timestamp: 2,
callID: "call-1",
output: "/tmp/demo\n",
},
})
expect(ended.commits).toEqual([
expect.objectContaining({
kind: "tool",
phase: "progress",
partID: "shell:call-1",
tool: "bash",
text: "/tmp/demo\n",
toolState: "completed",
shell: {
callID: "call-1",
command: "pwd",
},
}),
])
})
test("suppresses legacy bash part updates once shell events claim the call", () => {
let data = reduce(createSessionData(), {
type: "session.next.shell.started",
properties: {
sessionID: "session-1",
timestamp: 1,
callID: "call-1",
command: "pwd",
},
}).data
expect(
reduce(
data,
tool({
id: "tool-1",
messageID: "msg-1",
callID: "call-1",
tool: "bash",
state: {
status: "running",
input: {
command: "pwd",
},
time: { start: 1 },
},
}),
).commits,
).toEqual([])
data = reduce(data, {
type: "session.next.shell.ended",
properties: {
sessionID: "session-1",
timestamp: 2,
callID: "call-1",
output: "/tmp/demo\n",
},
}).data
expect(
reduce(
data,
tool({
id: "tool-1",
messageID: "msg-1",
callID: "call-1",
tool: "bash",
state: {
status: "completed",
input: {
command: "pwd",
},
output: "/tmp/demo\n",
title: "",
metadata: {
output: "/tmp/demo\n",
description: "",
},
time: { start: 1, end: 2 },
},
}),
).commits,
).toEqual([])
})
test("suppresses shell events when the legacy bash part claimed the call first", () => {
let data = reduce(
createSessionData(),
tool({
id: "tool-1",
messageID: "msg-1",
callID: "call-1",
tool: "bash",
state: {
status: "running",
input: {
command: "pwd",
},
time: { start: 1 },
},
}),
).data
expect(
reduce(data, {
type: "session.next.shell.started",
properties: {
sessionID: "session-1",
timestamp: 1,
callID: "call-1",
command: "pwd",
},
}).commits,
).toEqual([])
data = reduce(
data,
tool({
id: "tool-1",
messageID: "msg-1",
callID: "call-1",
tool: "bash",
state: {
status: "completed",
input: {
command: "pwd",
},
output: "/tmp/demo\n",
title: "",
metadata: {
output: "/tmp/demo\n",
description: "",
},
time: { start: 1, end: 2 },
},
}),
).data
expect(
reduce(data, {
type: "session.next.shell.ended",
properties: {
sessionID: "session-1",
timestamp: 2,
callID: "call-1",
output: "/tmp/demo\n",
},
}).commits,
).toEqual([])
})
test("synthesizes a glob start before an error when the running update is missed", () => {
expect(
reduce(