refactor(core): simplify direct instance wiring

AGENTS.md

@@ -177,7 +177,7 @@ const table = sqliteTable("session", {
 - Keep durable events minimal: record irreducible new facts and do not repeat state derivable by folding the ordered aggregate history. Enrich projections and read models with previous or derived state when consumers need self-contained views.
 - Keep durable prompt admission separate from model execution. `Session.prompt(...)` publishes `session.inbox.enqueued`, whose projection inserts one durable `session_inbox` row, before scheduling advisory `SessionExecution.wake(sessionID)` unless `resume: false` requests admit-only behavior. Delivery publishes `session.inbox.delivered`; its projection consumes the inbox row and inserts the visible message in the same transaction. `session_inbox` stores only unconsumed work.
 - Reusing a Session ID adopts the existing Session. Reusing a user or synthetic inbox item ID is idempotent when Session and type match: the first admission wins and the retried payload, metadata, and delivery mode are ignored, whether the item is still pending or already delivered (reconciled from the projected message without retained enqueue history). Cross-Session or cross-type reuse fails. Control items keep their operation-specific conflict behavior.
-- Keep `SessionExecution` process-global and Session-ID based. Its local implementation owns the process-local Session coordinator and discovers placement through `SessionStore` plus `LocationServiceMap.get(session.location)` only when a drain starts; no layer should take a Session ID. V2 interruption targets the active process-local ownership chain for that Session; interruption of a known but idle or locally unowned Session is a no-op, while the public API rejects an unknown Session.
+- Keep `SessionExecution` process-global and Session-ID based. Its local implementation owns the process-local Session coordinator and selects capabilities through `SessionStore` plus `SessionInstance.get(session)` only when a drain starts. The server adapter uses `LocationServiceMap.get(session.location)`; direct SDK bindings retain ready instances without a location map. No layer should take a Session ID. V2 interruption targets the active process-local ownership chain for that Session; interruption of a known but idle or locally unowned Session is a no-op, while the public API rejects an unknown Session.
 - Keep `SessionRunner`, model resolution, tool registry, permissions, and filesystem Location-scoped. Omitted `Location.workspaceID` means implicit-local placement; explicit workspace identity remains reserved for future placement semantics.
 - Preserve one explicit `llm.stream(request)` call per Physical Attempt and reload projected history before durable continuation. A logical Step may use generic pre-output retries, one full-context retry after continuation rejection, incomplete-stream continuation, or one overflow-compaction rebuild. Generic retries retain the logical step number and do not consume another agent-step allowance. Do not delegate orchestration to an in-memory tool loop.
 - Keep local Session drains process-local until clustering is implemented. `SessionRunCoordinator` joins explicit same-Session resumes, coalesces prompt wakeups, and allows different Sessions to run concurrently. A write-ahead execution claim marks a process-local busy period for restart recovery: terminal completion, failure, or user interruption releases it, while shutdown interruption and process death preserve it. Startup recovery resumes claimed top-level Sessions with durable per-execution attempt accounting. The claim is a recovery marker, not clustered ownership, fencing, or an exactly-once guarantee.

packages/core/src/bus.ts

@@ -1,6 +1,6 @@
 export * as Bus from "./bus.js"
 
-import { Cause, Clock, Context, Effect, Layer, Option, PubSub, Schema, Stream } from "effect"
+import { Cause, Clock, Context, Effect, Layer, Option, PubSub, Schema, type Scope, Stream } from "effect"
 import { Event } from "@opencode-ai/schema/event"
 import type { EventLog } from "@opencode-ai/schema/event-log"
 import { and, asc, eq, gt, lte, sql } from "drizzle-orm"
@@ -125,6 +125,8 @@ export interface Subscribe {
    * With an ambient Location, delivery is restricted to that Location and global
    * events. Unlocated Session events use the Session's owner at publication time.
    * Session moves reach both the old and new Location, without changing the event.
+   * Captured instance views restrict instance-local ephemerals to their private
+   * owner, unless published with `global: true`.
    */
   (): Stream.Stream<Event.Payload>
   <D extends Event.Definition>(definition: D): Stream.Stream<Event.Payload<D>>
@@ -146,6 +148,8 @@ export interface Interface {
     events: I,
   ) => Effect.Effect<PublishResult<I>>
   readonly subscribe: Subscribe
+  /** Acquire a live Session subscription immediately, retained until the caller's Scope closes. */
+  readonly observe: (sessionID: SessionID) => Effect.Effect<Stream.Stream<SessionEvent.Event>, never, Scope.Scope>
   /**
    * Durable, ordered per-aggregate log read. Forked aggregates may reserve an
    * inherited prefix before their first child-authored event. `follow: false`
@@ -171,6 +175,33 @@ export interface Interface {
 
 export class Service extends Context.Service<Service, Interface>()("@opencode/Bus") {}
 
+export const PrivateOwner = Context.Reference<symbol | undefined>("@opencode/Bus/PrivateOwner", {
+  defaultValue: () => undefined,
+})
+
+/** Bind instance-local ephemeral audiences without replacing the shared durable authority. */
+export function capture(bus: Interface, owner: symbol): Interface {
+  function subscribe(): Stream.Stream<Event.Payload>
+  function subscribe<D extends Event.Definition>(definition: D): Stream.Stream<Event.Payload<D>>
+  function subscribe<const D extends readonly [Event.Definition, ...Event.Definition[]]>(
+    definitions: D,
+  ): Stream.Stream<SubscribePayload<D>>
+  function subscribe(
+    input?: Event.Definition | readonly [Event.Definition, ...Event.Definition[]],
+  ): Stream.Stream<Event.Payload> {
+    const stream =
+      input === undefined ? bus.subscribe() : isDefinition(input) ? bus.subscribe(input) : bus.subscribe(input)
+    return stream.pipe(Stream.provideService(PrivateOwner, owner))
+  }
+
+  return {
+    ...bus,
+    publish: (definition, data, options) =>
+      bus.publish(definition, data, options).pipe(Effect.provideService(PrivateOwner, owner)),
+    subscribe,
+  }
+}
+
 interface Options {
   readonly beforeAggregateRead?: (aggregateID: string) => Effect.Effect<void>
   /** Maximum durable rows read per page while replaying or tailing an aggregate log. */
@@ -205,6 +236,36 @@ export function configured(options?: Options) {
         // Keep routing separate from the public event, and retain its snapshot
         // while a slow subscriber drains events queued before a move or deletion.
         const routes = new WeakMap<Event.Payload, readonly Location.Ref[]>()
+        const audiences = new WeakMap<Event.Payload, symbol | null>()
+        const localTypes = new Set([
+          "mcp.tools.changed",
+          "mcp.prompts.changed",
+          "mcp.resources.changed",
+          "mcp.status.changed",
+          "plugin.added",
+          "plugin.updated",
+          "config.updated",
+          "agent.updated",
+          "catalog.updated",
+          "integration.updated",
+          "command.updated",
+          "reference.updated",
+          "skill.updated",
+          "websearch.updated",
+          "instruction-discovery.updated",
+          "permission.asked",
+          "permission.replied",
+          "form.created",
+          "form.replied",
+          "form.cancelled",
+          "pty.created",
+          "pty.updated",
+          "pty.exited",
+          "pty.deleted",
+          "shell.created",
+          "shell.exited",
+          "shell.deleted",
+        ])
 
         const isSessionEvent = (event: Event.Payload): event is SessionEvent.Event =>
           Object.hasOwn(SessionEvent.All.cases, event.type)
@@ -489,7 +550,7 @@ export function configured(options?: Options) {
           })
         }
 
-        const observe = (event: Event.Payload, observer: (event: Event.Payload) => Effect.Effect<void>) =>
+        const observeListener = (event: Event.Payload, observer: (event: Event.Payload) => Effect.Effect<void>) =>
           Effect.suspend(() => observer(event)).pipe(
             Effect.catchCauseIf(
               (cause) => !Cause.hasInterrupts(cause),
@@ -501,7 +562,7 @@ export function configured(options?: Options) {
           return Effect.gen(function* () {
             yield* Effect.forEach(
               listeners,
-              (listener) => (isolateListeners ? observe(event, listener) : listener(event)),
+              (listener) => (isolateListeners ? observeListener(event, listener) : listener(event)),
               { discard: true },
             )
             const typed = pubsub.typed.get(event.type)
@@ -519,18 +580,19 @@ export function configured(options?: Options) {
                 (serviceLocation
                   ? { directory: serviceLocation.directory, workspaceID: serviceLocation.workspaceID }
                   : undefined))
-            return yield* publishEvent(
-              definition,
-              {
-                id: options?.id ?? Event.ID.create(),
-                created: yield* Clock.currentTimeMillis,
-                ...(options?.metadata ? { metadata: options.metadata } : {}),
-                type: definition.type,
-                ...(location ? { location } : {}),
-                data,
-              } as Event.Payload<D>,
-              options?.commit,
-            )
+            const event = {
+              id: options?.id ?? Event.ID.create(),
+              created: yield* Clock.currentTimeMillis,
+              ...(options?.metadata ? { metadata: options.metadata } : {}),
+              type: definition.type,
+              ...(location ? { location } : {}),
+              data,
+            } as Event.Payload<D>
+            if (!definition.durable && localTypes.has(definition.type)) {
+              const owner = options?.global ? null : yield* PrivateOwner
+              if (owner !== undefined) audiences.set(event as Event.Payload, owner)
+            }
+            return yield* publishEvent(definition, event, options?.commit)
           })
         }
 
@@ -731,24 +793,24 @@ export function configured(options?: Options) {
 
         const local = <A extends Event.Payload>(stream: Stream.Stream<A>) =>
           Stream.unwrap(
-            Effect.serviceOption(Location.Service).pipe(
-              Effect.map((location) =>
-                Option.match(location, {
-                  onNone: () => stream,
-                  onSome: (location) => {
-                    const matches = (ref: Location.Ref) =>
-                      ref.directory === location.directory && ref.workspaceID === location.workspaceID
-                    return stream.pipe(
-                      Stream.filter((event) => {
-                        const refs = routes.get(event)
-                        if (refs) return refs.some(matches)
-                        return !event.location || matches(event.location)
-                      }),
-                    )
-                  },
+            Effect.gen(function* () {
+              const owner = yield* PrivateOwner
+              const location = Option.getOrUndefined(yield* Effect.serviceOption(Location.Service))
+              const matches = (ref: Location.Ref) =>
+                ref.directory === location?.directory && ref.workspaceID === location?.workspaceID
+              return stream.pipe(
+                Stream.filter((event) => {
+                  if (!event.durable && localTypes.has(event.type)) {
+                    const audience = audiences.get(event)
+                    if (audience !== null && audience !== owner) return false
+                  }
+                  if (!location) return true
+                  const refs = routes.get(event)
+                  if (refs) return refs.some(matches)
+                  return !event.location || matches(event.location)
                 }),
-              ),
-            ),
+              )
+            }),
           )
 
         function subscribe(): Stream.Stream<Event.Payload>
@@ -767,6 +829,14 @@ export function configured(options?: Options) {
 
         const streamLive = (): Stream.Stream<Event.Payload> => local(Stream.fromPubSub(pubsub.live))
 
+        const observe = Effect.fn("Bus.observe")(function* (sessionID: SessionID) {
+          const subscription = yield* PubSub.subscribe(pubsub.live)
+          return Stream.fromSubscription(subscription).pipe(
+            Stream.filter(isSessionEvent),
+            Stream.filter((event) => event.data.sessionID === sessionID),
+          )
+        })
+
         const readAfter = (
           aggregateID: string,
           after: number,
@@ -903,6 +973,7 @@ export function configured(options?: Options) {
           publish,
           publishAll,
           subscribe,
+          observe,
           log,
           listen,
           project,

packages/core/src/instance.ts

@@ -1,9 +1,20 @@
-import { Effect, Layer } from "effect"
+import { Context, Effect, Layer } from "effect"
+import { CrossSpawnSpawner } from "@opencode-ai/util/cross-spawn-spawner"
+import { httpClient } from "@opencode-ai/util/effect/app-node-platform"
+import { FSUtil } from "@opencode-ai/util/fs-util"
+import { Global } from "@opencode-ai/util/global"
+import { Npm } from "@opencode-ai/util/npm"
+import { AppProcess } from "@opencode-ai/util/process"
 import { Agent } from "./agent.js"
 import { AISDK } from "./aisdk.js"
+import { App } from "./app.js"
+import { Bus } from "./bus.js"
 import { Catalog } from "./catalog.js"
 import { Command } from "./command.js"
 import { Config } from "./config.js"
+import { Credential } from "./credential.js"
+import { Database } from "./database/database.js"
+import { llmClient, webSocketConstructor } from "./effect/app-node-platform.js"
 import { LayerNode } from "@opencode-ai/util/effect/layer-node"
 import { Node } from "@opencode-ai/util/effect/app-node"
 import { FileMutation } from "./file-mutation.js"
@@ -12,24 +23,34 @@ import { Formatter } from "./formatter.js"
 import { FileSystem } from "./filesystem.js"
 import { FileSystemSearch } from "./filesystem/search.js"
 import { Generate } from "./generate.js"
+import { Git } from "./git.js"
 import { Form } from "./form.js"
 import { Image } from "./image.js"
 import { LocationWatcher } from "./filesystem/location-watcher.js"
 import { Integration } from "./integration.js"
+import { KV } from "./kv.js"
 import { Location } from "./location.js"
 import { LocationMutation } from "./location-mutation.js"
 import { ModelResolver } from "./model-resolver.js"
+import { ModelsDev } from "./models-dev.js"
 import { Mcp } from "./mcp/index.js"
 import { Permission } from "./permission.js"
+import { PermissionSaved } from "./permission/saved.js"
 import { Plugin } from "./plugin.js"
 import { PluginHooks } from "./plugin/hooks.js"
 import { InstancePlugins } from "./plugin/instance.js"
+import { PluginRuntime } from "./plugin/runtime.js"
+import { SdkPlugins } from "./plugin/sdk.js"
 import { PluginSupervisor } from "./plugin/supervisor.js"
+import { Project } from "./project.js"
+import { ProjectMarkers } from "./project/markers.js"
 import { Worktree } from "./worktree.js"
 import { Pty } from "./pty.js"
 import { Shell } from "./shell.js"
 import { ShellSelect } from "./shell/select.js"
 import { Reference } from "./reference.js"
+import { RepositoryCache } from "./repository-cache.js"
+import { RipgrepBinary } from "./ripgrep/binary.js"
 import { WebSearch } from "./websearch.js"
 import { ReferenceInstructions } from "./reference/instructions.js"
 import { SessionRunnerLLM } from "./session/runner/llm.js"
@@ -37,7 +58,10 @@ import { SessionRunnerModel } from "./session/runner/model.js"
 import { SessionModelTransport } from "./session/model-transport.js"
 import { SessionCompaction } from "./session/compaction.js"
 import { SessionTitle } from "./session/title.js"
+import { SessionEnvironment } from "./session/environment.js"
+import { SessionStore } from "./session/store.js"
 import { Skill } from "./skill.js"
+import { SkillDiscovery } from "./skill/discovery.js"
 import { SkillInstructions } from "./skill/instructions.js"
 import { Snapshot } from "./snapshot.js"
 import { InstructionDiscovery } from "./instruction-discovery.js"
@@ -50,6 +74,9 @@ import { ReadToolFileSystem } from "./tool/read-filesystem.js"
 import { Tool } from "./tool.js"
 import { ToolOutput } from "./tool-output.js"
 import { Vcs } from "./vcs.js"
+import { Watcher } from "./filesystem/watcher.js"
+import { WellKnown } from "./wellknown.js"
+import { Workspace } from "./workspace.js"
 
 export * as Instance from "./instance.js"
 
@@ -111,6 +138,46 @@ export const graph = LayerNode.group<typeof nodes>(nodes)
 export type Services = LayerNode.Output<typeof graph>
 export type Error = LayerNode.Error<typeof graph>
 
+const globalNodes = [
+  CrossSpawnSpawner.node,
+  Workspace.node,
+  Watcher.node,
+  Bus.node,
+  FSUtil.node,
+  Global.node,
+  Credential.node,
+  WellKnown.node,
+  RepositoryCache.node,
+  KV.node,
+  AppProcess.node,
+  Npm.node,
+  App.node,
+  llmClient,
+  SessionStore.node,
+  PermissionSaved.node,
+  SdkPlugins.node,
+  RipgrepBinary.node,
+  httpClient,
+  ProjectMarkers.node,
+  ModelsDev.node,
+  SessionEnvironment.node,
+  Git.node,
+  SkillDiscovery.node,
+  Worktree.node,
+  Database.node,
+  webSocketConstructor,
+  // Binding Location introduces Project even though the unbound graph does not.
+  Project.node,
+] as const satisfies readonly Node.GlobalNode<unknown, unknown>[]
+
+const globalJobs = new Map([Shell.cleanupNode, ToolOutput.cleanupNode].map((node) => [node.name, node] as const))
+
+/** Build and configure this graph once in the host scope, before composing instances. */
+export const globalsGraph = LayerNode.group([...globalNodes, ...globalJobs.values()])
+
+export type Globals = LayerNode.Output<typeof globalsGraph>
+export type GlobalsError = LayerNode.Error<typeof globalsGraph>
+
 export interface Options {
   // Plugins this instance is born with; empty and absent are equivalent.
   readonly plugins?: InstancePlugins.List
@@ -139,6 +206,66 @@ const vanillaReplacements: LayerNode.Replacements = [
   [InstructionDiscovery.node, InstructionDiscovery.configured({ project: false, global: false })],
 ]
 
+/**
+ * Reuse already-acquired host infrastructure while giving each instance fresh
+ * local services. Global replacements belong on globalsGraph; local replacements
+ * and a closed per-instance PluginRuntime replacement belong here.
+ */
+export function compose<const Items extends LayerNode.Replacements = readonly []>(
+  ref: Location.Ref,
+  options: Omit<Options, "replacements"> & { readonly replacements?: LayerNode.ComposableReplacements<Items> } = {},
+): Layer.Layer<Services, Error | LayerNode.ReplacementError<Items>, Globals | LayerNode.ReplacementServices<Items>> {
+  const startedAt = performance.now()
+  const replacements: LayerNode.Replacements = [
+    ...(options.discovery === false ? vanillaReplacements : []),
+    ...(options.replacements ?? []),
+    [Location.node, Location.boundNode(ref, { discovery: options.discovery })],
+    [InstancePlugins.node, InstancePlugins.bound(options.plugins ?? [])],
+  ]
+  const hoisted = LayerNode.hoist(graph, Node.tags.values.global, replacements).hoisted
+  // PluginRuntime itself is local to a direct instance, but a node replacement
+  // can still depend on shared globals. Inspect those edges before binding them.
+  const boundary = LayerNode.hoist(
+    LayerNode.group(
+      hoisted.dependencies.flatMap((node) => (node.name === PluginRuntime.node.name ? node.dependencies : [node])),
+    ),
+    Node.tags.values.global,
+  ).hoisted
+  const names = new Set(globalNodes.map((node) => node.name))
+  const unsupported = boundary.dependencies.filter((node) => !names.has(node.name) && !globalJobs.has(node.name))
+  if (unsupported.length > 0) {
+    throw new Error(`Unsupported instance globals: ${unsupported.map((node) => node.name).join(", ")}`)
+  }
+
+  return Layer.unwrap(
+    Effect.map(Effect.context<Globals>(), (globals) => {
+      const owner = Symbol()
+      const captured = Layer.succeedContext(
+        globals.pipe(
+          Context.add(Bus.PrivateOwner, owner),
+          Context.add(Bus.Service, Bus.capture(Context.get(globals, Bus.Service), owner)),
+        ),
+      )
+      const bindings: LayerNode.Replacements = boundary.dependencies.map((node) => [
+        node,
+        globalJobs.has(node.name) ? Layer.empty : captured,
+      ])
+      // Compile the original graph with real closed implementations, not the
+      // dependency-stripped hoist result that cannot honestly be a closed layer.
+      return LayerNode.compile(graph, [...replacements, ...bindings]).pipe(
+        Layer.fresh,
+        Layer.tap(() =>
+          Effect.logInfo("location services booted", {
+            directory: ref.directory,
+            workspaceID: ref.workspaceID,
+            durationMs: Math.round(performance.now() - startedAt),
+          }),
+        ),
+      )
+    }),
+  )
+}
+
 // One instance is one compiled, fresh copy of the graph standing on a directory.
 export function layer(ref: Location.Ref, options: Options = {}) {
   const startedAt = performance.now()

packages/core/src/plugin/runtime.ts

@@ -49,16 +49,19 @@ export class Service extends Context.Service<Service, Interface>()("@opencode/Pl
 
 export interface Cell {
   runtime?: Interface
+  readonly ready?: Effect.Effect<void>
 }
 
-export const makeCell = (): Cell => ({})
+export const makeCell = (ready?: Effect.Effect<void>): Cell => ({ ready })
 
 const require = <A, E, R>(cell: Cell, f: (runtime: Interface) => Effect.Effect<A, E, R>) =>
-  Effect.suspend(() => {
-    const runtime = cell.runtime
-    if (runtime === undefined) return Effect.die(new Error("Plugin runtime is unavailable"))
-    return f(runtime)
-  })
+  (cell.ready ?? Effect.void).pipe(
+    Effect.andThen(() => {
+      const runtime = cell.runtime
+      if (runtime === undefined) return Effect.die(new Error("Plugin runtime is unavailable"))
+      return f(runtime)
+    }),
+  )
 
 const defaultCell = makeCell()
 

packages/core/src/session.ts

@@ -1,7 +1,7 @@
 export * as Session from "./session.js"
 export * from "./session/schema.js"
 
-import { Cause, Effect, Layer, Schema, Context, RcMap, Stream, Scope } from "effect"
+import { Cause, Effect, Layer, Schema, Context, Stream, Scope } from "effect"
 import { ListAnchor } from "@opencode-ai/schema/session"
 import { and, asc, desc, eq, gt, isNull, like, lt, or, type SQL } from "drizzle-orm"
 import { Project } from "./project.js"
@@ -27,10 +27,9 @@ import { fromRow } from "./session/info.js"
 import { SessionRunner } from "./session/runner/index.js"
 import { SessionStore } from "./session/store.js"
 import { SessionExecution } from "./session/execution.js"
-import { SessionModelTransport } from "./session/model-transport.js"
 import { ForkEmptyError, MessageDecodeError, NotFoundError } from "./session/error.js"
 import { makeGlobalNode } from "@opencode-ai/util/effect/app-node"
-import { LocationServiceMap } from "./location-service-map.js"
+import { SessionInstance } from "./session/instance.js"
 import { SessionEvent } from "./session/event.js"
 import { SessionInbox } from "./session/inbox.js"
 import { InstructionState } from "./session/instruction-state.js"
@@ -99,6 +98,8 @@ type CreateBaseInput = {
   agent?: Agent.ID
   model?: Model.Ref
   metadata?: SessionSchema.Metadata
+  /** Runtime discovery policy; never recorded as a Session fact. */
+  discovery?: boolean
 }
 type CreateInput = CreateBaseInput &
   ({ location: Location.Ref; parentID?: never } | { parentID: SessionSchema.ID; location?: never })
@@ -342,23 +343,13 @@ const layer = Layer.effect(
     const global = yield* Global.Service
     const execution = yield* SessionExecution.Service
     const store = yield* SessionStore.Service
-    const locations = yield* LocationServiceMap.Service
+    const instances = yield* SessionInstance.Service
     const fs = yield* FSUtil.Service
     const jobs = yield* Job.Service
     const environments = yield* SessionEnvironment.Service
     const scope = yield* Scope.Scope
     const activeShells = new Set<SessionSchema.ID>()
     const shellLocks = KeyedMutex.makeUnsafe<SessionSchema.ID>()
-    const closeTransport = Effect.fn("Session.closeTransport")(function* (session: SessionSchema.Info) {
-      const location = Location.Ref.make({
-        directory: session.location.directory,
-        workspaceID: session.location.workspaceID,
-      })
-      if (!(yield* RcMap.has(locations.rcMap, location))) return
-      yield* SessionModelTransport.Service.use((transport) => transport.close(session.id)).pipe(
-        Effect.provide(locations.get(location)),
-      )
-    })
     const isDurableSessionEvent = Schema.is(SessionEvent.Durable)
     const persistProject = (project: Project.Resolved) => upsertProject(db, project).pipe(Effect.orDie)
 
@@ -395,7 +386,7 @@ const layer = Layer.effect(
         const location = parent?.location ?? input.location
         if (location === undefined)
           return yield* Effect.die(new Error("Session.create requires either location or an existing parentID"))
-        const project = yield* projects.resolve(location.directory)
+        const project = yield* projects.resolve(location.directory, { discovery: input.discovery })
         yield* persistProject(project)
         const projected = yield* bus
           .publish(
@@ -510,7 +501,7 @@ const layer = Layer.effect(
         const session = yield* result.get(sessionID)
         yield* execution.interrupt(sessionID)
         yield* execution.awaitIdle(sessionID)
-        yield* closeTransport(session)
+        yield* instances.closeTransport(session)
         const children = yield* result.list({ parentID: sessionID })
         yield* Effect.forEach(children.data, (child) => result.remove(child.id), { concurrency: 1, discard: true })
         yield* environments.clear(sessionID)
@@ -658,7 +649,7 @@ const layer = Layer.effect(
               if (existing) return existing
               const item = yield* restore(
                 preparePrompt(input, messageID).pipe(
-                  Effect.provide(locations.get(session.location)),
+                  Effect.provide(instances.get(session)),
                   Effect.provideService(FSUtil.Service, fs),
                 ),
               )
@@ -690,7 +681,7 @@ const layer = Layer.effect(
       ),
       generate: Effect.fn("Session.generate")(function* (input) {
         const session = yield* result.get(input.sessionID)
-        const generate = yield* SessionGenerate.Service.pipe(Effect.provide(locations.get(session.location)))
+        const generate = yield* SessionGenerate.Service.pipe(Effect.provide(instances.get(session)))
         return yield* generate.generate(input)
       }),
       command: Effect.fn("Session.command")(function* (input) {
@@ -699,7 +690,7 @@ const layer = Layer.effect(
           const plugins = yield* PluginSupervisor.Service
           yield* plugins.flush
           return yield* Command.Service
-        }).pipe(Effect.provide(locations.get(session.location)))
+        }).pipe(Effect.provide(instances.get(session)))
         const delivery = input.delivery ?? "steer"
         yield* commands.execute({
           name: input.command,
@@ -733,7 +724,7 @@ const layer = Layer.effect(
                   metadata: { sessionID: input.sessionID },
                 })
                 .pipe(Effect.orDie)
-            }).pipe(Effect.provide(locations.get(session.location)))
+            }).pipe(Effect.provide(instances.get(session)))
             yield* bus.publish(
               SessionEvent.Shell.Started,
               {
@@ -756,7 +747,7 @@ const layer = Layer.effect(
                     .pipe(Effect.catchTag("Shell.NotFoundError", () => Effect.succeed(missingShellOutput())))
                 : missingShellOutput()
               return { shell: terminal.info, output }
-            }).pipe(Effect.provide(locations.get(session.location)))
+            }).pipe(Effect.provide(instances.get(session)))
             yield* bus.publish(SessionEvent.Shell.Ended, {
               sessionID: input.sessionID,
               shell: completed.shell,
@@ -774,7 +765,7 @@ const layer = Layer.effect(
       }),
       skill: Effect.fn("Session.skill")(function* (input) {
         const session = yield* result.get(input.sessionID)
-        const skills = yield* Skill.Service.pipe(Effect.provide(locations.get(session.location)))
+        const skills = yield* Skill.Service.pipe(Effect.provide(instances.get(session)))
         const skill = yield* skills.get(input.skill)
         if (!skill) return yield* new SkillNotFoundError({ skill: input.skill })
         yield* bus.publish(
@@ -837,7 +828,7 @@ const layer = Layer.effect(
           subpath: RelativePath.make(path.relative(project.directory, directory).replaceAll("\\", "/")),
         }
         yield* Location.Service.pipe(
-          Effect.provide(locations.get(payload.location)),
+          Effect.provide(instances.destination(payload.location)),
           Effect.scoped,
           Effect.catchCause((cause) => {
             if (Cause.hasInterruptsOnly(cause)) return Effect.failCause(cause)
@@ -969,7 +960,7 @@ const layer = Layer.effect(
               Effect.provideService(Database.Service, database),
               Effect.provideService(Bus.Service, bus),
             )
-          }).pipe(Effect.provide(locations.get(session.location)))
+          }).pipe(Effect.provide(instances.get(session)))
         }),
         clear: Effect.fn("Session.revert.clear")(function* (sessionID) {
           const session = yield* result.get(sessionID)
@@ -978,7 +969,7 @@ const layer = Layer.effect(
             const plugins = yield* PluginSupervisor.Service
             yield* plugins.flush
             return yield* SessionRevert.clear(session).pipe(Effect.provideService(Bus.Service, bus))
-          }).pipe(Effect.provide(locations.get(session.location)))
+          }).pipe(Effect.provide(instances.get(session)))
           yield* execution.wake(sessionID)
           return revert
         }),
@@ -1222,7 +1213,7 @@ export const node = makeGlobalNode({
     Project.node,
     SessionExecution.node,
     SessionStore.node,
-    LocationServiceMap.node,
+    SessionInstance.node,
     SessionProjector.node,
     FSUtil.node,
     Global.node,

packages/core/src/session/bindings.ts

@@ -0,0 +1,126 @@
+export * as SessionBindings from "./bindings.js"
+
+import { Context, Effect, Layer, Schema, Scope } from "effect"
+import { makeGlobalNode } from "@opencode-ai/util/effect/app-node"
+import type { Instance } from "../instance.js"
+import { Location } from "../location.js"
+import type { SessionExecution } from "./execution.js"
+import { SessionInstance } from "./instance.js"
+import { SessionModelTransport } from "./model-transport.js"
+import { SessionSchema } from "./schema.js"
+import { SessionStore } from "./store.js"
+
+export class AlreadyBoundError extends Schema.TaggedError<AlreadyBoundError>()("Session.AlreadyBoundError", {
+  sessionID: SessionSchema.ID,
+}) {}
+
+export class ClosedError extends Schema.TaggedError<ClosedError>()("Session.ClosedError", {
+  sessionID: SessionSchema.ID,
+}) {}
+
+export interface Binding {
+  readonly check: Effect.Effect<void, ClosedError>
+  readonly activate: (context: Context.Context<Instance.Services>) => Effect.Effect<void>
+  readonly shutdown: (execution: SessionExecution.Interface) => Effect.Effect<void>
+}
+
+export interface Interface {
+  readonly reserve: (sessionID: SessionSchema.ID) => Effect.Effect<Binding, AlreadyBoundError, Scope.Scope>
+  readonly instances: SessionInstance.Interface
+}
+
+export class Service extends Context.Service<Service, Interface>()("@opencode/SessionBindings") {}
+
+type Entry = {
+  readonly ids: Set<SessionSchema.ID>
+  context?: Context.Context<Instance.Services>
+  closed: boolean
+}
+
+export const layer = Layer.effect(
+  Service,
+  Effect.gen(function* () {
+    const store = yield* SessionStore.Service
+    const entries = new Map<SessionSchema.ID, Entry>()
+    // Children use their nearest explicitly bound ancestor. Remember every used
+    // child so closing one instance settles its whole execution ownership chain.
+    const find = (session: SessionSchema.Info): Effect.Effect<Entry | undefined> =>
+      Effect.suspend(() => {
+        const entry = entries.get(session.id)
+        if (entry) return Effect.succeed(entry)
+        if (!session.parentID) return Effect.succeed(undefined)
+        return store
+          .get(session.parentID)
+          .pipe(Effect.flatMap((parent) => (parent ? find(parent) : Effect.succeed(undefined))))
+      })
+    const selected = Effect.fn("SessionBindings.selected")(function* (session: SessionSchema.Info) {
+      const entry = yield* find(session)
+      if (!entry || entry.closed || !entry.context)
+        return yield* Effect.die(new Error(`Session has no live bound instance: ${session.id}`))
+      const location = Context.get(entry.context, Location.Service)
+      if (session.location.directory !== location.directory || session.location.workspaceID !== location.workspaceID)
+        return yield* Effect.die(new Error(`Bound Session placement changed: ${session.id}`))
+      entry.ids.add(session.id)
+      entries.set(session.id, entry)
+      return entry.context
+    })
+    return Service.of({
+      reserve: (sessionID) =>
+        Effect.uninterruptible(
+          Effect.gen(function* () {
+            const session = yield* store.get(sessionID)
+            if (!session) return yield* Effect.die(new Error(`Session not found: ${sessionID}`))
+            if (yield* find(session)) return yield* new AlreadyBoundError({ sessionID })
+            const entry: Entry = { ids: new Set([sessionID]), closed: false }
+            entries.set(sessionID, entry)
+            const release = Effect.sync(() => {
+              entry.closed = true
+              entry.context = undefined
+              entry.ids.forEach((id) => {
+                if (entries.get(id) === entry) entries.delete(id)
+              })
+            })
+            yield* Effect.addFinalizer(() => release)
+            return {
+              check: Effect.suspend(() => (entry.closed ? Effect.fail(new ClosedError({ sessionID })) : Effect.void)),
+              activate: (context) =>
+                Effect.sync(() => {
+                  entry.context = context
+                }),
+              shutdown: (execution) =>
+                Effect.sync(() => {
+                  entry.closed = true
+                }).pipe(Effect.andThen(execution.shutdown(Array.from(entry.ids))), Effect.ensuring(release)),
+            }
+          }),
+        ),
+      instances: {
+        get: (session) => Layer.effectContext(selected(session)),
+        check: (sessionID) =>
+          store.get(sessionID).pipe(
+            Effect.flatMap((session) =>
+              session ? selected(session) : Effect.die(new Error(`Session not found: ${sessionID}`)),
+            ),
+            Effect.asVoid,
+          ),
+        closeTransport: (session) =>
+          selected(session).pipe(
+            Effect.flatMap((context) => Context.get(context, SessionModelTransport.Service).close(session.id)),
+          ),
+        destination: () =>
+          Layer.effect(Location.Service, Effect.die(new Error("Direct Sessions do not support movement"))),
+      },
+    })
+  }),
+)
+
+export const node = makeGlobalNode({ service: Service, layer, deps: [SessionStore.node] })
+
+export const instanceNode = makeGlobalNode({
+  service: SessionInstance.Service,
+  layer: Layer.effect(
+    SessionInstance.Service,
+    Effect.map(Service, (bindings) => bindings.instances),
+  ),
+  deps: [node],
+})

packages/core/src/session/direct.ts

@@ -0,0 +1,165 @@
+export * as DirectSession from "./direct.js"
+
+import { Context, Effect, Exit, Fiber, Latch, Layer, Scope, Stream } from "effect"
+import { LayerNode } from "@opencode-ai/util/effect/layer-node"
+import { Agent } from "../agent.js"
+import { Bus } from "../bus.js"
+import { Instance } from "../instance.js"
+import { Location } from "../location.js"
+import { Mcp } from "../mcp/index.js"
+import { PluginRuntime } from "../plugin/runtime.js"
+import { PluginSupervisor } from "../plugin/supervisor.js"
+import { Session } from "../session.js"
+import { Shared } from "../shared.js"
+import { McpTool } from "../tool/mcp.js"
+import { SessionEvent } from "./event.js"
+import { SessionSchema } from "./schema.js"
+
+export { AlreadyBoundError, ClosedError } from "./bindings.js"
+export { ID } from "./schema.js"
+
+type Facts = Omit<Parameters<Session.Interface["create"]>[0], "id" | "location" | "parentID" | "discovery">
+export type Options<Items extends Shared.Replacements = Shared.Replacements> = Facts &
+  Omit<Instance.Options, "replacements"> & {
+    readonly replacements?: LayerNode.ComposableReplacements<Items>
+  } & (
+    | { readonly id: SessionSchema.ID; readonly location?: Location.Ref }
+    | { readonly id?: SessionSchema.ID; readonly location: Location.Ref }
+  )
+
+/** Creates/adopts durable facts, then binds a private ready instance to the caller's Scope. */
+export const create = Effect.fn("DirectSession.create")(function* <
+  const Items extends Shared.Replacements = readonly [],
+>(options: Options<Items>) {
+  const shared = yield* Shared.Service
+  const discovery = options.discovery ?? false
+  const session =
+    options.location === undefined
+      ? yield* options.id === undefined
+          ? Effect.die(new Error("DirectSession.create requires a location or an existing Session ID"))
+          : shared.sessions.get(options.id)
+      : yield* shared.sessions.create({
+          id: options.id,
+          location: options.location,
+          title: options.title,
+          agent: options.agent,
+          model: options.model,
+          metadata: options.metadata,
+          discovery,
+        })
+  // The backing provider may close before the caller's Scope.
+  const scope = yield* Scope.fork(shared.scope)
+  yield* Effect.addFinalizer((exit) => Scope.close(scope, exit))
+  return yield* Effect.gen(function* () {
+    const binding = yield* shared.bindings.reserve(session.id)
+    const ready = yield* Latch.make()
+    const cell = PluginRuntime.makeCell(ready.await)
+    const replacements: Shared.Replacements = [
+      ...shared.replacements,
+      ...(options.replacements ?? []),
+      [PluginRuntime.node, PluginRuntime.layerWithCell(cell)],
+    ]
+    const context = yield* Layer.build(
+      Instance.compose(session.location, { ...options, discovery, replacements }),
+    ).pipe(Effect.provideContext(shared.globals))
+    const location = Context.get(context, Location.Service)
+    const info = new Location.Info({
+      directory: location.directory,
+      workspaceID: location.workspaceID,
+      project: location.project,
+    })
+    const bound = <A, E>(effect: Effect.Effect<A, E>) => binding.check.pipe(Effect.orDie, Effect.andThen(effect))
+    const at = <A, E>(ref: Location.Ref, effect: Effect.Effect<A, E>) =>
+      bound(
+        ref.directory === location.directory && ref.workspaceID === location.workspaceID
+          ? effect
+          : Effect.die(new Error("Direct instances can only inspect their bound Location")),
+      )
+    cell.runtime = {
+      session: {
+        ...shared.sessions,
+        create: (input) => bound(shared.sessions.create({ ...input, discovery })),
+        prompt: (input) => bound(shared.sessions.prompt(input)),
+        synthetic: (input) => bound(shared.sessions.synthetic(input)),
+        command: (input) => bound(shared.sessions.command(input)),
+        generate: (input) => bound(shared.sessions.generate(input)),
+        rename: (input) => bound(shared.sessions.rename(input)),
+        move: (input) => bound(shared.sessions.move(input)),
+        switchAgent: (input) => bound(shared.sessions.switchAgent(input)),
+        switchModel: (input) => bound(shared.sessions.switchModel(input)),
+      },
+      job: shared.jobs,
+      persistentPty: shared.persistentPty,
+      location: {
+        agent: {
+          list: (ref) =>
+            at(
+              ref,
+              Context.get(context, Agent.Service)
+                .list()
+                .pipe(Effect.map((data) => ({ location: info, data }))),
+            ),
+        },
+        mcp: {
+          list: (ref) =>
+            at(
+              ref,
+              Context.get(context, Mcp.Service)
+                .servers()
+                .pipe(Effect.map((data) => ({ location: info, data }))),
+            ),
+        },
+      },
+    }
+    yield* binding.activate(context)
+    yield* ready.open
+    const observations = yield* Scope.fork(scope)
+    yield* Effect.addFinalizer(() =>
+      binding.shutdown(shared.execution).pipe(
+        Effect.ensuring(
+          Effect.sync(() => {
+            cell.runtime = undefined
+          }),
+        ),
+      ),
+    )
+    yield* Context.get(context, PluginSupervisor.Service).flush
+    yield* Context.get(context, McpTool.Service).flush
+
+    const run = <A, E, R>(effect: Effect.Effect<A, E, R>) =>
+      binding.check.pipe(Effect.andThen(effect.pipe(Effect.forkIn(scope))), Effect.flatMap(Fiber.join))
+    const bus = Context.get(shared.globals, Bus.Service)
+    return {
+      id: session.id,
+      prompt: (input: Omit<Parameters<Session.Interface["prompt"]>[0], "sessionID">) =>
+        run(shared.sessions.prompt({ ...input, sessionID: session.id })),
+      resume: () => run(shared.sessions.resume(session.id)),
+      interrupt: () => run(shared.sessions.interrupt(session.id)),
+      wait: () => run(shared.sessions.wait(session.id)),
+      events: {
+        subscribe: <E, R>(callback: (event: SessionEvent.Event) => Effect.Effect<void, E, R>) =>
+          binding.check.pipe(
+            Effect.andThen(
+              Effect.gen(function* () {
+                const caller = yield* Scope.Scope
+                const observer = yield* Scope.fork(observations)
+                const events = yield* bus.observe(session.id).pipe(Scope.provide(observer))
+                return yield* events.pipe(
+                  Stream.runForEach(callback),
+                  Scope.provide(observer),
+                  Effect.onExit((exit) => Scope.close(observer, exit)),
+                  Effect.forkIn(observer),
+                  Effect.map(Fiber.runIn(caller)),
+                )
+              }),
+            ),
+          ),
+      },
+    }
+  }).pipe(
+    Scope.provide(scope),
+    Effect.onExit((exit) => (Exit.isFailure(exit) ? Scope.close(scope, exit) : Effect.void)),
+  )
+})
+
+export type Handle = Effect.Success<ReturnType<typeof create>>

packages/core/src/session/execution.ts

@@ -4,7 +4,7 @@ import { Cause, Context, Effect, Exit, Layer } from "effect"
 import { Bus } from "../bus.js"
 import { Database } from "../database/database.js"
 import { Job } from "../job.js"
-import { LocationServiceMap } from "../location-service-map.js"
+import { SessionInstance } from "./instance.js"
 import { makeGlobalNode } from "@opencode-ai/util/effect/app-node"
 import { SessionEvent } from "./event.js"
 import { SessionRunCoordinator } from "./run-coordinator.js"
@@ -31,9 +31,11 @@ export interface Interface {
   readonly interrupt: (sessionID: SessionSchema.ID, options?: { readonly continue?: boolean }) => Effect.Effect<boolean>
   /** 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>
+  /** Settles a scoped instance's work without releasing its restart claim. */
+  readonly shutdown: (sessionIDs: readonly SessionSchema.ID[]) => Effect.Effect<void>
 }
 
-/** Routes execution from a Session ID to the runner owned by that Session's Location. */
+/** Routes execution from a Session ID to its host-selected instance's runner. */
 export class Service extends Context.Service<Service, Interface>()("@opencode/SessionExecution") {}
 
 type InterruptReason = "user" | "shutdown"
@@ -46,12 +48,12 @@ export function terminal(exit: Exit.Exit<void, SessionRunner.RunError>, reason?:
   return { type: "failed" as const, error: toSessionError(failure) }
 }
 
-/** Process-local execution: drains run in this process, routed through the Session's Location graph. */
+/** One process-local coordinator; instance selection is separate from its drain policy. */
 export const layer = Layer.effect(
   Service,
   Effect.gen(function* () {
     const store = yield* SessionStore.Service
-    const locations = yield* LocationServiceMap.Service
+    const instances = yield* SessionInstance.Service
     const bus = yield* Bus.Service
     const jobs = yield* Job.Service
     const db = (yield* Database.Service).db
@@ -88,7 +90,7 @@ export const layer = Layer.effect(
       const result = yield* SessionRunner.Service.use((runner) =>
         runner.drain({ sessionID, force, continuation, promotable }),
       ).pipe(
-        Effect.provide(locations.get(session.location)),
+        Effect.provide(instances.get(session)),
         Effect.tapCause((cause) =>
           Cause.hasInterruptsOnly(cause)
             ? Effect.void
@@ -160,9 +162,17 @@ export const layer = Layer.effect(
             yield* coordinator.wake(sessionID, "steer")
           return interrupted
         }),
-      resume: coordinator.run,
-      wake: coordinator.wake,
+      resume: (sessionID) => instances.check(sessionID).pipe(Effect.andThen(coordinator.run(sessionID))),
+      wake: (sessionID) => instances.check(sessionID).pipe(Effect.andThen(coordinator.wake(sessionID))),
       awaitIdle: coordinator.awaitIdle,
+      shutdown: (sessionIDs) =>
+        coordinator
+          .interruptAll(sessionIDs, "shutdown")
+          .pipe(
+            Effect.andThen(
+              Effect.forEach(sessionIDs, coordinator.awaitIdle, { concurrency: "unbounded", discard: true }),
+            ),
+          ),
     })
   }),
 )
@@ -170,7 +180,7 @@ export const layer = Layer.effect(
 export const node = makeGlobalNode({
   service: Service,
   layer,
-  deps: [SessionStore.node, LocationServiceMap.node, Bus.node, Database.node, Job.node],
+  deps: [SessionStore.node, SessionInstance.node, Bus.node, Database.node, Job.node],
 })
 
 /** Low-level compatibility layer for callers that only need durable Session recording. */
@@ -182,5 +192,6 @@ export const noopLayer = Layer.succeed(
     wake: () => Effect.void,
     interrupt: () => Effect.succeed(false),
     awaitIdle: () => Effect.void,
+    shutdown: () => Effect.void,
   }),
 )

packages/core/src/session/instance.ts

@@ -0,0 +1,44 @@
+export * as SessionInstance from "./instance.js"
+
+import { Context, Effect, Layer, RcMap } from "effect"
+import { makeGlobalNode } from "@opencode-ai/util/effect/app-node"
+import type { Instance } from "../instance.js"
+import { Location } from "../location.js"
+import { LocationServiceMap } from "../location-service-map.js"
+import { SessionModelTransport } from "./model-transport.js"
+import { SessionSchema } from "./schema.js"
+
+/** Selects capabilities without owning Session admission or execution coordination. */
+export interface Interface {
+  readonly get: (session: SessionSchema.Info) => Layer.Layer<Instance.Services, Instance.Error>
+  readonly check: (sessionID: SessionSchema.ID) => Effect.Effect<void>
+  readonly closeTransport: (session: SessionSchema.Info) => Effect.Effect<void>
+  readonly destination: (ref: Location.Ref) => Layer.Layer<Location.Service, Instance.Error>
+}
+
+export class Service extends Context.Service<Service, Interface>()("@opencode/SessionInstance") {}
+
+/** The server keeps sharing a graph for each canonical Location. */
+export const layer = Layer.effect(
+  Service,
+  Effect.gen(function* () {
+    const locations = yield* LocationServiceMap.Service
+    return Service.of({
+      get: (session) => locations.get(session.location),
+      check: () => Effect.void,
+      destination: (ref) => locations.get(ref),
+      closeTransport: Effect.fn("SessionInstance.closeTransport")(function* (session) {
+        const ref = Location.Ref.make({
+          directory: session.location.directory,
+          workspaceID: session.location.workspaceID,
+        })
+        if (!(yield* RcMap.has(locations.rcMap, ref))) return
+        yield* SessionModelTransport.Service.use((transport) => transport.close(session.id)).pipe(
+          Effect.provide(locations.get(ref)),
+        )
+      }),
+    })
+  }),
+)
+
+export const node = makeGlobalNode({ service: Service, layer, deps: [LocationServiceMap.node] })

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

@@ -18,6 +18,8 @@ export interface Coordinator<Key, E, Reason = never> {
    * interrupted. Compose with `awaitIdle` for settlement.
    */
   readonly interrupt: (key: Key, reason?: Reason) => Effect.Effect<boolean>
+  /** Marks the whole ownership chain before signaling any of its fibers. */
+  readonly interruptAll: (keys: Iterable<Key>, reason?: Reason) => 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>
 }
@@ -135,28 +137,42 @@ export const make = <Key, E, Reason = never>(options: {
         start(key, false, scope)
       })
 
+    const stop = (key: Key, reason?: Reason) => {
+      const execution = executions.get(key)
+      if (execution === undefined || execution.stopping) return undefined
+      if (execution.owner === undefined) {
+        // Settlement window: the owner exited but the settled hook has not finished. The
+        // terminal outcome is already decided, so no reason attaches — but the interrupt
+        // still claims the recorded wakes so settle does not start a dead-intent successor.
+        execution.pendingWake = undefined
+        return undefined
+      }
+      execution.stopping = true
+      // Wakes recorded so far belong to the interrupted intent; the interrupt claims them.
+      // Wakes arriving during cleanup are new admissions and restart normally at settle.
+      execution.pendingWake = undefined
+      execution.interruptionReason = reason
+      return execution.owner
+    }
+
     const interrupt = (key: Key, reason?: Reason): Effect.Effect<boolean> =>
       Effect.sync(() => {
-        const execution = executions.get(key)
-        if (execution === undefined || execution.stopping) return false
-        if (execution.owner === undefined) {
-          // Settlement window: the owner exited but the settled hook has not finished. The
-          // terminal outcome is already decided, so no reason attaches — but the interrupt
-          // still claims the recorded wakes so settle does not start a dead-intent successor.
-          execution.pendingWake = undefined
-          return false
-        }
-        execution.stopping = true
-        // Wakes recorded so far belong to the interrupted intent; the interrupt claims them.
-        // Wakes arriving during cleanup are new admissions and restart normally at settle.
-        execution.pendingWake = undefined
-        execution.interruptionReason = reason
+        const owner = stop(key, reason)
+        if (owner === undefined) return false
         // Fire and forget: nobody benefits from waiting out cleanup here, and callers like
         // the interrupt endpoint must acknowledge immediately even when finalizers are slow.
-        fork(Fiber.interrupt(execution.owner))
+        fork(Fiber.interrupt(owner))
         return true
       })
 
+    const interruptAll = (keys: Iterable<Key>, reason?: Reason) =>
+      Effect.sync(() => {
+        Array.from(keys)
+          .map((key) => stop(key, reason))
+          .filter((owner) => owner !== undefined)
+          .forEach((owner) => fork(Fiber.interrupt(owner)))
+      })
+
     // One execution's `done` already spans coalesced continuations; re-check after it
     // settles to cover a successor execution started by a late doorbell.
     const awaitIdle = (key: Key): Effect.Effect<void> =>
@@ -166,5 +182,5 @@ export const make = <Key, E, Reason = never>(options: {
         return Deferred.await(execution.done).pipe(Effect.ignoreCause, Effect.andThen(awaitIdle(key)))
       })
 
-    return { active: Effect.sync(() => new Set(executions.keys())), run, wake, interrupt, awaitIdle }
+    return { active: Effect.sync(() => new Set(executions.keys())), run, wake, interrupt, interruptAll, awaitIdle }
   })

packages/core/src/shared.ts

@@ -0,0 +1,85 @@
+export * as Shared from "./shared.js"
+
+import path from "node:path"
+import { Context, Effect, Layer, Scope } from "effect"
+import { LayerNode } from "@opencode-ai/util/effect/layer-node"
+import { App } from "./app.js"
+import { Database } from "./database/database.js"
+import { Instance } from "./instance.js"
+import { Job } from "./job.js"
+import { PersistentPty } from "./persistent-pty.js"
+import { Session } from "./session.js"
+import { SessionBindings } from "./session/bindings.js"
+import { SessionExecution } from "./session/execution.js"
+import { SessionInstance } from "./session/instance.js"
+
+export interface Options<Items extends Replacements = Replacements> {
+  readonly database?: Database.Options
+  readonly app?: Partial<App.Info>
+  readonly replacements?: LayerNode.ComposableReplacements<Items>
+}
+
+/** Ready constructors accept only fully wired, infallible replacements. */
+export type Replacements = readonly (readonly [
+  LayerNode.Node<unknown, unknown, LayerNode.Tag | undefined>,
+  LayerNode.Node<unknown, never, LayerNode.Tag | undefined> | Layer.Layer<never>,
+])[]
+
+export interface Interface {
+  readonly scope: Scope.Scope
+  readonly globals: Context.Context<Instance.Globals>
+  readonly sessions: Session.Interface
+  readonly bindings: SessionBindings.Interface
+  readonly execution: SessionExecution.Interface
+  readonly jobs: Job.Interface
+  readonly persistentPty: PersistentPty.Interface
+  readonly replacements: Replacements
+}
+
+/** Supplied once by the host; contains no location map or embedded HTTP server. */
+export class Service extends Context.Service<Service, Interface>()("@opencode/Shared") {}
+
+export function layer<const Items extends Replacements = readonly []>(options: Options<Items> = {}) {
+  const replacements: Replacements = [
+    [
+      Database.node,
+      Database.configured({
+        path:
+          options.database?.path && options.database.path !== ":memory:"
+            ? path.resolve(options.database.path)
+            : ":memory:",
+      }),
+    ],
+    [App.node, App.configured(options.app)],
+    ...(options.replacements ?? []),
+    [SessionInstance.node, SessionBindings.instanceNode],
+  ]
+  const configured: LayerNode.Replacements = replacements
+  return LayerNode.compile(
+    LayerNode.group([
+      Instance.globalsGraph,
+      Session.node,
+      SessionBindings.node,
+      SessionExecution.node,
+      Job.node,
+      PersistentPty.node,
+    ]),
+    configured,
+  ).pipe(
+    Layer.flatMap((context) =>
+      Layer.effect(
+        Service,
+        Effect.map(Scope.Scope, (scope) => ({
+          scope,
+          globals: context,
+          sessions: Context.get(context, Session.Service),
+          bindings: Context.get(context, SessionBindings.Service),
+          execution: Context.get(context, SessionExecution.Service),
+          jobs: Context.get(context, Job.Service),
+          persistentPty: Context.get(context, PersistentPty.Service),
+          replacements,
+        })),
+      ),
+    ),
+  )
+}

packages/core/src/shell.ts

@@ -103,7 +103,7 @@ const cleanupLayer = Layer.effectDiscard(
   cleanup().pipe(Effect.repeat(Schedule.spaced(Duration.hours(1))), Effect.forkScoped),
 )
 
-const cleanupNode = makeGlobalNode({
+export const cleanupNode = makeGlobalNode({
   name: "shell-output-cleanup",
   layer: cleanupLayer,
   deps: [FSUtil.node, Global.node],

packages/core/src/tool-output.ts

@@ -142,7 +142,7 @@ const cleanupLayer = Layer.effectDiscard(
   }),
 )
 
-const cleanupNode = makeGlobalNode({
+export const cleanupNode = makeGlobalNode({
   name: "tool-output-cleanup",
   layer: cleanupLayer,
   deps: [FSUtil.node, Global.node],

packages/core/test/bus-capture.test.ts

@@ -0,0 +1,186 @@
+import { describe, expect } from "bun:test"
+import { Context, Effect, Fiber, Scope, Stream } from "effect"
+import { Bus } from "@opencode-ai/core/bus"
+import { Database } from "@opencode-ai/core/database/database"
+import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder"
+import { Location } from "@opencode-ai/core/location"
+import { Credential } from "@opencode-ai/schema/credential"
+import { Event } from "@opencode-ai/schema/event"
+import { FileSystem } from "@opencode-ai/schema/filesystem"
+import { IntegrationID } from "@opencode-ai/schema/integration-id"
+import { McpEvent } from "@opencode-ai/schema/mcp-event"
+import { Plugin } from "@opencode-ai/schema/plugin"
+import { AbsolutePath } from "@opencode-ai/schema/schema"
+import { SessionEvent } from "@opencode-ai/schema/session-event"
+import { SessionID } from "@opencode-ai/schema/session-id"
+import { VcsEvent } from "@opencode-ai/schema/vcs-event"
+import { LayerNode } from "@opencode-ai/util/effect/layer-node"
+import { location } from "./fixture/location"
+import { testEffect } from "./lib/effect"
+
+const it = testEffect(
+  AppNodeBuilder.build(LayerNode.group([Database.node, Bus.node]), [[Bus.node, Bus.configured({ persist: true })]]),
+)
+const here = Location.Ref.make({ directory: AbsolutePath.make("/capture") })
+const elsewhere = Location.Ref.make({ directory: AbsolutePath.make("/elsewhere") })
+
+describe("Bus.capture", () => {
+  ;(["wildcard", "typed", "multiple"] as const).forEach((mode) => {
+    it.effect(`restores private ownership for ${mode} streams in foreign and trimmed contexts`, () =>
+      Effect.gen(function* () {
+        const root = yield* Bus.Service
+        const owner = Symbol()
+        const first = Bus.capture(root, Symbol())
+        const second = Bus.capture(root, owner)
+        const foreign = (yield* Effect.context<Scope.Scope>()).pipe(Context.add(Bus.PrivateOwner, owner))
+        const trimmed = foreign.pipe(Context.pick(Scope.Scope))
+        const doneID = Event.ID.create()
+        const watch = (bus: Bus.Interface, context: Context.Context<Scope.Scope>) => {
+          const stream =
+            mode === "wildcard"
+              ? bus.subscribe()
+              : mode === "typed"
+                ? bus.subscribe(McpEvent.ToolsChanged)
+                : bus.subscribe([McpEvent.ToolsChanged, Plugin.Event.Added])
+          return stream.pipe(
+            Stream.takeUntil((event) => event.id === doneID),
+            Stream.runCollect,
+            Effect.setContext(context),
+            Effect.forkScoped({ startImmediately: true }),
+          )
+        }
+        expect(Context.get(trimmed, Bus.PrivateOwner)).toBeUndefined()
+        const inForeign = yield* watch(first, foreign)
+        const inTrimmed = yield* watch(first, trimmed)
+        const other = yield* watch(second, trimmed)
+        const shared = yield* watch(root, trimmed)
+
+        const one = yield* first.publish(McpEvent.ToolsChanged, { server: "foreign" }).pipe(Effect.setContext(foreign))
+        const two = yield* first.publish(McpEvent.ToolsChanged, { server: "trimmed" }).pipe(Effect.setContext(trimmed))
+        const added = yield* first.publish(Plugin.Event.Added, { id: Plugin.ID.make("capture-plugin") })
+        const privateOther = yield* second.publish(McpEvent.ToolsChanged, { server: "other" })
+        const unowned = yield* root.publish(McpEvent.ToolsChanged, { server: "shared" })
+        const done = yield* root.publish(McpEvent.ToolsChanged, { server: "done" }, { id: doneID, global: true })
+
+        const expected = mode === "typed" ? [one, two, done] : [one, two, added, done]
+        expect(Array.from(yield* Fiber.join(inForeign))).toEqual(expected)
+        expect(Array.from(yield* Fiber.join(inTrimmed))).toEqual(expected)
+        expect(Array.from(yield* Fiber.join(other))).toEqual([privateOther, done])
+        expect(Array.from(yield* Fiber.join(shared))).toEqual([unowned, done])
+        expect(Object.keys(one).sort()).toEqual(["created", "data", "id", "type"])
+      }),
+    )
+  })
+
+  it.effect("honors explicit global audiences and leaves credential notifications shared", () =>
+    Effect.gen(function* () {
+      const root = yield* Bus.Service
+      const first = Bus.capture(root, Symbol())
+      const second = Bus.capture(root, Symbol())
+      const doneID = Event.ID.create()
+      const watchers = yield* Effect.forEach([first, second, root], (bus, index) =>
+        bus.subscribe().pipe(
+          Stream.takeUntil((event) => event.id === doneID),
+          Stream.runCollect,
+          Effect.provideService(Location.Service, location(index === 0 ? here : elsewhere)),
+          Effect.forkScoped({ startImmediately: true }),
+        ),
+      )
+      const broadcast = yield* first
+        .publish(McpEvent.ToolsChanged, { server: "global" }, { global: true, location: here })
+        .pipe(Effect.provideService(Location.Service, location(here)))
+      const updated = yield* first.publish(Credential.Event.Updated, {})
+      const switched = yield* second.publish(Credential.Event.Switched, {
+        integrationID: IntegrationID.make("capture-integration"),
+        credentialID: null,
+      })
+      const done = yield* root.publish(McpEvent.ToolsChanged, { server: "done" }, { id: doneID, global: true })
+
+      expect(broadcast).not.toHaveProperty("location")
+      yield* Effect.forEach(watchers, (fiber) =>
+        Fiber.join(fiber).pipe(
+          Effect.tap((events) =>
+            Effect.sync(() => expect(Array.from(events)).toEqual([broadcast, updated, switched, done])),
+          ),
+        ),
+      )
+    }),
+  )
+
+  it.effect("keeps filesystem and VCS notifications placement-scoped rather than private", () =>
+    Effect.gen(function* () {
+      const root = yield* Bus.Service
+      const first = Bus.capture(root, Symbol())
+      const second = Bus.capture(root, Symbol())
+      const doneID = Event.ID.create()
+      const watch = (bus: Bus.Interface, ref: Location.Ref) =>
+        bus.subscribe().pipe(
+          Stream.takeUntil((event) => event.id === doneID),
+          Stream.runCollect,
+          Effect.provideService(Location.Service, location(ref)),
+          Effect.forkScoped({ startImmediately: true }),
+        )
+      const local = yield* watch(first, here)
+      const colocated = yield* watch(second, here)
+      const remote = yield* watch(second, elsewhere)
+      const shared = yield* watch(root, here)
+      const changed = yield* first.publish(
+        FileSystem.Event.Changed,
+        { file: "/capture/file", event: "change" },
+        { location: here },
+      )
+      const branch = yield* first.publish(VcsEvent.BranchUpdated, { branch: "capture-branch" }, { location: here })
+      const privateEvent = yield* first.publish(McpEvent.ToolsChanged, { server: "private" }, { location: here })
+      const wrongLocation = yield* first.publish(
+        McpEvent.ToolsChanged,
+        { server: "elsewhere" },
+        { location: elsewhere },
+      )
+      const done = yield* root.publish(McpEvent.ToolsChanged, { server: "done" }, { id: doneID, global: true })
+
+      expect(Array.from(yield* Fiber.join(local))).toEqual([changed, branch, privateEvent, done])
+      expect(Array.from(yield* Fiber.join(colocated))).toEqual([changed, branch, done])
+      expect(Array.from(yield* Fiber.join(remote))).toEqual([done])
+      expect(Array.from(yield* Fiber.join(shared))).toEqual([changed, branch, done])
+      expect(wrongLocation.location).toEqual(elsewhere)
+    }),
+  )
+
+  it.effect("delegates durable authority and Session audiences to the shared root", () =>
+    Effect.gen(function* () {
+      const root = yield* Bus.Service
+      const first = Bus.capture(root, Symbol())
+      const second = Bus.capture(root, Symbol())
+      ;(["publishAll", "observe", "project", "replay", "log", "claim", "remove", "listen"] as const).forEach((key) => {
+        expect(first[key]).toBe(root[key])
+        expect(second[key]).toBe(root[key])
+      })
+      const sessionID = SessionID.create()
+      const observer = yield* second.observe(sessionID)
+      const watchers = yield* Effect.forEach([first, second, root], (bus) =>
+        bus
+          .subscribe(SessionEvent.Renamed)
+          .pipe(Stream.take(4), Stream.runCollect, Effect.forkScoped({ startImmediately: true })),
+      )
+      const one = yield* first.publish(SessionEvent.Renamed, { sessionID, title: "first" })
+      const batch = yield* second.publishAll([
+        [SessionEvent.Renamed, { sessionID, title: "second" }],
+        [SessionEvent.Renamed, { sessionID, title: "third" }],
+      ])
+      const last = yield* root.publish(SessionEvent.Renamed, { sessionID, title: "fourth" })
+      const events = [one, ...batch, last]
+
+      expect(events.map((event) => event.durable.seq)).toEqual([0, 1, 2, 3].map((seq) => Event.Seq.make(seq)))
+      expect(Array.from(yield* observer.pipe(Stream.take(4), Stream.runCollect))).toEqual(events)
+      yield* Effect.forEach(watchers, (fiber) =>
+        Fiber.join(fiber).pipe(
+          Effect.tap((received) => Effect.sync(() => expect(Array.from(received)).toEqual(events))),
+        ),
+      )
+      expect(Array.from(yield* first.log({ aggregateID: sessionID }).pipe(Stream.runCollect))).toEqual([
+        ...events,
+        { type: "log.synced", aggregateID: sessionID, seq: Event.Seq.make(3) },
+      ])
+    }),
+  )
+})

packages/core/test/bus-observe.test.ts

@@ -0,0 +1,126 @@
+import { describe, expect } from "bun:test"
+import { Effect, Exit, Fiber, Scope, Stream } from "effect"
+import { Bus } from "@opencode-ai/core/bus"
+import { Database } from "@opencode-ai/core/database/database"
+import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder"
+import { Location } from "@opencode-ai/core/location"
+import { Event } from "@opencode-ai/schema/event"
+import { Permission } from "@opencode-ai/schema/permission"
+import { AbsolutePath } from "@opencode-ai/schema/schema"
+import { SessionEvent } from "@opencode-ai/schema/session-event"
+import { SessionID } from "@opencode-ai/schema/session-id"
+import { SessionMessage } from "@opencode-ai/schema/session-message"
+import { LayerNode } from "@opencode-ai/util/effect/layer-node"
+import { location } from "./fixture/location"
+import { testEffect } from "./lib/effect"
+
+const it = testEffect(AppNodeBuilder.build(LayerNode.group([Database.node, Bus.node])))
+const here = Location.Ref.make({ directory: AbsolutePath.make("/observer") })
+const elsewhere = Location.Ref.make({ directory: AbsolutePath.make("/publisher") })
+
+describe("Bus.observe", () => {
+  it.effect("acquires before consumption and filters exact Session events without Location or owner restrictions", () =>
+    Effect.gen(function* () {
+      const root = yield* Bus.Service
+      const first = Bus.capture(root, Symbol())
+      const second = Bus.capture(root, Symbol())
+      const sessionID = SessionID.create()
+      yield* second.publish(SessionEvent.Renamed, { sessionID, title: "before observation" }, { location: elsewhere })
+      const observer = yield* first.observe(sessionID).pipe(Effect.provideService(Location.Service, location(here)))
+
+      yield* second.publish(SessionEvent.Renamed, { sessionID: SessionID.create(), title: "other Session" })
+      yield* second.publish(Permission.Event.Asked, {
+        id: Permission.ID.create(),
+        sessionID,
+        action: "read",
+        resources: ["file"],
+      })
+      const renamed = yield* second.publish(
+        SessionEvent.Renamed,
+        { sessionID, title: "observed" },
+        { location: elsewhere },
+      )
+      const delta = yield* second.publish(
+        SessionEvent.Text.Delta,
+        {
+          sessionID,
+          assistantMessageID: SessionMessage.ID.create(),
+          ordinal: 0,
+          delta: "queued before consumption",
+        },
+        { location: elsewhere },
+      )
+
+      const events = yield* observer.pipe(
+        Stream.take(2),
+        Stream.runCollect,
+        Effect.provideService(Location.Service, location(here)),
+        Effect.provideService(Bus.PrivateOwner, Symbol()),
+      )
+      expect(Array.from(events)).toEqual([renamed, delta])
+      expect(renamed.durable.seq).toBe(Event.Seq.make(1))
+      expect(delta).not.toHaveProperty("durable")
+    }),
+  )
+
+  it.effect("lets consumer callbacks publish to the same aggregate without deadlocking publication", () =>
+    Effect.gen(function* () {
+      const bus = yield* Bus.Service
+      const sessionID = SessionID.create()
+      const observer = yield* bus.observe(sessionID)
+      const received: SessionEvent.Event[] = []
+      const consumer = yield* observer.pipe(
+        Stream.take(2),
+        Stream.runForEach((event) =>
+          Effect.gen(function* () {
+            received.push(event)
+            if (event.type === SessionEvent.Renamed.type && event.data.title === "before") {
+              yield* bus.publish(SessionEvent.Renamed, { sessionID, title: "after" })
+            }
+          }),
+        ),
+        Effect.forkScoped({ startImmediately: true }),
+      )
+      yield* bus.publish(SessionEvent.Renamed, { sessionID, title: "before" })
+      yield* Fiber.join(consumer)
+
+      expect(received.map((event) => ("durable" in event ? event.durable.seq : undefined))).toEqual([
+        Event.Seq.make(0),
+        Event.Seq.make(1),
+      ])
+      expect(
+        received.map((event) => (event.type === SessionEvent.Renamed.type ? event.data.title : undefined)),
+      ).toEqual(["before", "after"])
+    }),
+  )
+
+  it.effect("disposes an unconsumed subscription with its acquiring Scope without closing the shared bus", () =>
+    Effect.gen(function* () {
+      const bus = yield* Bus.Service
+      const sessionID = SessionID.create()
+      const owner = yield* Scope.Scope
+      const scope = yield* Scope.fork(owner)
+      const observer = yield* bus.observe(sessionID).pipe(Scope.provide(scope))
+      const survivor = yield* bus.observe(sessionID)
+      const before = yield* bus.publish(SessionEvent.Renamed, { sessionID, title: "before disposal" })
+      yield* Scope.close(scope, Exit.void)
+      const after = yield* bus.publish(SessionEvent.Renamed, { sessionID, title: "after disposal" })
+
+      expect(Array.from(yield* observer.pipe(Stream.runCollect))).toEqual([])
+      expect(Array.from(yield* survivor.pipe(Stream.take(2), Stream.runCollect))).toEqual([before, after])
+    }),
+  )
+
+  it.effect("ends a blocked consumer when the acquiring Scope closes", () =>
+    Effect.gen(function* () {
+      const bus = yield* Bus.Service
+      const owner = yield* Scope.Scope
+      const scope = yield* Scope.fork(owner)
+      const observer = yield* bus.observe(SessionID.create()).pipe(Scope.provide(scope))
+      const consumer = yield* observer.pipe(Stream.runCollect, Effect.forkScoped({ startImmediately: true }))
+      yield* Scope.close(scope, Exit.void)
+
+      expect(Array.from(yield* Fiber.join(consumer))).toEqual([])
+    }),
+  )
+})

packages/core/test/instance-bus-capture.test.ts

@@ -0,0 +1,98 @@
+import { describe, expect } from "bun:test"
+import { Context, Deferred, Effect, Fiber, Layer, Stream } from "effect"
+import { Bus } from "@opencode-ai/core/bus"
+import { Watcher } from "@opencode-ai/core/filesystem/watcher"
+import { Instance } from "@opencode-ai/core/instance"
+import { Location } from "@opencode-ai/core/location"
+import { Mcp } from "@opencode-ai/core/mcp/index"
+import { ModelsDev } from "@opencode-ai/core/models-dev"
+import { Plugin } from "@opencode-ai/core/plugin"
+import { InstancePlugins } from "@opencode-ai/core/plugin/instance"
+import { PluginSupervisor } from "@opencode-ai/core/plugin/supervisor"
+import { Credential } from "@opencode-ai/schema/credential"
+import { Event } from "@opencode-ai/schema/event"
+import { EventManifest } from "@opencode-ai/schema/event-manifest"
+import { AbsolutePath } from "@opencode-ai/schema/schema"
+import { LayerNode } from "@opencode-ai/util/effect/layer-node"
+import { Global } from "@opencode-ai/util/global"
+import { tempGlobalLayer } from "./fixture/global"
+import { tmpdirScoped } from "./fixture/tmpdir"
+import { it } from "./lib/effect"
+
+describe("Instance Bus capture", () => {
+  it.live("isolates real plugin and MCP notifications across same-directory instances", () =>
+    Effect.gen(function* () {
+      const directory = yield* tmpdirScoped()
+      const globals = yield* Layer.build(
+        LayerNode.compile(Instance.globalsGraph, [
+          [Global.node, tempGlobalLayer],
+          [ModelsDev.node, ModelsDev.configured({ fetch: false })],
+          [Watcher.node, Watcher.configured({ enabled: false })],
+        ]),
+      )
+      const root = Context.get(globals, Bus.Service)
+      const doneID = Event.ID.create()
+      const ids = ["capture-first", "capture-second"]
+      const received: EventManifest.ServerEvent[][] = [[], []]
+      const completed = yield* Effect.forEach(ids, () => Deferred.make<void>())
+      const selected = (event: EventManifest.ServerEvent) =>
+        (event.type === "plugin.added" && event.data.id.startsWith("capture-")) ||
+        event.type === "mcp.status.changed" ||
+        event.type === "credential.updated"
+      const shared = yield* root.subscribe().pipe(
+        Stream.filter(EventManifest.isServer),
+        Stream.filter(selected),
+        Stream.takeUntil((event) => event.id === doneID),
+        Stream.runCollect,
+        Effect.forkScoped({ startImmediately: true }),
+      )
+      const ref = Location.Ref.make({ directory: AbsolutePath.make(directory.path) })
+      const instances = yield* Effect.forEach(ids, (id, index) => {
+        const probe: InstancePlugins.List[number] = {
+          id,
+          effect: (ctx) =>
+            ctx.event.subscribe().pipe(
+              Stream.filter(EventManifest.isServer),
+              Stream.filter(selected),
+              Stream.takeUntil((event) => event.id === doneID),
+              Stream.runForEach((event) => Effect.sync(() => received[index].push(event))),
+              Effect.andThen(Deferred.succeed(completed[index], undefined)),
+              Effect.forkScoped({ startImmediately: true }),
+              Effect.asVoid,
+            ),
+        }
+        return Layer.build(
+          Instance.compose(ref, { discovery: false, plugins: [probe] }).pipe(
+            Layer.provide(Layer.succeedContext(globals)),
+          ),
+        )
+      })
+      yield* Effect.forEach(instances, (instance, index) =>
+        Effect.gen(function* () {
+          yield* Context.get(instance, PluginSupervisor.Service).flush
+          expect(
+            (yield* Context.get(instance, Plugin.Service).list()).find((plugin) => plugin.id === ids[index])?.status,
+          ).toBe("active")
+        }),
+      )
+      yield* Effect.forEach(instances, (instance, index) =>
+        Context.get(instance, Mcp.Service).transform((draft) =>
+          draft.set(ids[index], { type: "local", command: ["unused"], disabled: true }),
+        ),
+      )
+      const done = yield* root.publish(Credential.Event.Updated, {}, { id: doneID, global: true })
+      yield* Effect.forEach(completed, Deferred.await)
+
+      expect(Array.from(yield* Fiber.join(shared))).toEqual([done])
+      received.forEach((events, index) =>
+        expect(
+          events.map((event) => {
+            if (event.type === "plugin.added") return [event.type, event.data.id]
+            if (event.type === "mcp.status.changed") return [event.type, event.data.server]
+            return [event.type]
+          }),
+        ).toEqual([["plugin.added", ids[index]], ["mcp.status.changed", ids[index]], ["credential.updated"]]),
+      )
+    }),
+  )
+})

packages/core/test/instance-compose-types.test.ts

@@ -0,0 +1,32 @@
+import { test } from "bun:test"
+import { Context, Effect, Layer } from "effect"
+import { Image } from "../src/image"
+import { Instance } from "../src/instance"
+import { Location } from "../src/location"
+import { AbsolutePath } from "../src/schema"
+
+class External extends Context.Service<External, string>()("test/InstanceExternal") {}
+class BootError {
+  readonly _tag = "InstanceBootError"
+}
+
+const check = () => {
+  const ref = Location.Ref.make({ directory: AbsolutePath.make("/") })
+  const open = Instance.compose(ref)
+  const honest: Layer.Layer<Instance.Services, Instance.Error, Instance.Globals> = open
+  // @ts-expect-error Shared infrastructure is required, not secretly booted.
+  const closed: Layer.Layer<Instance.Services, Instance.Error> = open
+  const replacement = Layer.effect(Image.Service, External.pipe(Effect.andThen(Effect.fail(new BootError()))))
+  const advanced = Instance.compose(ref, { replacements: [[Image.node, replacement]] })
+  const requirements: Layer.Layer<Instance.Services, Instance.Error | BootError, Instance.Globals | External> = advanced
+  // @ts-expect-error Raw replacement layers retain their external requirements.
+  const missing: Layer.Layer<Instance.Services, Instance.Error | BootError, Instance.Globals> = advanced
+  // @ts-expect-error Raw replacement layers retain their acquisition errors.
+  const errors: Layer.Layer<Instance.Services, Instance.Error, Instance.Globals | External> = advanced
+  // @ts-expect-error Raw replacements must still provide the original service.
+  Instance.compose(ref, { replacements: [[Image.node, Layer.succeed(External, "wrong output")]] })
+  void [honest, closed, requirements, missing, errors]
+}
+void check
+
+test("instance composition types compile", () => {})

packages/core/test/instance-compose.test.ts

@@ -0,0 +1,182 @@
+import { describe, expect, test } from "bun:test"
+import { Context, Effect, Exit, Layer, Option, Scope } from "effect"
+import { Node } from "@opencode-ai/util/effect/app-node"
+import { LayerNode } from "@opencode-ai/util/effect/layer-node"
+import { Global } from "@opencode-ai/util/global"
+import { Plugin } from "@opencode-ai/plugin/effect"
+import { Agent } from "../src/agent"
+import { App } from "../src/app"
+import { Bus } from "../src/bus"
+import { Config } from "../src/config"
+import { Instance } from "../src/instance"
+import { InstructionDiscovery } from "../src/instruction-discovery"
+import { Location } from "../src/location"
+import { LocationServiceMap } from "../src/location-service-map"
+import { ModelsDev } from "../src/models-dev"
+import { InstancePlugins } from "../src/plugin/instance"
+import { PluginRuntime } from "../src/plugin/runtime"
+import { PluginSupervisor } from "../src/plugin/supervisor"
+import { Project } from "../src/project"
+import { AbsolutePath } from "../src/schema"
+import { Watcher } from "../src/filesystem/watcher"
+import { tempGlobalLayer } from "./fixture/global"
+import { tmpdirScoped } from "./fixture/tmpdir"
+import { it } from "./lib/effect"
+
+class Extra extends Context.Service<Extra, string>()("test/InstanceExtraGlobal") {}
+
+describe("Instance.compose", () => {
+  it.live("reuses configured globals across fresh, separately bound local graphs", () =>
+    Effect.gen(function* () {
+      const directory = yield* tmpdirScoped()
+      const acquired = { global: 0, app: 0, project: 0, runtime: 0 }
+      const released = { global: 0 }
+      const profile = [
+        [
+          Global.node,
+          Layer.effectContext(
+            Effect.gen(function* () {
+              acquired.global++
+              yield* Effect.addFinalizer(() => Effect.sync(() => released.global++))
+              return yield* Layer.build(tempGlobalLayer)
+            }),
+          ),
+        ],
+        [
+          App.node,
+          Layer.effect(
+            App.Metadata,
+            Effect.sync(() => {
+              acquired.app++
+              return App.make({ name: "compose-host", version: "test" })
+            }),
+          ),
+        ],
+        [ModelsDev.node, ModelsDev.configured({ fetch: false })],
+        [Watcher.node, Watcher.configured({ enabled: false })],
+      ] as const
+      const replacements = [
+        ...profile,
+        [
+          Project.node,
+          Layer.effectContext(
+            Effect.gen(function* () {
+              acquired.project++
+              return yield* Layer.build(LayerNode.compile(Project.node, profile))
+            }),
+          ),
+        ],
+      ] as const
+      const owner = yield* Effect.scope
+      const sharedScope = yield* Scope.fork(owner)
+      const memoMap = yield* Layer.makeMemoMap
+      const globals = yield* Layer.buildWithMemoMap(
+        LayerNode.compile(Instance.globalsGraph, replacements),
+        memoMap,
+        sharedScope,
+      )
+      expect(acquired).toEqual({ global: 1, app: 1, project: 1, runtime: 0 })
+      expect(Option.isNone(yield* Effect.serviceOption(LocationServiceMap.Service).pipe(Effect.provide(globals)))).toBe(
+        true,
+      )
+
+      const plugin = Plugin.define({
+        id: "compose-plugin",
+        effect: (ctx) => ctx.agent.transform((agents) => agents.update(Agent.ID.make("compose-agent"), () => {})),
+      })
+      const ref = Location.Ref.make({ directory: AbsolutePath.make(directory.path) })
+      const local = [
+        ...replacements,
+        [
+          Config.node,
+          Config.configured({ project: false, global: false, content: JSON.stringify({ shell: "compose-shell" }) }),
+        ],
+        [InstructionDiscovery.node, InstructionDiscovery.configured({ project: true, global: false })],
+        [
+          Location.node,
+          Location.boundNode(Location.Ref.make({ directory: AbsolutePath.make("/") }), { discovery: false }),
+        ],
+        [InstancePlugins.node, InstancePlugins.bound([plugin])],
+      ] as const
+      const build = (scope: Scope.Scope, plugins: InstancePlugins.List) =>
+        Layer.buildWithMemoMap(
+          Instance.compose(ref, {
+            discovery: false,
+            plugins,
+            replacements: [
+              ...local,
+              [
+                PluginRuntime.node,
+                Layer.effectContext(
+                  Effect.gen(function* () {
+                    acquired.runtime++
+                    return yield* Layer.build(PluginRuntime.layerWithCell(PluginRuntime.makeCell()))
+                  }),
+                ),
+              ],
+            ],
+          }).pipe(Layer.provide(Layer.succeedContext(globals))),
+          memoMap,
+          scope,
+        )
+      const firstScope = yield* Scope.fork(owner)
+      const secondScope = yield* Scope.fork(owner)
+      const first = yield* build(firstScope, [plugin])
+      const second = yield* build(secondScope, [])
+      yield* Context.get(first, PluginSupervisor.Service).flush
+      yield* Context.get(second, PluginSupervisor.Service).flush
+
+      expect(acquired).toEqual({ global: 1, app: 1, project: 1, runtime: 2 })
+      expect(Context.get(first, Location.Service).directory).toBe(ref.directory)
+      expect(Context.get(second, Location.Service).directory).toBe(ref.directory)
+      expect(Context.get(first, Config.Service)).not.toBe(Context.get(second, Config.Service))
+      expect(Context.get(first, Agent.Service)).not.toBe(Context.get(second, Agent.Service))
+      expect(Config.latest(yield* Context.get(first, Config.Service).entries(), "shell")).toBe("compose-shell")
+      expect(Context.get(first, InstructionDiscovery.Service).project).toBe(true)
+      expect(
+        Context.get(first, InstancePlugins.Service)
+          .all()
+          .map((item) => item.id),
+      ).toEqual([plugin.id])
+      expect(Context.get(second, InstancePlugins.Service).all()).toEqual([])
+      expect(yield* Context.get(first, Agent.Service).get(Agent.ID.make("compose-agent"))).toBeDefined()
+      expect(yield* Context.get(second, Agent.Service).get(Agent.ID.make("compose-agent"))).toBeUndefined()
+
+      yield* Scope.close(firstScope, Exit.void)
+      expect(released.global).toBe(0)
+      expect((yield* Context.get(second, Agent.Service).list()).length).toBeGreaterThan(0)
+      yield* Context.get(globals, Project.Service).list()
+      yield* Scope.close(secondScope, Exit.void)
+      yield* Scope.close(sharedScope, Exit.void)
+      expect(released.global).toBe(1)
+    }),
+  )
+
+  test("rejects globals introduced by a local replacement instead of acquiring them fresh", () => {
+    const extra = Node.makeGlobalNode({ service: Extra, layer: Layer.succeed(Extra, "extra"), deps: [] })
+    const discovery = Node.makeLocationNode({
+      service: InstructionDiscovery.Service,
+      layer: InstructionDiscovery.layer().pipe(Layer.tap(() => Extra)),
+      deps: [Bus.node, extra],
+    })
+    expect(() =>
+      Instance.compose(Location.Ref.make({ directory: AbsolutePath.make("/") }), {
+        replacements: [[InstructionDiscovery.node, discovery]],
+      }),
+    ).toThrow("Unsupported instance globals: test/InstanceExtraGlobal")
+  })
+
+  test("also checks shared dependencies of a per-instance runtime replacement", () => {
+    const extra = Node.makeGlobalNode({ service: Extra, layer: Layer.succeed(Extra, "extra"), deps: [] })
+    const runtime = Node.makeGlobalNode({
+      service: PluginRuntime.Service,
+      layer: PluginRuntime.layerWithCell(PluginRuntime.makeCell()).pipe(Layer.tap(() => Extra)),
+      deps: [extra],
+    })
+    expect(() =>
+      Instance.compose(Location.Ref.make({ directory: AbsolutePath.make("/") }), {
+        replacements: [[PluginRuntime.node, runtime]],
+      }),
+    ).toThrow("Unsupported instance globals: test/InstanceExtraGlobal")
+  })
+})

packages/core/test/session-execution.test.ts

@@ -17,6 +17,7 @@ import { SessionRestart } from "@opencode-ai/core/session/execution/restart"
 import { UserInterruptedError } from "@opencode-ai/core/session/error"
 import { SessionEvent } from "@opencode-ai/core/session/event"
 import { SessionInbox } from "@opencode-ai/core/session/inbox"
+import { SessionInstance } from "@opencode-ai/core/session/instance"
 import { SessionMessage } from "@opencode-ai/core/session/message"
 import { SessionRunner } from "@opencode-ai/core/session/runner/index"
 import { SessionInboxTable, SessionTable } from "@opencode-ai/core/session/sql"
@@ -1307,7 +1308,8 @@ function buildExecution(
     return yield* Layer.buildWithScope(
       SessionRestart.layer(options).pipe(
         Layer.provideMerge(sessionLayer),
-        Layer.provideMerge(Layer.fresh(SessionExecution.layer)),
+        // Capture the fixture's Location map instead of Session.node's memoized adapter.
+        Layer.provideMerge(Layer.fresh(SessionExecution.layer.pipe(Layer.provide(SessionInstance.layer)))),
         Layer.provide(Layer.succeed(Database.Service, database)),
         Layer.provide(Layer.succeed(Bus.Service, bus)),
         Layer.provide(Layer.succeed(SessionStore.Service, store)),

packages/core/test/session-message-update.test.ts

@@ -42,6 +42,7 @@ const it = testEffect(
             wake: () => Effect.void,
             interrupt: () => Effect.succeed(false),
             awaitIdle: () => Effect.void,
+            shutdown: () => Effect.void,
           }),
         ),
       ],

packages/core/test/session-prompt.test.ts

@@ -56,6 +56,7 @@ const execution = Layer.succeed(
         wakeCalls.push(sessionID)
       }),
     awaitIdle: () => Effect.void,
+    shutdown: () => Effect.void,
   }),
 )
 const locations = Layer.effect(

packages/core/test/session-runner-recorded.test.ts

@@ -129,6 +129,10 @@ const execution = (llmClient: Layer.Layer<typeof LLMClient.Service>) =>
         wake: coordinator.wake,
         interrupt: (sessionID) => coordinator.interrupt(sessionID),
         awaitIdle: coordinator.awaitIdle,
+        shutdown: (sessionIDs) =>
+          coordinator
+            .interruptAll(sessionIDs)
+            .pipe(Effect.andThen(Effect.forEach(sessionIDs, coordinator.awaitIdle, { discard: true }))),
       })
     }),
   ).pipe(Layer.provide(runnerLayer(llmClient)))

packages/core/test/session-runner.test.ts

@@ -449,6 +449,10 @@ const layer = Layer.unwrap(
           wake: coordinator.wake,
           interrupt: (sessionID) => coordinator.interrupt(sessionID),
           awaitIdle: coordinator.awaitIdle,
+          shutdown: (sessionIDs) =>
+            coordinator
+              .interruptAll(sessionIDs)
+              .pipe(Effect.andThen(Effect.forEach(sessionIDs, coordinator.awaitIdle, { discard: true }))),
         })
       }),
     ).pipe(Layer.provide(runnerLayer))

packages/core/test/tool-shell.test.ts

@@ -117,6 +117,7 @@ const executionNode = makeGlobalNode({
         wake: () => Effect.void,
         interrupt: () => Effect.succeed(false),
         awaitIdle: (id) => complete(id).pipe(Effect.exit, Effect.asVoid),
+        shutdown: () => Effect.void,
       })
     }),
   ),

packages/core/test/tool-subagent.test.ts

@@ -90,6 +90,7 @@ const executionNode = makeGlobalNode({
         wake: () => Effect.void,
         interrupt: () => Effect.succeed(false),
         awaitIdle: (sessionID) => complete(sessionID).pipe(Effect.exit, Effect.asVoid),
+        shutdown: () => Effect.void,
       })
     }),
   ),

