diff --git a/src/cli/print.ts b/src/cli/print.ts index 3e6ae3e8d..5abf2d7ae 100644 --- a/src/cli/print.ts +++ b/src/cli/print.ts @@ -54,6 +54,7 @@ import { getSessionState, notifySessionStateChanged, notifySessionMetadataChanged, + registerSdkEventConsumer, setPermissionModeChangedListener, type RequiresActionDetails, type SessionExternalMetadata, @@ -1029,6 +1030,13 @@ function runHeadlessStreaming( // Same queue sendRequest() enqueues to — one FIFO for everything. const output = structuredIO.outbound + // Push session_state_changed events directly to output (bypasses SDK + // event queue). Aligned with OpenCode: status transitions reach clients + // immediately via direct push, not delayed until drainSdkEvents(). + const unregisterSdkConsumer = registerSdkEventConsumer(event => { + output.enqueue(event as unknown as StdoutMessage) + }) + // Ctrl+C in -p mode: abort the in-flight query, then shut down gracefully. // gracefulShutdown persists session state and flushes analytics, with a // failsafe timer that force-exits if cleanup hangs. diff --git a/src/query.ts b/src/query.ts index 1ece8ce48..0fd3755a5 100644 --- a/src/query.ts +++ b/src/query.ts @@ -77,6 +77,7 @@ import { } from './utils/messageQueueManager.js' import { notifyCommandLifecycle } from './utils/commandLifecycle.js' import { headlessProfilerCheckpoint } from './utils/headlessProfiler.js' +import { notifySessionStateChanged } from './utils/sessionState.js' import { getRuntimeMainLoopModel, renderModelName, @@ -377,6 +378,11 @@ async function* queryLoop( toolUseContext, ) + // Emit heartbeat at each iteration start so SDK consumers see + // the session is still active during long tool execution pauses. + // Aligned with OpenCode: each AI loop iteration sets 'busy'. + notifySessionStateChanged('running') + yield { type: 'stream_request_start' } queryCheckpoint('query_fn_entry') diff --git a/src/server/eventBus.ts b/src/server/eventBus.ts index 95a0a638f..e93907c12 100644 --- a/src/server/eventBus.ts +++ b/src/server/eventBus.ts @@ -45,13 +45,6 @@ export class EventBus { data: JSON.stringify({ type: 'server.connected' }), }) - for (const buffered of this.buffer) { - void sendAndCleanup({ - event: buffered.event, - data: JSON.stringify(buffered.data), - }) - } - return id } @@ -60,6 +53,7 @@ export class EventBus { } publish(event: string, data: unknown): void { + const tPublish = event === 'session.status' ? Date.now() : 0 const entry: BusEvent = { event, data } this.buffer.push(entry) if (this.buffer.length > this.bufferSize) { @@ -67,6 +61,7 @@ export class EventBus { } const payload = JSON.stringify(data) const deadIds: string[] = [] + let clientCount = 0 for (const client of this.clients.values()) { if (client.writer) { if (client.sessionIdFilter) { @@ -85,6 +80,7 @@ export class EventBus { } } } + clientCount++ client.writer.writeSSE({ event, data: payload }).catch(() => { deadIds.push(client.id) }) @@ -95,6 +91,13 @@ export class EventBus { this.clients.delete(id) } } + if (event === 'session.status' && tPublish > 0) { + const dataObj = data as Record + const sessionId = dataObj?.sessionID ?? dataObj?.session_id + process.stderr.write( + `[serve:timing:${sessionId}] eventBus.publish('session.status') → ${clientCount} SSE clients, +${Date.now() - tPublish}ms in publish loop\n`, + ) + } } publishSessionEvent( diff --git a/src/server/routes/session.ts b/src/server/routes/session.ts index 5577281bf..d1584a469 100644 --- a/src/server/routes/session.ts +++ b/src/server/routes/session.ts @@ -1,6 +1,7 @@ import { Hono } from 'hono' import { streamSSE } from 'hono/streaming' import type { SessionManager } from '../sessionManager.js' +import type { EventBus } from '../eventBus.js' import { getScriptArgsForChild } from '../childSpawn.js' import { badRequest, @@ -10,6 +11,7 @@ import { conflict, } from '../errors.js' import { listSessionsImpl } from '../../utils/listSessionsImpl.js' +import { canonicalizePath } from '../../utils/sessionStoragePortable.js' /** 从磁盘历史记录查找 session 的原始 cwd，用于 resume 时恢复正确的工作目录 */ async function getHistoryCwd(sessionId: string): Promise { @@ -120,6 +122,7 @@ function ssePrompt( export function createSessionRoutes( sessionManager: SessionManager, + eventBus: EventBus, ): Hono { return new Hono() .post('/session', async c => { @@ -239,8 +242,13 @@ export function createSessionRoutes( // 补充内存中有但磁盘还没落盘的活跃 session（刚创建还没写过消息的） const historyIds = new Set(historySessions.map(s => s.sessionId)) + const canonicalDir = dir ? await canonicalizePath(dir).catch(() => dir) : undefined for (const handle of sessionManager.getAllSessions()) { if (!historyIds.has(handle.sessionId)) { + if (canonicalDir) { + const handleCwd = await canonicalizePath(handle.cwd).catch(() => handle.cwd) + if (handleCwd !== canonicalDir) continue + } const info = handle.getInfo() merged.push({ ...info, title: info.title ?? '' }) } @@ -346,7 +354,12 @@ export function createSessionRoutes( }) .delete('/session/:sessionID', async c => { const id = c.req.param('sessionID') +<<<<<<< HEAD await sessionManager.deleteSession(id) +======= + const deleted = sessionManager.deleteSession(id) + if (!deleted) throw notFound('session not found') +>>>>>>> dab72c6b (feat: 对齐 OpenCode 即时推送架构，session status 事件绕过队列直接推送) return c.json({ deleted: true }) }) .post('/session/:sessionID/prompt', async c => { @@ -401,27 +414,12 @@ export function createSessionRoutes( return ssePrompt(handle, id, content, c, body.parts, body.messageID) }) .post('/session/:sessionID/prompt_async', async c => { + const tEntry = Date.now() const id = c.req.param('sessionID') - let handle = sessionManager.getSession(id) - - // 历史 session 不在内存中，自动恢复 - if (!handle) { - const cwd = await getHistoryCwd(id) - try { - handle = await sessionManager.createSession({ - cwd, - sessionId: id, - resumeSessionId: id, - execPath: process.execPath, - scriptArgs: getScriptArgsForChild(), - }) - await handle.waitReady(30000) - } catch (err) { - throw sessionError(err instanceof Error ? err.message : 'Failed to resume session') - } - } - + // Parse body FIRST — independent of session lookup, so we can + // emit busy status immediately even for historical sessions that + // need to create+waitReady (3–30s). const body = await c.req.json<{ content?: string parts?: Array> @@ -444,22 +442,93 @@ export function createSessionRoutes( if (!content) { throw badRequest('content is required') } - if (handle.prompting) { + + // Emit busy IMMEDIATELY — before session recovery or child-process + // init. Aligned with OpenCode: status.set fires in the request + // handler synchronously, not after waiting for the child process. + eventBus.publish('session.status', { + sessionID: id, + status: { type: 'busy' }, + }) + const tBusyEmitted = Date.now() + process.stderr.write( + `[serve:timing:${id}] busy emitted at +${tBusyEmitted - tEntry}ms (immediate)\n`, + ) + + let handle = sessionManager.getSession(id) + + // 历史 session 不在内存中，自动恢复 + if (!handle) { + process.stderr.write( + `[serve:timing:${id}] historical session, starting recovery...\n`, + ) + const cwd = await getHistoryCwd(id) + try { + handle = await sessionManager.createSession({ + cwd, + sessionId: id, + resumeSessionId: id, + execPath: process.execPath, + scriptArgs: getScriptArgsForChild(), + }) + const tCreated = Date.now() + process.stderr.write( + `[serve:timing:${id}] session created at +${tCreated - tEntry}ms, waiting for ready...\n`, + ) + await handle.waitReady(30000) + const tReady = Date.now() + process.stderr.write( + `[serve:timing:${id}] session ready at +${tReady - tEntry}ms\n`, + ) + } catch (err) { + // Recovery failed — clear the busy status we emitted earlier + eventBus.publish('session.status', { + sessionID: id, + status: { type: 'idle' }, + }) + throw sessionError(err instanceof Error ? err.message : 'Failed to resume session') + } + } else if (handle.prompting) { throw conflict('session is already processing a prompt') } + const tQueued = Date.now() + process.stderr.write( + `[serve:timing:${id}] session ready, queuing prompt at +${tQueued - tEntry}ms\n`, + ) + void (async () => { try { + const tFire = Date.now() + process.stderr.write( + `[serve:timing:${id}] IIFE starts, status=${handle.status} at +${tFire - tEntry}ms\n`, + ) if (handle.status !== 'running') { await handle.waitReady() + const tReady = Date.now() + process.stderr.write( + `[serve:timing:${id}] IIFE waitReady done at +${tReady - tEntry}ms\n`, + ) } if (body.model?.modelID) { + const tModelStart = Date.now() try { await handle.setModel(body.model.modelID) } catch {} + const tModelDone = Date.now() + process.stderr.write( + `[serve:timing:${id}] setModel done at +${tModelDone - tEntry}ms (took ${tModelDone - tModelStart}ms)\n`, + ) } +<<<<<<< HEAD const effectiveAgent = body.agent ?? (agentParts[0] as Record | undefined)?.name as string | undefined if (effectiveAgent && effectiveAgent !== handle.agent) { try { await handle.setAgent(effectiveAgent) } catch {} } +======= + const tPromptCall = Date.now() + process.stderr.write( + `[serve:timing:${id}] calling prompt() at +${tPromptCall - tEntry}ms\n`, + ) +>>>>>>> dab72c6b (feat: 对齐 OpenCode 即时推送架构，session status 事件绕过队列直接推送) handle.prompt(content, { parts: body.parts, messageID: body.messageID }).catch(() => {}) } catch {} diff --git a/src/server/server.ts b/src/server/server.ts index 3140570b3..e258402a4 100644 --- a/src/server/server.ts +++ b/src/server/server.ts @@ -45,7 +45,7 @@ export function startServer( app.route('/', createHealthRoutes(sessionManager)) app.route('/', createInfoRoutes(sessionManager)) - app.route('/', createSessionRoutes(sessionManager)) + app.route('/', createSessionRoutes(sessionManager, eventBus)) app.route('/', createEventRoutes(eventBus)) app.route('/', createPermissionRoutes(sessionManager)) app.route('/', createQuestionRoutes(sessionManager)) diff --git a/src/server/sessionHandle.ts b/src/server/sessionHandle.ts index 933f3e309..234424716 100644 --- a/src/server/sessionHandle.ts +++ b/src/server/sessionHandle.ts @@ -485,13 +485,18 @@ export class SessionHandle { } async prompt(content: string, opts?: { parts?: Array>; messageID?: string }): Promise<{ done: boolean; error?: string }> { + const tEnter = Date.now() if (this._status === 'stopped') { throw new Error( `Session ${this.sessionId} is stopped`, ) } if (this._status !== 'running') { + const tWaitStart = Date.now() await this.waitReady() + process.stderr.write( + `[serve:timing:${this.sessionId}] prompt() waitReady took ${Date.now() - tWaitStart}ms\n`, + ) } if (this._prompting) { throw new Error( @@ -499,8 +504,18 @@ export class SessionHandle { ) } this._prompting = true - this._busyStatus = { type: 'busy' } - this.emitBusyStatus() + + // Busy status was already emitted from the route handler (immediate). + // Only emit here if this prompt() was called directly (not via prompt_async). + const alreadyBusy = this._busyStatus?.type === 'busy' + process.stderr.write( + `[serve:timing:${this.sessionId}] prompt() entry, alreadyBusy=${alreadyBusy}, +0ms from prompt() call\n`, + ) + + if (!alreadyBusy) { + this._busyStatus = { type: 'busy' } + this.emitBusyStatus() + } this.lastActiveAt = Date.now() const uuid = opts?.messageID ?? crypto.randomUUID() diff --git a/src/server/sessionMessageRouter.ts b/src/server/sessionMessageRouter.ts index 8eb057a6f..1cf0ed248 100644 --- a/src/server/sessionMessageRouter.ts +++ b/src/server/sessionMessageRouter.ts @@ -149,6 +149,9 @@ function handleResultMessage(msg: StdoutMessage, ctx: MessageRouterCtx): void { if (usage?.output_tokens) ctx.addOutputTokens(usage.output_tokens) ctx.setBusyStatus({ type: 'idle' }) ctx.emitBusyStatus() + process.stderr.write( + `[serve:timing:${ctx.sessionId}] result → idle emitted\n`, + ) ctx.emitEvent('result', msg) ctx.resolvePrompt({ done: true }) } diff --git a/src/utils/__tests__/sessionState.test.ts b/src/utils/__tests__/sessionState.test.ts index de047658b..76bd99d4f 100644 --- a/src/utils/__tests__/sessionState.test.ts +++ b/src/utils/__tests__/sessionState.test.ts @@ -1,10 +1,14 @@ import { beforeEach, describe, expect, test } from 'bun:test' import { notifyAutomationStateChanged, - notifySessionStateChanged, + notifyPermissionModeChanged, notifySessionMetadataChanged, + notifySessionStateChanged, + registerSdkEventConsumer, resetSessionStateForTests, + setPermissionModeChangedListener, setSessionMetadataChangedListener, + setSessionStateChangedListener, } from '../sessionState' describe('sessionState metadata replay', () => { @@ -172,3 +176,275 @@ describe('sessionState metadata replay', () => { ]) }) }) + +describe('sessionState multi-subscriber', () => { + beforeEach(() => { + resetSessionStateForTests() + }) + + test('setSessionStateChangedListener returns unsubscribe function', () => { + const calls: string[] = [] + const unsub = setSessionStateChangedListener(() => calls.push('L1')) + + notifySessionStateChanged('running') + expect(calls).toEqual(['L1']) + + unsub() + notifySessionStateChanged('idle') + expect(calls).toEqual(['L1']) + }) + + test('multiple state listeners all receive notifications', () => { + const calls: string[] = [] + setSessionStateChangedListener(() => calls.push('L1')) + setSessionStateChangedListener(() => calls.push('L2')) + setSessionStateChangedListener(() => calls.push('L3')) + + notifySessionStateChanged('running') + + expect(calls).toEqual(['L1', 'L2', 'L3']) + }) + + test('unsubscribed state listener does not receive subsequent notifications', () => { + const calls: string[] = [] + const unsub1 = setSessionStateChangedListener(() => calls.push('L1')) + setSessionStateChangedListener(() => calls.push('L2')) + + notifySessionStateChanged('running') + expect(calls).toEqual(['L1', 'L2']) + + unsub1() + calls.length = 0 + + notifySessionStateChanged('requires_action') + expect(calls).toEqual(['L2']) + }) + + test('state listeners receive state and details', () => { + const received: Array<{ state: string; details?: Record }> = [] + setSessionStateChangedListener((state, details) => { + received.push({ state, details: details as Record | undefined }) + }) + + notifySessionStateChanged('requires_action', { + tool_name: 'Bash', + action_description: 'Running tests', + tool_use_id: 'toolu_001', + request_id: 'req_001', + }) + + expect(received).toEqual([ + { + state: 'requires_action', + details: { + tool_name: 'Bash', + action_description: 'Running tests', + tool_use_id: 'toolu_001', + request_id: 'req_001', + }, + }, + ]) + }) + + test('multiple metadata listeners all receive notifications', () => { + const calls: string[] = [] + setSessionMetadataChangedListener(m => { + calls.push(`L1:${m.model ?? 'no-model'}`) + }) + setSessionMetadataChangedListener(m => { + calls.push(`L2:${m.model ?? 'no-model'}`) + }) + + notifySessionMetadataChanged({ model: 'claude-sonnet-4-6' }) + + expect(calls).toEqual(['L1:claude-sonnet-4-6', 'L2:claude-sonnet-4-6']) + }) + + test('unsubscribed metadata listener does not receive subsequent notifications', () => { + const calls: string[] = [] + const unsub = setSessionMetadataChangedListener(m => { + calls.push(`L1:${m.model ?? 'no-model'}`) + }) + setSessionMetadataChangedListener(m => { + calls.push(`L2:${m.model ?? 'no-model'}`) + }) + + notifySessionMetadataChanged({ model: 'opus-4' }) + expect(calls).toEqual(['L1:opus-4', 'L2:opus-4']) + + unsub() + calls.length = 0 + + notifySessionMetadataChanged({ model: 'sonnet-4' }) + expect(calls).toEqual(['L2:sonnet-4']) + }) + + test('metadata replay only fires for the subscribing listener', () => { + const calls: string[] = [] + + notifySessionMetadataChanged({ model: 'haiku-3.5' }) + + setSessionMetadataChangedListener( + m => calls.push(`L1:${m.model}`), + { replayCurrent: true }, + ) + setSessionMetadataChangedListener( + m => calls.push(`L2:${m.model}`), + { replayCurrent: true }, + ) + + expect(calls).toEqual(['L1:haiku-3.5', 'L2:haiku-3.5']) + }) + + test('multiple permission mode listeners all receive notifications', () => { + const calls: string[] = [] + setPermissionModeChangedListener(m => calls.push(`L1:${m}`)) + setPermissionModeChangedListener(m => calls.push(`L2:${m}`)) + + notifyPermissionModeChanged('plan' as never) + + expect(calls).toEqual(['L1:plan', 'L2:plan']) + }) + + test('unsubscribed permission mode listener does not receive subsequent notifications', () => { + const calls: string[] = [] + const unsub = setPermissionModeChangedListener(m => calls.push(`L1:${m}`)) + setPermissionModeChangedListener(m => calls.push(`L2:${m}`)) + + notifyPermissionModeChanged('default' as never) + expect(calls).toEqual(['L1:default', 'L2:default']) + + unsub() + calls.length = 0 + + notifyPermissionModeChanged('acceptEdits' as never) + expect(calls).toEqual(['L2:acceptEdits']) + }) + + test('resetSessionStateForTests clears all listeners and consumers', () => { + const stateCalls: string[] = [] + const metadataCalls: string[] = [] + const permCalls: string[] = [] + const sdkCalls: string[] = [] + + setSessionStateChangedListener(() => stateCalls.push('s')) + setSessionMetadataChangedListener(() => metadataCalls.push('m')) + setPermissionModeChangedListener(() => permCalls.push('p')) + registerSdkEventConsumer(() => sdkCalls.push('e')) + + notifySessionStateChanged('running') + notifySessionMetadataChanged({ model: 'test' }) + notifyPermissionModeChanged('plan' as never) + + expect(stateCalls).toEqual(['s']) + expect(metadataCalls).toEqual(['m']) + expect(permCalls).toEqual(['p']) + expect(sdkCalls.length).toBeGreaterThan(0) + + resetSessionStateForTests() + + stateCalls.length = 0 + metadataCalls.length = 0 + permCalls.length = 0 + sdkCalls.length = 0 + + notifySessionStateChanged('idle') + notifySessionMetadataChanged({ model: 'after-reset' }) + notifyPermissionModeChanged('default' as never) + + expect(stateCalls).toEqual([]) + expect(metadataCalls).toEqual([]) + expect(permCalls).toEqual([]) + expect(sdkCalls).toEqual([]) + }) +}) + +describe('sessionState SDK event consumer (direct push)', () => { + beforeEach(() => { + resetSessionStateForTests() + }) + + test('SDK consumer receives session_state_changed events directly', () => { + const events: Array<{ subtype: string; state: string }> = [] + registerSdkEventConsumer(event => { + events.push({ subtype: event.subtype, state: event.state }) + }) + + notifySessionStateChanged('running') + notifySessionStateChanged('requires_action') + notifySessionStateChanged('idle') + + expect(events).toEqual([ + { subtype: 'session_state_changed', state: 'running' }, + { subtype: 'session_state_changed', state: 'requires_action' }, + { subtype: 'session_state_changed', state: 'idle' }, + ]) + }) + + test('multiple SDK consumers all receive events', () => { + const calls: string[] = [] + registerSdkEventConsumer(() => calls.push('C1')) + registerSdkEventConsumer(() => calls.push('C2')) + registerSdkEventConsumer(() => calls.push('C3')) + + notifySessionStateChanged('running') + + expect(calls).toEqual(['C1', 'C2', 'C3']) + }) + + test('unsubscribed SDK consumer does not receive subsequent events', () => { + const calls: string[] = [] + const unsub = registerSdkEventConsumer(() => calls.push('C1')) + registerSdkEventConsumer(() => calls.push('C2')) + + notifySessionStateChanged('running') + expect(calls).toEqual(['C1', 'C2']) + + unsub() + calls.length = 0 + + notifySessionStateChanged('idle') + expect(calls).toEqual(['C2']) + }) + + test('SDK event contains correct shape matching OpenCode format', () => { + const events: Array> = [] + registerSdkEventConsumer(event => { + events.push(event as unknown as Record) + }) + + notifySessionStateChanged('running') + + expect(events).toEqual([ + { + type: 'system', + subtype: 'session_state_changed', + state: 'running', + }, + ]) + }) + + test('SDK consumer works independently of state listeners', () => { + const stateCalls: string[] = [] + const sdkCalls: string[] = [] + + setSessionStateChangedListener(() => stateCalls.push('state')) + registerSdkEventConsumer(() => sdkCalls.push('sdk')) + + notifySessionStateChanged('running') + + expect(stateCalls).toEqual(['state']) + expect(sdkCalls).toEqual(['sdk']) + }) + + test('SDK consumer is NOT called when only metadata changes', () => { + const sdkCalls: string[] = [] + registerSdkEventConsumer(() => sdkCalls.push('sdk')) + + notifySessionMetadataChanged({ model: 'sonnet-4' }) + notifyPermissionModeChanged('plan' as never) + notifyAutomationStateChanged({ enabled: true, phase: 'standby', next_tick_at: 100, sleep_until: null }) + + expect(sdkCalls).toEqual([]) + }) +}) diff --git a/src/utils/sdkEventQueue.ts b/src/utils/sdkEventQueue.ts index 2cf5ac085..9880c8961 100644 --- a/src/utils/sdkEventQueue.ts +++ b/src/utils/sdkEventQueue.ts @@ -53,23 +53,15 @@ type TaskNotificationSdkEvent = { } } -// Mirrors notifySessionStateChanged. The CCR bridge already receives this -// via its own listener; SDK consumers (scmuxd, VS Code) need the same signal -// to know when the main turn's generator is idle vs actively producing. -// The 'idle' transition fires AFTER heldBackResult flushes and the bg-agent -// do-while loop exits — so SDK consumers can trust it as the authoritative -// "turn is over" signal even when result was withheld for background agents. -type SessionStateChangedEvent = { - type: 'system' - subtype: 'session_state_changed' - state: 'idle' | 'running' | 'requires_action' -} +// SessionStateChangedEvent is now emitted directly by sessionState.ts's +// registerSdkEventConsumer() — it bypasses this queue entirely so status +// transitions reach SDK consumers (scmuxd, VS Code) without delay. +// See SdkEventConsumer in src/utils/sessionState.ts for the type. export type SdkEvent = | TaskStartedEvent | TaskProgressEvent | TaskNotificationSdkEvent - | SessionStateChangedEvent const MAX_QUEUE_SIZE = 1000 const queue: SdkEvent[] = [] diff --git a/src/utils/sessionState.ts b/src/utils/sessionState.ts index 9e0c8ee08..e9343068a 100644 --- a/src/utils/sessionState.ts +++ b/src/utils/sessionState.ts @@ -34,9 +34,7 @@ export type AutomationStateMetadata = { sleep_until: number | null } -import { isEnvTruthy } from './envUtils.js' import type { PermissionMode } from './permissions/PermissionMode.js' -import { enqueueSdkEvent } from './sdkEventQueue.js' // CCR external_metadata keys — push in onChangeAppState, restore in // externalMetadataToAppState. @@ -68,31 +66,55 @@ type SessionMetadataListenerOptions = { replayCurrent?: boolean } -let stateListener: SessionStateChangedListener | null = null -let metadataListener: SessionMetadataChangedListener | null = null -let permissionModeListener: PermissionModeChangedListener | null = null +/** + * Consumer for session_state_changed SDK events. When registered, every + * status transition pushes an event directly to this consumer (bypassing + * the SDK event queue), ensuring external clients see status changes + * immediately — not delayed until the next drainSdkEvents() call. + * + * Aligned with OpenCode's GlobalBus pattern: status events propagate + * in real-time without buffering. + */ +export type SdkEventConsumer = (event: { + type: 'system' + subtype: 'session_state_changed' + state: SessionState +}) => void + +const stateListeners = new Set() +const metadataListeners = new Set() +const permissionModeListeners = new Set() +const sdkEventConsumers = new Set() export function setSessionStateChangedListener( cb: SessionStateChangedListener | null, -): void { - stateListener = cb +): () => void { + if (cb) { + stateListeners.add(cb) + return () => { + stateListeners.delete(cb) + } + } + return () => {} } export function setSessionMetadataChangedListener( cb: SessionMetadataChangedListener | null, options?: SessionMetadataListenerOptions, -): void { - metadataListener = cb - if (!cb || !options?.replayCurrent) { - return +): () => void { + if (cb) { + metadataListeners.add(cb) + if (options?.replayCurrent) { + const snapshot = getSessionMetadataSnapshot() + if (Object.keys(snapshot).length > 0) { + cb(snapshot) + } + } + return () => { + metadataListeners.delete(cb) + } } - - const snapshot = getSessionMetadataSnapshot() - if (Object.keys(snapshot).length === 0) { - return - } - - cb(snapshot) + return () => {} } /** @@ -104,8 +126,29 @@ export function setSessionMetadataChangedListener( */ export function setPermissionModeChangedListener( cb: PermissionModeChangedListener | null, -): void { - permissionModeListener = cb +): () => void { + if (cb) { + permissionModeListeners.add(cb) + return () => { + permissionModeListeners.delete(cb) + } + } + return () => {} +} + +/** + * Register a consumer that receives session_state_changed events directly, + * bypassing the SDK event queue. Use this in headless/stream-json mode to + * push status transitions immediately to the output stream instead of + * waiting for the next drainSdkEvents() call. + * + * Returns an unsubscribe function. + */ +export function registerSdkEventConsumer(cb: SdkEventConsumer): () => void { + sdkEventConsumers.add(cb) + return () => { + sdkEventConsumers.delete(cb) + } } let hasPendingAction = false @@ -175,7 +218,22 @@ export function notifySessionStateChanged( details?: RequiresActionDetails, ): void { currentState = state - stateListener?.(state, details) + + // Notify all registered state listeners (CCR bridge, etc.) + for (const listener of stateListeners) { + listener(state, details) + } + + // Push session_state_changed directly to SDK consumers (bypasses queue). + // Aligned with OpenCode: status events propagate immediately via + // bus.publish → GlobalBus → SSE, not buffered. + for (const consumer of sdkEventConsumers) { + consumer({ + type: 'system', + subtype: 'session_state_changed', + state, + }) + } // Mirror details into external_metadata so GetSession carries the // pending-action context without proto changes. Cleared via RFC 7396 @@ -208,30 +266,15 @@ export function notifySessionStateChanged( : null, ) } - - // Mirror to the SDK event stream so non-CCR consumers (scmuxd, VS Code) - // see the same authoritative idle/running signal the CCR bridge does. - // 'idle' fires after heldBackResult flushes — lets scmuxd flip IDLE and - // show the bg-task dot instead of a stuck generating spinner. - // - // Opt-in until CCR web + mobile clients learn to ignore this subtype in - // their isWorking() last-message heuristics — the trailing idle event - // currently pins them at "Running...". - // https://anthropic.slack.com/archives/C093BJBD1CP/p1774152406752229 - if (isEnvTruthy(process.env.CLAUDE_CODE_EMIT_SESSION_STATE_EVENTS)) { - enqueueSdkEvent({ - type: 'system', - subtype: 'session_state_changed', - state, - }) - } } export function notifySessionMetadataChanged( metadata: SessionExternalMetadata, ): void { applyMetadataUpdate(metadata) - metadataListener?.(metadata) + for (const listener of metadataListeners) { + listener(metadata) + } } export function notifyAutomationStateChanged( @@ -246,7 +289,9 @@ export function notifyAutomationStateChanged( currentAutomationState = nextState applyMetadataUpdate({ automation_state: nextState }) - metadataListener?.({ automation_state: nextState }) + for (const listener of metadataListeners) { + listener({ automation_state: nextState }) + } } /** @@ -256,13 +301,16 @@ export function notifyAutomationStateChanged( * silently bypass them. */ export function notifyPermissionModeChanged(mode: PermissionMode): void { - permissionModeListener?.(mode) + for (const listener of permissionModeListeners) { + listener(mode) + } } export function resetSessionStateForTests(): void { - stateListener = null - metadataListener = null - permissionModeListener = null + stateListeners.clear() + metadataListeners.clear() + permissionModeListeners.clear() + sdkEventConsumers.clear() hasPendingAction = false currentState = 'idle' currentAutomationState = null