From 434ee9e806d87419e8d428b567ac39ebe3c7c516 Mon Sep 17 00:00:00 2001 From: DoSun Date: Mon, 11 May 2026 09:26:15 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20session=20status=20=E6=94=B9=E9=80=A0?= =?UTF-8?q?=E4=B8=BA=20prompt=20=E7=BA=A7=E5=88=AB=20busy/idle=20=E4=B8=89?= =?UTF-8?q?=E6=80=81=EF=BC=8CSSE=20=E4=BA=8B=E4=BB=B6=E5=92=8C=20status=20?= =?UTF-8?q?=E6=8E=A5=E5=8F=A3=E6=8C=89=20cwd=20=E8=BF=87=E6=BB=A4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增 SessionBusyStatus 类型（idle/busy/retry），替代进程生命周期状态 - SessionHandle 在 prompt/result/abort/close/error 中追踪 busyStatus 并通过 EventBus 发布 session.status SSE 事件 - /session/status 接口返回扁平 Record，支持按 cwd 过滤 - EventBus 支持客户端 cwdFilter，SSE 事件按工作空间过滤推送 - sessionManager 在 create/delete 时注册/注销 sessionCwd 映射 --- src/server/__tests__/sessionManager.test.ts | 4 +-- src/server/eventBus.ts | 24 +++++++++++-- src/server/routes/event.ts | 5 ++- src/server/routes/session.ts | 40 ++------------------- src/server/sessionHandle.ts | 25 ++++++++++++- src/server/sessionManager.ts | 28 ++++++++------- src/server/types.ts | 5 +++ 7 files changed, 75 insertions(+), 56 deletions(-) diff --git a/src/server/__tests__/sessionManager.test.ts b/src/server/__tests__/sessionManager.test.ts index 898bf1d13..79392cbf1 100644 --- a/src/server/__tests__/sessionManager.test.ts +++ b/src/server/__tests__/sessionManager.test.ts @@ -21,10 +21,10 @@ describe('SessionManager', () => { expect(mgr.getAllSessions()).toEqual([]) }) - test('getSessionStatuses returns empty object', () => { + test('getSessionStatuses returns empty object', async () => { const bus = new EventBus() const mgr = new SessionManager({ eventBus: bus }) - expect(mgr.getSessionStatuses()).toEqual({}) + expect(await mgr.getSessionStatuses()).toEqual({}) }) test('getAllPendingPermissions returns empty', () => { diff --git a/src/server/eventBus.ts b/src/server/eventBus.ts index ab4e27506..95a0a638f 100644 --- a/src/server/eventBus.ts +++ b/src/server/eventBus.ts @@ -7,6 +7,7 @@ type SSEClient = { writer: SSEWriter | null rawWrite: ((event: string, data: unknown) => void) | null sessionIdFilter?: string + cwdFilter?: string } type BusEvent = { @@ -19,10 +20,19 @@ export class EventBus { private buffer: BusEvent[] = [] private bufferSize = 100 private heartbeatInterval: ReturnType | null = null + private sessionCwds = new Map() - addClient(writer: SSEWriter, sessionIdFilter?: string): string { + registerSessionCwd(sessionId: string, cwd: string): void { + this.sessionCwds.set(sessionId, cwd) + } + + unregisterSessionCwd(sessionId: string): void { + this.sessionCwds.delete(sessionId) + } + + addClient(writer: SSEWriter, sessionIdFilter?: string, cwdFilter?: string): string { const id = crypto.randomUUID() - const client: SSEClient = { id, writer, rawWrite: null, sessionIdFilter } + const client: SSEClient = { id, writer, rawWrite: null, sessionIdFilter, cwdFilter } this.clients.set(id, client) const sendAndCleanup = (opts: { event: string; data: string }) => @@ -65,6 +75,16 @@ export class EventBus { continue } } + if (client.cwdFilter) { + const dataObj = data as Record | undefined + const sid = dataObj?.session_id ?? dataObj?.sessionID + if (typeof sid === 'string') { + const sessionCwd = this.sessionCwds.get(sid) + if (sessionCwd && sessionCwd !== client.cwdFilter) { + continue + } + } + } client.writer.writeSSE({ event, data: payload }).catch(() => { deadIds.push(client.id) }) diff --git a/src/server/routes/event.ts b/src/server/routes/event.ts index 8c8327c0c..9a32a9fbf 100644 --- a/src/server/routes/event.ts +++ b/src/server/routes/event.ts @@ -1,12 +1,15 @@ import { Hono } from 'hono' import { streamSSE } from 'hono/streaming' import type { EventBus } from '../eventBus.js' +import { canonicalizePath } from '../../utils/sessionStoragePortable.js' export function createEventRoutes(eventBus: EventBus): Hono { return new Hono().get('/event', async c => { const sessionIdFilter = c.req.query('session_id') ?? undefined + const headerDir = c.req.header('x-csc-directory') + const cwdFilter = headerDir ? await canonicalizePath(decodeURIComponent(headerDir)) : undefined return streamSSE(c, async stream => { - const clientId = eventBus.addClient(stream, sessionIdFilter) + const clientId = eventBus.addClient(stream, sessionIdFilter, cwdFilter) stream.onAbort(() => { eventBus.removeClient(clientId) }) diff --git a/src/server/routes/session.ts b/src/server/routes/session.ts index d76e328a4..73844ad8a 100644 --- a/src/server/routes/session.ts +++ b/src/server/routes/session.ts @@ -258,44 +258,10 @@ export function createSessionRoutes( return c.json({ sessions }) }) .get('/session/status', async c => { - // 内存中活跃 session 的状态 - const activeStatuses = sessionManager.getSessionStatuses() - - // 从请求头或 query 取目录过滤 const headerDir = c.req.header('x-csc-directory') - const statusDir = (headerDir ? decodeURIComponent(headerDir) : undefined) - ?? c.req.query('dir') - ?? undefined - - // 补充磁盘历史 session（全部视为 idle） - let historySessions: Awaited> = [] - try { - historySessions = await listSessionsImpl({ dir: statusDir, limit: 200 }) - } catch {} - - const sessions: Record = {} - - // 先把历史 session 全部标为 idle/stopped - for (const s of historySessions) { - sessions[s.sessionId] = { - status: 'stopped', - state: 'stopped', - has_pending_permission: false, - type: 'idle', - } - } - - // 用内存中活跃的 handle 状态覆盖 - for (const [id, st] of Object.entries(activeStatuses)) { - sessions[id] = { - status: st.status, - state: st.status, - has_pending_permission: st.has_pending_permission, - type: st.prompting ? 'busy' : 'idle', - } - } - - return c.json({ sessions }) + const dir = headerDir ? decodeURIComponent(headerDir) : undefined + const statuses = sessionManager.getSessionStatuses(dir) + return c.json(statuses) }) .get('/session/:sessionID', async c => { const id = c.req.param('sessionID') diff --git a/src/server/sessionHandle.ts b/src/server/sessionHandle.ts index 96c24d1cf..c5de9129d 100644 --- a/src/server/sessionHandle.ts +++ b/src/server/sessionHandle.ts @@ -3,7 +3,7 @@ import { createInterface } from 'readline' import { jsonParse, jsonStringify } from '../utils/slowOperations.js' import { logError } from '../utils/log.js' import type { EventBus } from './eventBus.js' -import type { SessionState } from './types.js' +import type { SessionBusyStatus, SessionState } from './types.js' type StdoutMessage = { type: string @@ -119,6 +119,7 @@ export class SessionHandle { private _spawnCwd: string | undefined private child: ChildProcess | null = null private _status: SessionState = 'starting' + private _busyStatus: SessionBusyStatus = { type: 'idle' } private _model?: string private _providerId?: string private _permissionMode?: string @@ -148,6 +149,9 @@ export class SessionHandle { get status(): SessionState { return this._status } + get busyStatus(): SessionBusyStatus { + return this._busyStatus + } get model(): string | undefined { return this._model } @@ -233,6 +237,14 @@ export class SessionHandle { this.eventBus.publishSessionEvent(this.sessionId, event, data) } + private emitBusyStatus(): void { + if (this.opts.silent) return + this.eventBus.publish('session.status', { + sessionID: this.sessionId, + status: this._busyStatus, + }) + } + onMessage(listener: MessageListener): () => void { this.listeners.add(listener) return () => { this.listeners.delete(listener) } @@ -289,6 +301,8 @@ export class SessionHandle { this.child.on('close', (code, signal) => { if (this._status !== 'stopped') { this._status = 'stopped' + this._busyStatus = { type: 'idle' } + this.emitBusyStatus() this.emitEvent('deleted', { status: 'stopped', exit_code: code, @@ -310,6 +324,8 @@ export class SessionHandle { this.child.on('error', err => { logError(err) this._status = 'stopped' + this._busyStatus = { type: 'idle' } + this.emitBusyStatus() if (this.initReject) { const reject = this.initReject this.initResolve = null @@ -503,6 +519,8 @@ export class SessionHandle { if (cost) this._costUsd += cost if (usage?.input_tokens) this._inputTokens += usage.input_tokens if (usage?.output_tokens) this._outputTokens += usage.output_tokens + this._busyStatus = { type: 'idle' } + this.emitBusyStatus() this.emitEvent('result', msg) if (this.promptResolve) { this.promptResolve({ done: true }) @@ -665,6 +683,8 @@ export class SessionHandle { ) } this._prompting = true + this._busyStatus = { type: 'busy' } + this.emitBusyStatus() this.lastActiveAt = Date.now() const userMsg = jsonStringify({ @@ -708,6 +728,9 @@ export class SessionHandle { }) this.writeStdin(interrupt) + this._busyStatus = { type: 'idle' } + this.emitBusyStatus() + if (!this.promptResolve) return await new Promise(resolve => { diff --git a/src/server/sessionManager.ts b/src/server/sessionManager.ts index 76618f464..07d558882 100644 --- a/src/server/sessionManager.ts +++ b/src/server/sessionManager.ts @@ -5,7 +5,8 @@ import { getClaudeConfigHomeDir } from '../utils/envUtils.js' import { logError } from '../utils/log.js' import type { EventBus } from './eventBus.js' import { SessionHandle, type InitData } from './sessionHandle.js' -import type { SessionIndex, SessionIndexEntry, SessionState } from './types.js' +import type { SessionIndex, SessionIndexEntry, SessionBusyStatus, SessionState } from './types.js' +import { canonicalizePath } from '../utils/sessionStoragePortable.js' const INDEX_FILE = 'server-sessions.json' @@ -108,6 +109,7 @@ export class SessionManager { reason: 'idle_timeout', }) this.sessions.delete(id) + this.eventBus.unregisterSessionCwd(id) this.scheduleIndexSave() } } @@ -126,20 +128,15 @@ export class SessionManager { return [...this.sessions.values()] } - getSessionStatuses(): Record< - string, - { status: SessionState; has_pending_permission: boolean; prompting: boolean } - > { - const result: Record< - string, - { status: SessionState; has_pending_permission: boolean; prompting: boolean } - > = {} + async getSessionStatuses(cwd?: string): Promise> { + const canonicalCwd = cwd ? await canonicalizePath(cwd) : undefined + const result: Record = {} for (const [id, handle] of this.sessions) { - result[id] = { - status: handle.status, - has_pending_permission: handle.getPendingPermissions().length > 0, - prompting: handle.prompting, + if (canonicalCwd) { + const handleCwd = await canonicalizePath(handle.cwd) + if (handleCwd !== canonicalCwd) continue } + result[id] = handle.busyStatus } return result } @@ -193,6 +190,9 @@ export class SessionManager { this.sessions.set(sessionId, handle) if (!opts.silent) { + void canonicalizePath(cwd).then(canonical => { + this.eventBus.registerSessionCwd(sessionId, canonical) + }) this.eventBus.publishSessionEvent(sessionId, 'created', { status: 'starting', }) @@ -203,6 +203,7 @@ export class SessionManager { handle.spawn() } catch (err) { this.sessions.delete(sessionId) + this.eventBus.unregisterSessionCwd(sessionId) this.scheduleIndexSave() throw err } @@ -266,6 +267,7 @@ export class SessionManager { const silent = handle.silent handle.forceKill() this.sessions.delete(id) + this.eventBus.unregisterSessionCwd(id) if (!silent) { this.eventBus.publishSessionEvent(id, 'deleted', { status: 'stopped' }) } diff --git a/src/server/types.ts b/src/server/types.ts index b305bc8fa..d1cd910a5 100644 --- a/src/server/types.ts +++ b/src/server/types.ts @@ -30,6 +30,11 @@ export type SessionState = | 'stopping' | 'stopped' +export type SessionBusyStatus = + | { type: 'idle' } + | { type: 'busy' } + | { type: 'retry'; attempt: number; message: string; next: number } + export type SessionInfo = { id: string status: SessionState