packages/sdk/README.md

@@ -1,6 +1,6 @@
 # @opencode-ai/sdk
 
-In-process OpenCode host for Promise and Effect applications. The SDK executes Server's assembled HTTP router in memory, opening no listener and adding no network hop.
+In-process OpenCode for Promise and Effect applications. The existing `OpenCode` host executes Server's assembled HTTP router in memory, opening no listener and adding no network hop. The direct Effect entrypoint binds Session handles to private instances without constructing a router or location map.
 
 ```ts
 import { OpenCode } from "@opencode-ai/sdk"
@@ -70,3 +70,40 @@ const session = yield * opencode.sessions.get({ sessionID })
 ```
 
 The Effect Workerd entrypoint is `@opencode-ai/sdk/workerd/effect`.
+
+## Direct Effect Sessions
+
+Use `@opencode-ai/sdk/direct/effect` when the application owns Session lifetimes and supplies imported Effect plugins. Provide `Shared.layer` once, then create a private instance for each active conversation. Two handles at the same directory have independent tools, agents, config, and plugin registrations while sharing the configured database, credentials, and execution coordinator.
+
+```ts
+import { AbsolutePath, Location, Session, Shared } from "@opencode-ai/sdk/direct/effect"
+import { Effect } from "effect"
+import threadTools from "./thread-tools"
+
+const program = Effect.gen(function* () {
+  const session = yield* Session.create({
+    location: Location.Ref.make({ directory: AbsolutePath.make("/workspace") }),
+    plugins: [threadTools],
+  })
+
+  yield* session.events.subscribe((event) => Effect.log(event.type))
+  yield* session.prompt({ text: "Inspect the latest request" })
+  yield* session.wait()
+})
+
+await Effect.runPromise(
+  program.pipe(Effect.provide(Shared.layer({ database: { path: "./bot.sqlite" } })), Effect.scoped),
+)
+```
+
+`Session.create` waits for plugin and initial MCP-tool readiness but does not execute the Session. Filesystem discovery defaults to `false` on this entrypoint; set `discovery: true` to opt into ambient config and instructions. Imported plugins can configure capabilities without filesystem discovery. Runtime options and plugins are not persisted as Session metadata.
+
+Pass an existing `id` to adopt stored history. The saved title, model, metadata, and location win over retry arguments; location may be omitted when adopting. Install an observer before explicitly calling `session.resume()` to recover admitted work using the new instance's capabilities. A second live handle for the same Session fails with `Session.AlreadyBoundError` rather than replacing the first instance.
+
+`prompt` durably admits input before scheduling execution; `resume: false` admits only. `resume()` starts or joins execution, `interrupt()` requests user interruption, and `wait()` waits for local execution to settle without starting it. Child Sessions created by subagent tools inherit their parent's instance. Closing either the caller's Scope or the supplied Shared layer settles bound execution before disposing capabilities, preserves shutdown recovery claims, and makes retained handle methods fail with `Session.ClosedError`. Private instances close before shared backing is released. Keep the provided program and Scope open for the lifetime of an active thread, not just one inbound message.
+
+`events.subscribe(callback)` is ready before returning. It returns the observation fiber, whose failure can be observed by the caller, and disposes with either the subscription's Scope or the handle's Scope. Events are live and Session-specific, not a replay feed or an exactly-once outbound-delivery guarantee. Callbacks run outside publication transactions.
+
+Direct `resume()` drains the selected Session. `Shared.layer` does not run the server's automatic restart or background-Job recovery sweep. Automatic restoration of background Jobs after reopening handles is not part of this entrypoint yet; shutdown retains their durable recovery markers rather than turning them into user cancellations.
+
+`Shared.layer` defaults to an in-memory database. Explicit relative database paths are resolved against the application working directory. Advanced `replacements` on `Shared.layer` configure fully wired, infallible shared implementations; constructor replacements configure instance-local services. For lower-level Effect composition, Core's `Instance.compose` requires its explicit shared infrastructure and preserves replacement dependencies and errors. The initial direct handle is placement-bound and does not expose movement or a Promise facade. Existing embedded and Workerd entrypoints retain their behavior.

