feat: 对齐 OpenCode 即时推送架构，session status 事件绕过队列直接推送

- sessionState.ts: 单 listener → Set 多订阅者，新增 registerSdkEventConsumer 绕过 SDK 事件队列
- sdkEventQueue.ts: SessionStateChangedEvent 从队列系统移除
- print.ts: runHeadlessStreaming 注册直接消费者，事件即时 output.enqueue()
- query.ts: queryLoop 每轮迭代发送 running 心跳
- server/routes/session.ts: prompt_async 立即 emit busy（不等子进程 init），历史会话先解析 body 再恢复
- server/sessionHandle.ts: prompt() 检测已 busy 时跳过重复 emit
- server/eventBus.ts: 添加 SSE publish 耗时日志
- server/server.ts: 传递 eventBus 给 sessionRoutes
- 新增 16 个 sessionState 多订阅者 + SDK consumer 测试

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.

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')

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<string, unknown>
+      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(

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<string | undefined> {
@@ -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)
+      // Parse body FIRST — independent of session lookup, so we can
-
+      // emit busy status immediately even for historical sessions that
-      // 历史 session 不在内存中，自动恢复
+      // need to create+waitReady (3–30s).
-      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')
-        }
-      }
-
       const body = await c.req.json<{
         content?: string
         parts?: Array<Record<string, unknown>>
@@ -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<string, unknown> | 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 {}

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

src/server/sessionHandle.ts

@@ -485,13 +485,18 @@ export class SessionHandle {
   }
 
   async prompt(content: string, opts?: { parts?: Array<Record<string, unknown>>; 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()

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

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<string, unknown> }> = []
+    setSessionStateChangedListener((state, details) => {
+      received.push({ state, details: details as Record<string, unknown> | 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<Record<string, unknown>> = []
+    registerSdkEventConsumer(event => {
+      events.push(event as unknown as Record<string, unknown>)
+    })
+
+    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([])
+  })
+})

src/utils/sdkEventQueue.ts

@@ -53,23 +53,15 @@ type TaskNotificationSdkEvent = {
   }
 }
 
-// Mirrors notifySessionStateChanged. The CCR bridge already receives this
+// SessionStateChangedEvent is now emitted directly by sessionState.ts's
-// via its own listener; SDK consumers (scmuxd, VS Code) need the same signal
+// registerSdkEventConsumer() — it bypasses this queue entirely so status
-// to know when the main turn's generator is idle vs actively producing.
+// transitions reach SDK consumers (scmuxd, VS Code) without delay.
-// The 'idle' transition fires AFTER heldBackResult flushes and the bg-agent
+// See SdkEventConsumer in src/utils/sessionState.ts for the type.
-// 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'
-}
 
 export type SdkEvent =
   | TaskStartedEvent
   | TaskProgressEvent
   | TaskNotificationSdkEvent
-  | SessionStateChangedEvent
 
 const MAX_QUEUE_SIZE = 1000
 const queue: SdkEvent[] = []

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
+ * Consumer for session_state_changed SDK events. When registered, every
-let permissionModeListener: PermissionModeChangedListener | null = null
+ * 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<SessionStateChangedListener>()
+const metadataListeners = new Set<SessionMetadataChangedListener>()
+const permissionModeListeners = new Set<PermissionModeChangedListener>()
+const sdkEventConsumers = new Set<SdkEventConsumer>()
 
 export function setSessionStateChangedListener(
   cb: SessionStateChangedListener | null,
-): void {
+): () => void {
-  stateListener = cb
+  if (cb) {
+    stateListeners.add(cb)
+    return () => {
+      stateListeners.delete(cb)
+    }
+  }
+  return () => {}
 }
 
 export function setSessionMetadataChangedListener(
   cb: SessionMetadataChangedListener | null,
   options?: SessionMetadataListenerOptions,
-): void {
+): () => void {
-  metadataListener = cb
+  if (cb) {
-  if (!cb || !options?.replayCurrent) {
+    metadataListeners.add(cb)
-    return
+    if (options?.replayCurrent) {
+      const snapshot = getSessionMetadataSnapshot()
+      if (Object.keys(snapshot).length > 0) {
+        cb(snapshot)
+      }
+    }
+    return () => {
+      metadataListeners.delete(cb)
+    }
   }
-
+  return () => {}
-  const snapshot = getSessionMetadataSnapshot()
-  if (Object.keys(snapshot).length === 0) {
-    return
-  }
-
-  cb(snapshot)
 }
 
 /**
@@ -104,8 +126,29 @@ export function setSessionMetadataChangedListener(
  */
 export function setPermissionModeChangedListener(
   cb: PermissionModeChangedListener | null,
-): void {
+): () => void {
-  permissionModeListener = cb
+  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
+  stateListeners.clear()
-  metadataListener = null
+  metadataListeners.clear()
-  permissionModeListener = null
+  permissionModeListeners.clear()
+  sdkEventConsumers.clear()
   hasPendingAction = false
   currentState = 'idle'
   currentAutomationState = null