From 5e90a68d6a7996c4d38a4f58bcefe7bac655626d Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Sat, 27 Jun 2026 23:46:07 -0400 Subject: [PATCH] wip(core): v2 subagent foundations (awaitIdle, wait, create parentID) Checkpoint of the safe subagent slice: - SessionExecution.awaitIdle (coordinator drain-to-quiescence loop) - SessionV2.wait delegates to awaitIdle; drop unavailable error + handler catch - SessionV2.create: parentID/title, optional location inherited from parent - subagent tool file (core/src/tool/subagent.ts) NOT yet wired into the node graph; wiring the global SessionV2 dep from a location-scoped tool is the open problem being experimented on next. --- packages/core/src/session.ts | 27 ++- packages/core/src/session/execution.ts | 3 + packages/core/src/session/execution/local.ts | 1 + packages/core/src/session/run-coordinator.ts | 14 +- packages/core/src/tool/subagent.ts | 180 ++++++++++++++++++ packages/core/test/session-prompt.test.ts | 1 + .../core/test/session-runner-recorded.test.ts | 1 + packages/core/test/session-runner.test.ts | 1 + packages/server/src/handlers/session.ts | 8 - 9 files changed, 218 insertions(+), 18 deletions(-) create mode 100644 packages/core/src/tool/subagent.ts diff --git a/packages/core/src/session.ts b/packages/core/src/session.ts index 0a53c8dd10..60e71a15b2 100644 --- a/packages/core/src/session.ts +++ b/packages/core/src/session.ts @@ -78,9 +78,12 @@ export type ListInput = typeof ListInput.Type type CreateInput = { id?: SessionSchema.ID + parentID?: SessionSchema.ID + title?: string agent?: AgentV2.ID model?: ModelV2.Ref - location: Location.Ref + // Optional when parentID is given: the child inherits the parent Session's location. + location?: Location.Ref } type CompactInput = { @@ -168,7 +171,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 +216,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) + // An explicit location wins; otherwise a child inherits its parent's location. + const parent = input.location === undefined && input.parentID ? yield* store.get(input.parentID) : undefined + const location = input.location ?? parent?.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 +234,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 ? { @@ -243,7 +252,7 @@ export const layer = Layer.effect( time: { created: now, updated: now }, }) const projected = yield* events - .publish(SessionV1.Event.Created, { sessionID, info }, { location: input.location }) + .publish(SessionV1.Event.Created, { sessionID, info }, { location }) .pipe( Effect.as({ type: "created" } as const), Effect.catchDefect((defect) => { @@ -432,7 +441,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) { 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/subagent.ts b/packages/core/src/tool/subagent.ts new file mode 100644 index 0000000000..06345d38bd --- /dev/null +++ b/packages/core/src/tool/subagent.ts @@ -0,0 +1,180 @@ +export * as SubagentTool from "./subagent" + +import { ToolFailure } from "@opencode-ai/llm" +import { DateTime, Effect, Layer, Schema } from "effect" +import { AgentV2 } from "../agent" +import { BackgroundJob } from "../background-job" +import { EventV2 } from "../event" +import { ModelV2 } from "../model" +import { SessionV2 } from "../session" +import { SessionEvent } from "../session/event" +import { SessionMessage } from "../session/message" +import { SessionSchema } from "../session/schema" +import { makeLocationNode, type LocationNode } from "../effect/app-node" +import { Tool } from "./tool" +import { Tools } from "./tools" +import { ToolRegistry } from "./registry" + +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" }), + model: Schema.String.pipe(Schema.optional).annotate({ + description: "Optional model override in 'providerID/modelID' form; defaults to the agent or session model", + }), + 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") + +// Accept "providerID/modelID" overrides; anything malformed falls back to the agent/session default. +const parseModel = (value: string | undefined): ModelV2.Ref | undefined => { + if (value === undefined || !value.includes("/")) return undefined + const parsed = ModelV2.parse(value) + return ModelV2.Ref.make({ providerID: parsed.providerID, id: parsed.modelID }) +} + +export const layer = Layer.effectDiscard( + Effect.gen(function* () { + const tools = yield* Tools.Service + const sessions = yield* SessionV2.Service + const jobs = yield* BackgroundJob.Service + const agents = yield* AgentV2.Service + const events = yield* EventV2.Service + + // 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, + text: string, + ) { + yield* events.publish(SessionEvent.Synthetic, { + sessionID: parentID, + messageID: SessionMessage.ID.create(), + timestamp: yield* DateTime.now, + text: `\n${text}\n`, + }) + }) + + 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 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` }) + + // Precedence: explicit input model -> agent's configured model -> parent session model. + const model = parseModel(input.model) ?? agent.model + + const child = yield* sessions.create({ + parentID: context.sessionID, + title: input.description, + agent: AgentV2.ID.make(input.agent), + model, + // No location: the child inherits the parent's location. + // TODO(opencode kkdvxn): derive restricted subagent permissions from the parent + // session (V1 deriveSubagentSessionPermission). MVP uses the agent's own permissions. + }) + + 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 } }) + yield* sessions.wait(child.id) + return yield* latestAssistantText(child.id) + }) + + const info = yield* jobs.start({ + id: child.id, + type: name, + title: input.description, + metadata: background ? { background: true } : {}, + onPromote: jobs + .wait({ id: child.id }) + .pipe( + Effect.flatMap((result) => + result.info?.status === "completed" + ? injectCompletion(context.sessionID, child.id, input.description, result.info.output ?? NO_TEXT) + : Effect.void, + ), + ), + run, + }) + + if (background) { + 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), + ) + 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 as a separate Location-scoped node rather than inside builtins, because its session +// dependencies would form a static import cycle through location-services -> tool/builtins -> session. +// Explicit annotation keeps SessionV2's type (which references LocationServiceMap) from +// expanding into the locationServices group inference and forming a type-level self-reference. +export const node: LocationNode = makeLocationNode({ + name: "subagent-tool", + layer, + deps: [ToolRegistry.toolsNode, SessionV2.node, AgentV2.node, BackgroundJob.node, EventV2.node], +}) 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..c3204bf665 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)) diff --git a/packages/server/src/handlers/session.ts b/packages/server/src/handlers/session.ts index a09444acfb..c216b09e3f 100644 --- a/packages/server/src/handlers/session.ts +++ b/packages/server/src/handlers/session.ts @@ -222,14 +222,6 @@ export const SessionHandler = HttpApiBuilder.group(Api, "server.session", (handl }), ), ), - Effect.catchTag("Session.OperationUnavailableError", (error) => - Effect.fail( - new ServiceUnavailableError({ - message: `Session ${error.operation} is not available yet`, - service: `session.${error.operation}`, - }), - ), - ), ) return HttpApiSchema.NoContent.make() }),