packages/sdk/package.json

@@ -18,6 +18,7 @@
   "exports": {
     ".": "./src/index.ts",
     "./effect": "./src/effect/index.ts",
+    "./direct/effect": "./src/direct/effect.ts",
     "./workerd": "./src/workerd.ts",
     "./workerd/effect": "./src/effect/workerd.ts"
   },

packages/sdk/script/verify-package.ts

@@ -87,6 +87,57 @@ try {
     JSON.stringify({ name: "opencode-sdk-consumer", private: true, type: "module" }),
   )
   await Promise.all([
+    Bun.write(
+      join(consumer, "direct.mjs"),
+      `import { AbsolutePath, Location, Session, Shared } from "@opencode-ai/sdk/direct/effect"
+import { ModelsDev } from "@opencode-ai/core/models-dev"
+import { Watcher } from "@opencode-ai/core/filesystem/watcher"
+import { Global } from "@opencode-ai/util/global"
+import { Deferred, Effect } from "effect"
+import { mkdtemp, rm } from "node:fs/promises"
+import { tmpdir } from "node:os"
+import { join } from "node:path"
+
+const directory = await mkdtemp(join(tmpdir(), "opencode-packed-direct-"))
+try {
+  await Effect.runPromise(Effect.gen(function* () {
+    const location = Location.Ref.make({ directory: AbsolutePath.make(directory) })
+    const first = yield* Session.create({ location, title: "Packed direct Session" })
+    const second = yield* Session.create({ location })
+    if (first.id === second.id) throw new Error("Private Sessions unexpectedly reused an ID")
+    const observed = yield* Deferred.make()
+    yield* first.events.subscribe((event) => event.type === "session.inbox.enqueued"
+      ? Deferred.succeed(observed, event.data.sessionID).pipe(Effect.asVoid)
+      : Effect.void)
+    const admitted = yield* first.prompt({ text: "Admit without model execution", resume: false })
+    if (admitted.sessionID !== first.id) throw new Error("Direct admission used the wrong Session")
+    const id = yield* Deferred.await(observed).pipe(Effect.timeout("5 seconds"))
+    if (id !== first.id) throw new Error("Direct observation used the wrong Session")
+    yield* first.wait()
+  }).pipe(Effect.provide(Shared.layer({
+    database: { path: join(directory, "direct.sqlite") },
+    replacements: [
+      [Global.node, Global.layerWith({
+        home: directory,
+        config: join(directory, "config"),
+        data: join(directory, "data"),
+        cache: join(directory, "cache"),
+        state: join(directory, "state"),
+        tmp: join(directory, "tmp"),
+        bin: join(directory, "bin"),
+        log: join(directory, "log"),
+        repos: join(directory, "repos"),
+      })],
+      [ModelsDev.node, ModelsDev.configured({ fetch: false })],
+      [Watcher.node, Watcher.configured({ enabled: false })],
+    ],
+  })), Effect.scoped))
+  console.log("packed direct Sessions OK")
+} finally {
+  await rm(directory, { recursive: true, force: true })
+}
+`,
+    ),
     Bun.write(
       join(consumer, "wrangler.jsonc"),
       JSON.stringify({
@@ -178,6 +229,7 @@ for (const module of modules) {
     throw new Error(`Packed SDK consumer resolved multiple Effect runtimes:\n${runtimes.join("\n")}`)
   }
   await $`bun imports.mjs`.cwd(consumer)
+  await $`bun direct.mjs`.cwd(consumer)
   await $`bun --conditions=workerd imports.mjs`.cwd(consumer)
   await $`node_modules/.bin/wrangler deploy --dry-run --config wrangler.jsonc --outdir dist`.cwd(consumer)
 

packages/sdk/src/direct/effect.ts

@@ -0,0 +1,4 @@
+export { DirectSession as Session } from "@opencode-ai/core/session/direct"
+export { Shared } from "@opencode-ai/core/shared"
+export { Location } from "@opencode-ai/schema/location"
+export { AbsolutePath } from "@opencode-ai/schema/schema"

packages/sdk/test/direct-session.test.ts

@@ -0,0 +1,985 @@
+import path from "node:path"
+import { describe, expect } from "bun:test"
+import { LLMClient, type LLMRequest } from "@opencode-ai/ai"
+import { TestLLM } from "@opencode-ai/ai/testing"
+import { Bus } from "@opencode-ai/core/bus"
+import { Database } from "@opencode-ai/core/database/database"
+import { llmClient } from "@opencode-ai/core/effect/app-node-platform"
+import { LocationWatcher } from "@opencode-ai/core/filesystem/location-watcher"
+import { Watcher } from "@opencode-ai/core/filesystem/watcher"
+import { LocationServiceMap } from "@opencode-ai/core/location-service-map"
+import { ModelResolver } from "@opencode-ai/core/model-resolver"
+import { ModelsDev } from "@opencode-ai/core/models-dev"
+import { PluginHooks } from "@opencode-ai/core/plugin/hooks"
+import { PluginRuntime } from "@opencode-ai/core/plugin/runtime"
+import { SessionStore } from "@opencode-ai/core/session/store"
+import { SessionTable } from "@opencode-ai/core/session/sql"
+import { Skill } from "@opencode-ai/core/skill"
+import { Tool } from "@opencode-ai/core/tool"
+import { SubagentTool } from "@opencode-ai/core/tool/plugin/subagent"
+import { Agent, Model, Plugin, Provider } from "@opencode-ai/plugin/effect"
+import { SessionEvent } from "@opencode-ai/schema/session-event"
+import { SessionMessage } from "@opencode-ai/schema/session-message"
+import { Global } from "@opencode-ai/util/global"
+import { makeGlobalNode } from "@opencode-ai/util/effect/app-node"
+import { Cause, Context, Deferred, Effect, Exit, Fiber, Layer, Option, Schema, Scope, Stream } from "effect"
+import { tempGlobalLayer } from "../../core/test/fixture/global"
+import { tmpdirScoped } from "../../core/test/fixture/tmpdir"
+import { testEffect } from "../../core/test/lib/effect"
+import { AbsolutePath, Location, Session, Shared } from "../src/direct/effect"
+
+const it = testEffect(Layer.empty)
+const model = Model.Ref.make({ providerID: Provider.ID.make("direct-test"), id: Model.ID.make("fictional-chat") })
+const modelPlugin = Plugin.define({
+  id: "direct-model",
+  effect: (ctx) =>
+    ctx.catalog.transform((catalog) => {
+      catalog.provider.update(model.providerID, (provider) => {
+        provider.activation = "enabled"
+        provider.package = "@opencode-ai/ai/providers/openai/chat"
+        provider.settings = { baseURL: "https://provider.example/v1" }
+      })
+      catalog.model.update(model.providerID, model.id, (draft) => {
+        draft.capabilities = { tools: true, input: ["text"], output: ["text"] }
+        draft.limit = { context: 100_000, output: 1_000 }
+      })
+      catalog.model.default.set(model.providerID, model.id)
+    }),
+})
+
+const reviewerPlugin = Plugin.define({
+  id: "direct-reviewer",
+  effect: (ctx) =>
+    ctx.agent.transform((agents) => {
+      agents.update("build", (agent) => {
+        agent.permissions.push({ action: "subagent", resource: "reviewer", effect: "allow" })
+      })
+      agents.update("reviewer", (agent) => {
+        agent.mode = "subagent"
+        agent.model = model
+      })
+    }),
+})
+
+const withShared = <A, E, R>(
+  body: (fixture: { readonly location: Location.Ref; readonly llm: TestLLM.TestInterface }) => Effect.Effect<A, E, R>,
+) =>
+  Effect.gen(function* () {
+    const directory = yield* tmpdirScoped("opencode-direct-session-")
+    // Acquire the client outside the fresh graphs so every Session uses these controls.
+    const llm = yield* TestLLM.Test.pipe(Effect.provide(TestLLM.testLayer()))
+    return yield* body({
+      location: Location.Ref.make({ directory: AbsolutePath.make(directory.path) }),
+      llm,
+    }).pipe(
+      Effect.provide(
+        Shared.layer({
+          database: { path: ":memory:" },
+          replacements: [
+            [Global.node, tempGlobalLayer],
+            [llmClient, Layer.succeed(LLMClient.Service, llm)],
+            [ModelsDev.node, ModelsDev.configured({ fetch: false })],
+            [Watcher.node, Watcher.configured({ enabled: false })],
+            [Bus.node, Bus.configured({ persist: true })],
+          ],
+        }),
+      ),
+    )
+  })
+
+type Execution = { readonly sessionID: Session.ID; readonly capability: string; readonly text: string }
+
+const capability = (name: string, executions: Execution[] = [], prompts: Session.ID[] = []) =>
+  Plugin.define({
+    // Deliberately reuse the plugin and skill IDs across private graphs.
+    id: "direct-capability",
+    effect: (ctx) =>
+      Effect.gen(function* () {
+        const skill = Skill.Info.make({
+          id: Skill.ID.make("direct-policy"),
+          name: Skill.Name.make("Direct Policy"),
+          description: `${name} policy`,
+          location: AbsolutePath.make(path.join(ctx.location.directory, "policy.md")),
+          content: `${name} skill guidance`,
+        })
+        yield* ctx.skill.transform((skills) => skills.add(skill))
+        yield* ctx.session.hook("prompt", (event) =>
+          Effect.sync(() => {
+            prompts.push(event.sessionID)
+            event.prompt.text = `${name}: ${event.prompt.text}`
+            event.prompt.skills = [{ id: skill.id }]
+          }),
+        )
+        yield* ctx.tool
+          .transform((tools) =>
+            tools.add({
+              name: `${name}_tool`,
+              description: `Execute the ${name} capability`,
+              options: { codemode: false },
+              input: Schema.Struct({ text: Schema.String }),
+              output: Schema.Struct({ capability: Schema.String, text: Schema.String }),
+              execute: (input, context) =>
+                Effect.sync(() => {
+                  executions.push({ sessionID: context.sessionID, capability: name, text: input.text })
+                  return { output: { capability: name, text: input.text } }
+                }),
+            }),
+          )
+          .pipe(Effect.orDie)
+      }),
+  })
+
+const userTexts = (request: LLMRequest) =>
+  request.messages.flatMap((message) =>
+    message.role === "user" ? message.content.flatMap((part) => (part.type === "text" ? [part.text] : [])) : [],
+  )
+
+class EventSink extends Context.Service<EventSink, SessionEvent.Event[]>()("test/direct-session/EventSink") {}
+
+const checkTypes = (handle: Session.Handle, options: Session.Options) => {
+  const create = Session.create(options)
+  const constructor: Effect.Effect<Session.Handle, unknown, Shared.Service | Scope.Scope> = create
+  // @ts-expect-error Creating a direct Session requires caller-owned shared infrastructure.
+  const missingShared: Effect.Effect<Session.Handle, unknown, Scope.Scope> = create
+  // @ts-expect-error Creating a direct Session also requires a lifetime Scope.
+  const missingScope: Effect.Effect<Session.Handle, unknown, Shared.Service> = create
+  const observe = handle.events.subscribe(() => EventSink.pipe(Effect.asVoid))
+  const subscription: Effect.Effect<Fiber.Fiber<void>, unknown, EventSink | Scope.Scope> = observe
+  // @ts-expect-error The callback's services must be supplied when subscribing.
+  const missingCallback: Effect.Effect<Fiber.Fiber<void>, unknown, Scope.Scope> = observe
+  const supplied: Effect.Effect<Fiber.Fiber<void>, unknown, Scope.Scope> = observe.pipe(
+    Effect.provideService(EventSink, []),
+  )
+  // @ts-expect-error New Session creation needs a Location; adoption needs an ID.
+  Session.create({})
+
+  const open = Layer.effectDiscard(EventSink)
+  const fallible = Layer.effectDiscard(Effect.fail(new Error("Replacement acquisition failed")))
+  const fallibleNode = makeGlobalNode({ name: "test/fallible-replacement", layer: fallible, deps: [] })
+  // @ts-expect-error Ready Shared constructors cannot silently acquire open replacement layers.
+  Shared.layer({ replacements: [[Bus.node, open]] })
+  // @ts-expect-error Ready Shared constructors cannot silently erase replacement errors.
+  Shared.layer({ replacements: [[Bus.node, fallible]] })
+  // @ts-expect-error Fallible replacement nodes are rejected too.
+  Shared.layer({ replacements: [[Bus.node, fallibleNode]] })
+  // @ts-expect-error Direct Session replacement layers must be closed.
+  Session.create({ ...options, replacements: [[PluginHooks.node, open]] })
+  // @ts-expect-error Direct Session replacement layers must be infallible.
+  Session.create({ ...options, replacements: [[PluginHooks.node, fallible]] })
+  // @ts-expect-error Direct Session replacement nodes must be infallible.
+  Session.create({ ...options, replacements: [[PluginHooks.node, fallibleNode]] })
+  // @ts-expect-error Shared replacements must provide the service they replace.
+  Shared.layer({ replacements: [[Bus.node, Layer.succeed(EventSink, [])]] })
+  // @ts-expect-error Direct Session replacements must provide the service they replace.
+  Session.create({ ...options, replacements: [[PluginHooks.node, Layer.succeed(EventSink, [])]] })
+  void [constructor, missingShared, missingScope, subscription, missingCallback, supplied]
+}
+void checkTypes
+
+describe("direct Session", () => {
+  it.live("closes private instances before Shared backing when the caller Scope outlives its provider", () =>
+    Effect.gen(function* () {
+      const directory = yield* tmpdirScoped("opencode-direct-owner-")
+      const lifecycle: string[] = []
+      const owner = yield* Scope.Scope
+      const caller = yield* Scope.fork(owner)
+      const handle = yield* Session.create({
+        location: Location.Ref.make({ directory: AbsolutePath.make(directory.path) }),
+        plugins: [
+          Plugin.define({
+            id: "direct-lifetime",
+            effect: () =>
+              Effect.addFinalizer(() =>
+                Effect.sync(() => {
+                  lifecycle.push("instance")
+                }),
+              ),
+          }),
+        ],
+      }).pipe(
+        Scope.provide(caller),
+        Effect.provide(
+          Shared.layer({
+            replacements: [
+              [
+                Global.node,
+                Layer.effectContext(
+                  Effect.gen(function* () {
+                    const context = yield* Layer.build(tempGlobalLayer)
+                    yield* Effect.addFinalizer(() =>
+                      Effect.sync(() => {
+                        lifecycle.push("shared")
+                      }),
+                    )
+                    return context
+                  }),
+                ),
+              ],
+              [ModelsDev.node, ModelsDev.configured({ fetch: false })],
+              [Watcher.node, Watcher.configured({ enabled: false })],
+            ],
+          }),
+        ),
+      )
+      expect(lifecycle).toEqual(["instance", "shared"])
+      expect(yield* handle.prompt({ text: "After provider close", resume: false }).pipe(Effect.exit)).toEqual(
+        Exit.fail(new Session.ClosedError({ sessionID: handle.id })),
+      )
+      yield* Scope.close(caller, Exit.void)
+      expect(lifecycle).toEqual(["instance", "shared"])
+    }),
+  )
+
+  it.live(
+    "waits for runtime installation when plugin setup calls Session APIs before the graph is acquired",
+    () =>
+      withShared((fixture) =>
+        Effect.gen(function* () {
+          const id = Session.ID.create()
+          const graphRelease = yield* Deferred.make<void>()
+          const pluginStarted = yield* Deferred.make<Fiber.Fiber<unknown>>()
+          const late = LocationWatcher.node.implementation
+          if (!Layer.isLayer(late)) throw new Error("LocationWatcher must have a layer implementation")
+          const held = {
+            ...LocationWatcher.node,
+            implementation: late.pipe(Layer.tap(() => Deferred.await(graphRelease))),
+          }
+          const creating = yield* Session.create({
+            id,
+            location: fixture.location,
+            title: "Startup readiness",
+            replacements: [[LocationWatcher.node, held]],
+            plugins: [
+              Plugin.define({
+                id: "direct-startup",
+                effect: (ctx) =>
+                  Effect.gen(function* () {
+                    // Start the call before signaling, so polling below observes a real suspended lookup.
+                    const lookup = yield* ctx.session
+                      .get({ sessionID: id })
+                      .pipe(Effect.orDie, Effect.forkScoped({ startImmediately: true }))
+                    yield* Deferred.succeed(pluginStarted, lookup)
+                    expect((yield* Fiber.join(lookup)).id).toBe(id)
+                    yield* ctx.session.hook("prompt", (event) =>
+                      Effect.sync(() => {
+                        event.prompt.text = "Plugin setup finished"
+                      }),
+                    )
+                  }),
+              }),
+            ],
+          }).pipe(Effect.forkScoped({ startImmediately: true }))
+
+          const lookup = yield* Deferred.await(pluginStarted).pipe(Effect.timeout("5 seconds"))
+          expect(lookup.pollUnsafe()).toBeUndefined()
+          expect(creating.pollUnsafe()).toBeUndefined()
+          yield* Deferred.succeed(graphRelease, undefined)
+          const handle = yield* Fiber.join(creating).pipe(Effect.timeout("5 seconds"))
+
+          expect((yield* handle.prompt({ text: "After readiness", resume: false })).payload.text).toBe(
+            "Plugin setup finished",
+          )
+          expect(yield* fixture.llm.requests()).toEqual([])
+        }),
+      ),
+    15_000,
+  )
+
+  it.live(
+    "executes isolated tools, prompt hooks, and skill attachments in the same directory",
+    () =>
+      withShared((fixture) =>
+        Effect.gen(function* () {
+          const shared = yield* Shared.Service
+          const executions: Execution[] = []
+          const firstPrompts: Session.ID[] = []
+          const secondPrompts: Session.ID[] = []
+          const first = yield* Session.create({
+            location: fixture.location,
+            title: "First",
+            plugins: [modelPlugin, capability("first", executions, firstPrompts)],
+          })
+          const second = yield* Session.create({
+            location: fixture.location,
+            title: "Second",
+            plugins: [modelPlugin, capability("second", executions, secondPrompts)],
+          })
+          yield* fixture.llm.push(
+            TestLLM.tool("call-first", "first_tool", { text: "one" }),
+            TestLLM.text("First complete", "answer-first"),
+            TestLLM.tool("call-second", "second_tool", { text: "two" }),
+            TestLLM.text("Second complete", "answer-second"),
+          )
+
+          const firstInput = yield* first.prompt({ text: "Use my tool" })
+          yield* first.wait()
+          const secondInput = yield* second.prompt({ text: "Use my tool" })
+          yield* second.wait()
+
+          expect(first.id).not.toBe(second.id)
+          expect(executions).toEqual([
+            { sessionID: first.id, capability: "first", text: "one" },
+            { sessionID: second.id, capability: "second", text: "two" },
+          ])
+          expect(firstPrompts).toEqual([first.id])
+          expect(secondPrompts).toEqual([second.id])
+          expect(firstInput.payload.text).toBe("first: Use my tool")
+          expect(secondInput.payload.text).toBe("second: Use my tool")
+          expect(firstInput.payload.skills?.[0]?.text).toContain("first skill guidance")
+          expect(firstInput.payload.skills?.[0]?.text).not.toContain("second skill guidance")
+          expect(secondInput.payload.skills?.[0]?.text).toContain("second skill guidance")
+          expect(secondInput.payload.skills?.[0]?.text).not.toContain("first skill guidance")
+
+          const requests = yield* fixture.llm.requests()
+          expect(requests).toHaveLength(4)
+          requests.slice(0, 2).forEach((request) => {
+            expect(request.tools.map((tool) => tool.name)).toContain("first_tool")
+            expect(request.tools.map((tool) => tool.name)).not.toContain("second_tool")
+            expect(userTexts(request).join("\n")).toContain("first skill guidance")
+          })
+          requests.slice(2).forEach((request) => {
+            expect(request.tools.map((tool) => tool.name)).toContain("second_tool")
+            expect(request.tools.map((tool) => tool.name)).not.toContain("first_tool")
+            expect(userTexts(request).join("\n")).toContain("second skill guidance")
+          })
+          expect(yield* shared.sessions.context(first.id)).toContainEqual(
+            expect.objectContaining({ id: firstInput.id, type: "user", text: "first: Use my tool" }),
+          )
+          expect(yield* shared.sessions.context(second.id)).toContainEqual(
+            expect.objectContaining({ id: secondInput.id, type: "user", text: "second: Use my tool" }),
+          )
+          expect(yield* shared.sessions.inbox(first.id)).toEqual([])
+          expect(yield* shared.sessions.inbox(second.id)).toEqual([])
+        }),
+      ),
+    20_000,
+  )
+
+  it.live(
+    "subscribes ready and isolates admission and runner events with the caller's services",
+    () =>
+      withShared((fixture) =>
+        Effect.gen(function* () {
+          const first = yield* Session.create({ location: fixture.location, title: "First", plugins: [modelPlugin] })
+          const second = yield* Session.create({ location: fixture.location, title: "Second", plugins: [modelPlugin] })
+          const firstEvents: SessionEvent.Event[] = []
+          const secondEvents: SessionEvent.Event[] = []
+          const firstEnqueued = yield* Deferred.make<SessionEvent.InboxEnqueued>()
+          const secondEnqueued = yield* Deferred.make<SessionEvent.InboxEnqueued>()
+          const firstDone = yield* Deferred.make<void>()
+          const secondDone = yield* Deferred.make<void>()
+          const observe =
+            (enqueued: Deferred.Deferred<SessionEvent.InboxEnqueued>, done: Deferred.Deferred<void>) =>
+            (event: SessionEvent.Event) =>
+              Effect.gen(function* () {
+                const events = yield* EventSink
+                events.push(event)
+                if (event.type === "session.inbox.enqueued") yield* Deferred.succeed(enqueued, event)
+                if (event.type === "session.execution.succeeded") yield* Deferred.succeed(done, undefined)
+              })
+
+          yield* first.events
+            .subscribe(observe(firstEnqueued, firstDone))
+            .pipe(Effect.provideService(EventSink, firstEvents))
+          // No connected marker, scheduler yield, or delay before the first prompt.
+          const firstInput = yield* first.prompt({ text: "First event", resume: false })
+          yield* second.events
+            .subscribe(observe(secondEnqueued, secondDone))
+            .pipe(Effect.provideService(EventSink, secondEvents))
+          const secondInput = yield* second.prompt({ text: "Second event", resume: false })
+          expect((yield* Deferred.await(firstEnqueued).pipe(Effect.timeout("5 seconds"))).data.inboxID).toBe(
+            firstInput.id,
+          )
+          expect((yield* Deferred.await(secondEnqueued).pipe(Effect.timeout("5 seconds"))).data.inboxID).toBe(
+            secondInput.id,
+          )
+          yield* fixture.llm.push(TestLLM.text("First", "answer-first"), TestLLM.text("Second", "answer-second"))
+          yield* first.resume()
+          yield* second.resume()
+          yield* Effect.all([Deferred.await(firstDone), Deferred.await(secondDone)]).pipe(Effect.timeout("5 seconds"))
+
+          expect(firstEvents.map((event) => event.data.sessionID)).toEqual(firstEvents.map(() => first.id))
+          expect(secondEvents.map((event) => event.data.sessionID)).toEqual(secondEvents.map(() => second.id))
+          ;[firstEvents, secondEvents].forEach((events) => {
+            expect(events.map((event) => event.type)).toEqual(
+              expect.arrayContaining([
+                "session.inbox.enqueued",
+                "session.inbox.delivered",
+                "session.step.started",
+                "session.execution.succeeded",
+              ]),
+            )
+          })
+        }),
+      ),
+    20_000,
+  )
+
+  it.live(
+    "keeps callback failures on the subscription fiber without failing prompt admission",
+    () =>
+      withShared((fixture) =>
+        Effect.gen(function* () {
+          const shared = yield* Shared.Service
+          const handle = yield* Session.create({ location: fixture.location, title: "Callback failure" })
+          const failure = new Error("Observer failed")
+          const released = yield* Deferred.make<void>()
+          const observer = yield* handle.events.subscribe(() =>
+            Effect.acquireRelease(Effect.void, () => Deferred.succeed(released, undefined).pipe(Effect.asVoid)).pipe(
+              Effect.andThen(Effect.fail(failure)),
+            ),
+          )
+
+          const input = yield* handle.prompt({ text: "Still admitted", resume: false })
+
+          expect(yield* Fiber.join(observer).pipe(Effect.flip, Effect.timeout("5 seconds"))).toBe(failure)
+          yield* Deferred.await(released).pipe(Effect.timeout("5 seconds"))
+          expect(yield* shared.sessions.inbox(handle.id)).toEqual([input])
+          expect(yield* fixture.llm.requests()).toEqual([])
+        }),
+      ),
+    15_000,
+  )
+
+  it.live(
+    "stops observations when their caller Scope or handle Scope closes",
+    () =>
+      withShared((fixture) =>
+        Effect.gen(function* () {
+          const scope = yield* Scope.Scope
+          const handleScope = yield* Scope.fork(scope)
+          const subscriptionScope = yield* Scope.fork(scope)
+          const handle = yield* Session.create({ location: fixture.location, title: "Scoped observations" }).pipe(
+            Scope.provide(handleScope),
+          )
+          const enqueued = yield* Deferred.make<void>()
+          const events: SessionEvent.Event[] = []
+          const observer = yield* handle.events
+            .subscribe((event) =>
+              Effect.gen(function* () {
+                events.push(event)
+                if (event.type === "session.inbox.enqueued") yield* Deferred.succeed(enqueued, undefined)
+              }),
+            )
+            .pipe(Scope.provide(subscriptionScope))
+          yield* handle.prompt({ text: "Observed", resume: false })
+          yield* Deferred.await(enqueued).pipe(Effect.timeout("5 seconds"))
+
+          yield* Scope.close(subscriptionScope, Exit.void)
+          const stopped = yield* Fiber.await(observer)
+          yield* handle.prompt({ text: "Not observed", resume: false })
+          expect(events.map((event) => event.type)).toEqual(["session.inbox.enqueued"])
+
+          const owned = yield* handle.events.subscribe(() => Effect.void)
+          yield* Scope.close(handleScope, Exit.void)
+          const closed = yield* Fiber.await(owned)
+          ;[stopped, closed].forEach((exit) => {
+            // fromSubscription can append its normal Done marker to the interruption cause.
+            expect(
+              Exit.isSuccess(exit) ||
+                exit.cause.reasons.every(
+                  (reason) =>
+                    Cause.isInterruptReason(reason) || (Cause.isFailReason(reason) && Cause.isDone(reason.error)),
+                ),
+            ).toBe(true)
+          })
+        }),
+      ),
+    15_000,
+  )
+
+  it.live("does not retain caller finalizers after observation fibers finish", () =>
+    withShared((fixture) =>
+      Effect.gen(function* () {
+        const caller = yield* Scope.Scope
+        const handle = yield* Session.create({ location: fixture.location, title: "Observer cleanup" })
+        if (caller.state._tag !== "Open") throw new Error("Expected the caller's handle cleanup")
+        const finalizers = caller.state.finalizers
+        if (!finalizers) throw new Error("Expected registered caller finalizers")
+        const before = finalizers.size
+        yield* Effect.forEach(Array.from({ length: 5 }), () =>
+          Effect.gen(function* () {
+            const observer = yield* handle.events.subscribe(() => Effect.void)
+            yield* Fiber.interrupt(observer)
+            expect(finalizers.size).toBe(before)
+          }),
+        )
+      }),
+    ),
+  )
+
+  it.live(
+    "creates and resumes a native subagent with its parent's private capabilities",
+    () =>
+      withShared((fixture) =>
+        Effect.gen(function* () {
+          const shared = yield* Shared.Service
+          const executions: Execution[] = []
+          const prompts: Session.ID[] = []
+          const parent = yield* Session.create({
+            location: fixture.location,
+            title: "Parent",
+            model,
+            metadata: { owner: "parent" },
+            plugins: [modelPlugin, capability("parent", executions, prompts), reviewerPlugin],
+          })
+          // A later sibling at the same Location must not become the child's instance.
+          yield* Session.create({
+            location: fixture.location,
+            title: "Sibling",
+            plugins: [modelPlugin, capability("sibling", executions)],
+          })
+          yield* fixture.llm.push(
+            TestLLM.tool("call-subagent", SubagentTool.name, {
+              agent: "reviewer",
+              description: "Child",
+              prompt: "Use my inherited tool",
+            }),
+            TestLLM.tool("call-child-tool", "parent_tool", { text: "child" }),
+            TestLLM.text("Child complete", "answer-child"),
+            TestLLM.text("Parent complete", "answer-parent"),
+          )
+          yield* parent.prompt({ text: "Delegate" })
+          yield* parent.wait()
+
+          const children = (yield* shared.sessions.list({ parentID: parent.id })).data
+          expect(children).toHaveLength(1)
+          const child = children[0]
+          expect(child).toMatchObject({
+            parentID: parent.id,
+            location: fixture.location,
+            agent: "reviewer",
+            model,
+            metadata: { owner: "parent" },
+          })
+          expect(executions).toEqual([{ sessionID: child.id, capability: "parent", text: "child" }])
+          expect(prompts).toEqual([parent.id, child.id])
+          expect((yield* shared.sessions.context(child.id)).find((message) => message.type === "user")).toMatchObject({
+            text: "parent: You are a subagent spawned by another session.\nUse my inherited tool",
+          })
+
+          yield* fixture.llm.push(
+            TestLLM.tool("call-continue-child", SubagentTool.name, {
+              sessionID: child.id,
+              agent: "reviewer",
+              description: "Continue",
+              prompt: "Use it again",
+            }),
+            TestLLM.tool("call-child-again", "parent_tool", { text: "continued" }),
+            TestLLM.text("Child continued", "answer-child-again"),
+            TestLLM.text("Parent continued", "answer-parent-again"),
+          )
+          yield* parent.prompt({ text: "Continue the same child" })
+          yield* parent.wait()
+
+          expect((yield* shared.sessions.list({ parentID: parent.id })).data.map((session) => session.id)).toEqual([
+            child.id,
+          ])
+          expect(executions).toEqual([
+            { sessionID: child.id, capability: "parent", text: "child" },
+            { sessionID: child.id, capability: "parent", text: "continued" },
+          ])
+          expect(prompts).toEqual([parent.id, child.id, parent.id, child.id])
+          const requests = yield* fixture.llm.requests()
+          expect(requests).toHaveLength(8)
+          requests.forEach((request) => {
+            expect(request.tools.map((tool) => tool.name)).toContain("parent_tool")
+            expect(request.tools.map((tool) => tool.name)).not.toContain("sibling_tool")
+          })
+          expect(yield* shared.sessions.context(parent.id)).toContainEqual(
+            expect.objectContaining({
+              type: "assistant",
+              content: expect.arrayContaining([expect.objectContaining({ type: "text", text: "Parent continued" })]),
+            }),
+          )
+        }),
+      ),
+    25_000,
+  )
+  ;[false, true].forEach((background) =>
+    it.live(
+      background
+        ? "preserves background child recovery without admitting a cancellation synthetic on handle shutdown"
+        : "retains foreground parent and child claims with shutdown interruption for the whole ownership chain",
+      () =>
+        withShared((fixture) =>
+          Effect.gen(function* () {
+            const shared = yield* Shared.Service
+            const owner = yield* Scope.Scope
+            const scope = yield* Scope.fork(owner)
+            const parent = yield* Session.create({
+              location: fixture.location,
+              title: "Parent",
+              model,
+              plugins: [modelPlugin, reviewerPlugin],
+            }).pipe(Scope.provide(scope))
+            let parentCalls = 0
+            yield* fixture.llm.serve((request) => {
+              // Match the child by its real prompt so background scheduling cannot reorder scripted replies.
+              if (userTexts(request).some((text) => text.includes("You are a subagent"))) return TestLLM.hangAfter()
+              return ++parentCalls === 1
+                ? TestLLM.tool("call-shutdown-subagent", SubagentTool.name, {
+                    agent: "reviewer",
+                    description: "Child",
+                    prompt: "Work",
+                    background,
+                  })
+                : TestLLM.text("Parent done", "parent-answer")
+            })
+            yield* parent.prompt({ text: "Delegate" })
+            yield* fixture.llm.wait(background ? 3 : 2).pipe(Effect.timeout("5 seconds"))
+            if (background) yield* parent.wait()
+
+            const children = (yield* shared.sessions.list({ parentID: parent.id })).data
+            expect(children).toHaveLength(1)
+            const child = children[0]
+            expect((yield* shared.jobs.get(child.id))?.status).toBe("running")
+            const beforeMarkers = yield* shared.jobs.pendingBackground
+            expect(beforeMarkers).toHaveLength(background ? 1 : 0)
+
+            yield* Scope.close(scope, Exit.void)
+
+            expect(yield* shared.execution.active).toEqual(new Set())
+            expect((yield* shared.jobs.get(child.id))?.status).toBe("cancelled")
+            expect(yield* shared.sessions.inbox(parent.id)).toEqual([])
+            expect(yield* shared.sessions.inbox(child.id)).toEqual([])
+            const markers = yield* shared.jobs.pendingBackground
+            expect(markers).toHaveLength(background ? 1 : 0)
+            if (background) {
+              expect(markers[0]).toMatchObject({
+                id: child.id,
+                status: "running",
+                notificationID: beforeMarkers[0].notificationID,
+                recovery: { kind: "subagent", parentSessionID: parent.id, childSessionID: child.id },
+              })
+            }
+
+            const database = Context.get(shared.globals, Database.Service)
+            const claims = yield* database.db
+              .select({ id: SessionTable.id, suspended: SessionTable.time_suspended })
+              .from(SessionTable)
+              .all()
+            expect(claims).toHaveLength(2)
+            expect(claims.find((row) => row.id === child.id)?.suspended).toEqual(expect.any(Number))
+            expect(claims.find((row) => row.id === parent.id)?.suspended).toEqual(
+              background ? null : expect.any(Number),
+            )
+            // Read durable logs after closing; handle observations have already been torn down.
+            const logs = yield* Effect.forEach([parent.id, child.id], (sessionID) =>
+              shared.sessions.log({ sessionID, follow: false }).pipe(Stream.runCollect),
+            )
+            const interruptions = logs.flat().filter((event) => event.type === "session.execution.interrupted")
+            expect(interruptions).toHaveLength(background ? 1 : 2)
+            interruptions.forEach((event) => expect(event.data.reason).toBe("shutdown"))
+            expect(
+              logs[0].filter(
+                (event) => event.type === "session.inbox.enqueued" && event.data.item.type === "synthetic",
+              ),
+            ).toEqual([])
+          }),
+        ),
+      20_000,
+    ),
+  )
+
+  it.live(
+    "adopts saved facts without auto-running pending work and explicitly resumes with new capabilities",
+    () =>
+      withShared((fixture) =>
+        Effect.gen(function* () {
+          const shared = yield* Shared.Service
+          const id = Session.ID.create()
+          const recorded = yield* Effect.gen(function* () {
+            const handle = yield* Session.create({
+              id,
+              location: fixture.location,
+              title: "Saved title",
+              agent: Agent.ID.make("build"),
+              model,
+              metadata: { owner: "saved" },
+              plugins: [modelPlugin, capability("original")],
+            })
+            const pending = yield* handle.prompt({
+              id: SessionMessage.ID.create(),
+              text: "Pending work",
+              resume: false,
+            })
+            return { info: yield* shared.sessions.get(id), pending }
+          }).pipe(Effect.scoped)
+          const executions: Execution[] = []
+          const adoptedPrompts: Session.ID[] = []
+
+          yield* Effect.gen(function* () {
+            const adopted = yield* Session.create({
+              id,
+              location: Location.Ref.make({
+                directory: AbsolutePath.make(path.join(fixture.location.directory, "ignored-location")),
+              }),
+              title: "Ignored title",
+              agent: Agent.ID.make("ignored-agent"),
+              model: Model.Ref.make({ providerID: model.providerID, id: Model.ID.make("fictional-other") }),
+              metadata: { owner: "ignored" },
+              plugins: [modelPlugin, capability("adopted", executions, adoptedPrompts)],
+            })
+            expect(adopted.id).toBe(id)
+            expect(yield* shared.sessions.get(id)).toEqual(recorded.info)
+            expect(yield* shared.sessions.inbox(id)).toEqual([recorded.pending])
+            yield* adopted.wait()
+            expect(yield* fixture.llm.requests()).toEqual([])
+
+            yield* fixture.llm.push(
+              TestLLM.tool("call-adopted", "adopted_tool", { text: "resumed" }),
+              TestLLM.text("Resumed", "answer-adopted"),
+            )
+            yield* adopted.resume()
+            yield* adopted.wait()
+            expect(yield* shared.sessions.inbox(id)).toEqual([])
+            expect(executions).toEqual([{ sessionID: id, capability: "adopted", text: "resumed" }])
+            expect(adoptedPrompts).toEqual([])
+            expect(yield* shared.sessions.context(id)).toContainEqual(
+              expect.objectContaining({ id: recorded.pending.id, type: "user", text: "original: Pending work" }),
+            )
+            const requests = yield* fixture.llm.requests()
+            expect(requests).toHaveLength(2)
+            expect(requests[0].model).toMatchObject({ provider: model.providerID, id: model.id })
+            expect(userTexts(requests[0]).join("\n")).toContain("original skill guidance")
+            expect(requests[0].tools.map((tool) => tool.name)).not.toContain("original_tool")
+          }).pipe(Effect.scoped)
+
+          const reopened = yield* Session.create({ id, plugins: [modelPlugin, capability("reopened")] })
+          const next = yield* reopened.prompt({ text: "Next input", resume: false })
+          expect(next.payload.text).toBe("reopened: Next input")
+          expect((yield* shared.sessions.get(id)).location).toEqual(recorded.info.location)
+        }),
+      ),
+    25_000,
+  )
+
+  it.live(
+    "rejects a conflicting live binding without changing the original handle",
+    () =>
+      withShared((fixture) =>
+        Effect.gen(function* () {
+          const shared = yield* Shared.Service
+          const first = yield* Session.create({
+            location: fixture.location,
+            title: "Original",
+            plugins: [capability("first")],
+          })
+          const conflict = yield* Session.create({
+            id: first.id,
+            location: fixture.location,
+            title: "Replacement",
+            plugins: [capability("replacement")],
+          }).pipe(Effect.flip)
+
+          expect(conflict).toMatchObject({ _tag: "Session.AlreadyBoundError", sessionID: first.id })
+          expect((yield* shared.sessions.get(first.id)).title).toBe("Original")
+          expect((yield* first.prompt({ text: "Still mine", resume: false })).payload.text).toBe("first: Still mine")
+          expect(yield* fixture.llm.requests()).toEqual([])
+        }),
+      ),
+    15_000,
+  )
+
+  it.live(
+    "rejects closed handles, including effects constructed before closure",
+    () =>
+      withShared((fixture) =>
+        Effect.gen(function* () {
+          const shared = yield* Shared.Service
+          const closed = yield* Effect.gen(function* () {
+            const handle = yield* Session.create({ location: fixture.location, title: "Closed" })
+            return { handle, prompt: handle.prompt({ text: "Constructed while open", resume: false }) }
+          }).pipe(Effect.scoped)
+
+          const errors = yield* Effect.all([
+            closed.prompt.pipe(Effect.flip),
+            closed.handle.prompt({ text: "After close", resume: false }).pipe(Effect.flip),
+            closed.handle.resume().pipe(Effect.flip),
+            closed.handle.interrupt().pipe(Effect.flip),
+            closed.handle.wait().pipe(Effect.flip),
+            closed.handle.events.subscribe(() => Effect.void).pipe(Effect.flip),
+          ])
+
+          errors.forEach((error) =>
+            expect(error).toMatchObject({ _tag: "Session.ClosedError", sessionID: closed.handle.id }),
+          )
+          expect(yield* shared.sessions.inbox(closed.handle.id)).toEqual([])
+          expect(yield* fixture.llm.requests()).toEqual([])
+        }),
+      ),
+    15_000,
+  )
+  ;(["interrupt", "close"] as const).forEach((operation) =>
+    it.live(
+      `${operation} settles only the selected Session and leaves its sibling runnable`,
+      () =>
+        withShared((fixture) =>
+          Effect.gen(function* () {
+            const shared = yield* Shared.Service
+            const scope = yield* Scope.Scope
+            const firstScope = yield* Scope.fork(scope)
+            const first = yield* Session.create({
+              location: fixture.location,
+              title: "First",
+              plugins: [modelPlugin],
+            }).pipe(Scope.provide(firstScope))
+            const second = yield* Session.create({
+              location: fixture.location,
+              title: "Second",
+              plugins: [modelPlugin],
+            })
+            yield* fixture.llm.push(
+              TestLLM.text("First complete", "answer-first"),
+              TestLLM.text("Second complete", "answer-second"),
+            )
+            const gate = yield* fixture.llm.gate()
+            yield* first.prompt({ text: "First" })
+            yield* gate.started.pipe(Effect.timeout("5 seconds"))
+            yield* second.prompt({ text: "Second" })
+            yield* gate.started.pipe(Effect.timeout("5 seconds"))
+            expect(yield* shared.execution.active).toEqual(new Set([first.id, second.id]))
+
+            yield* operation === "close"
+              ? Scope.close(firstScope, Exit.void)
+              : first.interrupt().pipe(Effect.andThen(first.wait()))
+            expect(yield* shared.execution.active).toEqual(new Set([second.id]))
+            yield* gate.release
+            yield* second.wait()
+
+            expect(yield* shared.execution.active).toEqual(new Set())
+            expect(yield* shared.sessions.context(second.id)).toContainEqual(
+              expect.objectContaining({
+                type: "assistant",
+                content: expect.arrayContaining([expect.objectContaining({ type: "text", text: "Second complete" })]),
+              }),
+            )
+            expect(yield* fixture.llm.requests()).toHaveLength(2)
+            if (operation === "interrupt") expect(yield* first.interrupt()).toBe(false)
+          }),
+        ),
+      20_000,
+    ),
+  )
+
+  it.live(
+    "reuses shared infrastructure identities while acquiring private registries and runtime bindings",
+    () =>
+      withShared((fixture) =>
+        Effect.gen(function* () {
+          const shared = yield* Shared.Service
+          const capture = Effect.all({
+            database: Database.Service,
+            bus: Bus.Service,
+            global: Global.Service,
+            llm: LLMClient.Service,
+            store: SessionStore.Service,
+            tools: Tool.Service,
+            hooks: PluginHooks.Service,
+            runtime: PluginRuntime.Service,
+          })
+          const instances: Array<Effect.Success<typeof capture>> = []
+          const probe = {
+            ...ModelResolver.node,
+            implementation: Layer.merge(
+              ModelResolver.layer,
+              Layer.effectDiscard(capture.pipe(Effect.tap((instance) => Effect.sync(() => instances.push(instance))))),
+            ),
+            dependencies: [
+              ...ModelResolver.node.dependencies,
+              Database.node,
+              Bus.node,
+              Global.node,
+              llmClient,
+              SessionStore.node,
+              Tool.node,
+              PluginHooks.node,
+              PluginRuntime.node,
+            ],
+          }
+          const scope = yield* Scope.Scope
+          const firstScope = yield* Scope.fork(scope)
+          const first = yield* Session.create({
+            location: fixture.location,
+            title: "First",
+            plugins: [modelPlugin],
+            replacements: [[ModelResolver.node, probe]],
+          }).pipe(Scope.provide(firstScope))
+          const second = yield* Session.create({
+            location: fixture.location,
+            title: "Second",
+            plugins: [modelPlugin],
+            replacements: [[ModelResolver.node, probe]],
+          })
+
+          expect(instances).toHaveLength(2)
+          instances.forEach((instance) => {
+            expect(instance.database).toBe(Context.get(shared.globals, Database.Service))
+            expect(instance.bus).not.toBe(Context.get(shared.globals, Bus.Service))
+            expect(instance.global).toBe(Context.get(shared.globals, Global.Service))
+            expect(instance.llm).toBe(fixture.llm)
+            expect(instance.store).toBe(Context.get(shared.globals, SessionStore.Service))
+          })
+          expect(instances[0].tools).not.toBe(instances[1].tools)
+          expect(instances[0].hooks).not.toBe(instances[1].hooks)
+          expect(instances[0].bus).not.toBe(instances[1].bus)
+          expect(instances[0].runtime).not.toBe(instances[1].runtime)
+          expect(Option.isNone(Context.getOption(shared.globals, LocationServiceMap.Service))).toBe(true)
+
+          const job = yield* instances[0].runtime.job.start({ type: "direct-test", run: Effect.never })
+          expect(yield* shared.jobs.get(job.id)).toEqual(job)
+          yield* instances[1].runtime.job.cancel(job.id)
+          expect((yield* shared.jobs.get(job.id))?.status).toBe("cancelled")
+          expect(yield* instances[1].runtime.session.get(first.id)).toEqual(yield* shared.sessions.get(first.id))
+
+          yield* Scope.close(firstScope, Exit.void)
+          yield* fixture.llm.push(TestLLM.text("Still shared", "answer-second"))
+          yield* second.prompt({ text: "Survive sibling closure" })
+          yield* second.wait()
+          expect((yield* shared.sessions.get(first.id)).id).toBe(first.id)
+          expect(yield* fixture.llm.requests()).toHaveLength(1)
+          const bus = Context.get(shared.globals, Bus.Service)
+          const log = yield* bus.log({ aggregateID: second.id }).pipe(Stream.runCollect)
+          expect(log).toContainEqual(
+            expect.objectContaining({
+              type: "session.inbox.enqueued",
+              durable: expect.objectContaining({ aggregateID: second.id }),
+              data: expect.objectContaining({ sessionID: second.id }),
+            }),
+          )
+        }),
+      ),
+    20_000,
+  )
+
+  it.live(
+    "defaults discovery off without importing ambient plugins or instructions",
+    () =>
+      withShared((fixture) =>
+        Effect.gen(function* () {
+          const marker = path.join(fixture.location.directory, "ambient-loaded")
+          yield* Effect.promise(() =>
+            Bun.write(path.join(fixture.location.directory, "AGENTS.md"), "Ambient instruction sentinel"),
+          )
+          yield* Effect.promise(() =>
+            Bun.write(
+              path.join(fixture.location.directory, ".opencode/plugins/ambient.ts"),
+              `await Bun.write(${JSON.stringify(marker)}, "loaded")\nexport default { id: "ambient-plugin", setup() {} }\n`,
+            ),
+          )
+          const handle = yield* Session.create({ location: fixture.location, title: "Vanilla", plugins: [modelPlugin] })
+          yield* fixture.llm.push(TestLLM.text("Vanilla", "answer-vanilla"))
+          yield* handle.prompt({ text: "Use only supplied capabilities" })
+          yield* handle.wait()
+
+          expect(yield* Effect.promise(() => Bun.file(marker).exists())).toBe(false)
+          const requests = yield* fixture.llm.requests()
+          expect(requests).toHaveLength(1)
+          expect(requests[0].system.map((part) => part.text).join("\n")).not.toContain("Ambient instruction sentinel")
+          expect(requests[0].tools.map((tool) => tool.name)).toContain(SubagentTool.name)
+        }),
+      ),
+    15_000,
+  )
+})

packages/sdk/test/import-boundaries.test.ts

@@ -17,6 +17,13 @@ test("bundles the Promise and Effect clients with the in-memory host", async ()
   }
 })
 
+test("bundles direct Effect Sessions without the embedded host or HTTP client", async () => {
+  const inputs = await bundleInputs("@opencode-ai/sdk/direct/effect")
+  expect(within(inputs, core).length).toBeGreaterThan(0)
+  expect(within(inputs, client)).toEqual([])
+  expect(within(inputs, server)).toEqual([])
+})
+
 async function bundleInputs(specifier: string) {
   const temporary = await mkdtemp(join(import.meta.dir, ".import-boundary-"))
   const entrypoint = join(temporary, "index.ts")

packages/server/test/session-message-update.test.ts

@@ -47,6 +47,7 @@ it.live("updates completed assistant message content through the session HTTP AP
             }),
           interrupt: () => Effect.succeed(false),
           awaitIdle: () => Effect.void,
+          shutdown: () => Effect.void,
         })
       }),
     )

