// Global event subscription and prompt turn coordination. // // Creates a long-lived global event stream subscription and feeds relevant // events for the current session tree through the reducers. The reducers // produce scrollback commits and footer patches, which get forwarded to the // footer through stream.ts. // // Prompt turns are one-at-a-time: runPromptTurn() sends the prompt to the // SDK, arms a deferred Wait, and resolves when the session becomes idle. // Prefer session.status idle events, but also poll session.status because some // transports can miss status events while still delivering message events. If // the turn is aborted (user interrupt), it flushes any in-progress parts as // interrupted entries. // // The tick counter prevents stale idle events from resolving the wrong turn. // We also re-check live session status before resolving an idle event so a // delayed idle from an older turn cannot complete a newer busy turn. import type { Event, GlobalEvent, OpencodeClient } from "@opencode-ai/sdk/v2" import { Context, Deferred, Effect, Exit, Layer, Scope, Stream } from "effect" import { makeRuntime } from "@/effect/run-service" import { blockerStatus, bootstrapSessionData, createSessionData, flushInterrupted, pickBlockerView, reduceSessionData, type SessionData, } from "./session-data" import { replaySession } from "./session-replay" import { bootstrapSubagentCalls, bootstrapSubagentData, clearFinishedSubagents, createSubagentData, listSubagentPermissions, listSubagentQuestions, listSubagentTabs, reduceSubagentData, sameSubagentTab, snapshotSelectedSubagentData, SUBAGENT_BOOTSTRAP_LIMIT, SUBAGENT_CALL_BOOTSTRAP_LIMIT, type SubagentData, } from "./subagent-data" import { traceFooterOutput, writeSessionOutput } from "./stream" import type { FooterApi, FooterOutput, FooterPatch, FooterSubagentState, FooterSubagentTab, FooterView, RunFilePart, RunInput, RunPrompt, RunPromptPart, StreamCommit, } from "./types" type Trace = { write(type: string, data?: unknown): void } const StreamClosed = undefined as never type StreamInput = { sdk: OpencodeClient directory?: string sessionID: string thinking: boolean replay?: boolean replayLimit?: number limits: () => Record footer: FooterApi trace?: Trace signal?: AbortSignal } type Wait = { tick: number armed: boolean live: boolean done: Deferred.Deferred } export type SessionTurnInput = { agent: string | undefined model: RunInput["model"] variant: string | undefined prompt: RunPrompt files: RunFilePart[] includeFiles: boolean signal?: AbortSignal } export type SessionTransport = { runPromptTurn(input: SessionTurnInput): Promise selectSubagent(sessionID: string | undefined): void close(): Promise } type State = { data: SessionData subagent: SubagentData wait?: Wait tick: number fault?: unknown footerView: FooterView blockerTick: number selectedSubagent?: string blockers: Map } type TransportService = { readonly runPromptTurn: (input: SessionTurnInput) => Effect.Effect readonly selectSubagent: (sessionID: string | undefined) => Effect.Effect readonly close: () => Effect.Effect } class Service extends Context.Service()("@opencode/RunStreamTransport") {} function sid(event: Event): string | undefined { if (event.type === "message.updated") { return event.properties.sessionID } if (event.type === "message.part.delta") { return event.properties.sessionID } if (event.type === "message.part.updated") { return event.properties.part.sessionID } if ( event.type === "session.next.shell.started" || event.type === "session.next.shell.ended" || event.type === "permission.asked" || event.type === "permission.replied" || event.type === "question.asked" || event.type === "question.replied" || event.type === "question.rejected" || event.type === "session.error" || event.type === "session.status" ) { return event.properties.sessionID } return undefined } function isEvent(value: unknown): value is Event { if (!value || typeof value !== "object" || Array.isArray(value)) { return false } const type = Reflect.get(value, "type") const properties = Reflect.get(value, "properties") return typeof type === "string" && !!properties && typeof properties === "object" } function isGlobalEvent(value: unknown): value is GlobalEvent { if (!value || typeof value !== "object" || Array.isArray(value)) { return false } const payload = Reflect.get(value, "payload") return !!payload && typeof payload === "object" } function globalPayloadEvent(value: unknown): Event | undefined { if (!isGlobalEvent(value)) { return undefined } const payload = value.payload if (payload.type === "sync") { return undefined } return isEvent(payload) ? payload : undefined } function isMatchingDisposeEvent(value: unknown, directory: string | undefined): boolean { if (!directory || !isGlobalEvent(value)) { return false } if (value.directory !== directory) { return false } return value.payload.type === "server.instance.disposed" } function active(event: Event, sessionID: string): boolean { if (sid(event) !== sessionID) { return false } if (event.type === "message.updated") { return event.properties.info.role === "assistant" } if (event.type === "message.part.delta" || event.type === "message.part.updated") { return false } if (event.type !== "session.status") { return true } return event.properties.status.type !== "idle" } // Races the turn's deferred completion against an abort signal. function waitTurn(done: Wait["done"], signal: AbortSignal) { return Effect.raceAll([ Deferred.await(done).pipe(Effect.as("idle" as const), Effect.exit), Effect.callback<"abort">((resume) => { if (signal.aborted) { resume(Effect.succeed("abort")) return Effect.void } const onAbort = () => { signal.removeEventListener("abort", onAbort) resume(Effect.succeed("abort")) } signal.addEventListener("abort", onAbort, { once: true }) return Effect.sync(() => signal.removeEventListener("abort", onAbort)) }).pipe(Effect.exit), ]).pipe(Effect.flatMap((exit) => (Exit.isFailure(exit) ? Effect.failCause(exit.cause) : Effect.succeed(exit.value)))) } export function formatUnknownError(error: unknown): string { if (typeof error === "string") { return error } if (error instanceof Error) { return error.message || error.name } if (error && typeof error === "object") { const value = error as { message?: unknown; name?: unknown } if (typeof value.message === "string" && value.message.trim()) { return value.message } if (typeof value.name === "string" && value.name.trim()) { return value.name } } return "unknown error" } function sameView(a: FooterView, b: FooterView) { if (a.type !== b.type) { return false } if (a.type === "prompt" && b.type === "prompt") { return true } if (a.type === "prompt" || b.type === "prompt") { return false } return a.request === b.request } function blockerOrder(order: Map, id: string) { return order.get(id) ?? Number.MAX_SAFE_INTEGER } function firstByOrder(left: T[], right: T[], order: Map) { return [...left, ...right].sort((a, b) => { const next = blockerOrder(order, a.id) - blockerOrder(order, b.id) if (next !== 0) { return next } return a.id.localeCompare(b.id) })[0] } function pickView(data: SessionData, subagent: SubagentData, order: Map): FooterView { return pickBlockerView({ permission: firstByOrder(data.permissions, listSubagentPermissions(subagent), order), question: firstByOrder(data.questions, listSubagentQuestions(subagent), order), }) } function composeFooter(input: { patch?: FooterPatch subagent?: FooterSubagentState current: FooterView previous: FooterView }) { let footer: FooterOutput | undefined if (input.subagent) { footer = { ...footer, subagent: input.subagent, } } if (!sameView(input.previous, input.current)) { footer = { ...footer, view: input.current, } } if (input.current.type !== "prompt") { footer = { ...footer, patch: { ...input.patch, status: blockerStatus(input.current), }, } return footer } if (input.patch) { footer = { ...footer, patch: input.patch, } return footer } if (input.previous.type !== "prompt") { footer = { ...footer, patch: { status: "", }, } } return footer } function traceTabs(trace: Trace | undefined, prev: FooterSubagentTab[], next: FooterSubagentTab[]) { const before = new Map(prev.map((item) => [item.sessionID, item])) const after = new Map(next.map((item) => [item.sessionID, item])) for (const [sessionID, tab] of after) { if (sameSubagentTab(before.get(sessionID), tab)) { continue } trace?.write("subagent.tab", { sessionID, tab, }) } for (const sessionID of before.keys()) { if (after.has(sessionID)) { continue } trace?.write("subagent.tab", { sessionID, cleared: true, }) } } function createLayer(input: StreamInput) { return Layer.fresh( Layer.effect( Service, Effect.gen(function* () { const scope = yield* Scope.make() const abort = yield* Scope.provide(scope)( Effect.acquireRelease( Effect.sync(() => new AbortController()), (abort) => Effect.sync(() => abort.abort()), ), ) let closed = false let closeStream = () => {} const halt = () => { abort.abort() } const stop = () => { input.signal?.removeEventListener("abort", halt) abort.abort() closeStream() } const closeScope = () => { if (closed) { return Effect.void } closed = true stop() return Scope.close(scope, Exit.void) } input.signal?.addEventListener("abort", halt, { once: true }) yield* Effect.addFinalizer(() => closeScope()) const events = yield* Scope.provide(scope)( Effect.acquireRelease( Effect.promise(() => input.sdk.global.event({ signal: abort.signal, }), ), (events) => Effect.sync(() => { void events.stream.return(StreamClosed).catch(() => {}) }), ), ) closeStream = () => { void events.stream.return(StreamClosed).catch(() => {}) } input.trace?.write("recv.subscribe", { sessionID: input.sessionID, }) const state: State = { data: createSessionData(), subagent: createSubagentData(), tick: 0, footerView: { type: "prompt" }, blockerTick: 0, blockers: new Map(), } let booting = true const buffered: Event[] = [] const replayedParts = new Set() const recovering = new Set() const tracked = (sessionID: string | undefined) => sessionID === input.sessionID || (!!sessionID && state.subagent.tabs.has(sessionID)) const currentSubagentState = () => { if (state.selectedSubagent && !state.subagent.tabs.has(state.selectedSubagent)) { state.selectedSubagent = undefined } return snapshotSelectedSubagentData(state.subagent, state.selectedSubagent) } const seedBlocker = (id: string) => { if (state.blockers.has(id)) { return } state.blockerTick += 1 state.blockers.set(id, state.blockerTick) } const trackBlocker = (event: Event) => { if (event.type !== "permission.asked" && event.type !== "question.asked") { return } if (event.properties.sessionID !== input.sessionID && !state.subagent.tabs.has(event.properties.sessionID)) { return } seedBlocker(event.properties.id) } const releaseBlocker = (event: Event) => { if ( event.type !== "permission.replied" && event.type !== "question.replied" && event.type !== "question.rejected" ) { return } state.blockers.delete(event.properties.requestID) } const syncFooter = (commits: StreamCommit[], patch?: FooterPatch, nextSubagent?: FooterSubagentState) => { const current = pickView(state.data, state.subagent, state.blockers) const footer = composeFooter({ patch, subagent: nextSubagent, current, previous: state.footerView, }) if (commits.length === 0 && !footer) { state.footerView = current return } input.trace?.write("reduce.output", { commits, footer: traceFooterOutput(footer), }) writeSessionOutput( { footer: input.footer, trace: input.trace, }, { commits, footer, }, ) state.footerView = current } const resolveShellAgent = Effect.fn("RunStreamTransport.resolveShellAgent")(function* ( agent: string | undefined, ) { if (agent) { return agent } const list = yield* Effect.promise(() => input.sdk.app.agents(input.directory ? { directory: input.directory } : undefined, { throwOnError: true }), ).pipe( Effect.map((item) => item.data ?? []), Effect.orElseSucceed(() => []), ) const next = list.find((item) => item.mode !== "subagent" && item.hidden !== true)?.name if (next) { return next } return yield* Effect.fail(new Error("no primary agent available for shell mode")) }) const recoverQuestion = Effect.fn("RunStreamTransport.recoverQuestion")(function* (partID: string) { if (recovering.has(partID)) { return } recovering.add(partID) try { while (!closed && !abort.signal.aborted && !input.footer.isClosed) { if (state.data.questions.length > 0 || !state.data.tools.has(partID)) { return } const questions = yield* Effect.promise(() => input.sdk.question.list()).pipe( Effect.map((item) => (item.data ?? []).filter((request) => request.sessionID === input.sessionID)), Effect.orElseSucceed(() => []), ) if (state.data.questions.length > 0 || !state.data.tools.has(partID)) { return } if (questions.length > 0) { bootstrapSessionData({ data: state.data, messages: [], permissions: [], questions, }) for (const request of questions) { seedBlocker(request.id) } input.trace?.write("question.recover", { sessionID: input.sessionID, requests: questions.map((request) => request.id), }) syncFooter([]) return } yield* Effect.sleep("250 millis") } } finally { recovering.delete(partID) } }) const messages = (sessionID: string, limit?: number) => Effect.promise(() => input.sdk.session.messages({ sessionID, ...(typeof limit === "number" ? { limit } : {}), }), ).pipe( Effect.map((item) => item.data ?? []), Effect.orElseSucceed(() => []), ) const bootstrapSubagentHistory = Effect.fn("RunStreamTransport.bootstrapSubagentHistory")(function* ( sessions: string[], ) { yield* Effect.forEach( sessions, (sessionID) => messages(sessionID, SUBAGENT_CALL_BOOTSTRAP_LIMIT).pipe( Effect.tap((messagesList) => Effect.sync(() => { if ( !bootstrapSubagentCalls({ data: state.subagent, sessionID, messages: messagesList, thinking: input.thinking, limits: input.limits(), }) ) { return } syncFooter([], undefined, currentSubagentState()) }), ), ), { concurrency: 4, discard: true, }, ) }) const bootstrap = Effect.fn("RunStreamTransport.bootstrap")(function* () { const [messagesList, children, permissions, questions] = yield* Effect.all( [ messages( input.sessionID, input.replay ? input.replayLimit === undefined ? undefined : Math.max(input.replayLimit, SUBAGENT_BOOTSTRAP_LIMIT) : SUBAGENT_BOOTSTRAP_LIMIT, ), Effect.promise(() => input.sdk.session.children({ sessionID: input.sessionID, }), ).pipe( Effect.map((item) => item.data ?? []), Effect.orElseSucceed(() => []), ), Effect.promise(() => input.sdk.permission.list()).pipe( Effect.map((item) => item.data ?? []), Effect.orElseSucceed(() => []), ), Effect.promise(() => input.sdk.question.list()).pipe( Effect.map((item) => item.data ?? []), Effect.orElseSucceed(() => []), ), ], { concurrency: "unbounded", }, ) const sessionPermissions = permissions.filter((item) => item.sessionID === input.sessionID) const sessionQuestions = questions.filter((item) => item.sessionID === input.sessionID) const history = input.replay ? replaySession({ messages: messagesList, permissions: sessionPermissions, questions: sessionQuestions, thinking: input.thinking, limits: input.limits(), }) : undefined const replay = history && input.replayLimit !== undefined && messagesList.length > input.replayLimit ? replaySession({ messages: messagesList.slice(-input.replayLimit), permissions: sessionPermissions, questions: sessionQuestions, thinking: input.thinking, limits: input.limits(), }) : history replayedParts.clear() if (history) { state.data = history.data } if (!history) { bootstrapSessionData({ data: state.data, messages: messagesList, permissions: sessionPermissions, questions: sessionQuestions, }) } if (replay) { for (const [partID] of replay.data.text) { if (!replay.data.part.has(partID)) { continue } replayedParts.add(partID) } } bootstrapSubagentData({ data: state.subagent, messages: messagesList, children, permissions, questions, }) clearFinishedSubagents(state.subagent) for (const request of [ ...state.data.permissions, ...listSubagentPermissions(state.subagent), ...state.data.questions, ...listSubagentQuestions(state.subagent), ].sort((a, b) => a.id.localeCompare(b.id))) { seedBlocker(request.id) } if (replay) { const activeCommitIDs = new Set([...state.data.part.keys(), ...state.data.tools]) for (const commit of replay.commits) { input.trace?.write("ui.commit", commit) input.footer.append(commit) if (commit.partID && activeCommitIDs.has(commit.partID)) { continue } yield* Effect.promise(() => input.footer.idle()).pipe(Effect.orElseSucceed(() => undefined)) } } const snapshot = currentSubagentState() traceTabs(input.trace, [], snapshot.tabs) syncFooter([], replay?.patch, snapshot) if (replay) { yield* Effect.promise(() => input.footer.idle()).pipe(Effect.orElseSucceed(() => undefined)) } booting = false yield* drainBuffered() const sessions = [...state.subagent.tabs.keys()] if (sessions.length === 0) { return } yield* bootstrapSubagentHistory(sessions).pipe( Effect.forkIn(scope, { startImmediately: true }), Effect.asVoid, ) }) const idle = Effect.fn("RunStreamTransport.idle")((fallback: boolean) => Effect.promise(() => input.sdk.session.status()).pipe( Effect.map((out) => { const item = out.data?.[input.sessionID] return !item || item.type === "idle" }), Effect.orElseSucceed(() => fallback), ), ) const fail = Effect.fn("RunStreamTransport.fail")(function* (error: unknown) { if (state.fault) { return } state.fault = error const next = state.wait state.wait = undefined if (!next) { return } yield* Deferred.fail(next.done, error).pipe(Effect.ignore) }) const touch = (event: Event) => { const next = state.wait if (!next || !active(event, input.sessionID)) { return } next.live = true } const complete = Effect.fn("RunStreamTransport.complete")(function* (next: Wait, fallback: boolean) { if (state.wait !== next || !next.armed || !next.live) { return } if (!(yield* idle(fallback)) || state.wait !== next) { return } state.tick = next.tick + 1 state.wait = undefined yield* Deferred.succeed(next.done, undefined).pipe(Effect.ignore) }) const mark = Effect.fn("RunStreamTransport.mark")(function* (event: Event) { if ( event.type !== "session.status" || event.properties.sessionID !== input.sessionID || event.properties.status.type !== "idle" ) { return } const next = state.wait if (!next) { return } yield* complete(next, true) }) const poll = Effect.fn("RunStreamTransport.poll")(function* (next: Wait, signal: AbortSignal) { while (state.wait === next && !signal.aborted && !input.footer.isClosed && !closed) { yield* Effect.sleep("250 millis") yield* complete(next, false) } }) const flush = (type: "turn.abort" | "turn.cancel") => { const commits: StreamCommit[] = [] flushInterrupted(state.data, commits) syncFooter(commits) input.trace?.write(type, { sessionID: input.sessionID, }) } const applyEvent = Effect.fn("RunStreamTransport.applyEvent")(function* (event: Event) { if (event.type === "message.part.delta" && event.properties.sessionID === input.sessionID) { if (replayedParts.has(event.properties.partID)) { const seen = state.data.text.get(event.properties.partID) ?? "" if (seen.endsWith(event.properties.delta)) { return } replayedParts.delete(event.properties.partID) } } trackBlocker(event) const prev = event.type === "message.part.updated" ? listSubagentTabs(state.subagent) : undefined const next = reduceSessionData({ data: state.data, event, sessionID: input.sessionID, thinking: input.thinking, limits: input.limits(), }) state.data = next.data if ( event.type === "message.part.updated" && event.properties.part.sessionID === input.sessionID && event.properties.part.type === "tool" && event.properties.part.tool === "question" && event.properties.part.state.status === "running" && state.data.questions.length === 0 ) { yield* recoverQuestion(event.properties.part.id).pipe( Effect.forkIn(scope, { startImmediately: true }), Effect.asVoid, ) } const changed = reduceSubagentData({ data: state.subagent, event, sessionID: input.sessionID, thinking: input.thinking, limits: input.limits(), }) if (changed && prev) { traceTabs(input.trace, prev, listSubagentTabs(state.subagent)) } releaseBlocker(event) syncFooter(next.commits, next.footer?.patch, changed ? currentSubagentState() : undefined) touch(event) yield* mark(event) }) const drainBuffered = Effect.fn("RunStreamTransport.drainBuffered")(function* () { let pending = buffered.splice(0) while (pending.length > 0) { const next: Event[] = [] let changed = false for (const event of pending) { if (!tracked(sid(event))) { next.push(event) continue } changed = true yield* applyEvent(event) } if (!changed) { buffered.push(...next) return } pending = next } }) const watch = Effect.fn("RunStreamTransport.watch")(() => Stream.fromAsyncIterable(events.stream, (error) => error instanceof Error ? error : new Error(String(error)), ).pipe( Stream.takeUntil(() => input.footer.isClosed || abort.signal.aborted), Stream.runForEach( Effect.fn("RunStreamTransport.event")(function* (item: unknown) { if (input.footer.isClosed) { abort.abort() return } if (isMatchingDisposeEvent(item, input.directory)) { yield* fail(new Error("instance disposed")) yield* closeScope() return } const event = globalPayloadEvent(item) if (!event) { return } const sessionID = sid(event) if (booting) { if (sessionID) { input.trace?.write("recv.event", event) buffered.push(event) } return } if (!tracked(sessionID)) { if (sessionID) { input.trace?.write("recv.event", event) buffered.push(event) } return } input.trace?.write("recv.event", event) yield* applyEvent(event) yield* drainBuffered() }), ), Effect.catch((error) => (abort.signal.aborted ? Effect.void : fail(error))), Effect.ensuring( Effect.gen(function* () { if (!abort.signal.aborted && !state.fault) { yield* fail(new Error("global event stream closed")) } closeStream() }), ), ), ) yield* Scope.provide(scope)(watch().pipe(Effect.forkScoped)) yield* bootstrap() const runPromptTurn = Effect.fn("RunStreamTransport.runPromptTurn")(function* (next: SessionTurnInput) { if (closed || next.signal?.aborted || input.footer.isClosed) { return } if (state.fault) { yield* Effect.fail(state.fault) return } if (state.wait) { yield* Effect.fail(new Error("prompt already running")) return } const prev = listSubagentTabs(state.subagent) if (clearFinishedSubagents(state.subagent)) { const snapshot = currentSubagentState() traceTabs(input.trace, prev, snapshot.tabs) syncFooter([], undefined, snapshot) } const item: Wait = { tick: state.tick, armed: false, live: false, done: yield* Deferred.make(), } state.wait = item state.data.announced = false const turn = new AbortController() const stop = () => { turn.abort() } next.signal?.addEventListener("abort", stop, { once: true }) abort.signal.addEventListener("abort", stop, { once: true }) yield* poll(item, turn.signal).pipe(Effect.forkIn(scope, { startImmediately: true }), Effect.asVoid) const req = { sessionID: input.sessionID, agent: next.agent, model: next.model, variant: next.variant, parts: [ ...(next.includeFiles ? next.files : []), { type: "text" as const, text: next.prompt.text }, ...next.prompt.parts, ], } const command = next.prompt.command const send = next.prompt.mode === "shell" ? Effect.sync(() => { input.trace?.write("send.shell", { sessionID: input.sessionID, command: next.prompt.text, }) }).pipe( Effect.andThen( resolveShellAgent(next.agent) .pipe( Effect.flatMap((agent) => Effect.promise(() => input.sdk.session.shell( { sessionID: input.sessionID, agent, model: next.model, command: next.prompt.text, }, { signal: turn.signal, throwOnError: true }, ), ), ), ) .pipe( Effect.tap(() => Effect.sync(() => { input.trace?.write("send.shell.ok", { sessionID: input.sessionID, }) item.armed = true item.live = true }), ), Effect.flatMap(() => Deferred.succeed(item.done, undefined).pipe(Effect.ignore)), Effect.catch((error) => Deferred.fail(item.done, error).pipe(Effect.ignore)), Effect.forkIn(scope, { startImmediately: true }), Effect.asVoid, ), ), ) : command ? Effect.sync(() => { input.trace?.write("send.command", { sessionID: input.sessionID, command: command.name }) }).pipe( Effect.andThen( Effect.promise(() => input.sdk.session.command( { sessionID: input.sessionID, agent: next.agent, model: next.model ? `${next.model.providerID}/${next.model.modelID}` : undefined, variant: next.variant, command: command.name, arguments: command.arguments, parts: [ ...(next.includeFiles ? next.files : []), ...next.prompt.parts.filter( (item): item is Extract => item.type === "file", ), ], }, { signal: turn.signal }, ), ).pipe( Effect.tap(() => Effect.sync(() => { input.trace?.write("send.command.ok", { sessionID: input.sessionID, command: command.name, }) item.armed = true item.live = true }), ), Effect.flatMap(() => Deferred.succeed(item.done, undefined).pipe(Effect.ignore)), Effect.catch((error) => Deferred.fail(item.done, error).pipe(Effect.ignore)), Effect.forkIn(scope, { startImmediately: true }), Effect.asVoid, ), ), ) : Effect.sync(() => { input.trace?.write("send.prompt", req) }).pipe( Effect.andThen( Effect.promise(() => input.sdk.session.promptAsync(req, { signal: turn.signal, }), ), ), Effect.tap(() => Effect.sync(() => { input.trace?.write("send.prompt.ok", { sessionID: input.sessionID, }) item.armed = true }), ), ) yield* send.pipe( Effect.flatMap(() => { if (turn.signal.aborted || next.signal?.aborted || input.footer.isClosed || closed) { if (state.wait === item) { state.wait = undefined } flush("turn.abort") return Effect.void } if (!input.footer.isClosed && !state.data.announced) { input.trace?.write("ui.patch", { phase: "running", status: "waiting for assistant", }) input.footer.event({ type: "turn.wait", }) } if (state.tick > item.tick) { if (state.wait === item) { state.wait = undefined } return Effect.void } return waitTurn(item.done, turn.signal).pipe( Effect.flatMap((status) => Effect.sync(() => { if (state.wait === item) { state.wait = undefined } if (status === "abort") { flush("turn.abort") } }), ), ) }), Effect.catch((error) => { if (state.wait === item) { state.wait = undefined } const canceled = turn.signal.aborted || next.signal?.aborted === true || input.footer.isClosed || closed if (canceled) { flush("turn.cancel") return Effect.void } if (error === state.fault) { return Effect.fail(error) } input.trace?.write("send.prompt.error", { sessionID: input.sessionID, error: formatUnknownError(error), }) return Effect.fail(error) }), Effect.ensuring( Effect.sync(() => { input.trace?.write("turn.end", { sessionID: input.sessionID, }) next.signal?.removeEventListener("abort", stop) abort.signal.removeEventListener("abort", stop) }), ), ) return }) const selectSubagent = Effect.fn("RunStreamTransport.selectSubagent")((sessionID: string | undefined) => Effect.sync(() => { if (closed) { return } const next = sessionID && state.subagent.tabs.has(sessionID) ? sessionID : undefined if (state.selectedSubagent === next) { return } state.selectedSubagent = next syncFooter([], undefined, currentSubagentState()) }), ) const close = Effect.fn("RunStreamTransport.close")(function* () { yield* closeScope() }) return Service.of({ runPromptTurn, selectSubagent, close, }) }), ), ) } // Opens an SDK event subscription and returns a SessionTransport. // // The background `watch` loop consumes every SDK event, runs it through the // reducer, and writes output to the footer. When a session.status idle // event arrives, it resolves the current turn's Wait so runPromptTurn() // can return. // // The transport is single-turn: only one runPromptTurn() call can be active // at a time. The prompt queue enforces this from above. export async function createSessionTransport(input: StreamInput): Promise { const runtime = makeRuntime(Service, createLayer(input)) await runtime.runPromise(() => Effect.void) return { runPromptTurn: (next) => runtime.runPromise((svc) => svc.runPromptTurn(next)), selectSubagent: (sessionID) => runtime.runSync((svc) => svc.selectSubagent(sessionID)), close: () => runtime.runPromise((svc) => svc.close()), } }