diff --git a/packages/core/src/session.ts b/packages/core/src/session.ts index 0a53c8dd10..26f00c29b8 100644 --- a/packages/core/src/session.ts +++ b/packages/core/src/session.ts @@ -76,12 +76,14 @@ const ListAllInput = Schema.Struct(ListInputBase) export const ListInput = Schema.Union([ListDirectoryInput, ListProjectInput, ListAllInput]) export type ListInput = typeof ListInput.Type -type CreateInput = { +type CreateBaseInput = { id?: SessionSchema.ID + title?: string agent?: AgentV2.ID model?: ModelV2.Ref - location: Location.Ref } +type CreateInput = CreateBaseInput & + ({ location: Location.Ref; parentID?: never } | { parentID: SessionSchema.ID; location?: never }) type CompactInput = { sessionID: SessionSchema.ID @@ -95,7 +97,7 @@ export class NotFoundError extends Schema.TaggedErrorClass()("Ses export class OperationUnavailableError extends Schema.TaggedErrorClass()( "Session.OperationUnavailableError", { - operation: Schema.Literals(["move", "shell", "skill", "switchAgent", "compact", "wait"]), + operation: Schema.Literals(["move", "shell", "skill", "switchAgent", "compact"]), }, ) {} @@ -115,7 +117,7 @@ export type Error = NotFoundError | MessageDecodeError | OperationUnavailableErr export interface Interface { readonly list: (input?: ListInput) => Effect.Effect - readonly create: (input: CreateInput) => Effect.Effect + readonly create: (input: CreateInput) => Effect.Effect readonly get: (sessionID: SessionSchema.ID) => Effect.Effect readonly messages: (input: { sessionID: SessionSchema.ID @@ -168,7 +170,7 @@ export interface Interface { resume?: boolean }) => Effect.Effect readonly compact: (input: CompactInput) => Effect.Effect - readonly wait: (id: SessionSchema.ID) => Effect.Effect + readonly wait: (id: SessionSchema.ID) => Effect.Effect readonly active: Effect.Effect> readonly resume: (sessionID: SessionSchema.ID) => Effect.Effect readonly interrupt: (sessionID: SessionSchema.ID) => Effect.Effect @@ -213,7 +215,12 @@ export const layer = Layer.effect( const sessionID = input.id ?? SessionSchema.ID.create() const recorded = yield* store.get(sessionID) if (recorded) return recorded - const project = yield* projects.resolve(input.location.directory) + const parent = input.parentID ? yield* store.get(input.parentID) : undefined + if (input.parentID && parent === undefined) return yield* new NotFoundError({ sessionID: input.parentID }) + const location = parent?.location ?? input.location + if (location === undefined) + return yield* Effect.die(new Error("V2Session.create requires either location or an existing parentID")) + const project = yield* projects.resolve(location.directory) yield* db .insert(ProjectTable) .values({ id: project.id, worktree: project.directory, vcs: project.vcs?.type, sandboxes: [] }) @@ -226,10 +233,11 @@ export const layer = Layer.effect( slug: Slug.create(), version: InstallationVersion, projectID: project.id, - directory: input.location.directory, - path: path.relative(project.directory, input.location.directory).replaceAll("\\", "/"), - workspaceID: input.location.workspaceID ? WorkspaceV2.ID.make(input.location.workspaceID) : undefined, - title: `New session - ${new Date(now).toISOString()}`, + parentID: input.parentID, + directory: location.directory, + path: path.relative(project.directory, location.directory).replaceAll("\\", "/"), + workspaceID: location.workspaceID ? WorkspaceV2.ID.make(location.workspaceID) : undefined, + title: input.title ?? `New session - ${new Date(now).toISOString()}`, agent: input.agent, model: input.model ? { @@ -242,24 +250,22 @@ export const layer = Layer.effect( tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } }, time: { created: now, updated: now }, }) - const projected = yield* events - .publish(SessionV1.Event.Created, { sessionID, info }, { location: input.location }) - .pipe( - Effect.as({ type: "created" } as const), - Effect.catchDefect((defect) => { - if (!(defect instanceof SessionProjector.SessionAlreadyProjected)) { - return Effect.die(defect) - } - // Concurrent creation lost the projection race. The existing Session identity wins. - return store - .get(sessionID) - .pipe( - Effect.flatMap((session) => - session ? Effect.succeed({ type: "existing", session } as const) : Effect.die(defect), - ), - ) - }), - ) + const projected = yield* events.publish(SessionV1.Event.Created, { sessionID, info }, { location }).pipe( + Effect.as({ type: "created" } as const), + Effect.catchDefect((defect) => { + if (!(defect instanceof SessionProjector.SessionAlreadyProjected)) { + return Effect.die(defect) + } + // Concurrent creation lost the projection race. The existing Session identity wins. + return store + .get(sessionID) + .pipe( + Effect.flatMap((session) => + session ? Effect.succeed({ type: "existing", session } as const) : Effect.die(defect), + ), + ) + }), + ) if (projected.type === "existing") return projected.session // TODO: Restore recorded sessions onto replacement synchronized workspaces in a future API slice. return yield* result.get(sessionID).pipe(Effect.orDie) @@ -432,7 +438,7 @@ export const layer = Layer.effect( }), wait: Effect.fn("V2Session.wait")(function* (sessionID) { yield* result.get(sessionID) - return yield* new OperationUnavailableError({ operation: "wait" }) + yield* execution.awaitIdle(sessionID) }), active: execution.active, resume: Effect.fn("V2Session.resume")(function* (sessionID) { @@ -456,7 +462,7 @@ export const layer = Layer.effect( clear: Effect.fn("V2Session.revert.clear")(function* (sessionID) { const session = yield* result.get(sessionID) if ((yield* execution.active).has(sessionID)) return yield* new BusyError({ sessionID }) - yield* SessionRevert.clear(session).pipe( + return yield* SessionRevert.clear(session).pipe( Effect.provideService(EventV2.Service, events), Effect.provide(locations.get(session.location)), ) @@ -464,7 +470,7 @@ export const layer = Layer.effect( commit: Effect.fn("V2Session.revert.commit")(function* (sessionID) { const session = yield* result.get(sessionID) if ((yield* execution.active).has(sessionID)) return yield* new BusyError({ sessionID }) - yield* SessionRevert.commit(session).pipe(Effect.provideService(EventV2.Service, events)) + return yield* SessionRevert.commit(session).pipe(Effect.provideService(EventV2.Service, events)) }), }, }) diff --git a/packages/core/src/session/execution.ts b/packages/core/src/session/execution.ts index 5938c37726..5dbba30776 100644 --- a/packages/core/src/session/execution.ts +++ b/packages/core/src/session/execution.ts @@ -15,6 +15,8 @@ export interface Interface { readonly wake: (sessionID: SessionSchema.ID) => Effect.Effect /** Interrupt active work owned by this process. Idle interruption is a no-op. */ readonly interrupt: (sessionID: SessionSchema.ID) => Effect.Effect + /** Resolves once this process owns no active execution for the Session. Returns immediately when idle and never starts work. */ + readonly awaitIdle: (sessionID: SessionSchema.ID) => Effect.Effect } /** Routes execution from a Session ID to the runner owned by that Session's Location. */ @@ -30,5 +32,6 @@ export const noopLayer = Layer.succeed( resume: () => Effect.void, wake: () => Effect.void, interrupt: () => Effect.void, + awaitIdle: () => Effect.void, }), ) diff --git a/packages/core/src/session/execution/local.ts b/packages/core/src/session/execution/local.ts index 281c5b7044..e4870e89d8 100644 --- a/packages/core/src/session/execution/local.ts +++ b/packages/core/src/session/execution/local.ts @@ -33,6 +33,7 @@ export const layer = Layer.effect( interrupt: coordinator.interrupt, resume: coordinator.run, wake: coordinator.wake, + awaitIdle: coordinator.awaitIdle, }) }), ) diff --git a/packages/core/src/session/run-coordinator.ts b/packages/core/src/session/run-coordinator.ts index 2f89aff9e3..8cba78b3f8 100644 --- a/packages/core/src/session/run-coordinator.ts +++ b/packages/core/src/session/run-coordinator.ts @@ -12,6 +12,8 @@ export interface Coordinator { readonly wake: (key: Key) => Effect.Effect /** Stops active execution and waits for its cleanup. */ readonly interrupt: (key: Key) => Effect.Effect + /** Resolves once no execution is active for the key. Returns immediately when already idle and never starts work. */ + readonly awaitIdle: (key: Key) => Effect.Effect } type Entry = { @@ -100,5 +102,15 @@ export const make = (options: { return Fiber.interrupt(entry.owner) }) - return { active: Effect.sync(() => new Set(active.keys())), run, wake, interrupt } + // Each successful drain reuses its entry.done across coalesced wakes, so one await + // already spans steered and queued continuation. Re-check after it settles to cover a + // fresh wake (or a failure/stopping successor) that installs a new entry. + const awaitIdle = (key: Key): Effect.Effect => + Effect.suspend(() => { + const entry = active.get(key) + if (entry === undefined) return Effect.void + return Deferred.await(entry.done).pipe(Effect.exit, Effect.andThen(awaitIdle(key))) + }) + + return { active: Effect.sync(() => new Set(active.keys())), run, wake, interrupt, awaitIdle } }) diff --git a/packages/core/src/tool/registry.ts b/packages/core/src/tool/registry.ts index 868ba44672..8110e29da7 100644 --- a/packages/core/src/tool/registry.ts +++ b/packages/core/src/tool/registry.ts @@ -2,7 +2,7 @@ export * as ToolRegistry from "./registry" import { ToolOutput, type ToolCall, type ToolDefinition, type ToolResultValue } from "@opencode-ai/llm" import { Context, Effect, Layer, Scope } from "effect" -import { AgentV2 } from "../agent" +import type { AgentV2 } from "../agent" import { PermissionV2 } from "../permission" import { SessionMessage } from "../session/message" import { SessionSchema } from "../session/schema" diff --git a/packages/core/src/tool/subagent.ts b/packages/core/src/tool/subagent.ts new file mode 100644 index 0000000000..23b502a90c --- /dev/null +++ b/packages/core/src/tool/subagent.ts @@ -0,0 +1,192 @@ +export * as SubagentTool from "./subagent" + +import { ToolFailure } from "@opencode-ai/llm" +import { DateTime, Effect, Layer, Schema, Scope } from "effect" +import { AgentV2 } from "../agent" +import { BackgroundJob } from "../background-job" +import { EventV2 } from "../event" +import { LocationServiceMap } from "../location-service-map" +import { SessionV2 } from "../session" +import { SessionEvent } from "../session/event" +import { SessionMessage } from "../session/message" +import { SessionSchema } from "../session/schema" +import { makeGlobalNode } from "../effect/app-node" +import { ApplicationTools } from "./application-tools" +import { Tool } from "./tool" + +export const name = "subagent" + +const NO_TEXT = "Subagent completed without a text response." +const BACKGROUND_STARTED = + "The subagent is working in the background. You will be notified automatically when it finishes. DO NOT sleep, poll, or proactively check on its progress." + +export const Input = Schema.Struct({ + agent: Schema.String.annotate({ description: "The configured agent to run as the subagent" }), + description: Schema.String.annotate({ description: "A short description of the subagent's task" }), + prompt: Schema.String.annotate({ description: "The task for the subagent to perform" }), + background: Schema.Boolean.pipe(Schema.optional).annotate({ + description: + "Run the subagent in the background and return immediately. You will be notified when it completes. DO NOT poll its progress.", + }), +}) + +export const Output = Schema.Struct({ + sessionID: SessionSchema.ID, + status: Schema.Literals(["completed", "running"]), + output: Schema.String, +}) + +export const description = [ + "Spawn a subagent: a child session running a configured agent with fresh context.", + "Foreground (default) runs the subagent to completion and returns its final response.", + "Background mode (background=true) launches it asynchronously and returns immediately; you are notified when it finishes.", + "Use background only for independent work that can run while you continue elsewhere.", +].join("\n") + +export const layer = Layer.effectDiscard( + Effect.gen(function* () { + const tools = yield* ApplicationTools.Service + const sessions = yield* SessionV2.Service + const jobs = yield* BackgroundJob.Service + const events = yield* EventV2.Service + const locations = yield* LocationServiceMap.Service + const scope = yield* Scope.Scope + + // Concatenate the child's final completed assistant text. Distinguishes "completed with no + // text" (generic string) from "failed" (the run effect fails, surfaced as a job error). + const latestAssistantText = Effect.fn("SubagentTool.latestAssistantText")(function* (sessionID: SessionSchema.ID) { + const messages = yield* sessions.messages({ sessionID, order: "desc", limit: 20 }) + const assistant = messages.find( + (message) => + message.type === "assistant" && message.time.completed !== undefined && message.error === undefined, + ) + if (assistant === undefined || assistant.type !== "assistant") return NO_TEXT + const text = assistant.content + .filter((part): part is Extract => part.type === "text") + .map((part) => part.text) + .join("") + return text.length > 0 ? text : NO_TEXT + }) + + const injectCompletion = Effect.fn("SubagentTool.injectCompletion")(function* ( + parentID: SessionSchema.ID, + childID: SessionSchema.ID, + description: string, + state: "completed" | "error" | "cancelled", + text: string, + ) { + yield* events.publish(SessionEvent.Synthetic, { + sessionID: parentID, + messageID: SessionMessage.ID.create(), + timestamp: yield* DateTime.now, + text: `\n${text}\n`, + }) + }) + + const injectWhenDone = Effect.fn("SubagentTool.injectWhenDone")(function* ( + parentID: SessionSchema.ID, + childID: SessionSchema.ID, + description: string, + ) { + yield* jobs.wait({ id: childID }).pipe( + Effect.flatMap((result) => { + if (result.info?.status === "completed") + return injectCompletion(parentID, childID, description, "completed", result.info.output ?? NO_TEXT) + if (result.info?.status === "error") + return injectCompletion(parentID, childID, description, "error", result.info.error ?? "Subagent failed") + if (result.info?.status === "cancelled") + return injectCompletion(parentID, childID, description, "cancelled", "Subagent cancelled") + return Effect.void + }), + Effect.forkIn(scope, { startImmediately: true }), + ) + }) + + yield* tools + .register({ + [name]: Tool.make({ + description, + input: Input, + output: Output, + toModelOutput: ({ output }) => [{ type: "text", text: output.output }], + execute: (input, context) => + Effect.gen(function* () { + const parent = yield* sessions + .get(context.sessionID) + .pipe( + Effect.mapError(() => new ToolFailure({ message: `Parent session not found: ${context.sessionID}` })), + ) + const agents = yield* AgentV2.Service.pipe(Effect.provide(locations.get(parent.location))) + const agent = yield* agents.resolve(input.agent) + if (agent === undefined) return yield* new ToolFailure({ message: `Unknown agent: ${input.agent}` }) + if (agent.mode === "primary") + return yield* new ToolFailure({ message: `Agent ${input.agent} cannot run as a subagent` }) + + // Model selection is policy/config/session state, not an LLM-facing tool argument. + const model = agent.model ?? parent.model + const child = yield* sessions + .create({ + parentID: context.sessionID, + title: input.description, + agent: AgentV2.ID.make(input.agent), + model, + // TODO(opencode kkdvxn): derive restricted subagent permissions from the parent + // session (V1 deriveSubagentSessionPermission). MVP uses the agent's own permissions. + }) + .pipe( + Effect.mapError(() => new ToolFailure({ message: `Parent session not found: ${context.sessionID}` })), + ) + + const background = input.background === true + + const run = Effect.gen(function* () { + // The child session owns its agent/model (set at create); prompt only admits input. + yield* sessions.prompt({ sessionID: child.id, prompt: { text: input.prompt }, resume: false }) + yield* sessions.resume(child.id) + return yield* latestAssistantText(child.id) + }) + + const info = yield* jobs.start({ + id: child.id, + type: name, + title: input.description, + metadata: {}, + onPromote: injectWhenDone(context.sessionID, child.id, input.description), + run, + }) + + if (background) { + if ((yield* jobs.promote(info.id)) === undefined) + yield* injectWhenDone(context.sessionID, child.id, input.description) + return { sessionID: child.id, status: "running" as const, output: BACKGROUND_STARTED } + } + + const result = yield* Effect.raceFirst( + jobs.wait({ id: child.id }).pipe(Effect.map((waited) => waited.info)), + jobs.waitForPromotion(child.id), + ).pipe( + Effect.onInterrupt(() => + Effect.all([sessions.interrupt(child.id), jobs.cancel(child.id)], { discard: true }), + ), + ) + if (result?.metadata?.background === true) + return { sessionID: child.id, status: "running" as const, output: BACKGROUND_STARTED } + if (result?.status === "error") + return yield* new ToolFailure({ message: result.error ?? "Subagent failed" }) + if (result?.status === "cancelled") return yield* new ToolFailure({ message: "Subagent cancelled" }) + return { sessionID: child.id, status: "completed" as const, output: result?.output ?? NO_TEXT } + }), + }), + }) + .pipe(Effect.orDie) + }), +) + +// Registered at the app root via ApplicationTools, not as a Location node: SessionV2 sits above +// LocationServiceMap, so a location-scoped subagent node would create a static dependency cycle. +// Agent lookup is resolved through the parent Session's location when the tool executes. +export const node = makeGlobalNode({ + name: "subagent-tool", + layer, + deps: [ApplicationTools.node, SessionV2.node, BackgroundJob.node, EventV2.node, LocationServiceMap.node], +}) diff --git a/packages/core/test/session-create.test.ts b/packages/core/test/session-create.test.ts index 81efeb7623..f573d968d6 100644 --- a/packages/core/test/session-create.test.ts +++ b/packages/core/test/session-create.test.ts @@ -56,6 +56,14 @@ const it = testEffect( const location = Location.Ref.make({ directory: AbsolutePath.make("/project") }) const id = SessionV2.ID.create() +const assertCreateInputTypes = (session: SessionV2.Interface) => { + // @ts-expect-error location or parentID is required. + session.create({}) + // @ts-expect-error child sessions inherit their parent's location. + session.create({ parentID: SessionV2.ID.create(), location }) +} +void assertCreateInputTypes + describe("SessionV2.create", () => { it.effect("creates a fresh projected session when the ID is omitted", () => Effect.gen(function* () { @@ -102,6 +110,27 @@ describe("SessionV2.create", () => { }), ) + it.effect("inherits location from an existing parent when omitted", () => + Effect.gen(function* () { + const session = yield* SessionV2.Service + const parent = yield* session.create({ location }) + const child = yield* session.create({ parentID: parent.id, title: "child" }) + + expect(child).toMatchObject({ parentID: parent.id, location }) + }), + ) + + it.effect("rejects child creation when the parent does not exist", () => + Effect.gen(function* () { + const session = yield* SessionV2.Service + const missing = SessionV2.ID.create() + + expect(yield* Effect.flip(session.create({ parentID: missing, title: "child" }))).toEqual( + new SessionV2.NotFoundError({ sessionID: missing }), + ) + }), + ) + it.effect("returns the existing Session when one ID is reused with different create arguments", () => Effect.gen(function* () { const session = yield* SessionV2.Service diff --git a/packages/core/test/session-prompt.test.ts b/packages/core/test/session-prompt.test.ts index 66c8f76914..2bd43226df 100644 --- a/packages/core/test/session-prompt.test.ts +++ b/packages/core/test/session-prompt.test.ts @@ -39,6 +39,7 @@ const execution = Layer.succeed( Effect.sync(() => { wakeCalls.push(sessionID) }), + awaitIdle: () => Effect.void, }), ) const sessions = SessionV2.layer.pipe( diff --git a/packages/core/test/session-runner-recorded.test.ts b/packages/core/test/session-runner-recorded.test.ts index a0610dd465..cf3f3801d6 100644 --- a/packages/core/test/session-runner-recorded.test.ts +++ b/packages/core/test/session-runner-recorded.test.ts @@ -99,6 +99,7 @@ const execution = Layer.effect( resume: coordinator.run, wake: coordinator.wake, interrupt: coordinator.interrupt, + awaitIdle: coordinator.awaitIdle, }) }), ).pipe(Layer.provide(runner)) diff --git a/packages/core/test/session-runner.test.ts b/packages/core/test/session-runner.test.ts index 3bfd7b04e4..fb6f5cdc4f 100644 --- a/packages/core/test/session-runner.test.ts +++ b/packages/core/test/session-runner.test.ts @@ -258,6 +258,7 @@ const execution = Layer.effect( resume: coordinator.run, wake: coordinator.wake, interrupt: coordinator.interrupt, + awaitIdle: coordinator.awaitIdle, }) }), ).pipe(Layer.provide(runner)) @@ -380,7 +381,10 @@ const recordedEventTypes = (id: SessionV2.ID) => .where(eq(EventTable.aggregate_id, id)) .orderBy(asc(EventTable.seq)) .all() - .pipe(Effect.orDie, Effect.map((rows) => rows.map((row) => row.type))) + .pipe( + Effect.orDie, + Effect.map((rows) => rows.map((row) => row.type)), + ) }) const replaySessionProjection = (id: SessionV2.ID) => diff --git a/packages/core/test/session-wait.test.ts b/packages/core/test/session-wait.test.ts new file mode 100644 index 0000000000..cd8399a534 --- /dev/null +++ b/packages/core/test/session-wait.test.ts @@ -0,0 +1,54 @@ +import { describe, expect } from "bun:test" +import { Effect, Layer } from "effect" +import { Database } from "@opencode-ai/core/database/database" +import { EventV2 } from "@opencode-ai/core/event" +import { Location } from "@opencode-ai/core/location" +import { ProjectV2 } from "@opencode-ai/core/project" +import { AbsolutePath } from "@opencode-ai/core/schema" +import { SessionV2 } from "@opencode-ai/core/session" +import { SessionProjector } from "@opencode-ai/core/session/projector" +import { SessionExecution } from "@opencode-ai/core/session/execution" +import { SessionStore } from "@opencode-ai/core/session/store" +import { locationServiceMapLayer } from "@opencode-ai/core/location-services" +import { testEffect } from "./lib/effect" + +const location = Location.Ref.make({ directory: AbsolutePath.make("/project") }) +const awaited: SessionV2.ID[] = [] +const projects = Layer.mock(ProjectV2.Service, { + resolve: (directory) => Effect.succeed({ id: ProjectV2.ID.global, directory }), +}) +const execution = Layer.mock(SessionExecution.Service, { + awaitIdle: (sessionID) => Effect.sync(() => awaited.push(sessionID)), +}) +const sessions = SessionV2.layer.pipe( + Layer.provide(locationServiceMapLayer), + Layer.provide(EventV2.defaultLayer), + Layer.provide(Database.defaultLayer), + Layer.provide(SessionStore.defaultLayer), + Layer.provide(projects), + Layer.provide(execution), +) +const it = testEffect( + Layer.mergeAll( + Database.defaultLayer, + EventV2.defaultLayer, + projects, + SessionProjector.defaultLayer, + SessionStore.defaultLayer, + sessions, + ), +) + +describe("SessionV2.wait", () => { + it.effect("delegates to SessionExecution.awaitIdle", () => + Effect.gen(function* () { + awaited.length = 0 + const sessions = yield* SessionV2.Service + const session = yield* sessions.create({ location }) + + yield* sessions.wait(session.id) + + expect(awaited).toEqual([session.id]) + }), + ) +}) diff --git a/packages/core/test/tool-subagent.test.ts b/packages/core/test/tool-subagent.test.ts new file mode 100644 index 0000000000..ae50ecc9d5 --- /dev/null +++ b/packages/core/test/tool-subagent.test.ts @@ -0,0 +1,283 @@ +import { describe, expect } from "bun:test" +import { DateTime, Effect, Layer, Schema } from "effect" +import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder" +import { LayerNode } from "@opencode-ai/core/effect/layer-node" +import { makeGlobalNode } from "@opencode-ai/core/effect/app-node" +import { Database } from "@opencode-ai/core/database/database" +import { EventV2 } from "@opencode-ai/core/event" +import { Location } from "@opencode-ai/core/location" +import { ModelV2 } from "@opencode-ai/core/model" +import { ProviderV2 } from "@opencode-ai/core/provider" +import { AbsolutePath } from "@opencode-ai/core/schema" +import { AgentV2 } from "@opencode-ai/core/agent" +import { BackgroundJob } from "@opencode-ai/core/background-job" +import { LocationServiceMap } from "@opencode-ai/core/location-service-map" +import { SessionV2 } from "@opencode-ai/core/session" +import { SessionEvent } from "@opencode-ai/core/session/event" +import { SessionExecution } from "@opencode-ai/core/session/execution" +import { SessionMessage } from "@opencode-ai/core/session/message" +import { SessionRunnerModel } from "@opencode-ai/core/session/runner/model" +import { SessionStore } from "@opencode-ai/core/session/store" +import { SubagentTool } from "@opencode-ai/core/tool/subagent" +import { ToolRegistry } from "@opencode-ai/core/tool/registry" +import { ToolOutputStore } from "@opencode-ai/core/tool-output-store" +import { tmpdir } from "./fixture/tmpdir" +import { testEffect } from "./lib/effect" +import { executeTool, settleTool, toolIdentity } from "./lib/tool" + +const childText = "child final response" +const childModel = ModelV2.Ref.make({ id: ModelV2.ID.make("child"), providerID: ProviderV2.ID.make("test") }) +const parentModel = ModelV2.Ref.make({ id: ModelV2.ID.make("parent"), providerID: ProviderV2.ID.make("test") }) +const tokens = { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } } + +const outputSessionID = (value: unknown) => Schema.decodeUnknownSync(SubagentTool.Output)(value).sessionID + +const executionNode = makeGlobalNode({ + service: SessionExecution.Service, + layer: Layer.effect( + SessionExecution.Service, + Effect.gen(function* () { + const events = yield* EventV2.Service + const store = yield* SessionStore.Service + const completed = new Set() + const complete = Effect.fn("SubagentTest.complete")(function* (sessionID: SessionV2.ID) { + if (completed.has(sessionID)) return + if ((yield* store.get(sessionID))?.title.includes("fail")) { + yield* new SessionRunnerModel.ModelNotSelectedError({ sessionID }) + return + } + completed.add(sessionID) + const assistantMessageID = SessionMessage.ID.create() + const textID = "text_subagent_test" + yield* events.publish(SessionEvent.Step.Started, { + sessionID, + assistantMessageID, + timestamp: yield* DateTime.now, + agent: AgentV2.ID.make("reviewer"), + model: childModel, + }) + yield* events.publish(SessionEvent.Text.Started, { + sessionID, + assistantMessageID, + timestamp: yield* DateTime.now, + textID, + }) + yield* events.publish(SessionEvent.Text.Ended, { + sessionID, + assistantMessageID, + timestamp: yield* DateTime.now, + textID, + text: childText, + }) + yield* events.publish(SessionEvent.Step.Ended, { + sessionID, + assistantMessageID, + timestamp: yield* DateTime.now, + finish: "stop", + cost: 0, + tokens, + }) + }) + return SessionExecution.Service.of({ + active: Effect.succeed(new Set()), + resume: complete, + wake: () => Effect.void, + interrupt: () => Effect.void, + awaitIdle: (sessionID) => complete(sessionID).pipe(Effect.exit, Effect.asVoid), + }) + }), + ), + deps: [EventV2.node, SessionStore.node], +}) + +const layer = AppNodeBuilder.build( + LayerNode.bind( + LayerNode.group([ + Database.node, + EventV2.node, + BackgroundJob.node, + ToolOutputStore.cleanupNode, + SessionV2.node, + SubagentTool.node, + LocationServiceMap.node, + ]), + SessionExecution.node, + executionNode, + ), +) + +const it = testEffect(layer) + +const withSubagent = (location: Location.Ref) => + Effect.gen(function* () { + const locations = yield* LocationServiceMap.Service + yield* AgentV2.Service.use((agents) => + agents.transform((draft) => { + draft.update(AgentV2.ID.make("reviewer"), (agent) => { + agent.mode = "subagent" + agent.model = childModel + }) + draft.update(AgentV2.ID.make("fallback"), (agent) => { + agent.mode = "subagent" + }) + draft.update(AgentV2.ID.make("primary"), (agent) => { + agent.mode = "primary" + }) + }), + ).pipe(Effect.provide(locations.get(location))) + }) + +describe("SubagentTool", () => { + it.live("registers globally while resolving agents from the caller location", () => + Effect.acquireRelease( + Effect.promise(() => tmpdir()), + (dir) => Effect.promise(() => dir[Symbol.asyncDispose]()), + ).pipe( + Effect.flatMap((dir) => + Effect.gen(function* () { + const location = Location.Ref.make({ directory: AbsolutePath.make(dir.path) }) + const session = yield* SessionV2.Service + const parent = yield* session.create({ location }) + yield* withSubagent(parent.location) + + const locations = yield* LocationServiceMap.Service + const registry = yield* ToolRegistry.Service.pipe(Effect.provide(locations.get(parent.location))) + expect((yield* registry.materialize()).definitions.map((tool) => tool.name)).toContain(SubagentTool.name) + expect( + yield* executeTool(registry, { + sessionID: parent.id, + ...toolIdentity, + call: { + type: "tool-call", + id: "call-primary", + name: SubagentTool.name, + input: { agent: "primary", description: "primary", prompt: "should fail" }, + }, + }), + ).toEqual({ type: "error", value: "Agent primary cannot run as a subagent" }) + }), + ), + ), + ) + + it.live("runs a foreground child session and returns the final assistant text", () => + Effect.acquireRelease( + Effect.promise(() => tmpdir()), + (dir) => Effect.promise(() => dir[Symbol.asyncDispose]()), + ).pipe( + Effect.flatMap((dir) => + Effect.gen(function* () { + const location = Location.Ref.make({ directory: AbsolutePath.make(dir.path) }) + const sessions = yield* SessionV2.Service + const parent = yield* sessions.create({ location, model: parentModel }) + yield* withSubagent(parent.location) + const locations = yield* LocationServiceMap.Service + const registry = yield* ToolRegistry.Service.pipe(Effect.provide(locations.get(parent.location))) + + const settled = yield* settleTool(registry, { + sessionID: parent.id, + ...toolIdentity, + call: { + type: "tool-call", + id: "call-subagent", + name: SubagentTool.name, + input: { agent: "reviewer", description: "review", prompt: "review this" }, + }, + }) + + expect(settled.output?.structured).toMatchObject({ status: "completed", output: childText }) + const child = yield* sessions.get(outputSessionID(settled.output?.structured)) + expect(child).toMatchObject({ + parentID: parent.id, + location: parent.location, + agent: "reviewer", + model: childModel, + }) + + const fallback = yield* settleTool(registry, { + sessionID: parent.id, + ...toolIdentity, + call: { + type: "tool-call", + id: "call-subagent-fallback", + name: SubagentTool.name, + input: { agent: "fallback", description: "fallback", prompt: "fallback" }, + }, + }) + const fallbackChild = yield* sessions.get(outputSessionID(fallback.output?.structured)) + expect(fallbackChild).toMatchObject({ parentID: parent.id, model: parentModel }) + }), + ), + ), + ) + + it.live("returns child runner failures as tool errors", () => + Effect.acquireRelease( + Effect.promise(() => tmpdir()), + (dir) => Effect.promise(() => dir[Symbol.asyncDispose]()), + ).pipe( + Effect.flatMap((dir) => + Effect.gen(function* () { + const location = Location.Ref.make({ directory: AbsolutePath.make(dir.path) }) + const sessions = yield* SessionV2.Service + const parent = yield* sessions.create({ location }) + yield* withSubagent(parent.location) + const locations = yield* LocationServiceMap.Service + const registry = yield* ToolRegistry.Service.pipe(Effect.provide(locations.get(parent.location))) + + expect( + yield* executeTool(registry, { + sessionID: parent.id, + ...toolIdentity, + call: { + type: "tool-call", + id: "call-subagent-failure", + name: SubagentTool.name, + input: { agent: "reviewer", description: "fail review", prompt: "please fail" }, + }, + }), + ).toEqual({ type: "error", value: expect.stringContaining("No model is available for session") }) + }), + ), + ), + ) + + it.live("promotes background work and injects one synthetic parent completion", () => + Effect.acquireRelease( + Effect.promise(() => tmpdir()), + (dir) => Effect.promise(() => dir[Symbol.asyncDispose]()), + ).pipe( + Effect.flatMap((dir) => + Effect.gen(function* () { + const location = Location.Ref.make({ directory: AbsolutePath.make(dir.path) }) + const sessions = yield* SessionV2.Service + const jobs = yield* BackgroundJob.Service + const parent = yield* sessions.create({ location }) + yield* withSubagent(parent.location) + const locations = yield* LocationServiceMap.Service + const registry = yield* ToolRegistry.Service.pipe(Effect.provide(locations.get(parent.location))) + + const settled = yield* settleTool(registry, { + sessionID: parent.id, + ...toolIdentity, + call: { + type: "tool-call", + id: "call-background-subagent", + name: SubagentTool.name, + input: { agent: "reviewer", description: "background review", prompt: "review", background: true }, + }, + }) + const childID = outputSessionID(settled.output?.structured) + expect(settled.output?.structured).toMatchObject({ status: "running" }) + + yield* jobs.promote(childID) + yield* Effect.yieldNow + const synthetic = (yield* sessions.context(parent.id)).filter((message) => message.type === "synthetic") + expect(synthetic).toHaveLength(1) + expect(synthetic[0]?.text).toContain(` - Effect.fail( - new ServiceUnavailableError({ - message: `Session ${error.operation} is not available yet`, - service: `session.${error.operation}`, - }), - ), - ), ) return HttpApiSchema.NoContent.make() }), diff --git a/packages/server/src/routes.ts b/packages/server/src/routes.ts index 45841ccb33..a9e1040c95 100644 --- a/packages/server/src/routes.ts +++ b/packages/server/src/routes.ts @@ -10,6 +10,7 @@ import { SessionV2 } from "@opencode-ai/core/session" import { SessionExecution } from "@opencode-ai/core/session/execution" import { LocationServiceMap } from "@opencode-ai/core/location-service-map" import { SessionExecutionLocal } from "@opencode-ai/core/session/execution/local" +import { SubagentTool } from "@opencode-ai/core/tool/subagent" import { ToolOutputStore } from "@opencode-ai/core/tool-output-store" import { HttpRouter, HttpServer } from "effect/unstable/http" import { HttpApiBuilder } from "effect/unstable/httpapi" @@ -29,6 +30,7 @@ const applicationServices = LayerNode.group([ httpClient, ToolOutputStore.cleanupNode, SessionV2.node, + SubagentTool.node, PermissionSaved.node, PtyTicket.node, Credential.node,