diff --git a/packages/core/src/plugin/runtime.ts b/packages/core/src/plugin/runtime.ts index 1e5a2e5d06..dab43abe6a 100644 --- a/packages/core/src/plugin/runtime.ts +++ b/packages/core/src/plugin/runtime.ts @@ -21,6 +21,8 @@ export interface Interface { | "command" | "rename" | "resume" + | "switchAgent" + | "switchModel" | "interrupt" | "synthetic" | "wait" @@ -75,6 +77,8 @@ export const layerWithCell = (cell: Cell) => command: (input) => require(cell, (runtime) => runtime.session.command(input)), rename: (input) => require(cell, (runtime) => runtime.session.rename(input)), resume: (sessionID) => require(cell, (runtime) => runtime.session.resume(sessionID)), + switchAgent: (input) => require(cell, (runtime) => runtime.session.switchAgent(input)), + switchModel: (input) => require(cell, (runtime) => runtime.session.switchModel(input)), interrupt: (sessionID) => require(cell, (runtime) => runtime.session.interrupt(sessionID)), synthetic: (input) => require(cell, (runtime) => runtime.session.synthetic(input)), wait: (sessionID) => require(cell, (runtime) => runtime.session.wait(sessionID)), diff --git a/packages/core/src/tool/plugin/subagent.ts b/packages/core/src/tool/plugin/subagent.ts index 8488ad1821..4d4c89de86 100644 --- a/packages/core/src/tool/plugin/subagent.ts +++ b/packages/core/src/tool/plugin/subagent.ts @@ -14,7 +14,7 @@ export const name = "subagent" const NO_TEXT = "Subagent completed without a text response." const backgroundStarted = (sessionID: SessionSchema.ID) => [ - `The subagent is working in the background (id: ${sessionID}). You will be notified automatically when it finishes.`, + `The subagent is working in the background (sessionID: ${sessionID}). You will be notified automatically when it finishes.`, "DO NOT sleep, poll for progress, ask the subagent for status, or duplicate this subagent's work; avoid working with the same files or topics it is using.", "Work on non-overlapping tasks, or briefly tell the user what you launched and end your response.", ].join("\n") @@ -23,6 +23,10 @@ export const Input = Schema.Struct({ agent: Schema.String.annotate({ description: "The type of specialized agent to use for this task" }), description: Schema.String.annotate({ description: "A short 3-5 word label for the task, displayed to the user" }), prompt: Schema.String.annotate({ description: "The task for the subagent to perform" }), + sessionID: Schema.optionalKey(SessionSchema.ID).annotate({ + description: + "Continue a specific previous subagent conversation by passing its sessionID. Calls without a sessionID start a new conversation.", + }), background: Schema.optionalKey(Schema.Boolean).annotate({ description: "Run the subagent in the background and return immediately. You will be notified when it completes. DO NOT sleep, poll, or proactively check on its progress.", @@ -36,7 +40,8 @@ export const Output = Schema.Struct({ }) export const description = [ "Spawns an agent in a child session to work on the specified task.", - "Include all relevant context and instructions in the prompt because the child starts with fresh context.", + "The output includes a sessionID you can pass back later to continue that specific conversation with the subagent.", + "New child sessions start with fresh context, so include all relevant context and instructions when you don't pass a sessionID.", "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.", @@ -50,6 +55,9 @@ export const Plugin = { const config = yield* Config.Service const permission = yield* Permission.Service const scope = yield* Scope.Scope + // One completion observer per job generation. Keyed by child plus start time so a fresh + // continuation job is observable even while a settled generation's observer is finalizing. + const notifications = new Set() // 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). @@ -77,7 +85,7 @@ export const Plugin = { ) { yield* runtime.session.synthetic({ sessionID: parentID, - text: `\n${text}\n`, + text: `\n${text}\n`, description, metadata: { source: "subagent", childID, agent, state }, }) @@ -88,7 +96,11 @@ export const Plugin = { childID: SessionSchema.ID, agent: string, description: string, + startedAt: number, ) { + const key = `${childID}:${startedAt}` + if (notifications.has(key)) return + notifications.add(key) yield* runtime.job.wait({ id: childID }).pipe( Effect.flatMap((result) => { if (result.info?.status === "completed") @@ -106,6 +118,7 @@ export const Plugin = { return injectCompletion(parentID, childID, agent, description, "cancelled", "Subagent cancelled") return Effect.void }), + Effect.ensuring(Effect.sync(() => notifications.delete(key))), Effect.forkIn(scope, { startImmediately: true }), ) }) @@ -163,33 +176,74 @@ export const Plugin = { }) .pipe(Effect.mapError((error) => new ToolFailure({ message: `Subagent denied: ${agent.id}`, error }))) - // Model selection is policy/config/session state, not an LLM-facing tool argument. - const model = agent.model ?? parent.model - const child = yield* runtime.session - .create({ - parentID: context.sessionID, - title: input.description, - agent: Agent.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. + const existing = + input.sessionID === undefined + ? undefined + : yield* runtime.session + .get(input.sessionID) + .pipe( + Effect.mapError( + (error) => + new ToolFailure({ message: `Subagent session not found: ${input.sessionID}`, error }), + ), + ) + if (existing !== undefined && existing.parentID !== context.sessionID) + return yield* new ToolFailure({ + message: `Session ${existing.id} is not a child of the current session`, }) - .pipe( + // Continuing with a different agent switches the child, mirroring create semantics + // where the agent's configured model wins over the inherited one. + if (existing !== undefined && existing.agent !== agent.id) { + yield* runtime.session.switchAgent({ sessionID: existing.id, agent: agent.id }).pipe( + Effect.andThen( + agent.model === undefined + ? Effect.void + : runtime.session.switchModel({ sessionID: existing.id, model: agent.model }), + ), Effect.mapError( - (error) => new ToolFailure({ message: `Parent session not found: ${context.sessionID}`, error }), + (error) => + new ToolFailure({ message: `Failed to switch subagent session agent: ${existing.id}`, error }), ), ) + } + + // Model selection is policy/config/session state, not an LLM-facing tool argument. + const model = agent.model ?? parent.model + const child = + existing ?? + (yield* runtime.session + .create({ + parentID: context.sessionID, + title: input.description, + agent: Agent.ID.make(input.agent), + model, + }) + .pipe( + Effect.mapError( + (error) => new ToolFailure({ message: `Parent session not found: ${context.sessionID}`, error }), + ), + )) const background = input.background === true yield* context.progress({ sessionID: child.id, status: "running" }) - const run = Effect.gen(function* () { - // The child session owns its agent/model (set at create); prompt only admits input. - yield* runtime.session.prompt({ + // Standard prompt admission outside the job: Job.start joining a running child skips + // its run effect, and the default wake starts an idle child or steers a running one. + yield* runtime.session + .prompt({ sessionID: child.id, - text: ["You are a subagent spawned by another session.", input.prompt].join("\n"), - resume: false, + text: + existing === undefined + ? ["You are a subagent spawned by another session.", input.prompt].join("\n") + : input.prompt, }) + .pipe( + Effect.mapError( + (error) => new ToolFailure({ message: `Failed to prompt subagent: ${child.id}`, error }), + ), + ) + + const run = Effect.gen(function* () { yield* runtime.session.resume(child.id) return yield* latestAssistantText(child.id) }).pipe(Effect.onInterrupt(() => runtime.session.interrupt(child.id))) @@ -204,7 +258,7 @@ export const Plugin = { if (background) { yield* runtime.job.background(info.id) - yield* notifyWhenDone(context.sessionID, child.id, agent.name, input.description) + yield* notifyWhenDone(context.sessionID, child.id, agent.name, input.description, info.started_at) return { sessionID: child.id, status: "running" as const, @@ -220,21 +274,34 @@ export const Plugin = { ), ) if (result?.type === "backgrounded") { - yield* notifyWhenDone(context.sessionID, child.id, agent.name, input.description) + yield* notifyWhenDone( + context.sessionID, + child.id, + agent.name, + input.description, + result.info.started_at, + ) return { sessionID: child.id, status: "running" as const, output: backgroundStarted(child.id), } } + // Failure surfaces keep the sessionID visible so the model can continue the child. if (result?.info.status === "error") - return yield* new ToolFailure({ message: result.info.error ?? "Subagent failed" }) - if (result?.info.status === "cancelled") return yield* new ToolFailure({ message: "Subagent cancelled" }) + return yield* new ToolFailure({ + message: `Subagent failed (sessionID: ${child.id}): ${result.info.error ?? "unknown error"}`, + }) + if (result?.info.status === "cancelled") + return yield* new ToolFailure({ message: `Subagent cancelled (sessionID: ${child.id})` }) return { sessionID: child.id, status: "completed" as const, output: result?.info.output ?? NO_TEXT } }).pipe( Effect.map((output) => ({ output, - content: output.output, + content: + output.status === "completed" + ? `\n${output.output}\n` + : output.output, metadata: { sessionID: output.sessionID, status: output.status }, })), ), diff --git a/packages/core/test/tool-subagent.test.ts b/packages/core/test/tool-subagent.test.ts index 4a137167be..42a3bb81a6 100644 --- a/packages/core/test/tool-subagent.test.ts +++ b/packages/core/test/tool-subagent.test.ts @@ -34,6 +34,8 @@ import { testEffect } from "./lib/effect" import { executeTool, registerToolPlugin, toolIdentity } from "./lib/tool" const childText = "child final response" +const completedOutput = (sessionID: Session.ID) => + `\n${childText}\n` const childModel = Model.Ref.make({ id: Model.ID.make("child"), providerID: Provider.ID.make("test") }) const parentModel = Model.Ref.make({ id: Model.ID.make("parent"), providerID: Provider.ID.make("test") }) const tokens = { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } } @@ -252,7 +254,7 @@ describe("SubagentTool", () => { expect(settled).toMatchObject({ status: "completed", metadata: { status: "completed" }, - content: [{ type: "text", text: childText }], + content: [{ type: "text", text: expect.stringContaining(childText) }], }) expect(settled.metadata).toEqual({ sessionID: outputSessionID(settled.metadata), @@ -294,9 +296,10 @@ describe("SubagentTool", () => { expect(settled).toMatchObject({ status: "completed", metadata: { status: "completed" }, - content: [{ type: "text", text: childText }], + content: [{ type: "text", text: expect.stringContaining(childText) }], }) const child = yield* sessions.get(outputSessionID(settled.metadata)) + expect(settled.content).toEqual([{ type: "text", text: completedOutput(child.id) }]) expect(settled.metadata).toEqual({ sessionID: child.id, status: "completed" }) expect(progress[0]).toEqual({ sessionID: child.id, status: "running" }) expect(child).toMatchObject({ @@ -326,6 +329,188 @@ describe("SubagentTool", () => { ), ) + it.live("continues an existing child session", () => + 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* Session.Service + const parent = yield* sessions.create({ location, model: parentModel }) + yield* withSubagent(parent.location) + const locations = yield* LocationServiceMap.Service + const registry = yield* Tool.Service.pipe(Effect.provide(locations.get(parent.location))) + + const first = yield* executeTool(registry, { + sessionID: parent.id, + ...toolIdentity, + call: { + type: "tool-call", + id: "call-subagent-first", + name: SubagentTool.name, + input: { agent: "reviewer", description: "review", prompt: "review this" }, + }, + }) + const childID = outputSessionID(first.metadata) + const second = yield* executeTool(registry, { + sessionID: parent.id, + ...toolIdentity, + call: { + type: "tool-call", + id: "call-subagent-second", + name: SubagentTool.name, + input: { + agent: "reviewer", + description: "follow up", + prompt: "continue this", + sessionID: childID, + }, + }, + }) + + expect(outputSessionID(second.metadata)).toBe(childID) + expect((yield* sessions.list({ parentID: parent.id })).data).toHaveLength(1) + expect((yield* sessions.get(childID)).title).toBe("review") + expect( + (yield* sessions.inbox(childID)).flatMap((message) => + message.type === "user" ? [message.payload.text] : [], + ), + ).toEqual(["You are a subagent spawned by another session.\nreview this", "continue this"]) + expect(second.content).toEqual([{ type: "text", text: completedOutput(childID) }]) + }), + ), + ), + ) + + it.live("steers a running child session in the background", () => + 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* Session.Service + const parent = yield* sessions.create({ location, model: parentModel }) + const child = yield* sessions.create({ + parentID: parent.id, + title: "review", + agent: Agent.ID.make("reviewer"), + model: childModel, + }) + yield* withSubagent(parent.location) + const locations = yield* LocationServiceMap.Service + const registry = yield* Tool.Service.pipe(Effect.provide(locations.get(parent.location))) + const jobs = yield* Job.Service + yield* jobs.start({ id: child.id, type: SubagentTool.name, run: Effect.never }) + + const result = yield* executeTool(registry, { + sessionID: parent.id, + ...toolIdentity, + call: { + type: "tool-call", + id: "call-running-subagent", + name: SubagentTool.name, + input: { + agent: "reviewer", + description: "follow up", + prompt: "continue while running", + sessionID: child.id, + background: true, + }, + }, + }) + + expect(result).toMatchObject({ + status: "completed", + metadata: { sessionID: child.id, status: "running" }, + }) + expect((yield* sessions.inbox(child.id)).find((message) => message.type === "user")?.payload.text).toBe( + "continue while running", + ) + expect((yield* jobs.get(child.id))?.status).toBe("running") + yield* jobs.cancel(child.id) + }), + ), + ), + ) + + it.live("rejects unrelated children and switches agents on continuation", () => + 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* Session.Service + const parent = yield* sessions.create({ location, model: parentModel }) + const otherParent = yield* sessions.create({ location, model: parentModel }) + const unrelated = yield* sessions.create({ + parentID: otherParent.id, + title: "other review", + agent: Agent.ID.make("reviewer"), + }) + const switched = yield* sessions.create({ + parentID: parent.id, + title: "fallback review", + agent: Agent.ID.make("fallback"), + model: parentModel, + }) + yield* withSubagent(parent.location) + const locations = yield* LocationServiceMap.Service + const registry = yield* Tool.Service.pipe(Effect.provide(locations.get(parent.location))) + const call = (sessionID: Session.ID, id: string, agent = "reviewer") => + executeTool(registry, { + sessionID: parent.id, + ...toolIdentity, + call: { + type: "tool-call" as const, + id, + name: SubagentTool.name, + input: { agent, description: "follow up", prompt: "continue", sessionID }, + }, + }) + + const missing = Session.ID.create() + expect(yield* call(missing, "call-missing-child")).toEqual({ + status: "error", + error: { + type: "tool.execution", + message: `Subagent session not found: ${missing}`, + }, + }) + expect(yield* call(unrelated.id, "call-unrelated-child")).toEqual({ + status: "error", + error: { + type: "tool.execution", + message: `Session ${unrelated.id} is not a child of the current session`, + }, + }) + expect(yield* call(switched.id, "call-switched-child")).toMatchObject({ + status: "completed", + metadata: { sessionID: switched.id, status: "completed" }, + }) + expect(yield* sessions.get(switched.id)).toMatchObject({ + agent: "reviewer", + model: childModel, + }) + // Switching to an agent without a configured model keeps the child's current model. + expect(yield* call(switched.id, "call-modelless-switch", "fallback")).toMatchObject({ + status: "completed", + metadata: { sessionID: switched.id, status: "completed" }, + }) + expect(yield* sessions.get(switched.id)).toMatchObject({ + agent: "fallback", + model: childModel, + }) + }), + ), + ), + ) + it.live("returns child runner failures as tool errors", () => Effect.acquireRelease( Effect.promise(() => tmpdir()), @@ -399,12 +584,12 @@ describe("SubagentTool", () => { status: "running", }) expect(settled.metadata).toEqual({ sessionID: childID, status: "running" }) - expect(settled.content).toEqual([{ type: "text", text: expect.stringContaining(`id: ${childID}`) }]) + expect(settled.content).toEqual([{ type: "text", text: expect.stringContaining(`sessionID: ${childID}`) }]) const admission = Array.from(yield* Fiber.join(admitted))[0] expect(admission?.data.item.type).toBe("synthetic") if (admission?.data.item.type !== "synthetic") return yield* Effect.die("Expected synthetic inbox item") - expect(admission?.data.item.payload.text).toContain(` { yield* SessionInbox.promote(database.db, bus, parent.id, "steer") const synthetic = (yield* sessions.context(parent.id)).filter((message) => message.type === "synthetic") expect(synthetic).toHaveLength(1) - expect(synthetic[0]?.text).toContain(`