feat(core): wire v2 subagent tool (#34320)

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<NotFoundError>()("Ses
 export class OperationUnavailableError extends Schema.TaggedErrorClass<OperationUnavailableError>()(
   "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<SessionSchema.Info[]>
-  readonly create: (input: CreateInput) => Effect.Effect<SessionSchema.Info>
+  readonly create: (input: CreateInput) => Effect.Effect<SessionSchema.Info, NotFoundError>
   readonly get: (sessionID: SessionSchema.ID) => Effect.Effect<SessionSchema.Info, NotFoundError>
   readonly messages: (input: {
     sessionID: SessionSchema.ID
@@ -168,7 +170,7 @@ export interface Interface {
     resume?: boolean
   }) => Effect.Effect<void, OperationUnavailableError>
   readonly compact: (input: CompactInput) => Effect.Effect<void, NotFoundError | OperationUnavailableError>
-  readonly wait: (id: SessionSchema.ID) => Effect.Effect<void, NotFoundError | OperationUnavailableError>
+  readonly wait: (id: SessionSchema.ID) => Effect.Effect<void, NotFoundError>
   readonly active: Effect.Effect<ReadonlySet<SessionSchema.ID>>
   readonly resume: (sessionID: SessionSchema.ID) => Effect.Effect<void, NotFoundError | SessionRunner.RunError>
   readonly interrupt: (sessionID: SessionSchema.ID) => Effect.Effect<void>
@@ -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))
         }),
       },
     })

packages/core/src/session/execution.ts

@@ -15,6 +15,8 @@ export interface Interface {
   readonly wake: (sessionID: SessionSchema.ID) => Effect.Effect<void>
   /** Interrupt active work owned by this process. Idle interruption is a no-op. */
   readonly interrupt: (sessionID: SessionSchema.ID) => Effect.Effect<void>
+  /** 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<void>
 }
 
 /** 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,
   }),
 )

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,
     })
   }),
 )

packages/core/src/session/run-coordinator.ts

@@ -12,6 +12,8 @@ export interface Coordinator<Key, E> {
   readonly wake: (key: Key) => Effect.Effect<void>
   /** Stops active execution and waits for its cleanup. */
   readonly interrupt: (key: Key) => Effect.Effect<void>
+  /** Resolves once no execution is active for the key. Returns immediately when already idle and never starts work. */
+  readonly awaitIdle: (key: Key) => Effect.Effect<void>
 }
 
 type Entry<E> = {
@@ -100,5 +102,15 @@ export const make = <Key, E>(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<void> =>
+      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 }
   })

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"

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<typeof part, { type: "text" }> => 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: `<subagent id="${childID}" state="${state}" description="${description}">\n${text}\n</subagent>`,
+      })
+    })
+
+    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],
+})