anomalyco/opencode
1
0
Fork 0
mirror of https://github.com/anomalyco/opencode.git synced 2026-08-24 06:33:01 -04:00
Code Issues 5.9k Packages Projects Releases 19 Wiki Activity

fix(command): simplify callback consumers

Browse Source
This commit is contained in:
Dax Raad
2026-08-19 20:13:52 -04:00
parent 2efe5d1034
commit 34c29df60d
15 changed files with 2458 additions and 781 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/cli/src/acp/event.ts
+6
View File
@@ -76,6 +76,7 @@ export async function streamTurn(input: {
readonly cwd: string
readonly start: TurnStart
readonly writeTextFile: boolean
readonly action?: boolean
readonly submit: (signal: AbortSignal) => Promise<unknown>
readonly control: TurnControl
readonly childSessionUpdate?: (update: ChildSessionUpdate) => Promise<void>
@@ -345,6 +346,11 @@ export async function streamTurn(input: {
await input.submit(control.admission.signal).catch((error) => {
if (!control.cancelled) throw error
})
if (input.action) {
streamController.abort()
await completed.catch(() => {})
return response(undefined, undefined, "succeeded", control.cancelled, undefined)
}
if (control.cancelled) {
await input.client.session.interrupt({ sessionID: input.sessionID }).catch(() => {})
if (!started) {
packages/cli/src/acp/service.ts
+1
View File
@@ -326,6 +326,7 @@ export function make(input: { readonly client: OpenCodeClient; readonly connecti
cwd: state.cwd,
start: prepared.start,
writeTextFile: capabilities.writeTextFile,
action: prepared.command !== undefined,
control,
connectionSignal: input.connection.signal,
sessionSignal: state.abort.signal,
packages/cli/test/acp/event.test.ts
+39
View File
@@ -121,3 +121,42 @@ test("acp prompt resolves after ordered turn updates", async () => {
controller.enqueue(encoder.encode(`data: ${JSON.stringify(event)}\n\n`))
}
})
test("acp action resolves without prompt lifecycle events", async () => {
const encoder = new TextEncoder()
const server = Bun.serve({
port: 0,
fetch(request) {
if (new URL(request.url).pathname !== "/api/event") return new Response(null, { status: 404 })
return new Response(
new ReadableStream<Uint8Array>({
start(controller) {
controller.enqueue(encoder.encode(`data: ${JSON.stringify({ type: "server.connected", data: {} })}\n\n`))
},
}),
{ headers: { "content-type": "text/event-stream" } },
)
},
})
try {
const response = await streamTurn({
client: OpenCode.make({ baseUrl: server.url.toString() }),
connection: {
sessionUpdate: async () => {},
requestPermission: async () => ({ outcome: { outcome: "cancelled" } }),
},
sessionID: "ses_test",
cwd: "/workspace",
start: { type: "input", id: "msg_action" },
writeTextFile: false,
action: true,
control: { cancelled: false, admission: new AbortController() },
submit: async () => {},
})
expect(response).toMatchObject({ stopReason: "end_turn" })
} finally {
await server.stop(true)
}
})
packages/core/src/command.ts
+11 -4
View File
@@ -5,7 +5,7 @@ import type { PromptInput } from "@opencode-ai/schema/prompt-input"
import type { Session } from "@opencode-ai/schema/session"
import type { SessionInbox } from "@opencode-ai/schema/session-inbox"
import { makeLocationNode } from "@opencode-ai/util/effect/app-node"
import { Cause, Context, Effect, Layer, Schema } from "effect"
import { Context, Effect, Layer, Schema } from "effect"
import { Bus } from "./bus.js"
import { State } from "./state.js"
@@ -83,9 +83,8 @@ export const layer = Layer.effect(
if (!definition)
return yield* new NotFoundError({ command: input.name, message: `Command not found: ${input.name}` })
return yield* definition.execute(input.invocation).pipe(
Effect.mapError(
(error) => new ExecutionError({ command: input.name, message: Cause.pretty(Cause.fail(error)) }),
),
Effect.tapError((error) => Effect.logError("command execution failed", { command: input.name, error })),
Effect.mapError((error) => new ExecutionError({ command: input.name, message: errorMessage(error) })),
)
}),
})
@@ -97,3 +96,11 @@ export const node = makeLocationNode({
layer,
deps: [Bus.node],
})
function errorMessage(error: unknown) {
if (error instanceof Error) return error.message
if (typeof error === "string") return error
if (error && typeof error === "object" && "message" in error && typeof error.message === "string")
return error.message
return "Command execution failed"
}
packages/core/src/config/plugin/command.ts
+20 -13
View File
@@ -6,13 +6,13 @@ import { Info, type Entry } from "@opencode-ai/schema/config"
import { ConfigCommand } from "@opencode-ai/schema/config/command"
import { Model } from "@opencode-ai/schema/model"
import { Provider } from "@opencode-ai/schema/provider"
import { Global } from "@opencode-ai/util/global"
import { AppProcess } from "@opencode-ai/util/process"
import path from "path"
import { Effect, Option, Schema, Stream } from "effect"
import { ChildProcess } from "effect/unstable/process"
import { Config } from "../../config.js"
import { Location } from "../../location.js"
import { Shell } from "../../shell.js"
import { ShellSelect } from "../../shell/select.js"
import { FSUtil } from "@opencode-ai/util/fs-util"
import { ConfigMarkdown } from "../markdown.js"
@@ -24,9 +24,9 @@ export const Plugin = define({
effect: Effect.fn(function* (ctx) {
const config = yield* Config.Service
const fs = yield* FSUtil.Service
const global = yield* Global.Service
const location = yield* Location.Service
const processes = yield* AppProcess.Service
const shell = yield* Shell.Service
const load = Effect.fn("ConfigCommandPlugin.load")(function* () {
return yield* Effect.forEach(yield* config.entries(), (entry) => {
if (entry.type === "document") return Effect.succeed([{ commands: entry.info.commands }])
@@ -69,10 +69,12 @@ export const Plugin = define({
execute: (input) =>
Effect.gen(function* () {
const agent = command.agent === undefined ? undefined : Agent.ID.make(command.agent)
const session = yield* ctx.session.get({ sessionID: input.sessionID })
if (agent !== undefined && session.agent !== agent)
yield* ctx.session.switchAgent({ sessionID: input.sessionID, agent })
const commandAgent = agent === undefined ? undefined : (yield* ctx.agent.get({ agentID: agent })).data
const commandAgent = yield* Effect.gen(function* () {
if (agent === undefined) return
const session = yield* ctx.session.get({ sessionID: input.sessionID })
if (session.agent !== agent) yield* ctx.session.switchAgent({ sessionID: input.sessionID, agent })
return (yield* ctx.agent.get({ agentID: agent })).data
})
const model =
command.model === undefined
? commandAgent?.model
@@ -91,7 +93,7 @@ export const Plugin = define({
config,
location,
processes,
bin: global.bin,
shell,
}),
delivery: input.delivery,
})
@@ -156,7 +158,7 @@ function evaluateTemplate(
readonly config: Config.Interface
readonly location: Location.Info
readonly processes: AppProcess.Interface
readonly bin: string
readonly shell: Shell.Interface
},
) {
return Effect.gen(function* () {
@@ -177,13 +179,14 @@ function evaluateTemplate(
: withArguments.trim()
const matches = Array.from(text.matchAll(shellRegex))
if (matches.length === 0) return text
const shell = ShellSelect.preferred(Config.latest(yield* services.config.entries(), "shell"), undefined, services.bin)
const shell = yield* services.shell.name()
const outputs = yield* Effect.forEach(
matches,
(match) =>
services.processes
(match) => {
const source = match[1] ?? ""
return services.processes
.run(
ChildProcess.make(shell, ShellSelect.args(shell, match[1] ?? ""), {
ChildProcess.make(shell, ShellSelect.args(shell, source), {
cwd: services.location.directory,
stdin: "ignore",
}),
@@ -191,7 +194,11 @@ function evaluateTemplate(
)
.pipe(
Effect.map((result) => (result.output ?? Buffer.concat([result.stdout, result.stderr])).toString("utf8")),
),
Effect.mapError((error) =>
new Error(`Shell interpolation failed for ${JSON.stringify(source)}: ${error.message}`),
),
)
},
{ concurrency: 2 },
)
const iterator = outputs[Symbol.iterator]()
packages/core/src/mcp/index.ts
+4 -2
View File
@@ -2,6 +2,7 @@ export * as MCP from "./index.js"
import { Mcp } from "@opencode-ai/schema/mcp"
import { McpEvent } from "@opencode-ai/schema/mcp-event"
import { ephemeral } from "@opencode-ai/schema/event"
import { createHash } from "node:crypto"
import { isDeepStrictEqual } from "node:util"
import { Cause, Context, Deferred, Effect, Exit, FiberSet, Layer, Schema, Scope, Stream, Types } from "effect"
@@ -18,6 +19,7 @@ import { State } from "../state.js"
import type { MCPClient } from "./client.js"
export const ServerName = Schema.String.pipe(Schema.brand("MCP.ServerName"))
export const PromptsChanged = ephemeral({ type: "mcp.prompts.changed", schema: { server: Schema.String } })
export type ServerName = typeof ServerName.Type
// The status union is a public wire contract, so it lives in @opencode-ai/schema and is re-exported here.
@@ -427,7 +429,7 @@ export const layer = (options?: Options) =>
Effect.map((defs) => {
entry.prompts = defs.map((def) => toPrompt(name, def))
}),
Effect.andThen(bus.publish(McpEvent.PromptsChanged, { server: name })),
Effect.andThen(bus.publish(PromptsChanged, { server: name })),
)
// Runs a connection callback under the server lock, dropping it if the connection is no longer
@@ -546,7 +548,7 @@ export const layer = (options?: Options) =>
yield* Scope.close(scope, Exit.void)
yield* bus.publish(McpEvent.ToolsChanged, { server: name }).pipe(Effect.ignore)
yield* bus.publish(McpEvent.ResourcesChanged, { server: name }).pipe(Effect.ignore)
yield* bus.publish(McpEvent.PromptsChanged, { server: name }).pipe(Effect.ignore)
yield* bus.publish(PromptsChanged, { server: name }).pipe(Effect.ignore)
})
const disposeServer = Effect.fnUntraced(function* (name: ServerName, entry: ServerEntry) {
packages/core/src/plugin/command.ts
+1 -2
View File
@@ -1,7 +1,6 @@
export * as CommandPlugin from "./command.js"
import { define } from "@opencode-ai/plugin/effect/plugin"
import { McpEvent } from "@opencode-ai/schema/mcp-event"
import { Effect, Stream } from "effect"
import { Bus } from "../bus.js"
import { Location } from "../location.js"
@@ -17,7 +16,7 @@ export const Plugin = define({
const bus = yield* Bus.Service
const loaded = { prompts: [] as MCP.Prompt[] }
yield* bus
.subscribe(McpEvent.PromptsChanged)
.subscribe(MCP.PromptsChanged)
.pipe(
Stream.runForEach(() =>
mcp.prompts().pipe(
packages/core/test/command.test.ts
+24
View File
@@ -44,4 +44,28 @@ describe("Command", () => {
expect(yield* command.list()).toEqual([Command.Info.make({ name: "goal", description: "Second" })])
}),
)
it.effect("returns callback error messages without stack traces", () =>
Effect.gen(function* () {
const command = yield* Command.Service
yield* command.transform((draft) => {
draft.add({
name: "fail",
execute: () => Effect.fail(new Error("command failed")),
})
})
const error = yield* command
.execute({
name: "fail",
invocation: {
sessionID: Session.ID.make("ses_test"),
prompt: { text: "" },
delivery: "steer",
},
})
.pipe(Effect.flip)
expect(error).toMatchObject({ _tag: "Command.ExecutionError", message: "command failed" })
}),
)
})
packages/core/test/config/command.test.ts
+75 -15
View File
@@ -1,9 +1,12 @@
import fs from "fs/promises"
import path from "path"
import { describe, expect } from "bun:test"
import { Deferred, Effect, Fiber, Layer, Option, PubSub, Schema, Stream } from "effect"
import { DateTime, Deferred, Effect, Fiber, Layer, Option, PubSub, Schema, Stream } from "effect"
import { advance, drain } from "../lib/clock"
import { Directory, Document, Event, Info } from "@opencode-ai/schema/config"
import { Session } from "@opencode-ai/schema/session"
import { SessionInbox } from "@opencode-ai/schema/session-inbox"
import { SessionMessage } from "@opencode-ai/schema/session-message"
import { Command } from "@opencode-ai/core/command"
import { Config } from "@opencode-ai/core/config"
import { ConfigCommandPlugin } from "@opencode-ai/core/config/plugin/command"
@@ -18,6 +21,7 @@ import { AppProcess } from "@opencode-ai/util/process"
import { Location } from "@opencode-ai/core/location"
import { MCP } from "@opencode-ai/core/mcp/index"
import { AbsolutePath } from "@opencode-ai/core/schema"
import { Shell } from "@opencode-ai/core/shell"
import { Watcher } from "@opencode-ai/core/filesystem/watcher"
import { emptyCredentialNode, emptyWellknownNode } from "../fixture/config-nodes"
import { emptyConfigLayer, emptyMcpLayer, testLocationLayer } from "../fixture/mcp"
@@ -26,13 +30,28 @@ import { tmpdir } from "../fixture/tmpdir"
import { testEffect } from "../lib/effect"
import { host } from "../plugin/host"
const shellLayer = Layer.succeed(
Shell.Service,
Shell.Service.of({
name: () => Effect.succeed("sh"),
create: () => Effect.die("unused shell.create"),
list: () => Effect.die("unused shell.list"),
get: () => Effect.die("unused shell.get"),
wait: () => Effect.die("unused shell.wait"),
timeout: () => Effect.die("unused shell.timeout"),
output: () => Effect.die("unused shell.output"),
remove: () => Effect.die("unused shell.remove"),
}),
)
const it = testEffect(
AppNodeBuilder.build(
LayerNode.group([Command.node, Bus.node, FSUtil.node, AppProcess.node, Global.node, Location.node]),
LayerNode.group([Command.node, Bus.node, FSUtil.node, AppProcess.node, Location.node, Shell.node]),
[
[MCP.node, emptyMcpLayer],
[Config.node, emptyConfigLayer],
[Location.node, testLocationLayer],
[MCP.node, emptyMcpLayer],
[Config.node, emptyConfigLayer],
[Location.node, testLocationLayer],
[Shell.node, shellLayer],
],
),
)
@@ -66,6 +85,7 @@ Review files`,
const bus = yield* Bus.Service
const update = yield* bus.publish(Event.Updated, {})
const updates = yield* PubSub.unbounded<typeof update>()
const prompts: { text: string; files?: readonly { readonly uri: string }[]; delivery?: string }[] = []
yield* ConfigCommandPlugin.Plugin.effect(
host({
command: {
@@ -74,6 +94,20 @@ Review files`,
reload: command.reload,
},
event: { subscribe: () => Stream.fromPubSub(updates) },
session: {
prompt: (input) =>
Effect.sync(() => {
prompts.push({ text: input.text, files: input.files, delivery: input.delivery })
return SessionInbox.User.make({
id: SessionMessage.ID.make("msg_test"),
sessionID: input.sessionID,
timeCreated: DateTime.makeUnsafe(0),
type: "user",
payload: { text: input.text },
delivery: input.delivery ?? "steer",
})
}),
},
}),
).pipe(
Effect.provide(
@@ -95,6 +129,21 @@ Review files`,
Command.Info.make({ name: "empty" }),
Command.Info.make({ name: "nested/docs" }),
])
yield* command.execute({
name: "nested/docs",
invocation: {
sessionID: Session.ID.make("ses_test"),
prompt: { text: "details", files: [{ uri: "file:///tmp/context.md" }] },
delivery: "queue",
},
})
expect(prompts).toEqual([
{
text: "Write docs\n\ndetails",
files: [{ uri: "file:///tmp/context.md" }],
delivery: "queue",
},
])
yield* Effect.promise(() =>
fs.writeFile(path.join(tmp.path, "commands", "review.md"), markdown("Review again", "Review again")),
@@ -106,6 +155,15 @@ Review files`,
yield* Effect.sleep("10 millis")
}
expect((yield* command.get("review"))?.description).toBe("Review again")
yield* command.execute({
name: "review",
invocation: {
sessionID: Session.ID.make("ses_test"),
prompt: { text: "latest" },
delivery: "steer",
},
})
expect(prompts.at(-1)?.text).toBe("Review again\n\nlatest")
}),
),
),
@@ -296,18 +354,20 @@ describeNative("ConfigCommandPlugin native watcher", () => {
AppProcess.node,
Global.node,
Location.node,
Shell.node,
]),
[
[
Location.node,
Layer.succeed(
Location.Service,
Location.Service.of(location({ directory: AbsolutePath.make(path.join(tmp, "project")) })),
),
],
[Global.node, Global.layerWith({ config: global, home: path.join(global, "home") })],
[Credential.node, emptyCredentialNode],
[WellKnown.node, emptyWellknownNode],
[
Location.node,
Layer.succeed(
Location.Service,
Location.Service.of(location({ directory: AbsolutePath.make(path.join(tmp, "project")) })),
),
],
[Global.node, Global.layerWith({ config: global, home: path.join(global, "home") })],
[Shell.node, shellLayer],
[Credential.node, emptyCredentialNode],
[WellKnown.node, emptyWellknownNode],
],
),
),
packages/protocol/openapi.json
+2265 -688
View File
File diff suppressed because it is too large Load Diff
packages/schema/src/mcp-event.ts
+1 -8
View File
@@ -17,13 +17,6 @@ export const ResourcesChanged = Event.ephemeral({
},
})
export const PromptsChanged = Event.ephemeral({
type: "mcp.prompts.changed",
schema: {
server: Schema.String,
},
})
// Emitted whenever a server's connection status settles (connected, failed, needs_auth, closed) so
// observers can refresh status without polling.
export const StatusChanged = Event.ephemeral({
@@ -33,4 +26,4 @@ export const StatusChanged = Event.ephemeral({
},
})
export const Definitions = Event.inventory(ToolsChanged, ResourcesChanged, PromptsChanged, StatusChanged)
export const Definitions = Event.inventory(ToolsChanged, ResourcesChanged, StatusChanged)
packages/schema/test/event-manifest.test.ts
+1 -6
View File
@@ -66,12 +66,7 @@ describe("public event manifest", () => {
expect(Form.Event.Definitions).toEqual([Form.Event.Created, Form.Event.Replied, Form.Event.Cancelled])
expect(Reference.Event.Definitions).toEqual([Reference.Event.Updated])
expect(Plugin.Event.Definitions).toEqual([Plugin.Event.Added, Plugin.Event.Updated])
expect(McpEvent.Definitions).toEqual([
McpEvent.ToolsChanged,
McpEvent.ResourcesChanged,
McpEvent.PromptsChanged,
McpEvent.StatusChanged,
])
expect(McpEvent.Definitions).toEqual([McpEvent.ToolsChanged, McpEvent.ResourcesChanged, McpEvent.StatusChanged])
expect(EventManifest.Latest.has("mcp.browser.open.failed")).toBe(false)
expect(EventManifest.Latest.has("ide.installed")).toBe(false)
expect(IdeEvent.Definitions).toEqual([IdeEvent.Installed])
packages/tui/src/component/prompt/index.tsx
+5 -21
View File
@@ -1230,23 +1230,8 @@ export function Prompt(props: PromptProps) {
})
setStore("mode", "normal")
} else if (slashHead && isCommand) {
move.startSubmit()
const model = { providerID: selection.providerID, id: selection.modelID, variant }
const cancelCommit = local.model.trackSessionCommit(sessionID, model)
void (async () => {
if (!session) {
await data.session.sync(sessionID)
session = data.session.get(sessionID)
}
if (session?.agent !== agent.id) await client.api.session.switchAgent({ sessionID, agent: agent.id })
if (
session?.model?.providerID !== model.providerID ||
session.model.id !== model.id ||
session.model.variant !== model.variant
)
await client.api.session.switchModel({ sessionID, model })
await client.api.session.command({
void client.api.session
.command({
sessionID,
command: slashHead.name,
text: slashHead.arguments,
@@ -1255,10 +1240,9 @@ export function Prompt(props: PromptProps) {
skills: store.prompt.skills?.length ? store.prompt.skills : undefined,
delivery,
})
})().catch((error) => {
cancelCommit()
toast.show({ title: "Failed to run command", message: errorMessage(error), variant: "error" })
})
.catch((error) => {
toast.show({ title: "Failed to run command", message: errorMessage(error), variant: "error" })
})
} else if (isSkill) {
move.startSubmit()
void client.api.session.skill({
packages/tui/src/mini/stream-v2.transport.ts
+3 -14
View File
@@ -1647,12 +1647,6 @@ export async function createSessionTransport(input: StreamInput): Promise<Sessio
)
}
const selected = await resolveSelectedModel(input, client, next)
if (next.variant && !selected) throw new Error("Cannot select a variant before selecting a model")
if (next.agent)
await client.session.switchAgent({ sessionID: input.sessionID, agent: next.agent }, { signal: next.signal })
if (selected)
await client.session.switchModel({ sessionID: input.sessionID, model: selected }, { signal: next.signal })
input.trace?.write("send.command", { sessionID: input.sessionID, messageID, command: command.name, delivery })
return client.session.command(
{
@@ -1699,7 +1693,7 @@ export async function createSessionTransport(input: StreamInput): Promise<Sessio
throw new Error("This prompt cannot be queued")
if (!state.connected) throw new Error("Event stream is reconnecting")
const client = sdk
if (next.agent)
if (!next.prompt.command && next.agent)
await client.session.switchAgent({ sessionID: input.sessionID, agent: next.agent }, { signal: next.signal })
if (!next.prompt.command) {
const selected = await resolveSelectedModel(input, client, next)
@@ -1746,13 +1740,8 @@ export async function createSessionTransport(input: StreamInput): Promise<Sessio
return
}
if (command) {
await runTurnWait(
next,
messageID,
client,
() => admitPrompt(next, client, next.prompt.delivery ?? "steer"),
admitted,
)
await admitPrompt(next, client, next.prompt.delivery ?? "steer")
admitted?.()
return
}
packages/tui/test/mini/stream-v2.transport.test.ts
+2 -8
View File
@@ -2858,14 +2858,8 @@ describe("V2 mini transport", () => {
skills: [{ id: "api-design", mention: { start: 13, end: 24, text: "/api-design" } }],
delivery: "steer",
})
expect(client.session.switchAgent).toHaveBeenCalledWith(
{ sessionID: "ses_1", agent: "build" },
expect.anything(),
)
expect(client.session.switchModel).toHaveBeenCalledWith(
{ sessionID: "ses_1", model: { providerID: "test", id: "model", variant: undefined } },
expect.anything(),
)
expect(client.session.switchAgent).not.toHaveBeenCalled()
expect(client.session.switchModel).not.toHaveBeenCalled()
await transport.close()
})