Compare commits

..
69 changed files with 1254 additions and 1203 deletions
+5
View File
@@ -0,0 +1,5 @@
---
"@opencode-ai/core": patch
---
Simplify interrupt continuation: the steer-scoped resume decision now lives in SessionExecution as a post-cleanup inbox check, and the run coordinator drops its continuation state machine. Wakes arriving during cancellation cleanup now restart a normal full drain, and interrupting an idle session with continue now resumes pending steering input. Recovery-applied moves now end with the same full wake as inbox-admitted moves, retrying any stranded inbox work at the new location.
+6
View File
@@ -0,0 +1,6 @@
---
"@opencode-ai/plugin": patch
"@opencode-ai/core": patch
---
Add transport-neutral Session model request hooks and provider-scoped hook registration so eligible OpenAI Responses requests can prefer WebSocket without bypassing HTTP-only middleware.
+1
View File
@@ -82,6 +82,7 @@ jobs:
build-cli:
needs: version
runs-on: blacksmith-4vcpu-ubuntu-2404
timeout-minutes: 30
if: github.repository == 'anomalyco/opencode'
steps:
- uses: actions/checkout@f43a0e5ff2bd294095638e18286ca9a3d1956744 # v3.6.0
@@ -905,6 +905,7 @@ const onContentBlockDelta = Effect.fn("AnthropicMessages.onContentBlockDelta")(f
if (delta?.type === "input_json_delta" && event.index !== undefined) {
if (!delta.partial_json) return [state, NO_EVENTS] satisfies StepResult
if (!state.tools[event.index]) return [state, NO_EVENTS] satisfies StepResult
const result = ToolStream.appendExisting(
ADAPTER,
state.tools,
@@ -1010,6 +1010,34 @@ describe("Anthropic Messages route", () => {
}),
)
it.effect("ignores tool input deltas without a matching tool start", () =>
Effect.gen(function* () {
const response = yield* LLMClient.generate(request).pipe(
Effect.provide(
fixedResponse(
sseEvents(
{ type: "message_start", message: { usage: { input_tokens: 5 } } },
{ type: "content_block_start", index: 0, content_block: { type: "text", text: "" } },
{ type: "content_block_delta", index: 0, delta: { type: "text_delta", text: "Hello" } },
{
type: "content_block_delta",
index: 1,
delta: { type: "input_json_delta", partial_json: '{"query":"orphaned"}' },
},
{ type: "content_block_stop", index: 0 },
{ type: "message_delta", delta: { stop_reason: "end_turn" }, usage: { output_tokens: 1 } },
{ type: "message_stop" },
),
),
),
)
expect(response.text).toBe("Hello")
expect(response.toolCalls).toEqual([])
expect(response.finishReason).toEqual({ normalized: "stop", raw: "end_turn" })
}),
)
it.effect("settles pending tool calls at message_stop", () =>
Effect.gen(function* () {
const response = yield* LLMClient.generate(request).pipe(
+1 -1
View File
@@ -42,8 +42,8 @@
"solid-js": "catalog:",
"tree-sitter-bash": "0.25.0",
"tree-sitter-powershell": "0.25.10",
"web-tree-sitter": "0.25.10",
"uqr": "0.1.3",
"web-tree-sitter": "0.25.10",
"ws": "8.21.0"
},
"devDependencies": {
+1 -23
View File
@@ -160,29 +160,7 @@ const Root = Spec.make(typeof OPENCODE_CLI_NAME === "string" ? OPENCODE_CLI_NAME
}),
Spec.make("plugin", {
description: "Manage plugins",
commands: [
Spec.make("list", {
description: "List plugins",
params: {
builtin: Flag.boolean("builtin").pipe(
Flag.withDescription("Include built-in server plugins"),
Flag.withDefault(false),
),
},
}),
Spec.make("add", {
description: "Install a plugin and add it to the global configuration",
params: {
package: Argument.string("package").pipe(Argument.withDescription("npm registry package specifier")),
},
}),
Spec.make("remove", {
description: "Remove a plugin from global configuration",
params: {
package: Argument.string("package").pipe(Argument.withDescription("configured package specifier")),
},
}),
],
commands: [Spec.make("list", { description: "List active plugins" })],
}),
Spec.make("models", {
description: "List all available models",
@@ -84,12 +84,8 @@ export default Runtime.handler(Commands, (input) =>
update: (update) => runPromise(config.update(update)),
},
packages: {
resolve: (spec, install = true) =>
runPromise(
(install ? npm.add(spec, { subpaths: ["tui"] }) : npm.resolve(spec, { subpaths: ["tui"] })).pipe(
Effect.map((result) => result.entrypoint),
),
),
resolve: (spec) =>
runPromise(npm.add(spec, { subpaths: ["tui"] }).pipe(Effect.map((result) => result.entrypoint))),
},
environment: requestedServer === undefined ? Env.session() : undefined,
terminalHandoff: () => preflight.finish(),
@@ -1,82 +0,0 @@
import { EOL } from "node:os"
import path from "node:path"
import { mkdir, readFile, rename, writeFile } from "node:fs/promises"
import { Effect } from "effect"
import { applyEdits, modify, parse, type ParseError } from "jsonc-parser"
import { Global } from "@opencode-ai/util/global"
import { Npm } from "@opencode-ai/util/npm"
import { Commands } from "../../commands"
import { Runtime } from "../../../framework/runtime"
import { resolveConfigPath } from "../mcp/add"
import { Config } from "../../../config"
export default Runtime.handler(
Commands.commands.plugin.commands.add,
Effect.fn("cli.plugin.add")(function* (input) {
if (!(yield* Effect.promise(() => Npm.isRegistryPackage(input.package))))
return yield* Effect.fail(
new Error("Plugin target must be an npm registry package name, version, tag, or semver range"),
)
const npm = yield* Npm.Service
const installed = yield* npm.add(input.package, { subpaths: ["server", ""] })
const tui = yield* npm.resolve(input.package, { subpaths: ["tui"] })
const target = configurationTarget(installed.entrypoint, tui.entrypoint)
if (!target)
return yield* Effect.fail(new Error(`Plugin package has no server or TUI entrypoint: ${input.package}`))
if (target === "server") {
const global = yield* Global.Service
const configPath = yield* Effect.promise(() => resolveConfigPath(global.config))
const changed = yield* Effect.promise(() => writePluginConfig(configPath, input.package))
process.stdout.write(
changed
? `Plugin "${input.package}" installed and added to ${configPath}${EOL}`
: `Plugin "${input.package}" is already configured in ${configPath}${EOL}`,
)
return
}
const config = yield* Config.Service
yield* config.update((draft) => {
if (configured(draft.plugins, input.package)) return
draft.plugins = [...(draft.plugins ?? []), input.package]
})
process.stdout.write(`TUI plugin "${input.package}" installed and added to ${config.path}${EOL}`)
}),
)
export function configurationTarget(server?: string, tui?: string) {
if (server) return "server" as const
if (tui) return "tui" as const
}
export async function writePluginConfig(configPath: string, spec: string) {
const text = await readFile(configPath, "utf8").catch((error) => {
if (typeof error === "object" && error !== null && "code" in error && error.code === "ENOENT") return "{}"
throw error
})
const errors: ParseError[] = []
const config: unknown = parse(text, errors, { allowTrailingComma: true })
if (errors.length || typeof config !== "object" || config === null || Array.isArray(config))
throw new Error(`Invalid global configuration: ${configPath}`)
const plugins = "plugins" in config ? config.plugins : undefined
if (plugins !== undefined && !Array.isArray(plugins)) throw new Error(`Invalid plugins configuration: ${configPath}`)
if (configured(plugins, spec)) return false
const updated = applyEdits(
text,
modify(text, ["plugins"], [...(plugins ?? []), spec], { formattingOptions: { tabSize: 2, insertSpaces: true } }),
)
await mkdir(path.dirname(configPath), { recursive: true })
const temporary = configPath + ".tmp"
await writeFile(temporary, updated.endsWith("\n") ? updated : updated + "\n", { mode: 0o600 })
await rename(temporary, configPath)
return true
}
function configured(plugins: readonly unknown[] | undefined, spec: string) {
return plugins?.some(
(entry) =>
entry === spec || (typeof entry === "object" && entry !== null && "package" in entry && entry.package === spec),
)
}
@@ -5,69 +5,24 @@ import { Service } from "@opencode-ai/client/effect/service"
import { Commands } from "../../commands"
import { Runtime } from "../../../framework/runtime"
import { ServiceConfig } from "../../../services/service-config"
import { Config } from "../../../config"
import { Global } from "@opencode-ai/util/global"
import { discoverTuiPlugins, tuiPluginDirectories } from "@opencode-ai/tui/plugin/discovery"
export default Runtime.handler(
Commands.commands.plugin.commands.list,
Effect.fn("cli.plugin.list")(function* (input) {
Effect.fn("cli.plugin.list")(function* () {
const options = yield* ServiceConfig.options()
const found = yield* Service.discover(options)
const endpoint = found ?? (yield* Service.ensure(options))
const client = OpenCode.make({ baseUrl: endpoint.url, headers: Service.headers(endpoint) })
const response = yield* Effect.promise(() => client.plugin.list({ location: { directory: process.cwd() } }))
const config = yield* Config.Service
const global = yield* Global.Service
const info = yield* config.get()
const discovered = yield* Effect.promise(() =>
tuiPluginDirectories(process.cwd(), global.config).then(discoverTuiPlugins),
)
const output = format(
response.data,
[
...(info.plugins ?? []).flatMap((entry) => {
const target = typeof entry === "string" ? entry : entry.package
return target.startsWith("-") ? [] : [{ target, source: "configured" as const }]
}),
...discovered.map((target) => ({ target, source: "discovered" as const })),
],
input.builtin,
)
if (!output) {
process.stdout.write("No plugins found" + EOL)
const plugins = response.data.toSorted((a, b) => name(a).localeCompare(name(b)))
if (plugins.length === 0) {
process.stdout.write("No plugins loaded" + EOL)
return
}
process.stdout.write(output + EOL)
process.stdout.write(plugins.map(name).join(EOL) + EOL)
}),
)
export function format(
plugins: readonly PluginInfo[],
tui: ReadonlyArray<{ readonly target: string; readonly source: "configured" | "discovered" }>,
builtin = false,
) {
const server = plugins
.filter((plugin) => builtin || plugin.source.type !== "builtin")
.toSorted((a, b) => name(a).localeCompare(name(b)))
.map((plugin) => `${name(plugin)} (${plugin.status})`)
const advertised = plugins.flatMap((plugin) =>
plugin.status === "active" && plugin.tui && plugin.source.type === "package"
? [{ target: plugin.source.package, source: "advertised" as const }]
: [],
)
const targets = [...tui, ...advertised]
.filter((plugin, index, all) => all.findIndex((candidate) => candidate.target === plugin.target) === index)
.toSorted((a, b) => a.target.localeCompare(b.target))
.map((plugin) => `${plugin.target} (${plugin.source})`)
return [
targets.length ? ["TUI", ...targets].join(EOL) : undefined,
server.length ? ["Server", ...server].join(EOL) : undefined,
]
.filter((section) => section !== undefined)
.join(EOL + EOL)
}
function name(plugin: PluginInfo) {
if (plugin.id) return plugin.id
if (plugin.source.type === "package") return plugin.source.package
@@ -1,74 +0,0 @@
import { EOL } from "node:os"
import path from "node:path"
import { readFile, rename, writeFile } from "node:fs/promises"
import { Effect } from "effect"
import { applyEdits, modify, parse, type ParseError } from "jsonc-parser"
import { Global } from "@opencode-ai/util/global"
import { Commands } from "../../commands"
import { Runtime } from "../../../framework/runtime"
import { Config } from "../../../config"
import { resolveConfigPath } from "../mcp/add"
export default Runtime.handler(
Commands.commands.plugin.commands.remove,
Effect.fn("cli.plugin.remove")(function* (input) {
const global = yield* Global.Service
const configPath = yield* Effect.promise(() => resolveConfigPath(global.config))
const server = yield* Effect.promise(() => removePluginConfig(configPath, input.package))
const config = yield* Config.Service
const info = yield* config.get()
const tui = configured(info.plugins, input.package)
if (tui)
yield* config.update((draft) => {
draft.plugins = draft.plugins?.filter((entry) => !matches(entry, input.package))
})
const removed = [server ? configPath : undefined, tui ? config.path : undefined].filter(
(file) => file !== undefined,
)
process.stdout.write(
removed.length
? `Plugin "${input.package}" removed from ${removed.join(", ")}${EOL}`
: `Plugin "${input.package}" is not configured${EOL}`,
)
}),
)
export async function removePluginConfig(configPath: string, spec: string) {
const text = await readFile(configPath, "utf8").catch((error) => {
if (typeof error === "object" && error !== null && "code" in error && error.code === "ENOENT") return undefined
throw error
})
if (text === undefined) return false
const errors: ParseError[] = []
const config: unknown = parse(text, errors, { allowTrailingComma: true })
if (errors.length || typeof config !== "object" || config === null || Array.isArray(config))
throw new Error(`Invalid global configuration: ${configPath}`)
const plugins = "plugins" in config ? config.plugins : undefined
if (plugins !== undefined && !Array.isArray(plugins)) throw new Error(`Invalid plugins configuration: ${configPath}`)
if (!configured(plugins, spec)) return false
const updated = applyEdits(
text,
modify(
text,
["plugins"],
plugins?.filter((entry) => !matches(entry, spec)),
{
formattingOptions: { tabSize: 2, insertSpaces: true },
},
),
)
const temporary = configPath + ".tmp"
await writeFile(temporary, updated.endsWith("\n") ? updated : updated + "\n", { mode: 0o600 })
await rename(temporary, configPath)
return true
}
function configured(plugins: readonly unknown[] | undefined, spec: string) {
return plugins?.some((entry) => matches(entry, spec)) ?? false
}
function matches(entry: unknown, spec: string) {
return entry === spec || (typeof entry === "object" && entry !== null && "package" in entry && entry.package === spec)
}
-2
View File
@@ -38,8 +38,6 @@ const Handlers = Runtime.handlers(Commands, {
},
plugin: {
list: () => import("./commands/handlers/plugin/list"),
add: () => import("./commands/handlers/plugin/add"),
remove: () => import("./commands/handlers/plugin/remove"),
},
models: () => import("./commands/handlers/models"),
export: () => import("./commands/handlers/export"),
-30
View File
@@ -1,30 +0,0 @@
import { expect, test } from "bun:test"
import path from "node:path"
import { parse } from "jsonc-parser"
import { configurationTarget, writePluginConfig } from "../src/commands/handlers/plugin/add"
test("routes packages according to their exported runtimes", () => {
expect(configurationTarget("server.js", "tui.js")).toBe("server")
expect(configurationTarget("server.js", undefined)).toBe("server")
expect(configurationTarget(undefined, "tui.js")).toBe("tui")
expect(configurationTarget(undefined, undefined)).toBeUndefined()
})
test("adds a package to global plugin config without replacing unrelated settings", async () => {
const directory = await Bun.$`mktemp -d`.text().then((value) => value.trim())
const file = path.join(directory, "opencode.jsonc")
await Bun.write(file, '{\n // retained\n "model": "provider/model",\n "plugins": ["first"]\n}\n')
try {
expect(await writePluginConfig(file, "second@1.0.0")).toBe(true)
expect(await writePluginConfig(file, "second@1.0.0")).toBe(false)
const text = await Bun.file(file).text()
expect(text).toContain("// retained")
expect(parse(text)).toEqual({
model: "provider/model",
plugins: ["first", "second@1.0.0"],
})
} finally {
await Bun.$`rm -rf ${directory}`
}
})
-41
View File
@@ -1,41 +0,0 @@
import { expect, test } from "bun:test"
import { format } from "../src/commands/handlers/plugin/list"
test("formats server and TUI plugins in sections without builtins", () => {
expect(
format(
[
{ id: "opencode.agent", source: { type: "builtin" }, status: "active", tui: false },
{
id: "acme.dual",
source: { type: "package", package: "acme-plugin@1.0.0" },
status: "active",
tui: true,
},
{
source: { type: "package", package: "broken-plugin" },
status: "failed",
error: "broken",
tui: false,
},
],
[
{ target: "tui-only", source: "configured" },
{ target: "/tmp/local.ts", source: "discovered" },
],
),
).toBe(`TUI
/tmp/local.ts (discovered)
acme-plugin@1.0.0 (advertised)
tui-only (configured)
Server
acme.dual (active)
broken-plugin (failed)`)
})
test("includes builtins when requested", () => {
expect(format([{ id: "opencode.agent", source: { type: "builtin" }, status: "active", tui: false }], [], true))
.toBe(`Server
opencode.agent (active)`)
})
-23
View File
@@ -1,23 +0,0 @@
import { expect, test } from "bun:test"
import path from "node:path"
import { parse } from "jsonc-parser"
import { removePluginConfig } from "../src/commands/handlers/plugin/remove"
test("removes string and object package entries without replacing unrelated settings", async () => {
const directory = await Bun.$`mktemp -d`.text().then((value) => value.trim())
const file = path.join(directory, "opencode.jsonc")
await Bun.write(
file,
'{\n // retained\n "model": "provider/model",\n "plugins": ["remove-me", { "package": "remove-me", "options": {} }, "keep-me"]\n}\n',
)
try {
expect(await removePluginConfig(file, "remove-me")).toBe(true)
expect(await removePluginConfig(file, "remove-me")).toBe(false)
const text = await Bun.file(file).text()
expect(text).toContain("// retained")
expect(parse(text)).toEqual({ model: "provider/model", plugins: ["keep-me"] })
} finally {
await Bun.$`rm -rf ${directory}`
}
})
@@ -0,0 +1,70 @@
export * as ConfigFormatterPlugin from "./formatter.js"
import { define } from "@opencode-ai/plugin/effect/plugin"
import { FSUtil } from "@opencode-ai/util/fs-util"
import { Global } from "@opencode-ai/util/global"
import { Npm } from "@opencode-ai/util/npm"
import { AppProcess } from "@opencode-ai/util/process"
import { Effect, Stream } from "effect"
import { Config } from "../../config.js"
import { Formatter } from "../../formatter.js"
import { make, type Info } from "../../formatter/builtins.js"
import { Location } from "../../location.js"
export const Plugin = define({
id: "opencode.config.formatter",
effect: Effect.fn(function* (ctx) {
const config = yield* Config.Service
const formatter = yield* Formatter.Service
const fs = yield* FSUtil.Service
const global = yield* Global.Service
const location = yield* Location.Service
const npm = yield* Npm.Service
const processes = yield* AppProcess.Service
const loaded = { entries: yield* config.entries() }
const reload = config.entries().pipe(
Effect.tap((entries) => Effect.sync(() => (loaded.entries = entries))),
Effect.andThen(formatter.reload()),
)
yield* ctx.event.subscribe().pipe(
Stream.filter((event) => event.type === "config.updated"),
Stream.runForEach(() => reload),
Effect.forkScoped({ startImmediately: true }),
)
// Refetch after subscribing so a config update between the first read and
// the live subscription cannot leave the transform on a stale snapshot.
loaded.entries = yield* config.entries()
yield* formatter.transform((draft) => {
const configured = Config.latest(loaded.entries, "formatter")
if (!configured) return
const builtIns = make({
directory: location.directory,
worktree: location.project.directory,
fs,
npm,
processes,
bin: global.bin,
})
builtIns.forEach(draft.set)
if (configured === true) return
for (const [name, entry] of Object.entries(configured)) {
if (entry.disabled) {
draft.remove(name)
continue
}
const builtIn = builtIns.find((formatter) => formatter.name === name)
const current: Info = {
name,
extensions: entry.extensions ?? builtIn?.extensions ?? [],
environment: { ...builtIn?.environment, ...entry.environment },
enabled:
builtIn && !entry.command ? builtIn.enabled : Effect.succeed(entry.command ? [...entry.command] : false),
}
draft.set(current)
}
})
}),
})
+40
View File
@@ -0,0 +1,40 @@
export * as ConfigImagePlugin from "./image.js"
import { define } from "@opencode-ai/plugin/effect/plugin"
import { Effect, Stream } from "effect"
import { Config } from "../../config.js"
import { Image } from "../../image.js"
export const Plugin = define({
id: "opencode.config.image",
effect: Effect.fn(function* (ctx) {
const config = yield* Config.Service
const image = yield* Image.Service
const loaded = { entries: yield* config.entries() }
const reload = config.entries().pipe(
Effect.tap((entries) => Effect.sync(() => (loaded.entries = entries))),
Effect.andThen(image.reload()),
)
yield* ctx.event.subscribe().pipe(
Stream.filter((event) => event.type === "config.updated"),
Stream.runForEach(() => reload),
Effect.forkScoped({ startImmediately: true }),
)
// Refetch after subscribing so a config update between the first read and
// the live subscription cannot leave the transform on a stale snapshot.
loaded.entries = yield* config.entries()
yield* image.transform((draft) => {
for (const entry of loaded.entries) {
if (entry.type !== "document") continue
const configured = entry.info.media?.image
if (!configured) continue
draft.configure({
...(configured.auto_resize === undefined ? {} : { autoResize: configured.auto_resize }),
...(configured.max_width === undefined ? {} : { maxWidth: configured.max_width }),
...(configured.max_height === undefined ? {} : { maxHeight: configured.max_height }),
...(configured.max_base64_bytes === undefined ? {} : { maxBase64Bytes: configured.max_base64_bytes }),
})
}
})
}),
})
+34 -58
View File
@@ -4,15 +4,21 @@ import { Context, Effect, Layer } from "effect"
import { ChildProcess } from "effect/unstable/process"
import path from "path"
import { makeLocationNode } from "@opencode-ai/util/effect/app-node"
import { FSUtil } from "@opencode-ai/util/fs-util"
import { Npm } from "@opencode-ai/util/npm"
import { AppProcess } from "@opencode-ai/util/process"
import { Global } from "@opencode-ai/util/global"
import { Config } from "./config.js"
import { Location } from "./location.js"
import { make, type Info } from "./formatter/builtins.js"
import type { Info } from "./formatter/builtins.js"
import { State } from "./state.js"
export interface Interface {
type Data = {
formatters: Info[]
}
export type Draft = {
set: (formatter: Info) => void
remove: (name: string) => void
}
export interface Interface extends State.Transformable<Draft> {
readonly file: (filepath: string) => Effect.Effect<boolean>
}
@@ -21,66 +27,36 @@ export class Service extends Context.Service<Service, Interface>()("@opencode/v2
const layer = Layer.effect(
Service,
Effect.gen(function* () {
const config = yield* Config.Service
const fs = yield* FSUtil.Service
const location = yield* Location.Service
const npm = yield* Npm.Service
const processes = yield* AppProcess.Service
const global = yield* Global.Service
const commands = new Map<string, string[] | false>()
let formatters: Info[] = []
const load = yield* Effect.cached(
Effect.gen(function* () {
const configured = Config.latest(yield* config.entries(), "formatter")
if (!configured) {
yield* Effect.logInfo("all formatters are disabled")
return
}
const builtIns = make({
directory: location.directory,
worktree: location.project.directory,
fs,
npm,
processes,
bin: global.bin,
})
formatters = builtIns
if (configured === true) return
for (const [name, entry] of Object.entries(configured)) {
const index = formatters.findIndex((formatter) => formatter.name === name)
if (entry.disabled) {
if (index !== -1) formatters.splice(index, 1)
continue
}
const builtIn = builtIns.find((formatter) => formatter.name === name)
const formatter: Info = {
name,
extensions: entry.extensions ?? builtIn?.extensions ?? [],
environment: { ...builtIn?.environment, ...entry.environment },
enabled:
builtIn && !entry.command ? builtIn.enabled : Effect.succeed(entry.command ? [...entry.command] : false),
}
if (index === -1) formatters.push(formatter)
else formatters[index] = formatter
}
}).pipe(Effect.withSpan("Formatter.load")),
)
const commands = new WeakMap<Info, string[] | false>()
const state = State.create<Data, Draft>({
name: "formatter",
initial: () => ({ formatters: [] }),
draft: (draft) => ({
set: (formatter) => {
const index = draft.formatters.findIndex((item) => item.name === formatter.name)
if (index === -1) draft.formatters.push(formatter)
else draft.formatters[index] = formatter
},
remove: (name) => {
draft.formatters = draft.formatters.filter((formatter) => formatter.name !== name)
},
}),
})
const command = Effect.fnUntraced(function* (formatter: Info) {
const cached = commands.get(formatter.name)
const cached = commands.get(formatter)
if (cached !== undefined) return cached
const result = yield* formatter.enabled
if (result !== false) commands.set(formatter.name, result)
if (result !== false) commands.set(formatter, result)
return result
})
const file = Effect.fn("Formatter.file")(function* (filepath: string) {
yield* load
const matching = formatters.filter((formatter) => formatter.extensions.includes(path.extname(filepath)))
const matching = state
.get()
.formatters.filter((formatter) => formatter.extensions.includes(path.extname(filepath)))
for (const formatter of matching) {
const enabled = yield* command(formatter)
@@ -118,12 +94,12 @@ const layer = Layer.effect(
return false
})
return Service.of({ file })
return Service.of({ transform: state.transform, reload: state.reload, file })
}),
)
export const node = makeLocationNode({
service: Service,
layer,
deps: [Config.node, FSUtil.node, Location.node, Npm.node, AppProcess.node, Global.node],
deps: [Location.node, AppProcess.node],
})
+33 -17
View File
@@ -2,8 +2,8 @@ export * as Image from "./image.js"
import { makeLocationNode } from "@opencode-ai/util/effect/app-node"
import { Context, Effect, Layer, Schema } from "effect"
import { Config } from "./config.js"
import { FileSystem } from "./filesystem.js"
import { State } from "./state.js"
export class ResizerUnavailableError extends Schema.TaggedError<ResizerUnavailableError>()(
"Image.ResizerUnavailableError",
@@ -32,7 +32,18 @@ export class SizeError extends Schema.TaggedError<SizeError>()("Image.SizeError"
}
}
export interface Interface {
export type Limits = {
autoResize: boolean
maxWidth: number
maxHeight: number
maxBase64Bytes: number
}
export type Draft = {
configure: (limits: Partial<Limits>) => void
}
export interface Interface extends State.Transformable<Draft> {
readonly normalize: (
resource: string,
content: FileSystem.Content & { readonly encoding: "base64" },
@@ -47,7 +58,23 @@ export class Service extends Context.Service<Service, Interface>()("@opencode/Im
const layer = Layer.effect(
Service,
Effect.gen(function* () {
const config = yield* Config.Service
const state = State.create<Limits, Draft>({
name: "image",
initial: () => ({
autoResize: true,
maxWidth: 2_000,
maxHeight: 2_000,
maxBase64Bytes: 5 * 1024 * 1024,
}),
draft: (draft) => ({
configure: (limits) => {
if (limits.autoResize !== undefined) draft.autoResize = limits.autoResize
if (limits.maxWidth !== undefined) draft.maxWidth = limits.maxWidth
if (limits.maxHeight !== undefined) draft.maxHeight = limits.maxHeight
if (limits.maxBase64Bytes !== undefined) draft.maxBase64Bytes = limits.maxBase64Bytes
},
}),
})
const loadAdapter = yield* Effect.cached(
Effect.tryPromise({
try: () => import("./image/photon.js"),
@@ -58,22 +85,11 @@ const layer = Layer.effect(
resource: string,
content: FileSystem.Content & { readonly encoding: "base64" },
) {
const image = Object.assign(
{},
...(yield* config.entries()).flatMap((entry) =>
entry.type === "document" && entry.info.media?.image ? [entry.info.media.image] : [],
),
)
const normalize = yield* loadAdapter
return yield* normalize(resource, content, {
autoResize: image.auto_resize ?? true,
maxWidth: image.max_width ?? 2_000,
maxHeight: image.max_height ?? 2_000,
maxBase64Bytes: image.max_base64_bytes ?? 5 * 1024 * 1024,
})
return yield* normalize(resource, content, state.get())
})
return Service.of({ normalize })
return Service.of({ transform: state.transform, reload: state.reload, normalize })
}),
)
export const node = makeLocationNode({ service: Service, layer, deps: [Config.node] })
export const node = makeLocationNode({ service: Service, layer, deps: [] })
+43 -21
View File
@@ -4,6 +4,7 @@ import type { AISDKHooks } from "@opencode-ai/plugin/effect/aisdk"
import type { SessionHooks } from "@opencode-ai/plugin/effect/session"
import type { ShellHooks } from "@opencode-ai/plugin/effect/shell"
import type { ToolFailures, ToolHooks } from "@opencode-ai/plugin/effect/tool"
import type { ModelHookOptions } from "@opencode-ai/plugin/effect/registration"
import { Context, Effect, Layer, Scope } from "effect"
import { makeLocationNode } from "@opencode-ai/util/effect/app-node"
import { State } from "../state.js"
@@ -26,16 +27,26 @@ interface Failures extends Record<keyof Domains, unknown> {
}
type Callback<Event, Error> = (event: Event) => Effect.Effect<void, Error>
type Entry = { readonly callback: Function; readonly options?: ModelHookOptions }
const eventProviderID = (event: unknown) => {
if (typeof event !== "object" || event === null || !("model" in event)) return undefined
const model = event.model
if (typeof model !== "object" || model === null || !("providerID" in model)) return undefined
return typeof model.providerID === "string" ? model.providerID : undefined
}
export interface Interface {
readonly has: <Domain extends keyof Domains>(
domain: Domain,
name: keyof Domains[Domain] & keyof Failures[Domain],
providerID?: string,
) => Effect.Effect<boolean>
readonly register: <Domain extends keyof Domains, Name extends keyof Domains[Domain] & keyof Failures[Domain]>(
domain: Domain,
name: Name,
callback: Callback<Domains[Domain][Name], Failures[Domain][Name]>,
options?: ModelHookOptions,
) => Effect.Effect<State.Registration, never, Scope.Scope>
readonly trigger: <Domain extends keyof Domains, Name extends keyof Domains[Domain] & keyof Failures[Domain]>(
domain: Domain,
@@ -49,36 +60,47 @@ export class Service extends Context.Service<Service, Interface>()("@opencode/Pl
const layer = Layer.effect(
Service,
Effect.gen(function* () {
const callbacks = new Map<string, Function[]>()
const callbacks = new Map<string, Entry[]>()
const key = (domain: keyof Domains, name: PropertyKey) => `${domain}.${String(name)}`
const register: Interface["register"] = Effect.fn("PluginHooks.register")(function* (domain, name, callback) {
const scope = yield* Scope.Scope
const id = key(domain, name)
let active = true
callbacks.set(id, [...(callbacks.get(id) ?? []), callback])
const dispose = Effect.sync(() => {
if (!active) return
active = false
const next = (callbacks.get(id) ?? []).filter((item) => item !== callback)
if (next.length === 0) callbacks.delete(id)
else callbacks.set(id, next)
})
yield* Scope.addFinalizer(scope, dispose)
return { dispose }
})
const register: Interface["register"] = Effect.fn("PluginHooks.register")(
function* (domain, name, callback, options) {
const scope = yield* Scope.Scope
const id = key(domain, name)
let active = true
const entry = { callback, options }
callbacks.set(id, [...(callbacks.get(id) ?? []), entry])
const dispose = Effect.sync(() => {
if (!active) return
active = false
const next = (callbacks.get(id) ?? []).filter((item) => item !== entry)
if (next.length === 0) callbacks.delete(id)
else callbacks.set(id, next)
})
yield* Scope.addFinalizer(scope, dispose)
return { dispose }
},
)
const trigger: Interface["trigger"] = Effect.fnUntraced(function* (domain, name, event) {
for (const callback of callbacks.get(key(domain, name)) ?? []) {
const result: Effect.Effect<void, Failures[typeof domain][typeof name]> = Reflect.apply(callback, undefined, [
event,
])
for (const entry of callbacks.get(key(domain, name)) ?? []) {
if (entry.options?.providerID !== undefined && entry.options.providerID !== eventProviderID(event)) continue
const result: Effect.Effect<void, Failures[typeof domain][typeof name]> = Reflect.apply(
entry.callback,
undefined,
[event],
)
yield* result
}
return event
})
const has: Interface["has"] = (domain, name) => Effect.sync(() => callbacks.has(key(domain, name)))
const has: Interface["has"] = (domain, name, providerID) =>
Effect.sync(() =>
(callbacks.get(key(domain, name)) ?? []).some(
(entry) => entry.options?.providerID === undefined || entry.options.providerID === providerID,
),
)
return Service.of({ has, register, trigger })
}),
+4 -2
View File
@@ -104,9 +104,10 @@ export const make = Effect.fn("PluginHost.make")(function* (plugin: import("../p
}),
},
aisdk: {
hook: (name, callback) => {
hook: (name, callback, options) => {
if (name === "sdk") {
return aisdk.hook.sdk((event) => {
if (options?.providerID !== undefined && options.providerID !== event.model.providerID) return Effect.void
const output = {
model: mutable(event.model),
package: event.package,
@@ -119,6 +120,7 @@ export const make = Effect.fn("PluginHost.make")(function* (plugin: import("../p
})
}
return aisdk.hook.language((event) => {
if (options?.providerID !== undefined && options.providerID !== event.model.providerID) return Effect.void
const output = {
model: mutable(event.model),
options: event.options,
@@ -382,7 +384,7 @@ export const make = Effect.fn("PluginHost.make")(function* (plugin: import("../p
}),
},
session: {
hook: (name, callback) => hooks.register("session", name, callback),
hook: (name, callback, options) => hooks.register("session", name, callback, options),
create: (input) =>
runtime.session.create({
id: input?.id,
+8
View File
@@ -3,6 +3,7 @@ export * as PluginInternal from "./internal.js"
import type { Plugin } from "@opencode-ai/plugin/effect/plugin"
import { LayerNode } from "@opencode-ai/util/effect/layer-node"
import { httpClient } from "@opencode-ai/util/effect/app-node-platform"
import { AppProcess } from "@opencode-ai/util/process"
import { Context, Effect, Scope } from "effect"
import { HttpClient } from "effect/unstable/http"
import { Agent } from "../agent.js"
@@ -12,6 +13,8 @@ import { Config } from "../config.js"
import { Credential } from "../credential.js"
import { ConfigAgentPlugin } from "../config/plugin/agent.js"
import { ConfigCommandPlugin } from "../config/plugin/command.js"
import { ConfigFormatterPlugin } from "../config/plugin/formatter.js"
import { ConfigImagePlugin } from "../config/plugin/image.js"
import { ConfigInstructionPlugin } from "../config/plugin/instruction.js"
import { ConfigMCPPlugin } from "../config/plugin/mcp.js"
import { ConfigProviderPlugin } from "../config/plugin/provider.js"
@@ -77,6 +80,7 @@ import { WellKnownPlugin } from "../wellknown/plugin.js"
const services = Effect.fn("PluginInternal.services")(function* () {
const agent = yield* Agent.Service
const processes = yield* AppProcess.Service
const catalog = yield* Catalog.Service
const command = yield* Command.Service
const config = yield* Config.Service
@@ -115,6 +119,7 @@ const services = Effect.fn("PluginInternal.services")(function* () {
const wellknown = yield* WellKnown.Service
return Context.mergeAll(
Context.make(Agent.Service, agent),
Context.make(AppProcess.Service, processes),
Context.make(Catalog.Service, catalog),
Context.make(Command.Service, command),
Context.make(Config.Service, config),
@@ -160,6 +165,7 @@ export type Requirements = ContextServices<Effect.Success<ReturnType<typeof serv
export const requirements = LayerNode.group([
Agent.node,
AppProcess.node,
Catalog.node,
Command.node,
Config.node,
@@ -232,6 +238,8 @@ const post = [
ConfigReferencePlugin.Plugin,
ConfigAgentPlugin.Plugin,
ConfigCommandPlugin.Plugin,
ConfigFormatterPlugin.Plugin,
ConfigImagePlugin.Plugin,
ConfigSkillPlugin.Plugin,
ConfigProviderPlugin.Plugin,
ConfigWebSearchPlugin.Plugin,
@@ -241,19 +241,22 @@ export const GithubCopilotPlugin = define({
evt.sdk = mod.createOpenaiCompatible(evt.options)
}),
)
yield* ctx.session.hook("http.request", (evt) =>
Effect.gen(function* () {
if (evt.model.providerID !== Provider.ID.githubCopilot) return
if (evt.agent === Agent.ID.make("title"))
evt.request.headers.set("X-Interaction-Type", "conversation-background")
if (evt.agent === Agent.ID.make("compaction"))
evt.request.headers.set("X-Interaction-Type", "conversation-compaction")
const token = evt.request.headers.get("x-api-key")
if (!token) return
const text = yield* Effect.promise(() => evt.request.clone().text())
const body = Option.getOrUndefined(decodeBody(text))
applyHeaders(evt.request.headers, token, ctx.app, requestMetadata(evt.request.url, body), true)
}),
yield* ctx.session.hook(
"http.request",
(evt) =>
Effect.gen(function* () {
if (evt.model.providerID !== Provider.ID.githubCopilot) return
if (evt.agent === Agent.ID.make("title"))
evt.request.headers.set("X-Interaction-Type", "conversation-background")
if (evt.agent === Agent.ID.make("compaction"))
evt.request.headers.set("X-Interaction-Type", "conversation-compaction")
const token = evt.request.headers.get("x-api-key")
if (!token) return
const text = yield* Effect.promise(() => evt.request.clone().text())
const body = Option.getOrUndefined(decodeBody(text))
applyHeaders(evt.request.headers, token, ctx.app, requestMetadata(evt.request.url, body), true)
}),
{ providerID: Provider.ID.githubCopilot },
)
yield* ctx.aisdk.hook(
"language",
+11 -10
View File
@@ -5,7 +5,6 @@ import { App } from "../../app.js"
import { Credential } from "../../credential.js"
import { Bus } from "../../bus.js"
import { Integration } from "../../integration.js"
import { Model } from "../../model.js"
import { OauthCallbackPage } from "../../oauth/page.js"
import { Provider } from "../../provider.js"
import type { PluginInternal } from "../internal.js"
@@ -230,15 +229,17 @@ export const OpenAIPlugin = define({
})
}
})
yield* ctx.session.hook("http.request", (evt) =>
Effect.sync(() => {
if (!chatgpt || evt.model.providerID !== Provider.ID.openai) return
const url = new URL(evt.request.url)
evt.request.headers.set("originator", "opencode")
evt.request.headers.set("session-id", evt.sessionID)
if (url.origin !== "https://api.openai.com") return
evt.request = new Request(`${codexBaseURL}${url.pathname.replace(/^\/v1/, "")}${url.search}`, evt.request)
}),
yield* ctx.session.hook(
"model.request",
(evt) =>
Effect.sync(() => {
if (!chatgpt) return
if (evt.baseURL && URL.canParse(evt.baseURL) && new URL(evt.baseURL).origin === "https://api.openai.com")
evt.baseURL = codexBaseURL
evt.headers.originator = "opencode"
evt.headers["session-id"] = evt.sessionID
}),
{ providerID: Provider.ID.openai },
)
const refresh = () => loading.withPermit(load().pipe(Effect.andThen(ctx.catalog.reload())))
yield* bus.subscribe(Integration.Event.ConnectionUpdated).pipe(
@@ -0,0 +1,14 @@
export * as PluginSupervisor from "./supervisor-service.js"
import { Context, Effect } from "effect"
/**
* Dependency-only supervisor seam. Keep this module free of implementation
* imports: the supervisor reaches PluginRuntime, which depends on Session.
*/
export interface Interface {
/** Wait for the initial plugin generation and startup updates to settle. */
readonly flush: Effect.Effect<void>
}
export class Service extends Context.Service<Service, Interface>()("@opencode/PluginSupervisor") {}
+3 -8
View File
@@ -1,8 +1,9 @@
export * as PluginSupervisor from "./supervisor.js"
export { Service, type Interface } from "./supervisor-service.js"
import type { Plugin as PluginDefinition } from "@opencode-ai/plugin/effect/plugin"
import { Event } from "@opencode-ai/schema/config"
import { Cause, Context, Deferred, Effect, Layer, Schema, Stream } from "effect"
import { Cause, Deferred, Effect, Layer, Schema, Stream } from "effect"
import path from "path"
import { pathToFileURL } from "url"
import { ConfigPluginSource } from "../config/plugin/source.js"
@@ -14,6 +15,7 @@ import { PluginPromise } from "../plugin/promise.js"
import { PluginInternal } from "./internal.js"
import { SdkPlugins } from "./sdk.js"
import { importModule } from "@opencode-ai/util/runtime-import"
import { Service } from "./supervisor-service.js"
const PluginModule = Schema.Struct({
default: Schema.Union([
@@ -128,13 +130,6 @@ const load = Effect.fn("PluginSupervisor.load")(function* (
} satisfies Plugin.Versioned
})
export interface Interface {
/** Wait for the initial plugin generation and startup updates to settle. */
readonly flush: Effect.Effect<void>
}
export class Service extends Context.Service<Service, Interface>()("@opencode/PluginSupervisor") {}
export const layer = Layer.effect(
Service,
Effect.gen(function* () {
+9 -13
View File
@@ -40,6 +40,7 @@ import { SessionRevert } from "./session/revert.js"
import { Session } from "@opencode-ai/schema/session"
import { FSUtil } from "@opencode-ai/util/fs-util"
import { Image } from "./image.js"
import { PluginSupervisor } from "./plugin/supervisor-service.js"
import { Mime } from "./mime.js"
import type { EventLog } from "@opencode-ai/schema/event-log"
import { Event } from "@opencode-ai/schema/event"
@@ -579,7 +580,11 @@ const layer = Layer.effect(
if (session.revert) yield* SessionRevert.commit(session).pipe(Effect.provideService(Bus.Service, bus))
// Resolved lazily so prompt admission only boots location services when an
// image attachment actually needs the resizer.
const image = Image.Service.pipe(Effect.provide(locations.get(session.location)))
const image = Effect.gen(function* () {
const plugins = yield* PluginSupervisor.Service
yield* plugins.flush
return yield* Image.Service
}).pipe(Effect.provide(locations.get(session.location)))
const skills = Skill.Service.pipe(Effect.provide(locations.get(session.location)))
const prompt = yield* resolvePrompt(
{ text: input.text, files: input.files, agents: input.agents, skills: input.skills },
@@ -780,7 +785,7 @@ const layer = Layer.effect(
payload,
delivery: input.delivery ?? "steer",
})
const recovered = yield* SessionInbox.serialized(
yield* SessionInbox.serialized(
input.sessionID,
Effect.gen(function* () {
const latest = yield* result.get(input.sessionID)
@@ -791,25 +796,16 @@ const layer = Layer.effect(
)
const moved = [SessionEvent.Moved, { sessionID: input.sessionID, ...payload }] as const
const first = cancellations[0]
if (!first) {
yield* bus.publish(...moved)
return true
}
yield* bus.publishAll([first, ...cancellations.slice(1), moved])
return true
if (!first) return yield* bus.publish(...moved).pipe(Effect.asVoid)
return yield* bus.publishAll([first, ...cancellations.slice(1), moved])
}
yield* SessionInbox.admit(db, bus, {
id: SessionMessage.ID.create(),
sessionID: input.sessionID,
item,
})
return false
}),
)
if (recovered) {
yield* execution.wakeActive(input.sessionID)
return
}
yield* execution.wake(input.sessionID)
}),
compact: Effect.fn("Session.compact")(function* (input) {
+18 -15
View File
@@ -12,6 +12,7 @@ import { llmClient } from "../effect/app-node-platform.js"
import { SessionEvent } from "./event.js"
import type { SessionMessage } from "./message.js"
import { SessionModelHeaders } from "./model-headers.js"
import { SessionModelHook } from "./model-hook.js"
import { SessionModelHttp } from "./model-http.js"
import { SessionPromptCacheKey } from "./prompt-cache-key.js"
import { App } from "../app.js"
@@ -270,23 +271,25 @@ const make = (dependencies: Dependencies) => {
})
: Effect.void,
)
const request = yield* SessionModelHook.apply(
dependencies.hooks,
{ sessionID: plan.session.id, agent: Agent.ID.make("compaction"), model: plan.ref },
LLM.request({
model: plan.model,
promptCacheKey: SessionPromptCacheKey.make(plan.session.id),
http: { headers: SessionModelHeaders.make(plan.session, dependencies.app) },
messages: [Message.user(plan.prompt)],
tools: [],
}),
)
yield* dependencies.llm
.stream(
LLM.request({
model: plan.model,
promptCacheKey: SessionPromptCacheKey.make(plan.session.id),
http: { headers: SessionModelHeaders.make(plan.session, dependencies.app) },
messages: [Message.user(plan.prompt)],
tools: [],
.stream(request, {
http: SessionModelHttp.middleware(dependencies.hooks, {
sessionID: plan.session.id,
agent: Agent.ID.make("compaction"),
model: plan.ref,
}),
{
http: SessionModelHttp.middleware(dependencies.hooks, {
sessionID: plan.session.id,
agent: Agent.ID.make("compaction"),
model: plan.ref,
}),
},
)
})
.pipe(
Stream.runForEach((event) => {
if (LLMEvent.is.providerError(event))
+7 -11
View File
@@ -21,8 +21,6 @@ export interface Interface {
readonly resume: (sessionID: SessionSchema.ID) => Effect.Effect<void, SessionRunner.RunError>
/** Registers newly recorded work. Repeated wakeups may coalesce. */
readonly wake: (sessionID: SessionSchema.ID) => Effect.Effect<void>
/** Wakes only an active execution, preserving its current input eligibility. */
readonly wakeActive: (sessionID: SessionSchema.ID) => Effect.Effect<void>
/** Interrupt active work owned by this process. Idle interruption is a no-op. */
readonly interrupt: (sessionID: SessionSchema.ID, options?: { readonly continue?: boolean }) => Effect.Effect<void>
/** Resolves once this process owns no active execution for the Session. Returns immediately when idle and never starts work. */
@@ -137,16 +135,15 @@ export const layer = Layer.effect(
return Service.of({
active: coordinator.active,
interrupt: (sessionID, options) =>
coordinator.interrupt(
sessionID,
"user",
options?.continue
? { continue: { request: "steer", when: SessionInbox.has(db, sessionID, "steer") } }
: undefined,
),
Effect.gen(function* () {
yield* coordinator.interrupt(sessionID, "user")
if (!options?.continue) return
// Resume only steering input from the interrupted intent. Queued next-turn work
// stays parked: a steer-scoped drain never promotes queue-delivery rows.
if (yield* SessionInbox.has(db, sessionID, "steer")) yield* coordinator.wake(sessionID, "steer")
}),
resume: coordinator.run,
wake: coordinator.wake,
wakeActive: coordinator.wakeActive,
awaitIdle: coordinator.awaitIdle,
})
}),
@@ -165,7 +162,6 @@ export const noopLayer = Layer.succeed(
active: Effect.succeed(new Set()),
resume: () => Effect.void,
wake: () => Effect.void,
wakeActive: () => Effect.void,
interrupt: () => Effect.void,
awaitIdle: () => Effect.void,
}),
+11 -8
View File
@@ -11,6 +11,7 @@ import { SessionContext } from "./context.js"
import { SessionGenerate } from "./generate.js"
import { SessionHistory } from "./history.js"
import { SessionModelHeaders } from "./model-headers.js"
import { SessionModelHook } from "./model-hook.js"
import { SessionModelHttp } from "./model-http.js"
import { SessionPromptCacheKey } from "./prompt-cache-key.js"
import { SessionRunnerModel } from "./runner/model.js"
@@ -71,7 +72,9 @@ export const layer = Layer.effect(
providerID: model.ref.providerID,
modelID: model.ref.id,
})
const response = yield* llm.generate(
const request = yield* SessionModelHook.apply(
hooks,
{ sessionID: selection.session.id, agent: selection.agent.id, model: model.ref },
LLM.request({
model: model.model,
http: { headers: SessionModelHeaders.make(selection.session, app) },
@@ -80,14 +83,14 @@ export const layer = Layer.effect(
messages: contextEvent.messages,
tools: hookedTools,
}),
{
http: SessionModelHttp.middleware(hooks, {
sessionID: selection.session.id,
agent: selection.agent.id,
model: model.ref,
}),
},
)
const response = yield* llm.generate(request, {
http: SessionModelHttp.middleware(hooks, {
sessionID: selection.session.id,
agent: selection.agent.id,
model: model.ref,
}),
})
yield* Effect.logInfo("session generation usage diagnostic", { usage: response.usage })
return response.text
}),
+21 -7
View File
@@ -180,13 +180,27 @@ export const admitCompaction = Effect.fn("SessionInbox.admitCompaction")(functio
bus: Bus.Interface,
input: { readonly id: SessionMessage.ID; readonly sessionID: SessionSchema.ID; readonly delivery: Delivery },
) {
const admitted = yield* admit(db, bus, {
id: input.id,
sessionID: input.sessionID,
item: Item.make({ type: "compaction", payload: {}, delivery: input.delivery }),
})
if (admitted.type === "compaction") return admitted
return yield* Effect.die(new LifecycleConflict({ id: input.id }))
return yield* serialized(
input.sessionID,
Effect.gen(function* () {
const exact = yield* find(db, input.id)
if (exact) {
if (exact.type === "compaction" && exact.sessionID === input.sessionID) return exact
return yield* Effect.die(new LifecycleConflict({ id: input.id }))
}
if (yield* promotedFromMessage(db, input.sessionID, input.id, input.delivery))
return yield* Effect.die(new LifecycleConflict({ id: input.id }))
const pending = (yield* list(db, input.sessionID)).find((item) => item.type === "compaction")
if (pending) return pending
const admitted = yield* admit(db, bus, {
id: input.id,
sessionID: input.sessionID,
item: Item.make({ type: "compaction", payload: {}, delivery: input.delivery }),
})
if (admitted.type === "compaction") return admitted
return yield* Effect.die(new LifecycleConflict({ id: input.id }))
}),
)
})
export const projectAdmitted = Effect.fn("SessionInbox.projectAdmitted")(function* (
+34
View File
@@ -0,0 +1,34 @@
export * as SessionModelHook from "./model-hook.js"
import { HttpOptions, LanguageModel, LLMRequest } from "@opencode-ai/ai"
import type { Agent } from "@opencode-ai/schema/agent"
import type { Model } from "@opencode-ai/schema/model"
import type { Session } from "@opencode-ai/schema/session"
import { Effect } from "effect"
import { PluginHooks } from "../plugin/hooks.js"
export const apply = (
hooks: PluginHooks.Interface,
input: { readonly sessionID: Session.ID; readonly agent: Agent.ID; readonly model: Model.Ref },
request: LLMRequest,
) =>
Effect.gen(function* () {
const currentBaseURL = request.model.route.endpoint.baseURL
const event = yield* hooks.trigger("session", "model.request", {
...input,
baseURL: typeof currentBaseURL === "string" ? currentBaseURL : undefined,
headers: { ...request.http?.headers },
})
const route =
event.baseURL !== undefined && event.baseURL !== currentBaseURL
? request.model.route.with({ endpoint: { baseURL: event.baseURL } })
: request.model.route
return LLMRequest.update(request, {
model: route === request.model.route ? request.model : LanguageModel.update(request.model, { route }),
http: new HttpOptions({
body: request.http?.body,
headers: Object.keys(event.headers).length === 0 ? undefined : event.headers,
query: request.http?.query,
}),
})
})
+20 -14
View File
@@ -14,6 +14,7 @@ import { QuestionTool } from "../tool/plugin/question.js"
import { Tool } from "../tool.js"
import { SessionContext } from "./context.js"
import { SessionModelHeaders } from "./model-headers.js"
import { SessionModelHook } from "./model-hook.js"
import { SessionModelHttp } from "./model-http.js"
import { SessionModelTransport } from "./model-transport.js"
import { SessionPromptCacheKey } from "./prompt-cache-key.js"
@@ -226,20 +227,25 @@ export const layer = Layer.effect(
return [[name, { ...tool, description: definition.description, inputSchema: definition.input }] as const]
}),
)
const request = LLM.request({
model,
http: {
headers: SessionModelHeaders.make(session, app),
},
// TODO: Persist cache lineage so nested forks reuse the root session's cache key.
promptCacheKey: SessionPromptCacheKey.make(session.fork?.sessionID ?? session.id),
system: context.system,
messages: boundImages(unsupportedParts(context.messages, resolved.capabilities)),
tools: Array.from(hooked, ([name, tool]) => ({ ...tool, name })),
toolChoice: stepLimitReached ? "none" : undefined,
})
const request = yield* SessionModelHook.apply(
hooks,
{ sessionID: session.id, agent: agent.id, model: resolved.ref },
LLM.request({
model,
http: {
headers: SessionModelHeaders.make(session, app),
},
// TODO: Persist cache lineage so nested forks reuse the root session's cache key.
promptCacheKey: SessionPromptCacheKey.make(session.fork?.sessionID ?? session.id),
system: context.system,
messages: boundImages(unsupportedParts(context.messages, resolved.capabilities)),
tools: Array.from(hooked, ([name, tool]) => ({ ...tool, name })),
toolChoice: stepLimitReached ? "none" : undefined,
}),
)
const webSocketEligible =
!(yield* hooks.has("session", "http.request")) && !(yield* hooks.has("session", "http.response"))
!(yield* hooks.has("session", "http.request", resolved.ref.providerID)) &&
!(yield* hooks.has("session", "http.response", resolved.ref.providerID))
const http = webSocketEligible
? undefined
: SessionModelHttp.middleware(hooks, {
@@ -252,7 +258,7 @@ export const layer = Layer.effect(
...(webSocket &&
webSocketEligible &&
resolved.ref.providerID === Provider.ID.openai &&
model.route.id === "openai-responses"
request.model.route.id === "openai-responses"
? { webSocket: transport.bind(session.id) }
: {}),
}
+22 -74
View File
@@ -10,25 +10,17 @@ export interface Coordinator<Key, E, Reason = never> {
/** Starts an execution while idle, or joins the active execution and returns its exit. */
readonly run: (key: Key) => Effect.Effect<void, E>
/** Rings the doorbell: an idle key starts an execution; an active one drains again before settling. */
readonly wake: (key: Key, request?: Request) => Effect.Effect<void>
/** Rings the current execution's doorbell with its existing request. Idle keys remain idle. */
readonly wakeActive: (key: Key) => Effect.Effect<void>
readonly wake: (key: Key, scope?: Promotable) => Effect.Effect<void>
/** Stops the active execution, clears its doorbell, and waits for cleanup. No-op when idle. */
readonly interrupt: (
key: Key,
reason?: Reason,
options?: { readonly continue?: { readonly request: Request; readonly when: Effect.Effect<boolean> } },
) => Effect.Effect<void>
readonly interrupt: (key: Key, reason?: Reason) => Effect.Effect<void>
/** Resolves once no execution is active for the key. Returns immediately when already idle and never starts work. */
readonly awaitIdle: (key: Key) => Effect.Effect<void>
}
export type Request = Promotable
/**
* One execution is a busy period for one key: one fiber that drains from the first wake
* until the key would stay idle. `pendingWake` is the doorbell: work recorded during the
* execution rings it with its eligibility request, and the execution loop drains again
* execution rings it with the scope that work needs, and the execution loop drains again
* instead of ending. The doorbell closes the gap between a drain's last eligibility check
* and the idle transition, since those cannot be one atomic step. `done` resolves joiners
* with this execution's exit.
@@ -36,15 +28,10 @@ export type Request = Promotable
type Execution<E, Reason> = {
readonly done: Deferred.Deferred<void, E>
owner?: Fiber.Fiber<void>
request: Request
pendingWake?: Request
scope: Promotable
pendingWake?: Promotable
stopping: boolean
interruptionReason?: Reason
continuation?: {
readonly request: Request
readonly when: Effect.Effect<boolean>
signaled: boolean
}
}
/**
@@ -59,7 +46,7 @@ type Execution<E, Reason> = {
* ```
*/
export const make = <Key, E, Reason = never>(options: {
readonly drain: (key: Key, force: boolean, request: Request) => Effect.Effect<void, E>
readonly drain: (key: Key, force: boolean, scope: Promotable) => Effect.Effect<void, E>
/** Runs once when a process-local busy period begins, before its first drain. */
readonly started?: (key: Key) => Effect.Effect<void>
/**
@@ -73,11 +60,11 @@ export const make = <Key, E, Reason = never>(options: {
const fork = yield* FiberSet.makeRuntime<never, void, never>()
const loop = (key: Key, execution: Execution<E, Reason>, force: boolean): Effect.Effect<void, E> =>
Effect.suspend(() => options.drain(key, force, execution.request)).pipe(
Effect.suspend(() => options.drain(key, force, execution.scope)).pipe(
Effect.flatMap(() =>
Effect.suspend(() => {
if (execution.stopping || execution.pendingWake === undefined) return Effect.void
execution.request = execution.pendingWake
execution.scope = execution.pendingWake
execution.pendingWake = undefined
// Trampoline so drains that complete synchronously cannot grow the stack.
return Effect.yieldNow.pipe(Effect.andThen(loop(key, execution, false)))
@@ -85,10 +72,10 @@ export const make = <Key, E, Reason = never>(options: {
),
)
const start = (key: Key, force: boolean, request: Request) => {
const start = (key: Key, force: boolean, scope: Promotable) => {
const execution: Execution<E, Reason> = {
done: Deferred.makeUnsafe<void, E>(),
request,
scope,
stopping: false,
}
executions.set(key, execution)
@@ -104,7 +91,7 @@ export const make = <Key, E, Reason = never>(options: {
execution.owner = undefined
}).pipe(Effect.andThen(options.settled?.(key, exit, execution.interruptionReason) ?? Effect.void)),
),
Effect.onExit((exit) => finish(key, execution, exit)),
Effect.onExit((exit) => Effect.sync(() => settle(key, execution, exit))),
Effect.exit,
Effect.asVoid,
),
@@ -114,22 +101,12 @@ export const make = <Key, E, Reason = never>(options: {
// A doorbell that survives the execution loop (rung after the loop decided to end, or
// during failure or interruption cleanup) starts a fresh execution for the remaining work.
const settle = (key: Key, execution: Execution<E, Reason>, exit: Exit.Exit<void, E>, resume: boolean) => {
if (resume && execution.continuation) start(key, false, execution.continuation.request)
else if (execution.pendingWake) start(key, false, execution.pendingWake)
const settle = (key: Key, execution: Execution<E, Reason>, exit: Exit.Exit<void, E>) => {
if (execution.pendingWake) start(key, false, execution.pendingWake)
else executions.delete(key)
Deferred.doneUnsafe(execution.done, exit)
}
const finish = (key: Key, execution: Execution<E, Reason>, exit: Exit.Exit<void, E>) => {
if (!execution.continuation) return Effect.sync(() => settle(key, execution, exit, false))
return execution.continuation.when.pipe(
Effect.flatMap((ready) =>
Effect.sync(() => settle(key, execution, exit, ready || execution.continuation?.signaled === true)),
),
)
}
const run = (key: Key): Effect.Effect<void, E> =>
Effect.suspend(() => {
const execution = executions.get(key)
@@ -141,55 +118,26 @@ export const make = <Key, E, Reason = never>(options: {
return Deferred.await(start(key, true, "input").done)
})
const wake = (key: Key, request: Request = "input") =>
const wake = (key: Key, scope: Promotable = "input") =>
Effect.sync(() => {
const execution = executions.get(key)
if (execution !== undefined) {
if (execution.stopping) {
if (execution.continuation) execution.continuation.signaled = true
else execution.continuation = { request, when: Effect.succeed(true), signaled: true }
return
}
// Coalesced wakes keep the widest request: "input" subsumes "steer".
execution.pendingWake = execution.pendingWake === "input" ? "input" : request
// Coalesced wakes keep the widest scope: "input" subsumes "steer".
execution.pendingWake = execution.pendingWake === "input" ? "input" : scope
return
}
start(key, false, request)
start(key, false, scope)
})
const wakeActive = (key: Key) =>
const interrupt = (key: Key, reason?: Reason): Effect.Effect<void> =>
Effect.suspend(() => {
const execution = executions.get(key)
return execution ? wake(key, execution.request) : Effect.void
})
const interrupt = (
key: Key,
reason?: Reason,
options?: { readonly continue?: { readonly request: Request; readonly when: Effect.Effect<boolean> } },
): Effect.Effect<void> =>
Effect.suspend(() => {
const execution = executions.get(key)
if (execution === undefined) return Effect.void
if (execution.stopping) {
if (options?.continue)
execution.continuation = {
...options.continue,
signaled: execution.continuation?.signaled ?? false,
}
return Deferred.await(execution.done).pipe(Effect.exit, Effect.asVoid)
}
if (execution.owner === undefined) {
if (!options?.continue) return Effect.void
execution.stopping = true
execution.pendingWake = undefined
execution.continuation = { ...options.continue, signaled: false }
return Deferred.await(execution.done).pipe(Effect.exit, Effect.asVoid)
}
if (execution?.owner === undefined || execution.stopping) return Effect.void
execution.stopping = true
// Wakes recorded so far belong to the interrupted intent; the interrupt claims them.
// Wakes arriving during cleanup are new admissions and restart normally at settle.
execution.pendingWake = undefined
execution.interruptionReason = reason
if (options?.continue) execution.continuation = { ...options.continue, signaled: false }
return Fiber.interrupt(execution.owner)
})
@@ -202,5 +150,5 @@ export const make = <Key, E, Reason = never>(options: {
return Deferred.await(execution.done).pipe(Effect.exit, Effect.andThen(awaitIdle(key)))
})
return { active: Effect.sync(() => new Set(executions.keys())), run, wake, wakeActive, interrupt, awaitIdle }
return { active: Effect.sync(() => new Set(executions.keys())), run, wake, interrupt, awaitIdle }
})
+18 -15
View File
@@ -14,6 +14,7 @@ import { PluginHooks } from "../plugin/hooks.js"
import { SessionEvent } from "./event.js"
import { SessionHistory } from "./history.js"
import { SessionModelHeaders } from "./model-headers.js"
import { SessionModelHook } from "./model-hook.js"
import { SessionModelHttp } from "./model-http.js"
import { SessionRunnerModel } from "./runner/model.js"
import { SessionSchema } from "./schema.js"
@@ -80,23 +81,25 @@ const make = (dependencies: Dependencies) => {
})
: Effect.void,
)
const request = yield* SessionModelHook.apply(
dependencies.hooks,
{ sessionID: session.id, agent: agent.id, model: resolved.ref },
LLM.request({
model: resolved.model,
http: { headers: SessionModelHeaders.make(session, dependencies.app) },
system: agent.system,
messages: [Message.user(firstUser.text)],
tools: [],
}),
)
const streamed = yield* dependencies.llm
.stream(
LLM.request({
model: resolved.model,
http: { headers: SessionModelHeaders.make(session, dependencies.app) },
system: agent.system,
messages: [Message.user(firstUser.text)],
tools: [],
.stream(request, {
http: SessionModelHttp.middleware(dependencies.hooks, {
sessionID: session.id,
agent: agent.id,
model: resolved.ref,
}),
{
http: SessionModelHttp.middleware(dependencies.hooks, {
sessionID: session.id,
agent: agent.id,
model: resolved.ref,
}),
},
)
})
.pipe(
Stream.runForEach((event) => {
if (LLMEvent.is.providerError(event)) failed = true
+86
View File
@@ -0,0 +1,86 @@
import { describe, expect } from "bun:test"
import { Bus } from "@opencode-ai/core/bus"
import { Config } from "@opencode-ai/core/config"
import { ConfigImagePlugin } from "@opencode-ai/core/config/plugin/image"
import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder"
import { Image } from "@opencode-ai/core/image"
import { Plugin } from "@opencode-ai/core/plugin"
import { PluginHost } from "@opencode-ai/core/plugin/host"
import { Document, Event, Info, type Entry } from "@opencode-ai/schema/config"
import { Effect, Layer, Schema, Stream } from "effect"
import { testEffect } from "../lib/effect"
import { PluginTestLayer } from "../plugin/fixture"
const it = testEffect(Layer.merge(PluginTestLayer, AppNodeBuilder.build(Image.node)))
const decode = Schema.decodeUnknownSync(Info)
const content = {
uri: "file:///pixel.png",
content: "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mNk+A8AAQUBAScY42YAAAAASUVORK5CYII=",
encoding: "base64" as const,
mime: "image/png",
}
describe("ConfigImagePlugin.Plugin", () => {
it.live("merges image limits and reloads changed config", () =>
Effect.gen(function* () {
const image = yield* Image.Service
const bus = yield* Bus.Service
const config = yield* Config.Test
const plugins = yield* Plugin.Service
yield* ConfigImagePlugin.Plugin.effect(yield* PluginHost.make(plugins))
expect(yield* limits(image)).toEqual({ maxWidth: 1_200, maxHeight: 900, maxBytes: 1 })
yield* config.setEntries([document({ auto_resize: false, max_width: 700, max_base64_bytes: 1 })])
yield* bus.publish(Event.Updated, {})
yield* waitUntil(
limits(image).pipe(
Effect.map((current) => current.maxWidth === 700 && current.maxHeight === 2_000 && current.maxBytes === 1),
),
)
}).pipe(
Effect.provide(
Config.testLayer([
document({ auto_resize: false, max_width: 1_200 }),
document({ max_height: 900, max_base64_bytes: 1 }),
]),
),
),
)
it.live("refetches config after subscribing to updates", () =>
Effect.gen(function* () {
const image = yield* Image.Service
const plugins = yield* Plugin.Service
let reads = 0
const config = Config.Service.of({
entries: () => Effect.sync(() => [document({ max_width: reads++ === 0 ? 1_200 : 700, max_base64_bytes: 1 })]),
update: () => Effect.die(new Error("Config update is unavailable")),
changes: () => Stream.empty,
})
yield* ConfigImagePlugin.Plugin.effect(yield* PluginHost.make(plugins)).pipe(
Effect.provideService(Config.Service, config),
)
expect(yield* limits(image)).toEqual({ maxWidth: 700, maxHeight: 2_000, maxBytes: 1 })
}),
)
})
function document(image: NonNullable<typeof Info.Encoded.media>["image"]): Entry {
return new Document({ type: "document", info: decode({ media: { image } }) })
}
const limits = Effect.fnUntraced(function* (image: Image.Interface) {
const error = yield* image.normalize("pixel.png", content).pipe(Effect.flip, Effect.orDie)
if (error._tag !== "Image.SizeError") return yield* Effect.die(error)
return { maxWidth: error.maxWidth, maxHeight: error.maxHeight, maxBytes: error.maxBytes }
})
const waitUntil = Effect.fnUntraced(function* (condition: Effect.Effect<boolean>) {
for (let attempt = 0; attempt < 200; attempt++) {
if (yield* condition) return
yield* Effect.sleep("10 millis")
}
yield* Effect.die(new Error("Timed out waiting for image config reload"))
})
+193 -118
View File
@@ -1,41 +1,30 @@
import fs from "fs/promises"
import path from "path"
import { describe, expect } from "bun:test"
import { Effect, Layer, Schema } from "effect"
import { Deferred, Effect, Fiber } from "effect"
import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder"
import { Bus } from "@opencode-ai/core/bus"
import { Database } from "@opencode-ai/core/database/database"
import { LocationServiceMap } from "@opencode-ai/core/location-services"
import { PluginSupervisor } from "@opencode-ai/core/plugin/supervisor"
import { SdkPlugins } from "@opencode-ai/core/plugin/sdk"
import { AbsolutePath } from "@opencode-ai/core/schema"
import { Npm } from "@opencode-ai/util/npm"
import { Document, Info } from "@opencode-ai/schema/config"
import { Config } from "../src/config"
import { Info } from "@opencode-ai/schema/config"
import { Global } from "@opencode-ai/util/global"
import { LayerNode } from "@opencode-ai/util/effect/layer-node"
import { Formatter } from "../src/formatter"
import { Location } from "../src/location"
import { location } from "./fixture/location"
import { tempGlobalLayer } from "./fixture/global"
import { tmpdir } from "./fixture/tmpdir"
import { testEffect } from "./lib/effect"
const it = testEffect(Layer.empty)
const it = testEffect(
AppNodeBuilder.build(LayerNode.group([Database.node, Bus.node, SdkPlugins.node, LocationServiceMap.node]), [
[Global.node, tempGlobalLayer],
]),
)
type ConfigInput = typeof Info.Encoded
function formatterLayer(directory: string, configured?: ConfigInput["formatter"]) {
const entries =
configured === undefined
? []
: [
new Document({
type: "document",
info: Schema.decodeUnknownSync(Info)({ formatter: configured }),
}),
]
return AppNodeBuilder.build(Formatter.node, [
[Config.node, Config.testLayer(entries)],
[
Location.node,
Layer.succeed(Location.Service, Location.Service.of(location({ directory: AbsolutePath.make(directory) }))),
],
[Npm.node, Layer.mock(Npm.Service, { which: () => Effect.succeed(undefined) })],
])
}
function withTemp<A, E, R>(body: (directory: string) => Effect.Effect<A, E, R>) {
return Effect.acquireUseRelease(
Effect.promise(() => tmpdir()),
@@ -44,122 +33,208 @@ function withTemp<A, E, R>(body: (directory: string) => Effect.Effect<A, E, R>)
)
}
describe("Formatter", () => {
it.live("does not run formatters marked as disabled in config", () =>
withTemp((directory) =>
Effect.gen(function* () {
const file = path.join(directory, "test.disabled")
expect(yield* Formatter.Service.use((formatter) => formatter.file(file))).toBe(false)
}).pipe(
Effect.provide(
formatterLayer(directory, {
disabled: {
disabled: true,
command: [process.execPath, "-e", "process.exit(0)", "$FILE"],
extensions: [".disabled"],
},
}),
function withFormatter<A, E, R>(
configured: ConfigInput["formatter"],
body: (formatter: Formatter.Interface, directory: string) => Effect.Effect<A, E, R>,
) {
return withTemp((directory) =>
Effect.promise(() =>
fs.writeFile(path.join(directory, "opencode.json"), JSON.stringify({ formatter: configured })),
).pipe(
Effect.andThen(
Effect.gen(function* () {
const plugins = yield* PluginSupervisor.Service
yield* plugins.flush
return yield* body(yield* Formatter.Service, directory)
}).pipe(
Effect.scoped,
Effect.provide(
LocationServiceMap.Service.get(Location.Ref.make({ directory: AbsolutePath.make(directory) })),
),
),
),
),
)
}
describe("Formatter", () => {
it.live("does not run formatters marked as disabled in config", () =>
withFormatter(
{
disabled: {
disabled: true,
command: [process.execPath, "-e", "process.exit(0)", "$FILE"],
extensions: [".disabled"],
},
},
(formatter, directory) =>
Effect.gen(function* () {
const file = path.join(directory, "test.disabled")
expect(yield* formatter.file(file)).toBe(false)
}),
),
)
it.live("file() returns false when no formatter runs", () =>
withTemp((directory) =>
withFormatter(false, (formatter, directory) =>
Effect.gen(function* () {
const file = path.join(directory, "test.txt")
yield* Effect.promise(() => fs.writeFile(file, "x"))
expect(yield* Formatter.Service.use((formatter) => formatter.file(file))).toBe(false)
}).pipe(Effect.provide(formatterLayer(directory, false))),
expect(yield* formatter.file(file)).toBe(false)
}),
),
)
it.live("loads formatter state per directory", () =>
withTemp((off) =>
withTemp((on) =>
Effect.gen(function* () {
const offFile = path.join(off, "test.isolated")
const onFile = path.join(on, "test.isolated")
const disabled = yield* Formatter.Service.use((formatter) => formatter.file(offFile)).pipe(
Effect.provide(formatterLayer(off, false)),
)
const enabled = yield* Formatter.Service.use((formatter) => formatter.file(onFile)).pipe(
Effect.provide(
formatterLayer(on, {
isolated: {
command: [process.execPath, "-e", "process.exit(0)", "$FILE"],
extensions: [".isolated"],
},
}),
),
)
expect(disabled).toBe(false)
expect(enabled).toBe(true)
}),
withFormatter(false, (disabledFormatter, off) =>
withFormatter(
{
isolated: {
command: [process.execPath, "-e", "process.exit(0)", "$FILE"],
extensions: [".isolated"],
},
},
(enabledFormatter, on) =>
Effect.gen(function* () {
const offFile = path.join(off, "test.isolated")
const onFile = path.join(on, "test.isolated")
const disabled = yield* disabledFormatter.file(offFile)
const enabled = yield* enabledFormatter.file(onFile)
expect(disabled).toBe(false)
expect(enabled).toBe(true)
}),
),
),
)
it.live("stops after the first matching formatter succeeds", () =>
withTemp((directory) =>
Effect.gen(function* () {
const file = path.join(directory, "test.seq")
yield* Effect.promise(() => fs.writeFile(file, "x"))
expect(yield* Formatter.Service.use((formatter) => formatter.file(file))).toBe(true)
expect(yield* Effect.promise(() => fs.readFile(file, "utf8"))).toBe("xA")
}).pipe(
Effect.provide(
formatterLayer(directory, {
first: {
command: [
process.execPath,
"-e",
"const fs = require('fs'); const file = process.argv.at(-1); fs.appendFileSync(file, 'A')",
"$FILE",
],
extensions: [".seq"],
},
second: {
command: [
process.execPath,
"-e",
"const fs = require('fs'); const file = process.argv.at(-1); fs.appendFileSync(file, 'B')",
"$FILE",
],
extensions: [".seq"],
},
}),
),
),
withFormatter(
{
first: {
command: [
process.execPath,
"-e",
"const fs = require('fs'); const file = process.argv.at(-1); fs.appendFileSync(file, 'A')",
"$FILE",
],
extensions: [".seq"],
},
second: {
command: [
process.execPath,
"-e",
"const fs = require('fs'); const file = process.argv.at(-1); fs.appendFileSync(file, 'B')",
"$FILE",
],
extensions: [".seq"],
},
},
(formatter, directory) =>
Effect.gen(function* () {
const file = path.join(directory, "test.seq")
yield* Effect.promise(() => fs.writeFile(file, "x"))
expect(yield* formatter.file(file)).toBe(true)
expect(yield* Effect.promise(() => fs.readFile(file, "utf8"))).toBe("xA")
}),
),
)
it.live("tries the next matching formatter when the first fails", () =>
withTemp((directory) =>
withFormatter(
{
first: {
command: [process.execPath, "-e", "process.exit(1)", "$FILE"],
extensions: [".fallback"],
},
second: {
command: [
process.execPath,
"-e",
"const fs = require('fs'); const file = process.argv.at(-1); fs.appendFileSync(file, 'B')",
"$FILE",
],
extensions: [".fallback"],
},
},
(formatter, directory) =>
Effect.gen(function* () {
const file = path.join(directory, "test.fallback")
yield* Effect.promise(() => fs.writeFile(file, "x"))
expect(yield* formatter.file(file)).toBe(true)
expect(yield* Effect.promise(() => fs.readFile(file, "utf8"))).toBe("xB")
}),
),
)
it.live("rebuilds formatter state and clears resolved commands", () =>
withFormatter(false, (formatter, directory) =>
Effect.gen(function* () {
const file = path.join(directory, "test.fallback")
const command = { suffix: "A" }
yield* formatter.transform((draft) => {
const suffix = command.suffix
draft.set({
name: "reload",
extensions: [".reload"],
enabled: Effect.succeed([
process.execPath,
"-e",
`const fs = require('fs'); const file = process.argv.at(-1); fs.appendFileSync(file, '${suffix}')`,
"$FILE",
]),
})
})
const file = path.join(directory, "test.reload")
yield* Effect.promise(() => fs.writeFile(file, "x"))
expect(yield* Formatter.Service.use((formatter) => formatter.file(file))).toBe(true)
expect(yield* Effect.promise(() => fs.readFile(file, "utf8"))).toBe("xB")
}).pipe(
Effect.provide(
formatterLayer(directory, {
first: {
command: [process.execPath, "-e", "process.exit(1)", "$FILE"],
extensions: [".fallback"],
},
second: {
command: [
process.execPath,
"-e",
"const fs = require('fs'); const file = process.argv.at(-1); fs.appendFileSync(file, 'B')",
"$FILE",
],
extensions: [".fallback"],
},
}),
),
),
expect(yield* formatter.file(file)).toBe(true)
command.suffix = "B"
yield* formatter.reload()
expect(yield* formatter.file(file)).toBe(true)
expect(yield* Effect.promise(() => fs.readFile(file, "utf8"))).toBe("xAB")
}),
),
)
it.live("does not cache a command resolved before reload", () =>
withFormatter(false, (formatter, directory) =>
Effect.gen(function* () {
const resolving = yield* Deferred.make<void>()
const release = yield* Deferred.make<void>()
const command = { suffix: "A" }
yield* formatter.transform((draft) => {
const suffix = command.suffix
const resolved = [
process.execPath,
"-e",
`const fs = require('fs'); const file = process.argv.at(-1); fs.appendFileSync(file, '${suffix}')`,
"$FILE",
]
draft.set({
name: "reload-race",
extensions: [".race"],
enabled:
suffix === "A"
? Deferred.succeed(resolving, undefined).pipe(
Effect.andThen(Deferred.await(release)),
Effect.as(resolved),
)
: Effect.succeed(resolved),
})
})
const file = path.join(directory, "test.race")
yield* Effect.promise(() => fs.writeFile(file, "x"))
const first = yield* formatter.file(file).pipe(Effect.forkChild({ startImmediately: true }))
yield* Deferred.await(resolving)
command.suffix = "B"
yield* formatter.reload()
yield* Deferred.succeed(release, undefined)
expect(yield* Fiber.join(first)).toBe(true)
expect(yield* formatter.file(file)).toBe(true)
expect(yield* Effect.promise(() => fs.readFile(file, "utf8"))).toBe("xAB")
}),
),
)
})
-39
View File
@@ -35,17 +35,6 @@ describe("Npm.sanitize", () => {
})
})
describe("Npm.isRegistryPackage", () => {
test("accepts registry packages and rejects unsupported install targets", async () => {
expect(await Npm.isRegistryPackage("plugin")).toBe(true)
expect(await Npm.isRegistryPackage("@acme/plugin@beta")).toBe(true)
expect(await Npm.isRegistryPackage("plugin@^1.2.0")).toBe(true)
expect(await Npm.isRegistryPackage("./plugin")).toBe(false)
expect(await Npm.isRegistryPackage("github:acme/plugin")).toBe(false)
expect(await Npm.isRegistryPackage("alias@npm:plugin@1.0.0")).toBe(false)
})
})
describe("Npm.add", () => {
test("resolves cached scoped package specs without reifying", async () => {
await using tmp = await tmpdir()
@@ -117,31 +106,3 @@ describe("Npm.add", () => {
expect(entries.fallback.entrypoint).toEndWith("/index.js")
})
})
describe("Npm.resolve", () => {
test("resolves a TUI entrypoint only when the package is already cached", async () => {
await using tmp = await tmpdir()
const cache = path.join(tmp.path, "cache")
const spec = "fixture-plugin@1.0.0"
const directory = path.join(cache, "packages", Npm.sanitize(spec), "node_modules", "fixture-plugin")
const missing = await Effect.gen(function* () {
const npm = yield* Npm.Service
return yield* npm.resolve(spec, { subpaths: ["tui"] })
}).pipe(Effect.scoped, Effect.provide(npmLayer(cache)), Effect.runPromise)
expect(missing.entrypoint).toBeUndefined()
await fs.mkdir(directory, { recursive: true })
await writePackage(directory, {
name: "fixture-plugin",
exports: { ".": "./index.js", "./tui": "./tui.js" },
})
await Bun.write(path.join(directory, "index.js"), "export default {}\n")
await Bun.write(path.join(directory, "tui.js"), "export default {}\n")
const resolved = await Effect.gen(function* () {
const npm = yield* Npm.Service
return yield* npm.resolve(spec, { subpaths: ["tui"] })
}).pipe(Effect.scoped, Effect.provide(npmLayer(cache)), Effect.runPromise)
expect(resolved.entrypoint).toEndWith("/tui.js")
})
})
-1
View File
@@ -31,7 +31,6 @@ const npmLayer = Layer.succeed(
Npm.Service,
Npm.Service.of({
add: () => Effect.succeed({ directory: "", entrypoint: undefined }),
resolve: () => Effect.succeed({ directory: "", entrypoint: undefined }),
which: () => Effect.succeed(undefined),
}),
)
+16 -4
View File
@@ -320,10 +320,14 @@ describe("fromPromise", () => {
define({
id: "promise-session-http",
setup: async (ctx) => {
await ctx.session.hook("http.request", (event) => {
event.request = new Request("https://provider.test/changed", event.request)
event.request.headers.set("x-hook", "promise")
})
await ctx.session.hook(
"http.request",
(event) => {
event.request = new Request("https://provider.test/changed", event.request)
event.request.headers.set("x-hook", "promise")
},
{ providerID: "test" },
)
await ctx.session.hook("http.response", async (event) => {
event.response = new Response(`${await event.response.text()}-response`, {
status: event.response.status,
@@ -342,6 +346,11 @@ describe("fromPromise", () => {
...context,
request: new Request("https://provider.test", { method: "POST", body: "payload" }),
})
const ignored = yield* hooks.trigger("session", "http.request", {
...context,
model: Model.Ref.make({ providerID: Provider.ID.make("other"), id: Model.ID.make("model") }),
request: new Request("https://other.test"),
})
const response = yield* hooks.trigger("session", "http.response", {
...context,
request: request.request,
@@ -349,6 +358,9 @@ describe("fromPromise", () => {
})
expect(request.request.url).toBe("https://provider.test/changed")
expect(ignored.request.url).toBe("https://other.test/")
expect(yield* hooks.has("session", "http.request", Provider.ID.make("test"))).toBe(true)
expect(yield* hooks.has("session", "http.request", Provider.ID.make("other"))).toBe(false)
expect(yield* Effect.promise(() => response.response.text())).toBe("promise-response")
}),
)
@@ -23,7 +23,6 @@ const itWithAISDK = testEffect(Layer.mergeAll(PluginTestLayer, AppNodeBuilder.bu
function npmEntrypoint(entrypoint?: string) {
return Npm.Service.of({
add: () => Effect.succeed({ directory: "", entrypoint }),
resolve: () => Effect.succeed({ directory: "", entrypoint }),
which: () => Effect.succeed(undefined),
})
}
@@ -1,17 +1,25 @@
import { Money } from "@opencode-ai/schema/money"
import { Agent } from "@opencode-ai/schema/agent"
import { Session } from "@opencode-ai/schema/session"
import { OpenAIResponses } from "@opencode-ai/ai/protocols/openai-responses"
import { describe, expect } from "bun:test"
import { Effect } from "effect"
import { ConfigProvider, DateTime, Effect } from "effect"
import { Catalog } from "@opencode-ai/core/catalog"
import { Credential } from "@opencode-ai/core/credential"
import { Integration } from "@opencode-ai/core/integration"
import { Location } from "@opencode-ai/core/location"
import { Model } from "@opencode-ai/core/model"
import { Plugin } from "@opencode-ai/core/plugin"
import { PluginHost } from "@opencode-ai/core/plugin/host"
import { PluginHooks } from "@opencode-ai/core/plugin/hooks"
import { GithubCopilotPlugin } from "@opencode-ai/core/plugin/provider/github-copilot"
import { OpenAIPlugin } from "@opencode-ai/core/plugin/provider/openai"
import { Project } from "@opencode-ai/core/project"
import { Provider } from "@opencode-ai/core/provider"
import { AbsolutePath } from "@opencode-ai/core/schema"
import { SessionModelRequest } from "@opencode-ai/core/session/model-request"
import { SessionModelTransport } from "@opencode-ai/core/session/model-transport"
import { SessionRunnerModel } from "@opencode-ai/core/session/runner/model"
import { testEffect } from "../lib/effect"
import { PluginTestLayer } from "./fixture"
@@ -24,19 +32,33 @@ const addPlugin = Effect.fn(function* () {
yield* OpenAIPlugin.effect(host).pipe(Effect.provideService(Integration.Service, integrations))
})
const addGithubCopilotPlugin = Effect.fn(function* () {
const plugin = yield* Plugin.Service
const host = yield* PluginHost.make(plugin)
yield* GithubCopilotPlugin.effect(host)
})
function required<T>(value: T | undefined): T {
if (value === undefined) throw new Error("Expected value")
return value
}
const http = Effect.fn(function* (providerID: Provider.ID, url: string) {
const event = yield* (yield* PluginHooks.Service).trigger("session", "http.request", {
const request = Effect.fn(function* (providerID: Provider.ID, baseURL: string) {
const hooks = yield* PluginHooks.Service
const event = yield* hooks.trigger("session", "model.request", {
sessionID: Session.ID.make("ses_test"),
agent: Agent.ID.make("build"),
model: Model.Ref.make({ providerID, id: Model.ID.make("gpt-5.5") }),
request: new Request(url, { method: "POST", body: "{}" }),
baseURL,
headers: {},
})
return { url: event.request.url, headers: Object.fromEntries(event.request.headers.entries()) }
return {
baseURL: event.baseURL,
headers: event.headers,
hasHttpHooks:
(yield* hooks.has("session", "http.request", providerID)) ||
(yield* hooks.has("session", "http.response", providerID)),
}
})
describe("OpenAIPlugin", () => {
@@ -110,18 +132,19 @@ describe("OpenAIPlugin", () => {
})
yield* addPlugin()
const request = yield* http(Provider.ID.openai, "https://api.openai.com/v1/responses")
const custom = yield* http(Provider.ID.make("custom-openai"), "https://custom.example/v1/responses")
const proxy = yield* http(Provider.ID.openai, "https://proxy.example/v1/responses?region=us")
const direct = yield* request(Provider.ID.openai, "https://api.openai.com/v1")
const custom = yield* request(Provider.ID.make("custom-openai"), "https://custom.example/v1")
const proxy = yield* request(Provider.ID.openai, "https://proxy.example/v1?region=us")
const provider = required(yield* catalog.provider.get(Provider.ID.openai))
expect(provider.package).toBe("@opencode-ai/ai/providers/openai")
expect(provider.settings).toMatchObject({ baseURL: "https://chatgpt.com/backend-api/codex" })
expect(provider.headers).toMatchObject({ originator: "opencode", "chatgpt-account-id": "acct_123" })
expect(request.url).toBe("https://chatgpt.com/backend-api/codex/responses")
expect(request.headers).toMatchObject({ originator: "opencode", "session-id": "ses_test" })
expect(direct.baseURL).toBe("https://chatgpt.com/backend-api/codex")
expect(direct.headers).toMatchObject({ originator: "opencode", "session-id": "ses_test" })
expect(direct.hasHttpHooks).toBe(false)
expect(custom.headers).not.toHaveProperty("originator")
expect(proxy.url).toBe("https://proxy.example/v1/responses?region=us")
expect(proxy.baseURL).toBe("https://proxy.example/v1?region=us")
expect(proxy.headers).toMatchObject({ originator: "opencode", "session-id": "ses_test" })
const eligible = required(yield* catalog.model.get(Provider.ID.openai, Model.ID.make("gpt-5.5")))
expect(eligible.package).toBe("@opencode-ai/ai/providers/openai")
@@ -167,16 +190,77 @@ describe("OpenAIPlugin", () => {
})
yield* addPlugin()
const request = yield* http(Provider.ID.openai, "https://api.openai.com/v1/responses")
const direct = yield* request(Provider.ID.openai, "https://api.openai.com/v1")
const provider = required(yield* catalog.provider.get(Provider.ID.openai))
const model = required(yield* catalog.model.get(Provider.ID.openai, Model.ID.make("gpt-5.5")))
expect(model.package).toBe("@opencode-ai/ai/providers/openai")
expect(model.enabled).toBe(true)
expect(model.limit).toEqual({ context: 1_050_000, input: 922_000, output: 128_000 })
expect(request.headers).not.toHaveProperty("originator")
expect(direct.headers).not.toHaveProperty("originator")
expect(direct.hasHttpHooks).toBe(false)
expect(provider.headers).not.toHaveProperty("originator")
expect(required(yield* catalog.model.get(Provider.ID.openai, Model.ID.make("gpt-4.1"))).enabled).toBe(true)
}),
)
it.effect("selects WebSocket with the built-in provider hooks enabled", () =>
Effect.gen(function* () {
const credentials = yield* Credential.Service
yield* credentials.create({
integrationID: Integration.ID.make("openai"),
value: Credential.Key.make({ type: "key", key: "sk-test" }),
})
yield* addPlugin()
yield* addGithubCopilotPlugin()
const executor = { execute: () => Effect.die("unused WebSocket execution") }
const transport = SessionModelTransport.Service.of({
bind: () => executor,
close: () => Effect.void,
closeAll: Effect.void,
})
const sessionID = Session.ID.make("ses_websocket_hooks")
const agentID = Agent.ID.make("build")
const agent = Agent.Info.make(Agent.Info.default(agentID))
const model = SessionRunnerModel.resolved(OpenAIResponses.route.model({ id: "gpt-5.5" }), {
capabilities: { tools: true, input: ["text"], output: ["text"] },
cost: [],
})
const program = Effect.gen(function* () {
const requests = yield* SessionModelRequest.Service
return yield* requests.prepare({
context: {
session: Session.Info.make({
id: sessionID,
projectID: Project.ID.global,
cost: Money.USD.zero,
tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } },
time: { created: DateTime.makeUnsafe(0), updated: DateTime.makeUnsafe(0) },
location: Location.Ref.make({ directory: AbsolutePath.make("/project") }),
}),
agent: { id: agentID, info: agent },
model,
initial: "",
messages: [],
tools: { definitions: [], execute: () => Effect.die("unused tool execution") },
},
step: 1,
})
}).pipe(
Effect.provide(SessionModelRequest.layer),
Effect.provideService(SessionModelTransport.Service, transport),
Effect.provide(
ConfigProvider.layer(
ConfigProvider.fromEnv({ env: { OPENCODE_EXPERIMENTAL_OPENAI_RESPONSES_WEBSOCKET: "true" } }),
),
),
)
const prepared = yield* program
expect(prepared.webSocketEligible).toBe(true)
expect(prepared.options.webSocket).toBe(executor)
expect(prepared.options.http).toBeUndefined()
}),
)
})
@@ -14,7 +14,6 @@ const fixtureProvider = new URL("./fixtures/provider-factory.ts", import.meta.ur
const it = testEffect(PluginTestLayer)
const npm = Npm.Service.of({
add: () => Effect.succeed({ directory: "", entrypoint: undefined }),
resolve: () => Effect.succeed({ directory: "", entrypoint: undefined }),
which: () => Effect.succeed(undefined),
})
+18 -3
View File
@@ -73,7 +73,7 @@ const it = testEffect(
)
describe("Session.compact", () => {
it.effect("durably stacks manual compaction", () =>
it.effect("durably coalesces manual compaction", () =>
Effect.gen(function* () {
requests = []
const session = yield* Session.Service
@@ -102,11 +102,10 @@ describe("Session.compact", () => {
const first = yield* session.compact({ sessionID: created.id })
const second = yield* session.compact({ sessionID: created.id })
expect(second.id).not.toBe(first.id)
expect(second.id).toBe(first.id)
expect(requests).toHaveLength(0)
expect(yield* session.inbox(created.id)).toEqual([
expect.objectContaining({ id: first.id, type: "compaction", delivery: "queue" }),
expect.objectContaining({ id: second.id, type: "compaction", delivery: "queue" }),
])
expect((yield* session.context(created.id)).find((message) => message.id === first.id)).toBeUndefined()
@@ -115,4 +114,20 @@ describe("Session.compact", () => {
expect(steer).toMatchObject({ type: "compaction", delivery: "steer" })
}),
)
it.effect("coalesces concurrent manual compaction", () =>
Effect.gen(function* () {
const session = yield* Session.Service
const created = yield* session.create({ location })
const admitted = yield* Effect.all(
[SessionMessage.ID.create(), SessionMessage.ID.create()].map((id) =>
session.compact({ id, sessionID: created.id }),
),
{ concurrency: "unbounded" },
)
expect(admitted[1]?.id).toBe(admitted[0]?.id)
expect(yield* session.inbox(created.id)).toHaveLength(1)
}),
)
})
+110 -1
View File
@@ -14,8 +14,10 @@ import { SessionExecution } from "@opencode-ai/core/session/execution"
import { SessionRestart } from "@opencode-ai/core/session/execution/restart"
import { UserInterruptedError } from "@opencode-ai/core/session/error"
import { SessionEvent } from "@opencode-ai/core/session/event"
import { SessionInbox } from "@opencode-ai/core/session/inbox"
import { SessionMessage } from "@opencode-ai/core/session/message"
import { SessionRunner } from "@opencode-ai/core/session/runner/index"
import { SessionTable } from "@opencode-ai/core/session/sql"
import { SessionInboxTable, SessionTable } from "@opencode-ai/core/session/sql"
import { SessionStore } from "@opencode-ai/core/session/store"
import { Context, Deferred, Effect, Exit, Fiber, Layer, LayerMap, Scope } from "effect"
import { eq } from "drizzle-orm"
@@ -290,6 +292,113 @@ describe("SessionExecution lifecycle", () => {
)
})
describe("SessionExecution interrupt continuation", () => {
it.effect("resumes only steering input after an interrupt with continue", () =>
Effect.gen(function* () {
const database = yield* Database.Service
const sessionID = Session.ID.make("ses_continue_steer")
yield* seedSessions(database, [sessionID])
yield* seedInbox(database, sessionID, ["steer", "queue"])
const draining = yield* Deferred.make<void>()
const drains: Array<{ force: boolean; promotable?: SessionInbox.Promotable }> = []
const scope = yield* Scope.make()
yield* Effect.addFinalizer(() => Scope.close(scope, Exit.void))
const context = yield* buildExecution(scope, (input) =>
Effect.suspend(() => {
drains.push({ force: input.force, promotable: input.promotable })
if (drains.length > 1) return Effect.void
return Deferred.succeed(draining, undefined).pipe(Effect.andThen(Effect.never))
}),
)
const execution = Context.get(context, SessionExecution.Service)
yield* execution.resume(sessionID).pipe(Effect.forkScoped)
yield* Deferred.await(draining)
yield* execution.interrupt(sessionID, { continue: true })
yield* execution.awaitIdle(sessionID)
// The successor drain is steer-scoped: queued next-turn work stays parked.
expect(drains).toEqual([
{ force: true, promotable: "input" },
{ force: false, promotable: "steer" },
])
}),
)
it.effect("stays parked after an interrupt with continue when only queued work remains", () =>
Effect.gen(function* () {
const database = yield* Database.Service
const sessionID = Session.ID.make("ses_continue_parked")
yield* seedSessions(database, [sessionID])
yield* seedInbox(database, sessionID, ["queue"])
const draining = yield* Deferred.make<void>()
const drains: Array<SessionInbox.Promotable | undefined> = []
const scope = yield* Scope.make()
yield* Effect.addFinalizer(() => Scope.close(scope, Exit.void))
const context = yield* buildExecution(scope, (input) =>
Effect.suspend(() => {
drains.push(input.promotable)
return Deferred.succeed(draining, undefined).pipe(Effect.andThen(Effect.never))
}),
)
const execution = Context.get(context, SessionExecution.Service)
yield* execution.resume(sessionID).pipe(Effect.forkScoped)
yield* Deferred.await(draining)
yield* execution.interrupt(sessionID, { continue: true })
yield* execution.awaitIdle(sessionID)
expect(drains).toEqual(["input"])
expect(yield* execution.active).toEqual(new Set())
}),
)
it.effect("an idle interrupt with continue resumes pending steers", () =>
Effect.gen(function* () {
const database = yield* Database.Service
const sessionID = Session.ID.make("ses_continue_idle")
yield* seedSessions(database, [sessionID])
yield* seedInbox(database, sessionID, ["steer"])
const drains: Array<{ force: boolean; promotable?: SessionInbox.Promotable }> = []
const scope = yield* Scope.make()
yield* Effect.addFinalizer(() => Scope.close(scope, Exit.void))
const context = yield* buildExecution(scope, (input) =>
Effect.sync(() => void drains.push({ force: input.force, promotable: input.promotable })),
)
const execution = Context.get(context, SessionExecution.Service)
yield* execution.interrupt(sessionID, { continue: true })
yield* execution.awaitIdle(sessionID)
expect(drains).toEqual([{ force: false, promotable: "steer" }])
}),
)
})
function seedInbox(
database: Database.Service["Service"],
sessionID: Session.ID,
deliveries: ReadonlyArray<SessionInbox.Delivery>,
) {
return database.db
.insert(SessionInboxTable)
.values(
deliveries.map((delivery, index) => ({
id: SessionMessage.ID.create(),
session_id: sessionID,
type: "compaction" as const,
payload: {},
delivery,
enqueued_seq: index + 1,
})),
)
.run()
.pipe(Effect.orDie)
}
function seedSessions(
database: Database.Service["Service"],
sessionIDs: ReadonlyArray<Session.ID>,
@@ -291,18 +291,25 @@ it.effect(
instruction = "Changed context"
const before = yield* durableState(db, sessionID)
const hooks = yield* PluginHooks.Service
let modelRequestHook = false
yield* hooks.register("session", "context", (event) =>
Effect.sync(() => {
event.system = [SystemPart.make("Hooked system"), ...event.system]
if (event.tools.lookup) event.tools.lookup.description = "Hooked lookup"
}),
)
yield* hooks.register("session", "model.request", () =>
Effect.sync(() => {
modelRequestHook = true
}),
)
const generate = yield* SessionGenerate.Service
const result = yield* generate.generate({ sessionID, prompt: "Summarize privately" })
expect(result).toBe("Transient answer")
expect(requests).toHaveLength(1)
expect(modelRequestHook).toBe(true)
expect(hasHttpMiddleware).toBe(true)
expect(requests[0]?.model).toBe(model)
expect(requests[0]?.system[0]?.text).toBe("Hooked system")
+19 -6
View File
@@ -27,6 +27,7 @@ import { SessionStore } from "@opencode-ai/core/session/store"
import { LocationServiceMap } from "@opencode-ai/core/location-service-map"
import type { LocationServices } from "@opencode-ai/core/location-services"
import { Image } from "@opencode-ai/core/image"
import { PluginSupervisor } from "@opencode-ai/core/plugin/supervisor"
import { testEffect } from "./lib/effect"
const executionCalls: Session.ID[] = []
@@ -51,7 +52,6 @@ const execution = Layer.succeed(
Effect.sync(() => {
wakeCalls.push(sessionID)
}),
wakeActive: () => Effect.void,
awaitIdle: () => Effect.void,
}),
)
@@ -59,12 +59,25 @@ const locations = Layer.effect(
LocationServiceMap.Service,
LayerMap.make(
() =>
// Attachment admission only needs the location-scoped Image service.
// Attachment admission only needs image normalization and plugin readiness.
// oxlint-disable-next-line typescript-eslint/no-unsafe-type-assertion
Layer.mock(Image.Service, {
normalize: (_resource, content) =>
Effect.succeed(content.content.length > 5 * 1024 * 1024 ? { ...content, content: "AA==" } : content),
}) as unknown as Layer.Layer<LocationServices>,
Layer.unwrap(
Effect.sync(() => {
let ready = false
return Layer.mergeAll(
Layer.mock(Image.Service, {
normalize: (_resource, content) =>
ready
? Effect.succeed(content.content.length > 5 * 1024 * 1024 ? { ...content, content: "AA==" } : content)
: Effect.die(new Error("Image service used before plugins were ready")),
}),
Layer.succeed(
PluginSupervisor.Service,
PluginSupervisor.Service.of({ flush: Effect.sync(() => (ready = true)) }),
),
)
}),
) as unknown as Layer.Layer<LocationServices>,
),
)
const it = testEffect(
@@ -1,5 +1,6 @@
import { describe, expect } from "bun:test"
import { Cause, Deferred, Effect, Exit, Fiber, Layer } from "effect"
import { SessionInbox } from "@opencode-ai/core/session/inbox"
import { SessionRunCoordinator } from "@opencode-ai/core/session/run-coordinator"
import { testEffect } from "./lib/effect"
@@ -269,31 +270,24 @@ describe("SessionRunCoordinator", () => {
),
)
it.effect("replaces a settlement-window wake with a steer continuation", () =>
it.effect("a settlement-window wake starts a fresh execution with its own scope", () =>
Effect.scoped(
Effect.gen(function* () {
const settling = yield* Deferred.make<void>()
const release = yield* Deferred.make<void>()
const requests: SessionRunCoordinator.Request[] = []
const scopes: SessionInbox.Promotable[] = []
const coordinator = yield* SessionRunCoordinator.make({
drain: (_key, _force, request) => Effect.sync(() => requests.push(request)),
drain: (_key, _force, scope) => Effect.sync(() => scopes.push(scope)),
settled: () => Deferred.succeed(settling, undefined).pipe(Effect.andThen(Deferred.await(release))),
})
yield* coordinator.wake("session", "input")
yield* coordinator.wake("session", "steer")
yield* Deferred.await(settling)
yield* coordinator.wake("session", "input")
const interrupted = yield* coordinator
.interrupt("session", undefined, {
continue: { request: "steer", when: Effect.succeed(true) },
})
.pipe(Effect.forkChild)
yield* Effect.yieldNow
yield* Deferred.succeed(release, undefined)
yield* Fiber.join(interrupted)
yield* coordinator.awaitIdle("session")
expect(requests).toEqual(["input", "steer"])
expect(scopes).toEqual(["steer", "input"])
}),
),
)
@@ -371,17 +365,17 @@ describe("SessionRunCoordinator", () => {
),
)
it.effect("coalesces drain requests with input taking precedence", () =>
it.effect("coalesces drain scopes with input taking precedence", () =>
Effect.scoped(
Effect.gen(function* () {
const firstStarted = yield* Deferred.make<void>()
const release = yield* Deferred.make<void>()
const requests: SessionRunCoordinator.Request[] = []
const scopes: SessionInbox.Promotable[] = []
const coordinator = yield* SessionRunCoordinator.make({
drain: (_key, _force, request) =>
drain: (_key, _force, scope) =>
Effect.gen(function* () {
requests.push(request)
if (requests.length !== 1) return
scopes.push(scope)
if (scopes.length !== 1) return
yield* Deferred.succeed(firstStarted, undefined)
yield* Deferred.await(release)
}),
@@ -394,22 +388,22 @@ describe("SessionRunCoordinator", () => {
yield* Deferred.succeed(release, undefined)
yield* coordinator.awaitIdle("session")
expect(requests).toEqual(["steer", "input"])
expect(scopes).toEqual(["steer", "input"])
}),
),
)
it.effect("does not carry a completed input request into a steer drain", () =>
it.effect("does not carry a completed input scope into a steer drain", () =>
Effect.scoped(
Effect.gen(function* () {
const firstStarted = yield* Deferred.make<void>()
const release = yield* Deferred.make<void>()
const requests: SessionRunCoordinator.Request[] = []
const scopes: SessionInbox.Promotable[] = []
const coordinator = yield* SessionRunCoordinator.make({
drain: (_key, _force, request) =>
drain: (_key, _force, scope) =>
Effect.gen(function* () {
requests.push(request)
if (requests.length !== 1) return
scopes.push(scope)
if (scopes.length !== 1) return
yield* Deferred.succeed(firstStarted, undefined)
yield* Deferred.await(release)
}),
@@ -421,89 +415,23 @@ describe("SessionRunCoordinator", () => {
yield* Deferred.succeed(release, undefined)
yield* coordinator.awaitIdle("session")
expect(requests).toEqual(["input", "steer"])
expect(scopes).toEqual(["input", "steer"])
}),
),
)
it.effect("an active wake inherits scope without starting idle work", () =>
Effect.scoped(
Effect.gen(function* () {
const firstStarted = yield* Deferred.make<void>()
const release = yield* Deferred.make<void>()
const requests: SessionRunCoordinator.Request[] = []
const coordinator = yield* SessionRunCoordinator.make({
drain: (_key, _force, request) =>
Effect.gen(function* () {
requests.push(request)
if (requests.length !== 1) return
yield* Deferred.succeed(firstStarted, undefined)
yield* Deferred.await(release)
}),
})
yield* coordinator.wakeActive("session")
yield* coordinator.wake("session", "steer")
yield* Deferred.await(firstStarted)
yield* coordinator.wakeActive("session")
yield* Deferred.succeed(release, undefined)
yield* coordinator.awaitIdle("session")
expect(requests).toEqual(["steer", "steer"])
}),
),
)
it.effect("coalesces overlapping interrupt continuations into one steer successor", () =>
it.effect("a cleanup-era wake starts a successor with its own scope", () =>
Effect.scoped(
Effect.gen(function* () {
const firstStarted = yield* Deferred.make<void>()
const cleanupStarted = yield* Deferred.make<void>()
const cleanupGate = yield* Deferred.make<void>()
const requests: SessionRunCoordinator.Request[] = []
const scopes: SessionInbox.Promotable[] = []
const coordinator = yield* SessionRunCoordinator.make({
drain: (_key, _force, request) =>
drain: (_key, _force, scope) =>
Effect.gen(function* () {
requests.push(request)
if (requests.length !== 1) return
yield* Deferred.succeed(firstStarted, undefined)
yield* Effect.never.pipe(
Effect.onInterrupt(() =>
Deferred.succeed(cleanupStarted, undefined).pipe(Effect.andThen(Deferred.await(cleanupGate))),
),
)
}),
})
const continuation = { continue: { request: "steer" as const, when: Effect.succeed(false) } }
yield* coordinator.wake("session")
yield* Deferred.await(firstStarted)
const first = yield* coordinator.interrupt("session", undefined, continuation).pipe(Effect.forkChild)
yield* Deferred.await(cleanupStarted)
const second = yield* coordinator.interrupt("session", undefined, continuation).pipe(Effect.forkChild)
yield* Effect.yieldNow
yield* coordinator.wake("session", "input")
yield* Deferred.succeed(cleanupGate, undefined)
yield* Effect.all([Fiber.join(first), Fiber.join(second)])
yield* coordinator.awaitIdle("session")
expect(requests).toEqual(["input", "steer"])
}),
),
)
it.effect("a continuing interrupt replaces a cleanup-era input wake", () =>
Effect.scoped(
Effect.gen(function* () {
const firstStarted = yield* Deferred.make<void>()
const cleanupStarted = yield* Deferred.make<void>()
const cleanupGate = yield* Deferred.make<void>()
const requests: SessionRunCoordinator.Request[] = []
const coordinator = yield* SessionRunCoordinator.make({
drain: (_key, _force, request) =>
Effect.gen(function* () {
requests.push(request)
if (requests.length !== 1) return
scopes.push(scope)
if (scopes.length !== 1) return
yield* Deferred.succeed(firstStarted, undefined)
yield* Effect.never.pipe(
Effect.onInterrupt(() =>
@@ -515,45 +443,16 @@ describe("SessionRunCoordinator", () => {
yield* coordinator.wake("session", "input")
yield* Deferred.await(firstStarted)
const plain = yield* coordinator.interrupt("session").pipe(Effect.forkChild)
const interrupt = yield* coordinator.interrupt("session").pipe(Effect.forkChild)
yield* Deferred.await(cleanupStarted)
// A new admission during cancellation restarts normally: interruption only
// claims the wakes recorded before it.
yield* coordinator.wake("session", "input")
const continuing = yield* coordinator
.interrupt("session", undefined, {
continue: { request: "steer", when: Effect.succeed(false) },
})
.pipe(Effect.forkChild)
yield* Effect.yieldNow
yield* Deferred.succeed(cleanupGate, undefined)
yield* Effect.all([Fiber.join(plain), Fiber.join(continuing)])
yield* Fiber.join(interrupt)
yield* coordinator.awaitIdle("session")
expect(requests).toEqual(["input", "steer"])
}),
),
)
it.effect("does not start a conditional continuation without eligible work", () =>
Effect.scoped(
Effect.gen(function* () {
const started = yield* Deferred.make<void>()
const requests: SessionRunCoordinator.Request[] = []
const coordinator = yield* SessionRunCoordinator.make({
drain: (_key, _force, request) =>
Effect.sync(() => requests.push(request)).pipe(
Effect.andThen(Deferred.succeed(started, undefined)),
Effect.andThen(Effect.never),
),
})
yield* coordinator.wake("session")
yield* Deferred.await(started)
yield* coordinator.interrupt("session", undefined, {
continue: { request: "steer", when: Effect.succeed(false) },
})
yield* coordinator.awaitIdle("session")
expect(requests).toEqual(["input"])
expect(scopes).toEqual(["input", "input"])
}),
),
)
@@ -126,7 +126,6 @@ const execution = (llmClient: Layer.Layer<typeof LLMClient.Service>) =>
active: coordinator.active,
resume: coordinator.run,
wake: coordinator.wake,
wakeActive: coordinator.wakeActive,
interrupt: (sessionID) => coordinator.interrupt(sessionID),
awaitIdle: coordinator.awaitIdle,
})
+30 -1
View File
@@ -436,7 +436,6 @@ const execution = Layer.effect(
active: coordinator.active,
resume: coordinator.run,
wake: coordinator.wake,
wakeActive: coordinator.wakeActive,
interrupt: (sessionID) => coordinator.interrupt(sessionID),
awaitIdle: coordinator.awaitIdle,
})
@@ -1020,6 +1019,36 @@ describe("SessionRunnerLLM", () => {
}),
)
it.effect("keeps WebSocket eligibility after model request hooks", () =>
Effect.gen(function* () {
yield* setup
const hooks = yield* PluginHooks.Service
yield* hooks.register("session", "model.request", (event) =>
Effect.sync(() => {
event.headers["x-model-request-hook"] = "active"
}),
)
yield* hooks.register("session", "http.request", () => Effect.die("Other-provider HTTP hook should not apply"), {
providerID: Provider.ID.githubCopilot,
})
const context = yield* SessionContext.Service
const modelRequests = yield* SessionModelRequest.Service
const selected = yield* context.select(sessionID)
const database = yield* Database.Service
const bus = yield* Bus.Service
yield* InstructionState.prepare(database.db, bus, selected.instructions, sessionID)
const prepared = yield* modelRequests.prepare({
context: yield* context.load(selected),
step: 1,
})
expect(prepared.request.http?.headers?.["x-model-request-hook"]).toBe("active")
expect(prepared.webSocketEligible).toBe(true)
expect(prepared.options.http).toBeUndefined()
}),
)
it.effect("forces HTTP and triggers active request and response hooks once", () =>
Effect.gen(function* () {
yield* setup
+13 -40
View File
@@ -1,9 +1,7 @@
import { beforeEach, describe, expect } from "bun:test"
import path from "path"
import { Effect, Exit, Layer, Stream } from "effect"
import { Effect, Exit, Layer } from "effect"
import { Config } from "@opencode-ai/core/config"
import { Document, Info } from "@opencode-ai/schema/config"
import { ConfigMedia } from "@opencode-ai/schema/config/media"
import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder"
import { LayerNode } from "@opencode-ai/util/effect/layer-node"
import { FileSystem } from "@opencode-ai/core/filesystem"
@@ -90,7 +88,7 @@ const permission = permissionLayer({
),
})
const config = Config.testLayer()
const imageLayer = AppNodeBuilder.build(Image.node, [[Config.node, config]])
const imageLayer = AppNodeBuilder.build(Image.node)
const testFileSystem = Layer.effect(
FSUtil.Service,
FSUtil.Service.use((fs) =>
@@ -130,10 +128,9 @@ const mutation = Layer.succeed(
},
}),
)
const unavailableImage = Layer.succeed(
Image.Service,
Image.Service.of({ normalize: () => Effect.fail(new Image.ResizerUnavailableError()) }),
)
const unavailableImage = Layer.mock(Image.Service, {
normalize: () => Effect.fail(new Image.ResizerUnavailableError()),
})
const readLayer = (imageLayer: Layer.Layer<Image.Service>) =>
Layer.mergeAll(
AppNodeBuilder.build(LayerNode.group([Tool.node, readToolNode]), [
@@ -146,8 +143,9 @@ const readLayer = (imageLayer: Layer.Layer<Image.Service>) =>
[Location.node, locationLayer],
[Global.node, Global.layerWith({ data: Global.Path.data })],
]),
// Merge by reference so Config.Test resolves to the memoized instance.
// Merge by reference so Config.Test and Image.Service resolve to the memoized instances.
config,
imageLayer,
)
const it = testEffect(readLayer(imageLayer))
const itWithoutResizer = testEffect(readLayer(unavailableImage))
@@ -384,17 +382,8 @@ describe("ReadTool", () => {
encoding: "base64",
mime: "image/png",
}
const configTest = yield* Config.Test
yield* configTest.setEntries([
new Document({
type: "document",
info: new Info({
media: new ConfigMedia.Info({
image: new ConfigMedia.Image({ auto_resize: false, max_width: 4 }),
}),
}),
}),
])
const image = yield* Image.Service
yield* image.transform((draft) => draft.configure({ autoResize: false, maxWidth: 4 }))
const registry = yield* Tool.Service
expect(
@@ -427,15 +416,8 @@ describe("ReadTool", () => {
encoding: "base64",
mime: "image/png",
}
const configTest = yield* Config.Test
yield* configTest.setEntries([
new Document({
type: "document",
info: new Info({
media: new ConfigMedia.Info({ image: new ConfigMedia.Image({ max_width: 4 }) }),
}),
}),
])
const image = yield* Image.Service
yield* image.transform((draft) => draft.configure({ maxWidth: 4 }))
const registry = yield* Tool.Service
const result = yield* executeTool(registry, {
sessionID,
@@ -466,17 +448,8 @@ describe("ReadTool", () => {
encoding: "base64",
mime: "image/png",
}
const configTest = yield* Config.Test
yield* configTest.setEntries([
new Document({
type: "document",
info: new Info({
media: new ConfigMedia.Info({
image: new ConfigMedia.Image({ max_base64_bytes: 1 }),
}),
}),
}),
])
const image = yield* Image.Service
yield* image.transform((draft) => draft.configure({ maxBase64Bytes: 1 }))
const registry = yield* Tool.Service
expect(
-1
View File
@@ -115,7 +115,6 @@ const executionNode = makeGlobalNode({
active: Effect.succeed(new Set()),
resume: complete,
wake: () => Effect.void,
wakeActive: () => Effect.void,
interrupt: () => Effect.void,
awaitIdle: (id) => complete(id).pipe(Effect.exit, Effect.asVoid),
})
-1
View File
@@ -88,7 +88,6 @@ const executionNode = makeGlobalNode({
active: Effect.succeed(new Set()),
resume: complete,
wake: () => Effect.void,
wakeActive: () => Effect.void,
interrupt: () => Effect.void,
awaitIdle: (sessionID) => complete(sessionID).pipe(Effect.exit, Effect.asVoid),
})
+2 -2
View File
@@ -1,6 +1,6 @@
import type { LanguageModelV3 } from "@ai-sdk/provider"
import type { Model } from "@opencode-ai/schema/model"
import type { Hooks } from "./registration.js"
import type { ModelHooks } from "./registration.js"
export interface AISDKHooks {
sdk: {
@@ -18,5 +18,5 @@ export interface AISDKHooks {
}
export interface AISDKDomain {
readonly hook: Hooks<AISDKHooks>
readonly hook: ModelHooks<AISDKHooks>
}
@@ -4,6 +4,11 @@ export interface Registration {
readonly dispose: Effect.Effect<void>
}
export interface ModelHookOptions {
/** Limits the hook to one provider. Unscoped hooks apply to every provider. */
readonly providerID?: string
}
export type Hooks<Spec, Failures extends Record<keyof Spec, unknown> = Record<keyof Spec, never>> = <
Name extends keyof Spec,
>(
@@ -11,4 +16,12 @@ export type Hooks<Spec, Failures extends Record<keyof Spec, unknown> = Record<ke
callback: (input: Spec[Name]) => Effect.Effect<void, Failures[Name]>,
) => Effect.Effect<Registration, never, Scope.Scope>
export type ModelHooks<Spec, Failures extends Record<keyof Spec, unknown> = Record<keyof Spec, never>> = <
Name extends keyof Spec,
>(
name: Name,
callback: (input: Spec[Name]) => Effect.Effect<void, Failures[Name]>,
options?: ModelHookOptions,
) => Effect.Effect<Registration, never, Scope.Scope>
export type Transform<Input> = (callback: (input: Input) => void) => Effect.Effect<Registration, never, Scope.Scope>
+11 -2
View File
@@ -4,7 +4,7 @@ import type { Agent } from "@opencode-ai/schema/agent"
import type { Model } from "@opencode-ai/schema/model"
import type { Session } from "@opencode-ai/schema/session"
import type { JsonSchema } from "effect"
import type { Hooks } from "./registration.js"
import type { ModelHooks } from "./registration.js"
export interface SessionContext {
readonly sessionID: Session.ID
@@ -15,6 +15,14 @@ export interface SessionContext {
tools: Record<string, { description: string; input: JsonSchema.JsonSchema }>
}
export interface SessionModelRequest {
readonly sessionID: Session.ID
readonly agent: Agent.ID
readonly model: Model.Ref
baseURL?: string
headers: Record<string, string>
}
export interface SessionHttpRequest {
readonly sessionID: Session.ID
readonly agent: Agent.ID
@@ -32,6 +40,7 @@ export interface SessionHttpResponse {
export interface SessionHooks {
readonly context: SessionContext
readonly "model.request": SessionModelRequest
readonly "http.request": SessionHttpRequest
readonly "http.response": SessionHttpResponse
}
@@ -40,5 +49,5 @@ export type SessionDomain = Pick<
SessionApi<unknown>,
"create" | "get" | "prompt" | "generate" | "command" | "synthetic" | "interrupt" | "rename" | "wait"
> & {
readonly hook: Hooks<SessionHooks>
readonly hook: ModelHooks<SessionHooks>
}
+8 -4
View File
@@ -130,8 +130,10 @@ export function fromPromise(plugin: Plugin) {
reload: () => run(host.agent.reload()),
},
aisdk: {
hook: (name, callback) =>
register(host.aisdk.hook(name, (event) => Effect.promise(() => Promise.resolve(callback(event))))),
hook: (name, callback, options) =>
register(
host.aisdk.hook(name, (event) => Effect.promise(() => Promise.resolve(callback(event))), options),
),
},
catalog: {
provider: {
@@ -295,8 +297,10 @@ export function fromPromise(plugin: Plugin) {
),
},
session: {
hook: (name, callback) =>
register(host.session.hook(name, (event) => Effect.promise(() => Promise.resolve(callback(event))))),
hook: (name, callback, options) =>
register(
host.session.hook(name, (event) => Effect.promise(() => Promise.resolve(callback(event))), options),
),
create: adaptApiMethod(SessionEndpoints["session.create"], host.session.create),
get: adaptApiMethod(SessionEndpoints["session.get"], host.session.get),
prompt: adaptApiMethod(SessionEndpoints["session.prompt"], host.session.prompt),
+2 -2
View File
@@ -1,6 +1,6 @@
import type { LanguageModelV3 } from "@ai-sdk/provider"
import type { Model } from "@opencode-ai/schema/model"
import type { Hooks } from "./registration.js"
import type { ModelHooks } from "./registration.js"
export interface AISDKHooks {
sdk: {
@@ -18,5 +18,5 @@ export interface AISDKHooks {
}
export interface AISDKDomain {
readonly hook: Hooks<AISDKHooks>
readonly hook: ModelHooks<AISDKHooks>
}
@@ -2,9 +2,20 @@ export interface Registration {
readonly dispose: () => Promise<void>
}
export interface ModelHookOptions {
/** Limits the hook to one provider. Unscoped hooks apply to every provider. */
readonly providerID?: string
}
export type Hooks<Spec> = <Name extends keyof Spec>(
name: Name,
callback: (input: Spec[Name]) => Promise<void> | void,
) => Promise<Registration>
export type ModelHooks<Spec> = <Name extends keyof Spec>(
name: Name,
callback: (input: Spec[Name]) => Promise<void> | void,
options?: ModelHookOptions,
) => Promise<Registration>
export type Transform<Input> = (callback: (input: Input) => void) => Promise<Registration>
+11 -2
View File
@@ -4,7 +4,7 @@ import type { Agent } from "@opencode-ai/schema/agent"
import type { Model } from "@opencode-ai/schema/model"
import type { Session } from "@opencode-ai/schema/session"
import type { JsonSchema } from "effect"
import type { Hooks } from "./registration.js"
import type { ModelHooks } from "./registration.js"
export interface SessionContext {
readonly sessionID: Session.ID
@@ -15,6 +15,14 @@ export interface SessionContext {
tools: Record<string, { description: string; input: JsonSchema.JsonSchema }>
}
export interface SessionModelRequest {
readonly sessionID: Session.ID
readonly agent: Agent.ID
readonly model: Model.Ref
baseURL?: string
headers: Record<string, string>
}
export interface SessionHttpRequest {
readonly sessionID: Session.ID
readonly agent: Agent.ID
@@ -32,6 +40,7 @@ export interface SessionHttpResponse {
export interface SessionHooks {
readonly context: SessionContext
readonly "model.request": SessionModelRequest
readonly "http.request": SessionHttpRequest
readonly "http.response": SessionHttpResponse
}
@@ -40,5 +49,5 @@ export type SessionDomain = Pick<
SessionApi,
"create" | "get" | "prompt" | "generate" | "command" | "synthetic" | "interrupt" | "rename" | "wait"
> & {
readonly hook: Hooks<SessionHooks>
readonly hook: ModelHooks<SessionHooks>
}
-1
View File
@@ -24,7 +24,6 @@
"./context/client": "./src/context/client.tsx",
"./context/theme": "./src/context/theme.tsx",
"./theme/discovery": "./src/theme/discovery.ts",
"./plugin/discovery": "./src/plugin/discovery.ts",
"./context/editor": "./src/context/editor.ts",
"./context/clipboard": "./src/context/clipboard.tsx",
"./attention": "./src/attention.ts",
+6 -47
View File
@@ -1,4 +1,3 @@
import type { PluginInfo } from "@opencode-ai/client"
import type { Plugin } from "@opencode-ai/plugin/tui"
import { createMarkdownCodeBlockRenderer, type MarkdownCodeBlockRenderer, type MarkdownOptions } from "@opentui/core"
import {
@@ -6,7 +5,6 @@ import {
createContext,
createEffect,
createMemo,
createSignal,
on,
onCleanup,
onMount,
@@ -23,8 +21,6 @@ import { isDeepEqual } from "remeda"
import "#runtime-plugin-support"
import { useConfig } from "../config"
import { useTuiLifecycle } from "../context/runtime"
import { useClient } from "../context/client"
import { useData } from "../context/data"
import { errorMessage } from "../util/error"
import { builtins } from "./builtins"
import { createPluginContext, usePluginHost, type Dispose, type RegisteredSlot, type SlotRender } from "./api"
@@ -32,7 +28,7 @@ import { createSourceWatcher } from "./watch"
import { discoverTuiPlugins, freshSpecifier, localSource } from "./discovery"
export interface PackageResolver {
readonly resolve: (spec: string, install?: boolean) => Promise<string | undefined>
readonly resolve: (spec: string) => Promise<string | undefined>
}
type State =
@@ -94,13 +90,6 @@ export function PluginProvider(props: ParentProps<{ packages: PackageResolver; d
const host = usePluginHost()
const config = useConfig()
const lifecycle = useTuiLifecycle()
const client = useClient()
const data = useData()
const [serverPlugins, setServerPlugins] = createSignal<
ReadonlyArray<
Extract<PluginInfo, { readonly status: "active" }> & { readonly source: { readonly type: "package" } }
>
>([])
const directory = config.path ? path.dirname(config.path) : process.cwd()
const [store, setStore] = createStore({
ready: false,
@@ -241,11 +230,7 @@ export function PluginProvider(props: ParentProps<{ packages: PackageResolver; d
const npmFailures = new Map<string, string>()
const reconcile = async () => {
await Promise.all(props.directories.map(watcher.wait))
const entries = [
...(await discoverTuiPlugins(props.directories)).map((entry) => ({ entry, install: true, server: false })),
...serverPlugins().map((plugin) => ({ entry: plugin.source.package, install: false, server: true })),
...(config.data.plugins ?? []).map((entry) => ({ entry, install: true, server: false })),
]
const entries = [...(await discoverTuiPlugins(props.directories)), ...(config.data.plugins ?? [])]
// Resolve: fold entries into one desired generation. A source that fails
// to import keeps its running previous version and only reports failure.
@@ -253,8 +238,7 @@ export function PluginProvider(props: ParentProps<{ packages: PackageResolver; d
for (const plugin of builtins)
desired.set(plugin.id, { plugin, source: "builtin", version: "builtin", enabled: true })
const failures: State[] = []
for (const source of entries) {
const entry = source.entry
for (const entry of entries) {
const target = typeof entry === "string" ? entry : entry.package
if (target.startsWith("-")) {
for (const item of desired.values()) if (matches(target.slice(1), item.plugin.id)) item.enabled = false
@@ -275,12 +259,11 @@ export function PluginProvider(props: ParentProps<{ packages: PackageResolver; d
const memo = local ? undefined : npmFailures.get(target)
const resolved = memo
? { status: "failed" as const, error: memo }
: await resolvePlugin(target, local, options, previous, props.packages, source.install).catch((error) => ({
: await resolvePlugin(target, local, options, previous, props.packages).catch((error) => ({
status: "failed" as const,
error: errorMessage(error),
}))
if (resolved.status === "unsupported") {
if (source.server) continue
failures.push({ target, status: "unsupported" })
continue
}
@@ -456,7 +439,7 @@ export function PluginProvider(props: ParentProps<{ packages: PackageResolver; d
const resolved = createMemo(() => resolveSlots({ paths: new Set(Object.keys(mounted)), claims: claims() }))
createEffect(
on(
() => JSON.stringify([serverPlugins(), config.data.plugins ?? []]),
() => JSON.stringify(config.data.plugins ?? []),
() => {
npmFailures.clear()
void enqueue(reconcile).then(
@@ -466,29 +449,6 @@ export function PluginProvider(props: ParentProps<{ packages: PackageResolver; d
},
),
)
const syncServerPlugins = () =>
client.api.plugin
.list({ location: data.location.default() })
.then((response) =>
setServerPlugins(
response.data.filter(
(
plugin,
): plugin is Extract<PluginInfo, { readonly status: "active" }> & {
readonly source: { readonly type: "package" }
} => plugin.status === "active" && plugin.tui && plugin.source.type === "package",
),
),
)
.catch(() => undefined)
createEffect(
on(
() => JSON.stringify(data.location.default()),
() => void syncServerPlugins(),
),
)
onCleanup(client.event.on("plugin.updated", syncServerPlugins))
onCleanup(client.event.on("server.connected", syncServerPlugins))
onMount(() => {
let disposing: Promise<void> | undefined
const dispose = () => {
@@ -563,13 +523,12 @@ async function resolvePlugin(
options: Readonly<Record<string, any>> | undefined,
previous: Registration | undefined,
packages: PackageResolver,
install: boolean,
) {
// Package entrypoints never change within a session, so a loaded previous
// version needs no re-resolution (which could otherwise hit npm).
if (!local && previous && sameOptions(previous.options, options))
return { status: "unchanged" as const, plugin: previous.plugin, version: previous.version }
const entrypoint = local ? await resolveLocal(local) : await packages.resolve(spec, install)
const entrypoint = local ? await resolveLocal(local) : await packages.resolve(spec)
if (!entrypoint) return { status: "unsupported" as const }
// The cache-busted specifier doubles as the version: unique per entrypoint
// and mtime, so equal versions mean an identical module.
@@ -1,6 +1,7 @@
import { createMemo, For, Show, createEffect, onMount, onCleanup } from "solid-js"
import { createStore } from "solid-js/store"
import { TextAttributes, ScrollBoxRenderable } from "@opentui/core"
import type { SessionInfo } from "@opencode-ai/client"
import { useRoute, useRouteData } from "../../../context/route"
import { useData } from "../../../context/data"
import { useClient } from "../../../context/client"
@@ -9,6 +10,7 @@ import { Locale } from "../../../util/locale"
import { Keymap } from "../../../context/keymap"
import { useComposerTab } from "./index"
import { withTimestampedFallback } from "@opencode-ai/util/session-title-fallback"
import { sessionFamily } from "../../../util/session"
interface SubagentEntry {
sessionID: string
@@ -16,6 +18,7 @@ interface SubagentEntry {
title: string
status: string
current: boolean
prefix: string
}
export function SubagentsTab(props: { sessionID: string }) {
@@ -34,47 +37,24 @@ export function SubagentsTab(props: { sessionID: string }) {
const current = session()
if (!current) return []
const result: SubagentEntry[] = []
if (current.parentID) {
const siblings = data.session.list().filter((s) => s.parentID === current.parentID)
for (const sibling of siblings) {
const title = withTimestampedFallback(sibling)
const result = sessionFamily<SessionInfo>(data.session.list(), current.id).map(
({ session, prefix }): SubagentEntry => {
const title = withTimestampedFallback(session)
const agentMatch = title.match(/@(\w+) subagent/)
const agent = sibling.agent
? Locale.titlecase(sibling.agent)
: agentMatch
? Locale.titlecase(agentMatch[1])
: "Subagent"
const name = agentMatch ? title.replace(agentMatch[0], "").trim() || title : title
result.push({
sessionID: sibling.id,
agent,
title: name,
status: data.session.status(sibling.id),
current: sibling.id === route.sessionID,
})
}
} else {
const children = data.session.list().filter((s) => s.parentID === props.sessionID)
for (const child of children) {
const title = withTimestampedFallback(child)
const agentMatch = title.match(/@(\w+) subagent/)
const agent = child.agent
? Locale.titlecase(child.agent)
: agentMatch
? Locale.titlecase(agentMatch[1])
: "Subagent"
const name = agentMatch ? title.replace(agentMatch[0], "").trim() || title : title
result.push({
sessionID: child.id,
agent,
title: name,
status: data.session.status(child.id),
current: child.id === route.sessionID,
})
}
}
return {
sessionID: session.id,
agent: session.agent
? Locale.titlecase(session.agent)
: agentMatch
? Locale.titlecase(agentMatch[1])
: "Subagent",
title: agentMatch ? title.replace(agentMatch[0], "").trim() || title : title,
status: data.session.status(session.id),
current: session.id === route.sessionID,
prefix,
}
},
)
return result.filter((entry) => (store.active ? entry.status === "running" : entry.status !== "running"))
})
@@ -264,6 +244,7 @@ export function SubagentsTab(props: { sessionID: string }) {
attributes={active() ? TextAttributes.BOLD : undefined}
wrapMode="none"
>
{entry.prefix}
{entry.agent}: {entry.title}
</text>
</box>
+41
View File
@@ -1,6 +1,47 @@
import type { ModelInfo, SessionMessageAssistant, SessionMessageInfo } from "@opencode-ai/client"
import { Locale } from "./locale"
type SessionNode = {
id: string
parentID?: string | null
}
export function sessionFamily<T extends SessionNode>(sessions: readonly T[], sessionID: string) {
const byID = new Map(sessions.map((session) => [session.id, session]))
const current = byID.get(sessionID)
if (!current) return []
const children = new Map<string, T[]>()
sessions.forEach((session) => {
if (!session.parentID) return
const group = children.get(session.parentID)
if (group) group.push(session)
else children.set(session.parentID, [session])
})
function root(session: T): T {
const parent = session.parentID ? byID.get(session.parentID) : undefined
return parent ? root(parent) : session
}
function walk(parentID: string, ancestors: boolean[]): Array<{ session: T; prefix: string }> {
const group = children.get(parentID) ?? []
return group.flatMap((session, index) => {
const last = index === group.length - 1
const prefix =
ancestors.length === 0
? ""
: ancestors
.slice(1)
.map((ancestor) => (ancestor ? " " : "│ "))
.join("") + (last ? "└─ " : "├─ ")
return [{ session, prefix }, ...walk(session.id, [...ancestors, last])]
})
}
return walk(root(current).id, [])
}
export function lastAssistantWithUsage(messages: ReadonlyArray<SessionMessageInfo>, boundary?: string) {
const boundaryIndex = boundary ? messages.findIndex((message) => message.id === boundary) : -1
if (boundary && boundaryIndex === -1) return undefined
-5
View File
@@ -95,11 +95,6 @@ export function createFetch(override?: FetchHandler, events?: ReturnType<typeof
if (url.pathname === "/path") return json({ home: "", state: "", config: "", worktree, directory })
if (url.pathname === "/api/location")
return json({ directory, project: { id: "proj_test", directory: worktree, canonical: worktree } })
if (url.pathname === "/api/plugin")
return json({
location: { directory, project: { id: "proj_test", directory: worktree, canonical: worktree } },
data: [],
})
if (url.pathname === "/api/vcs")
return json({
location: { directory, project: { id: "proj_test", directory: worktree, canonical: worktree } },
+2 -51
View File
@@ -4,7 +4,6 @@ import { Effect, FileSystem } from "effect"
import { Global } from "@opencode-ai/util/global"
import { mkdir, readFile, symlink, writeFile } from "node:fs/promises"
import path from "node:path"
import { pathToFileURL } from "node:url"
import { createEventStream, createFetch, json } from "./fixture/tui-client"
import { tmpdir } from "./fixture/fixture"
@@ -31,26 +30,12 @@ async function until(read: () => Promise<string>, expected: (value: string | und
return value
}
async function bootApp(
directory: string,
options?: {
plugins?: unknown[]
resolve?: (spec: string, install?: boolean) => Promise<string | undefined>
},
) {
async function bootApp(directory: string) {
const setup = await createTestRenderer({ width: 80, height: 24, useThread: false })
const core = await import("@opentui/core")
mock.module("@opentui/core", () => ({ ...core, createCliRenderer: async () => setup.renderer }))
const events = createEventStream()
const calls = createFetch((url) => {
if (url.pathname === "/api/plugin")
return json({
location: {
directory,
project: { id: "proj_test", directory, canonical: directory },
},
data: options?.plugins ?? [],
})
if (url.pathname !== "/api/fs/list") return
return json({
location: {
@@ -69,7 +54,7 @@ async function bootApp(
app: { name: "test", version: "test", channel: "test" },
server: { endpoint: { url: server.url.toString() } },
config: { get: async () => ({}), update: async () => ({}) },
packages: { resolve: options?.resolve ?? (async () => undefined) },
packages: { resolve: async () => undefined },
args: {},
log: () => {},
}).pipe(
@@ -88,40 +73,6 @@ async function bootApp(
}
}
test("loads an advertised package TUI entrypoint only from the local cache", async () => {
await using tmp = await tmpdir()
const marker = path.join(tmp.path, "marker.txt")
const entrypoint = path.join(tmp.path, "tui.ts")
await writeFile(entrypoint, lifecycleSource(marker, "test.package", "package"))
const resolutions: Array<{ spec: string; install?: boolean }> = []
await using app = await bootApp(tmp.path, {
plugins: [
{
id: "test.server",
source: { type: "package", package: "test-plugin@1.0.0" },
status: "active",
tui: true,
},
],
resolve: async (spec, install) => {
resolutions.push({ spec, install })
return pathToFileURL(entrypoint).href
},
})
expect(
await until(
() => readFile(marker, "utf8"),
(value) => value === "package:setup\n",
),
).toBe("package:setup\n")
expect(resolutions).toContainEqual({ spec: "test-plugin@1.0.0", install: false })
process.emit("SIGHUP")
await app.task
})
test("discovers an ancestor TUI plugin directory created after startup", async () => {
await using tmp = await tmpdir()
const cwd = path.join(tmp.path, "repo", "packages", "app")
+22 -1
View File
@@ -1,6 +1,6 @@
import { describe, expect, test } from "bun:test"
import type { SessionMessageInfo } from "@opencode-ai/client"
import { lastAssistantWithUsage } from "../../src/util/session"
import { lastAssistantWithUsage, sessionFamily } from "../../src/util/session"
const assistant = (id: string, input: number): SessionMessageInfo => ({
id,
@@ -13,6 +13,27 @@ const assistant = (id: string, input: number): SessionMessageInfo => ({
})
describe("util.session", () => {
test("flattens nested subagents from any session in the family", () => {
const sessions = [
{ id: "root" },
{ id: "child-a", parentID: "root" },
{ id: "grandchild-a", parentID: "child-a" },
{ id: "great-grandchild-a", parentID: "grandchild-a" },
{ id: "grandchild-a2", parentID: "child-a" },
{ id: "child-b", parentID: "root" },
{ id: "grandchild-b", parentID: "child-b" },
]
expect(sessionFamily(sessions, "great-grandchild-a")).toEqual([
{ session: sessions[1], prefix: "" },
{ session: sessions[2], prefix: "├─ " },
{ session: sessions[3], prefix: "│ └─ " },
{ session: sessions[4], prefix: "└─ " },
{ session: sessions[5], prefix: "" },
{ session: sessions[6], prefix: "└─ " },
])
})
test("tracks usage across undo and redo boundaries", () => {
const messages = [assistant("msg_z", 10), assistant("msg_a", 30)]
-33
View File
@@ -29,7 +29,6 @@ export interface Interface {
pkg: string,
options?: { readonly subpaths?: readonly string[] },
) => Effect.Effect<EntryPoint, InstallFailedError | EffectFlock.LockError>
readonly resolve: (pkg: string, options?: { readonly subpaths?: readonly string[] }) => Effect.Effect<EntryPoint>
readonly which: (pkg: string, bin?: string) => Effect.Effect<string | undefined>
}
@@ -42,16 +41,6 @@ export function sanitize(pkg: string) {
return Array.from(pkg, (char) => (illegal.has(char) || char.charCodeAt(0) < 32 ? "_" : char)).join("")
}
export async function isRegistryPackage(pkg: string) {
const { default: npa } = await import("npm-package-arg")
try {
const result = npa(pkg)
return result.name !== undefined && ["version", "range", "tag"].includes(result.type)
} catch {
return false
}
}
const resolveEntryPoint = (name: string, dir: string, subpaths: readonly string[] = [""]): EntryPoint => {
const entrypoint = subpaths
.map((subpath) => {
@@ -145,23 +134,6 @@ const layer = Layer.effect(
return resolveEntryPoint(first.name, first.path, options?.subpaths)
}, Effect.scoped)
const resolve = Effect.fn("Npm.resolve")(function* (
pkg: string,
options?: { readonly subpaths?: readonly string[] },
) {
const { default: npa } = yield* Effect.promise(() => import("npm-package-arg"))
const name = (() => {
try {
return npa(pkg).name ?? pkg
} catch {
return pkg
}
})()
const dir = path.join(directory(pkg), "node_modules", name)
if (!(yield* afs.existsSafe(dir))) return { directory: dir }
return resolveEntryPoint(name, dir, options?.subpaths)
})
const which = Effect.fn("Npm.which")(function* (pkg: string, bin?: string) {
const dir = directory(pkg)
const binDir = path.join(dir, "node_modules", ".bin")
@@ -215,7 +187,6 @@ const layer = Layer.effect(
return Service.of({
add,
resolve,
which,
})
}),
@@ -233,10 +204,6 @@ export async function add(...args: Parameters<Interface["add"]>) {
return runPromise((svc) => svc.add(...args))
}
export async function resolve(...args: Parameters<Interface["resolve"]>) {
return runPromise((svc) => svc.resolve(...args))
}
export async function which(...args: Parameters<Interface["which"]>) {
return runPromise((svc) => svc.which(...args))
}
+1 -35
View File
@@ -99,32 +99,6 @@ an isolated cache. Package installation does not run lifecycle scripts.
Published packages should expose their plugin entrypoint and include every
runtime import in `dependencies`.
Install a package plugin globally with the CLI:
```sh
opencode2 plugin add opencode-acme-plugin@1.2.0
```
This installs and inspects the package before changing configuration. Packages
with a server entrypoint are added to global `opencode.json(c)`. Packages that
only expose `./tui` are added to global `cli.json` instead.
The command accepts npm registry package names with an optional version,
dist-tag, or semver range. Configure local paths directly instead; Git, tarball,
and npm alias targets are not accepted by `plugin add`.
List configured and active plugins, or remove a package from both global server
and TUI configuration:
```sh
opencode2 plugin list
opencode2 plugin list --builtin
opencode2 plugin remove opencode-acme-plugin@1.2.0
```
Built-in server plugins are hidden from the default list. Removing a plugin
keeps its package cache available for later reuse.
Local files and local package directories are imported directly. OpenCode does
**not** install their dependencies. Install dependencies in a `package.json`
visible from the plugin file, for example:
@@ -423,21 +397,13 @@ manifest is:
"name": "opencode-acme-plugin",
"version": "1.0.0",
"type": "module",
"exports": {
".": "./src/index.ts",
"./tui": "./src/tui.tsx"
},
"exports": "./src/index.ts",
"dependencies": {
"@opencode-ai/plugin": "beta"
}
}
```
Packages with a TUI entrypoint should set `tui: true` on their server plugin
definition. A locally connected TUI loads the package's `./tui` export from the
existing OpenCode package cache. A TUI connected to a remote server skips it
when that package is not installed locally.
Use versions compatible with the OpenCode release you target and test the
installed package, not only a workspace-linked copy. Because the plugin API is
beta, publish compatible plugin updates when V2 entrypoints or contracts