packages/util/src/effect/layer-node.ts

@@ -113,6 +113,26 @@ export function group<const Items extends readonly AnyNode[]>(
 
 export type Replacement = readonly [source: AnyNode, replacement: AnyNode | Layer.Any]
 export type Replacements = readonly Replacement[]
+export type ReplacementError<Items extends Replacements> = Items[number][1] extends infer Item
+  ? Item extends Layer.Any
+    ? Layer.Error<Item>
+    : Error<Item>
+  : never
+export type ReplacementServices<Items extends Replacements> = Items[number][1] extends infer Item
+  ? Item extends Layer.Any
+    ? Layer.Services<Item>
+    : never
+  : never
+
+// Open composition can retain new errors and requirements, but replacements
+// must still provide the original output and preserve node placement tags.
+export type ComposableReplacements<Items extends Replacements> = Items & {
+  readonly [K in keyof Items]: Items[K] extends readonly [Node<infer A, unknown, infer T>, infer Replacement]
+    ? Replacement extends Node<NoInfer<A>, unknown, T> | Layer.Layer<NoInfer<A>, unknown, unknown>
+      ? unknown
+      : { readonly "Invalid replacement": Replacement }
+    : { readonly "Invalid replacement": Items[K] }
+}
 
 type CheckReplacementErrors<SourceError, ReplacementError> = [Exclude<ReplacementError, SourceError>] extends [never]
   ? unknown

specs/v2/session.md

@@ -24,7 +24,7 @@ Manual compaction and Session movement use the same inbox as control items. Each
 
 ## Execution Is Process-Local
 
-`SessionExecution` is process-global and keyed only by Session ID. At drain start it loads the Session, enters its Location through `LocationServiceMap`, and invokes the Location-scoped runner. The runner, model resolution, tools, permissions, plugins, and filesystem remain Location-scoped.
+`SessionExecution` is process-global and keyed only by Session ID. At drain start it loads the Session and selects its capabilities through `SessionInstance`. The server adapter enters its Location through the existing `LocationServiceMap`; direct SDK handles retain an already-built private instance instead. Both use the same coordinator and runner algorithms. The runner, model resolution, tools, permissions, plugins, and filesystem remain instance-local and placement-bound.
 
 `SessionRunCoordinator` provides the local ownership rules: