diff --git a/packages/core/src/public/opencode.ts b/packages/core/src/public/opencode.ts index 80c69bdb06..b39b0d52c4 100644 --- a/packages/core/src/public/opencode.ts +++ b/packages/core/src/public/opencode.ts @@ -2,6 +2,7 @@ export * as OpenCode from "./opencode" import { Context, Effect, Layer } from "effect" import { Catalog } from "../catalog" +import { AgentV2 } from "../agent" import { Database } from "../database/database" import { EventV2 } from "../event" import { LocationServiceMap } from "../location-layer" @@ -12,12 +13,13 @@ import * as SessionExecutionLocal from "../session/execution/local" import { SessionProjector } from "../session/projector" import { SessionStore } from "../session/store" import { ApplicationTools } from "../tool/application-tools" +import { TaskTool } from "../tool/task" import { Session } from "./session" import { Tool } from "./tool" export interface Interface { - readonly sessions: Session.Interface - readonly tools: Tool.Interface + readonly session: Session.Interface + readonly tool: Tool.Interface } /** Intentional public native API for Effect applications embedding OpenCode. */ @@ -77,7 +79,7 @@ const SessionsLayer = Layer.merge( Layer.orDie, ), SessionModelValidationLayer, -).pipe(Layer.provide(LocationServicesLayer)) +).pipe(Layer.provideMerge(LocationServicesLayer)) const ApplicationToolsLayer = ApplicationTools.layer // TODO: Accept explicit storage so tests and embeddings can select disposable or application-owned persistence. @@ -85,14 +87,24 @@ export const layer = Layer.effect( Service, Effect.gen(function* () { const sessions = yield* SessionV2.Service + const locations = yield* LocationServiceMap const tools = yield* ApplicationTools.Service const validation = yield* SessionModelValidation + yield* tools.register({ + task: yield* TaskTool.make(sessions, (location, id) => + AgentV2.Service.pipe( + Effect.flatMap((agents) => agents.get(id)), + Effect.provide(locations.get(location)), + ), + ), + }) return Service.of({ - tools: { register: tools.register }, - sessions: { + tool: { register: tools.register }, + session: { create: (input) => sessions.create({ id: input.id, + parentID: input.parentID, agent: input.agent, model: input.model, location: input.location, @@ -111,7 +123,9 @@ export const layer = Layer.effect( sessionID: input.sessionID, prompt: input.prompt, delivery: input.delivery, + resume: input.resume, }), + resume: sessions.resume, messages: (input) => sessions.messages({ sessionID: input.sessionID, diff --git a/packages/core/src/public/session.ts b/packages/core/src/public/session.ts index 6c61aff3b6..1c1f4d9380 100644 --- a/packages/core/src/public/session.ts +++ b/packages/core/src/public/session.ts @@ -9,6 +9,7 @@ import { SessionEvent } from "../session/event" import { SessionInput } from "../session/input" import { SessionMessage } from "../session/message" import { Prompt } from "../session/prompt" +import type { SessionRunner } from "../session/runner" import { Agent } from "./agent" import { Location } from "./location" import { Model } from "./model" @@ -65,6 +66,7 @@ export { MessageDecodeError } export interface CreateInput { readonly id?: ID + readonly parentID?: ID readonly agent?: Agent.ID readonly model?: Model.Ref readonly location: Location.Ref @@ -75,6 +77,8 @@ export interface PromptInput { readonly sessionID: ID readonly prompt: Prompt readonly delivery?: Delivery + /** Admit durably without scheduling execution. */ + readonly resume?: boolean } export interface SwitchModelInput { @@ -107,6 +111,8 @@ export interface Interface { readonly get: (sessionID: ID) => Effect.Effect readonly list: (input?: ListInput) => Effect.Effect readonly prompt: (input: PromptInput) => Effect.Effect + /** Explicitly drain one Session and wait for the current execution chain to settle. */ + readonly resume: (sessionID: ID) => Effect.Effect readonly switchModel: ( input: SwitchModelInput, ) => Effect.Effect diff --git a/packages/core/src/session.ts b/packages/core/src/session.ts index 8ea4593304..32d011dfb4 100644 --- a/packages/core/src/session.ts +++ b/packages/core/src/session.ts @@ -71,6 +71,7 @@ export type ListInput = typeof ListInput.Type type CreateInput = { id?: SessionSchema.ID + parentID?: SessionSchema.ID agent?: AgentV2.ID model?: ModelV2.Ref location: Location.Ref @@ -216,6 +217,7 @@ export const layer = Layer.effect( slug: Slug.create(), version: InstallationVersion, projectID: project.id, + parentID: input.parentID, directory: input.location.directory, path: path.relative(project.directory, input.location.directory).replaceAll("\\", "/"), workspaceID: input.location.workspaceID ? WorkspaceV2.ID.make(input.location.workspaceID) : undefined, diff --git a/packages/core/src/tool/task.ts b/packages/core/src/tool/task.ts new file mode 100644 index 0000000000..c5f43de810 --- /dev/null +++ b/packages/core/src/tool/task.ts @@ -0,0 +1,122 @@ +export * as TaskTool from "./task" + +import { ToolFailure } from "@opencode-ai/llm" +import { Cause, Effect, Schema, Scope } from "effect" +import { AgentV2 } from "../agent" +import { Location } from "../location" +import { SessionV2 } from "../session" +import { SessionMessage } from "../session/message" +import { Prompt } from "../session/prompt" +import { Tool } from "./tool" + +export const Input = Schema.Struct({ + description: Schema.String.annotate({ description: "A short description of the task" }), + prompt: Schema.String.annotate({ description: "The task for the agent to perform" }), + subagent_type: Schema.String.annotate({ description: "The specialized agent to use" }), + background: Schema.optional(Schema.Boolean).annotate({ + description: "Return immediately and notify the parent Session when the task finishes", + }), +}) + +export const Output = Schema.Struct({ + sessionID: SessionV2.ID, + status: Schema.Literals(["running", "completed"]), + output: Schema.String.pipe(Schema.optional), +}) + +type Sessions = Pick + +export const make = Effect.fn("TaskTool.make")(function* ( + sessions: Sessions, + resolveAgent: (location: Location.Ref, id: AgentV2.ID) => Effect.Effect, +) { + const scope = yield* Scope.Scope + + return Tool.make({ + description: + "Delegate focused work to a specialized child agent. Foreground calls wait for the result; background calls return immediately and notify this Session when complete.", + input: Input, + output: Output, + execute: (parameters, context) => + Effect.gen(function* () { + const parent = yield* sessions.get(context.sessionID) + const agent = yield* resolveAgent(parent.location, AgentV2.ID.make(parameters.subagent_type)) + if (!agent || (agent.mode !== "subagent" && agent.mode !== "all") || agent.hidden) + return yield* new ToolFailure({ message: `Unknown subagent: ${parameters.subagent_type}` }) + const child = yield* sessions.create({ + parentID: parent.id, + location: parent.location, + agent: agent.id, + model: agent.model ?? parent.model, + }) + + // TODO: Replace this fresh-child-only composition once Session execution exposes a bounded + // activity/result identity. An admission ID alone cannot correlate a response when one drain + // processes later queued work. + const run = Effect.gen(function* () { + yield* sessions.prompt({ + sessionID: child.id, + prompt: new Prompt({ text: parameters.prompt }), + delivery: "steer", + resume: false, + }) + yield* sessions.resume(child.id) + const messages = yield* sessions.messages({ sessionID: child.id, order: "desc", limit: 1 }) + const assistant = messages.find( + (message): message is SessionMessage.Assistant => message.type === "assistant" && !!message.time.completed, + ) + if (!assistant) return "" + return assistant.content + .filter((part): part is SessionMessage.AssistantText => part.type === "text") + .map((part) => part.text) + .join("\n") + }) + + if (parameters.background !== true) { + const output = yield* run.pipe( + Effect.onInterrupt(() => sessions.interrupt(child.id)), + Effect.mapError((error) => new ToolFailure({ message: `Task failed: ${String(error)}`, error })), + ) + return { sessionID: child.id, status: "completed" as const, output } + } + + yield* run.pipe( + Effect.matchCauseEffect({ + onSuccess: (output) => notify("completed", output), + onFailure: (cause) => + Cause.hasInterruptsOnly(cause) ? Effect.void : notify("error", String(Cause.squash(cause))), + }), + Effect.tapCause((cause) => Effect.logError("Background task notification failed", Cause.squash(cause))), + Effect.ignore, + Effect.forkIn(scope, { startImmediately: true }), + ) + return { sessionID: child.id, status: "running" as const } + + function notify(state: "completed" | "error", text: string) { + const tag = state === "completed" ? "task_result" : "task_error" + return sessions.prompt({ + sessionID: parent.id, + prompt: new Prompt({ + text: `\nBackground task ${state}: ${parameters.description}\n<${tag}>\n${text}\n\n`, + }), + delivery: "steer", + }) + } + }).pipe( + Effect.mapError((error) => + error instanceof ToolFailure + ? error + : new ToolFailure({ message: `Unable to run task: ${String(error)}`, error }), + ), + ), + toModelOutput: ({ output }) => [ + { + type: "text", + text: + output.status === "running" + ? `\nThe task is working in the background. You will be notified automatically when it finishes.\n` + : `\n\n${output.output ?? ""}\n\n`, + }, + ], + }) +}) diff --git a/packages/core/test/public-opencode.test.ts b/packages/core/test/public-opencode.test.ts index c5f90e92c4..acd679307d 100644 --- a/packages/core/test/public-opencode.test.ts +++ b/packages/core/test/public-opencode.test.ts @@ -13,9 +13,9 @@ describe("public native OpenCode API", () => { Effect.gen(function* () { const opencode = yield* OpenCode.Service - expect(Object.keys(opencode).sort()).toEqual(["sessions", "tools"]) + expect(Object.keys(opencode).sort()).toEqual(["session", "tool"]) - expect(Object.keys(opencode.sessions).sort()).toEqual([ + expect(Object.keys(opencode.session).sort()).toEqual([ "context", "create", "events", @@ -25,12 +25,13 @@ describe("public native OpenCode API", () => { "message", "messages", "prompt", + "resume", "switchModel", ]) expect(Session.ID.create()).toStartWith("ses_") expect(Session.MessageID.create()).toStartWith("msg_") - expect(yield* opencode.sessions.list()).toBeArray() - yield* opencode.tools.register({ + expect(yield* opencode.session.list()).toBeArray() + yield* opencode.tool.register({ public_tool: Tool.make({ description: "Public tool", input: Schema.Struct({}), @@ -52,14 +53,14 @@ describe("public native OpenCode API", () => { const opencode = yield* OpenCode.Service const sessionID = Session.ID.make("ses_public_switch_available") const model = ref({ variant: "fast" }) - yield* opencode.sessions.create({ + yield* opencode.session.create({ id: sessionID, location: Location.Ref.make({ directory: AbsolutePath.make(tmp.path) }), }) - yield* opencode.sessions.switchModel({ sessionID, model }) + yield* opencode.session.switchModel({ sessionID, model }) - expect((yield* opencode.sessions.get(sessionID)).model).toEqual(model) + expect((yield* opencode.session.get(sessionID)).model).toEqual(model) }), ), ), @@ -77,27 +78,27 @@ describe("public native OpenCode API", () => { const opencode = yield* OpenCode.Service const availableID = Session.ID.make("ses_public_switch_exact_available") const disabledID = Session.ID.make("ses_public_switch_exact_disabled") - yield* opencode.sessions.create({ + yield* opencode.session.create({ id: availableID, location: Location.Ref.make({ directory: AbsolutePath.make(available.path) }), }) - yield* opencode.sessions.create({ + yield* opencode.session.create({ id: disabledID, location: Location.Ref.make({ directory: AbsolutePath.make(disabled.path) }), }) - yield* opencode.sessions.switchModel({ sessionID: availableID, model: ref({ variant: "default" }) }) - const disabledError = yield* opencode.sessions + yield* opencode.session.switchModel({ sessionID: availableID, model: ref({ variant: "default" }) }) + const disabledError = yield* opencode.session .switchModel({ sessionID: disabledID, model: ref() }) .pipe(Effect.flip) - const missingError = yield* opencode.sessions + const missingError = yield* opencode.session .switchModel({ sessionID: disabledID, model: ref({ id: "missing" }) }) .pipe(Effect.flip) expect(disabledError).toBeInstanceOf(Session.ModelUnavailableError) expect(missingError).toBeInstanceOf(Session.ModelUnavailableError) - expect((yield* opencode.sessions.get(availableID)).model).toEqual(ref({ variant: "default" })) - expect((yield* opencode.sessions.get(disabledID)).model).toBeUndefined() + expect((yield* opencode.session.get(availableID)).model).toEqual(ref({ variant: "default" })) + expect((yield* opencode.session.get(disabledID)).model).toBeUndefined() }), ), ), @@ -114,18 +115,18 @@ describe("public native OpenCode API", () => { const opencode = yield* OpenCode.Service const sessionID = Session.ID.make("ses_public_switch_variant") const selected = ref({ variant: "fast" }) - yield* opencode.sessions.create({ + yield* opencode.session.create({ id: sessionID, location: Location.Ref.make({ directory: AbsolutePath.make(tmp.path) }), }) - yield* opencode.sessions.switchModel({ sessionID, model: selected }) + yield* opencode.session.switchModel({ sessionID, model: selected }) - const error = yield* opencode.sessions + const error = yield* opencode.session .switchModel({ sessionID, model: ref({ variant: "unknown" }) }) .pipe(Effect.flip) expect(error).toBeInstanceOf(Session.VariantUnavailableError) - expect((yield* opencode.sessions.get(sessionID)).model).toEqual(selected) + expect((yield* opencode.session.get(sessionID)).model).toEqual(selected) }), ), ), @@ -135,7 +136,7 @@ describe("public native OpenCode API", () => { Effect.gen(function* () { const opencode = yield* OpenCode.Service const sessionID = Session.ID.make("ses_public_switch_missing") - const error = yield* opencode.sessions + const error = yield* opencode.session .switchModel({ sessionID, model: Schema.decodeUnknownSync(Model.Ref)({ id: "claude-sonnet-4-5", providerID: "anthropic" }), diff --git a/packages/core/test/session-create.test.ts b/packages/core/test/session-create.test.ts index 3551ec52f3..0beb568266 100644 --- a/packages/core/test/session-create.test.ts +++ b/packages/core/test/session-create.test.ts @@ -94,6 +94,7 @@ describe("SessionV2.create", () => { it.effect("stores supplied immutable create attributes", () => Effect.gen(function* () { const session = yield* SessionV2.Service + const parentID = SessionV2.ID.make("ses_parent") const workspaceID = WorkspaceV2.ID.make("wrk_test") const model = ModelV2.Ref.make({ id: ModelV2.ID.make("sonnet"), @@ -104,10 +105,11 @@ describe("SessionV2.create", () => { expect( yield* session.create({ location: Location.Ref.make({ directory: location.directory, workspaceID }), + parentID, agent: AgentV2.ID.make("build"), model, }), - ).toMatchObject({ location: { directory: location.directory, workspaceID }, agent: "build", model }) + ).toMatchObject({ parentID, location: { directory: location.directory, workspaceID }, agent: "build", model }) }), ) diff --git a/packages/core/test/tool-task.test.ts b/packages/core/test/tool-task.test.ts new file mode 100644 index 0000000000..be0c58a874 --- /dev/null +++ b/packages/core/test/tool-task.test.ts @@ -0,0 +1,206 @@ +import { describe, expect } from "bun:test" +import { AbsolutePath } from "@opencode-ai/core/schema" +import { AgentV2 } from "@opencode-ai/core/agent" +import { Location } from "@opencode-ai/core/location" +import { ModelV2 } from "@opencode-ai/core/model" +import { ProjectV2 } from "@opencode-ai/core/project" +import { ProviderV2 } from "@opencode-ai/core/provider" +import { SessionV2 } from "@opencode-ai/core/session" +import { SessionInput } from "@opencode-ai/core/session/input" +import { SessionMessage } from "@opencode-ai/core/session/message" +import { TaskTool } from "@opencode-ai/core/tool/task" +import { Tool } from "@opencode-ai/core/tool/tool" +import { DateTime, Deferred, Effect, Fiber, Layer } from "effect" +import { toolIdentity } from "./lib/tool" +import { testEffect } from "./lib/effect" + +const parentID = SessionV2.ID.make("ses_task_parent") +const childID = SessionV2.ID.make("ses_task_child") +const location = Location.Ref.make({ directory: AbsolutePath.make("/project") }) +const parent = new SessionV2.Info({ + id: parentID, + projectID: ProjectV2.ID.make("project"), + agent: AgentV2.ID.make("build"), + model: { id: ModelV2.ID.make("model"), providerID: ProviderV2.ID.make("provider") }, + cost: 0, + tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } }, + time: { created: DateTime.makeUnsafe(0), updated: DateTime.makeUnsafe(0) }, + title: "Parent", + location, +}) +const child = new SessionV2.Info({ + id: childID, + parentID, + projectID: parent.projectID, + agent: parent.agent, + model: parent.model, + cost: 0, + tokens: parent.tokens, + time: parent.time, + title: "Child", + location, +}) +const assistant = new SessionMessage.Assistant({ + id: SessionMessage.ID.make("msg_task_assistant"), + type: "assistant", + agent: "explore", + model: parent.model!, + content: [new SessionMessage.AssistantText({ type: "text", id: "text", text: "Task output" })], + time: { created: DateTime.makeUnsafe(1), completed: DateTime.makeUnsafe(2) }, +}) +const input = { + description: "Map auth", + prompt: "Map the authentication flow", + subagent_type: "explore", +} + +describe("TaskTool", () => { + const it = testEffect(Layer.empty) + const resolveAgent = (): Effect.Effect => + Effect.succeed(AgentV2.Info.empty(AgentV2.ID.make("explore"))) + + it.effect("runs a foreground child with an admit-only steer and explicit resume", () => + Effect.gen(function* () { + const inputs: Parameters[0][] = [] + let resumed = 0 + const sessions = mockSessions({ + prompt: (value) => { + inputs.push(value) + return Effect.succeed(admission(value)) + }, + resume: () => Effect.sync(() => resumed++), + }) + const tool = yield* TaskTool.make(sessions, resolveAgent) + + const result = yield* execute(tool, { ...input, background: false }, "call_task") + + expect(result.structured).toEqual({ sessionID: childID, status: "completed", output: "Task output" }) + expect(inputs).toHaveLength(1) + expect(inputs[0]).toMatchObject({ sessionID: childID, delivery: "steer", resume: false }) + expect(resumed).toBe(1) + }), + ) + + it.effect("rejects an unknown subagent before creating a child", () => + Effect.gen(function* () { + let created = false + const sessions = mockSessions({ + create: () => + Effect.sync(() => { + created = true + return child + }), + }) + const tool = yield* TaskTool.make(sessions, () => Effect.succeed(undefined)) + + const error = yield* execute(tool, { ...input, subagent_type: "missing" }, "call_task_unknown").pipe(Effect.flip) + + expect(error.message).toBe("Unknown subagent: missing") + expect(created).toBe(false) + }), + ) + + it.live("does not notify the parent when background work is interrupted", () => + Effect.gen(function* () { + let notified = false + const sessions = mockSessions({ + prompt: (value) => { + if (value.sessionID === parentID) notified = true + return Effect.succeed(admission(value)) + }, + resume: () => Effect.interrupt, + }) + const tool = yield* TaskTool.make(sessions, resolveAgent) + + const result = yield* execute(tool, { ...input, background: true }, "call_task_background_interrupt") + + expect(result.structured).toEqual({ sessionID: childID, status: "running" }) + yield* Effect.yieldNow + expect(notified).toBe(false) + }), + ) + + it.live("interrupts the child when foreground waiting is interrupted", () => + Effect.gen(function* () { + const resumed = yield* Deferred.make() + const interrupts: SessionV2.ID[] = [] + const sessions = mockSessions({ + resume: () => Deferred.succeed(resumed, undefined).pipe(Effect.andThen(Effect.never)), + interrupt: (sessionID) => Effect.sync(() => interrupts.push(sessionID)), + }) + const tool = yield* TaskTool.make(sessions, resolveAgent) + const fiber = yield* execute(tool, input, "call_task_interrupt").pipe(Effect.forkChild) + + yield* Deferred.await(resumed) + yield* Fiber.interrupt(fiber) + expect(interrupts).toEqual([childID]) + }), + ) + + it.live("returns before background completion and steers the result into the parent", () => + Effect.gen(function* () { + const gate = yield* Deferred.make() + const notified = yield* Deferred.make[0]>() + const inputs: Parameters[0][] = [] + const sessions = mockSessions({ + prompt: (value) => { + inputs.push(value) + return value.sessionID === parentID + ? Deferred.succeed(notified, value).pipe(Effect.as(admission(value))) + : Effect.succeed(admission(value)) + }, + resume: () => Deferred.await(gate), + }) + const tool = yield* TaskTool.make(sessions, resolveAgent) + + const result = yield* execute(tool, { ...input, background: true }, "call_task_background") + + expect(result.structured).toEqual({ sessionID: childID, status: "running" }) + expect(inputs).toHaveLength(1) + yield* Deferred.succeed(gate, undefined) + const notification = yield* Deferred.await(notified) + expect(notification).toMatchObject({ sessionID: parentID, delivery: "steer" }) + expect(notification.prompt.text).toContain("Background task completed: Map auth") + expect(notification.prompt.text).toContain("Task output") + }), + ) +}) + +function mockSessions(overrides: { + create?: SessionV2.Interface["create"] + interrupt?: SessionV2.Interface["interrupt"] + prompt?: SessionV2.Interface["prompt"] + resume?: SessionV2.Interface["resume"] +}): Pick { + return { + create: overrides.create ?? (() => Effect.succeed(child)), + get: (id) => Effect.succeed(id === parentID ? parent : child), + prompt: overrides.prompt ?? ((value) => Effect.succeed(admission(value))), + resume: overrides.resume ?? (() => Effect.void), + messages: () => Effect.succeed([assistant]), + interrupt: overrides.interrupt ?? (() => Effect.void), + } +} + +function admission(input: Parameters[0]) { + return new SessionInput.Admitted({ + admittedSeq: 1, + id: input.id ?? SessionMessage.ID.create(), + sessionID: input.sessionID, + prompt: input.prompt, + delivery: input.delivery ?? "steer", + timeCreated: DateTime.makeUnsafe(0), + }) +} + +function execute(tool: Tool.AnyTool, input: unknown, toolCallID: string) { + return Tool.settle( + tool, + { type: "tool-call", id: toolCallID, name: "task", input }, + { + sessionID: parentID, + ...toolIdentity, + toolCallID, + }, + ) +}