fix: 服务重启卡住 90 秒与关闭时崩溃——关闭时先结束 SSE 推送（其他长连接 5 秒后强制断开、10 秒后直接退出）；codex app-server 输入管道出错不再使服务崩溃（EPIPE）；systemd 改为 KillMode=mixed、停止超时 30 秒；监听器过多的警告记录调用栈以便定位

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

apps/server/src/http/server.ts

@@ -16,10 +16,17 @@ import { registerGateway } from '../mcp-gateway/server'
 import { registerMobileApp } from './mobile-app'
 import { NpmRegistry, registerNpmRegistry } from './npm-registry'
 import registryRoutes from './routes/registry'
+import { closeAllStreams } from './sse'
 
 export async function buildServer(app: App, opts: { logger: FastifyServerOptions['loggerInstance'] }) {
   const f = Fastify({ loggerInstance: opts.logger, trustProxy: config.trustProxy, bodyLimit: 2 * 1048576 })
 
+  // 关闭时先结束 SSE 推送，否则 server.close() 会一直等这些长连接断开（systemd 等 90 秒后强制结束）
+  f.addHook('preClose', done => {
+    closeAllStreams()
+    done()
+  })
+
   await f.register(cookie)
   await f.register(multipart)
 

apps/server/src/http/sse.ts

@@ -4,6 +4,13 @@ import type { App } from '../app'
 
 const PING_MS = 15_000
 
+/** 正在推送的 SSE 连接：服务关闭时主动结束（否则 server.close() 会一直等它们断开），客户端按 retry 自动重连到新进程 */
+const openStreams = new Set<() => void>()
+
+export function closeAllStreams() {
+  for (const end of [...openStreams]) end()
+}
+
 export function streamSessionEvents(app: App, req: FastifyRequest, reply: FastifyReply, sessionId: string, lastId: number) {
   streamEvents(req, reply, lastId, after => app.events.list(sessionId, after), fn => app.events.subscribe(sessionId, fn))
 }
@@ -41,8 +48,11 @@ export function streamEvents<E extends { seq: number; type: string }>(
   for (const e of pending) write(e)
 
   const ping = setInterval(() => res.write(': ping\n\n'), PING_MS)
+  const end = () => res.end()
+  openStreams.add(end)
   req.raw.on('close', () => {
     clearInterval(ping)
     unsubscribe()
+    openStreams.delete(end)
   })
 }

apps/server/src/main.ts

@@ -11,8 +11,20 @@ const logger = pino({
 const app = createApp(logger)
 const server = await buildServer(app, { logger })
 
+let stopping = false
 const shutdown = async (signal: string) => {
+  if (stopping) return
+  stopping = true
   logger.info({ signal }, 'shutting down')
+  // 其他长连接（如 MCP 客户端保持的流）：5 秒后强制断开；10 秒仍未退出则直接退出
+  setTimeout(() => {
+    logger.warn({}, '仍有连接未关闭，强制断开')
+    server.server.closeAllConnections()
+  }, 5000).unref()
+  setTimeout(() => {
+    logger.warn({}, '关闭超时，直接退出')
+    process.exit(0)
+  }, 10_000).unref()
   await server.close()
   await app.mcpUpstream.closeAll()
   app.db.$client.close()
@@ -20,6 +32,13 @@ const shutdown = async (signal: string) => {
 }
 process.on('SIGINT', () => void shutdown('SIGINT'))
 process.on('SIGTERM', () => void shutdown('SIGTERM'))
+// 诊断：监听器过多的警告记下调用栈（每种只记一次），用于查找是哪里在累积
+const warned = new Set<string>()
+process.on('warning', w => {
+  if (w.name !== 'MaxListenersExceededWarning' || warned.has(w.name)) return
+  warned.add(w.name)
+  logger.warn({ stack: w.stack }, w.message)
+})
 
 await server.listen({ host: config.host, port: config.port })
 logger.info({ dataDir: config.dataDir }, `AgentServer 已启动：http://${config.host}:${config.port}`)

apps/server/src/providers/codex/rpc.ts

@@ -56,6 +56,8 @@ export class AppServerClient extends EventEmitter<{ notification: [string, unkno
       this.emit('exit', code)
     })
     child.on('error', err => this.emit('stderr', err.message))
+    // app-server 先于本进程退出时（例如服务停止时一起收到 SIGTERM），写入会 EPIPE；不处理会使整个服务崩溃
+    child.stdin.on('error', err => this.emit('stderr', `stdin：${err.message}`))
     const res = await this.request('initialize', {
       clientInfo,
       capabilities: { experimentalApi: true, requestAttestation: false },
@@ -86,7 +88,9 @@ export class AppServerClient extends EventEmitter<{ notification: [string, unkno
   }
 
   private write(msg: object) {
-    this.child?.stdin.write(`${JSON.stringify(msg)}\n`)
+    const stdin = this.child?.stdin
+    if (!stdin || stdin.destroyed || !stdin.writable) return
+    stdin.write(`${JSON.stringify(msg)}\n`)
   }
 
   private onLine(line: string) {

apps/server/test/shutdown.test.ts

@@ -0,0 +1,28 @@
+// 服务关闭：有 SSE 推送连接时 server.close() 不能一直等待（否则 systemd 等 90 秒后强制结束）
+import type { AddressInfo } from 'node:net'
+import pino from 'pino'
+import { expect, it } from 'vitest'
+import { createApp } from '../src/app'
+import { buildServer } from '../src/http/server'
+
+it('关闭服务时主动结束 SSE 推送', async () => {
+  const log = pino({ level: 'silent' })
+  const app = createApp(log, { providers: {} })
+  const server = await buildServer(app, { logger: log })
+  await server.listen({ host: '127.0.0.1', port: 0 })
+  const base = `http://127.0.0.1:${(server.server.address() as AddressInfo).port}`
+  const builtin = app.profiles.getBuiltin().id
+  const key = app.apiKeys.create('shutdown', { allowedProfileIds: [builtin], defaultProfileId: builtin })
+  const s = await app.sessions.create(key.apiKey.id, {})
+  const res = await fetch(`${base}/v1/sessions/${s.id}/stream`, { headers: { authorization: `Bearer ${key.key}` } })
+  expect(res.status).toBe(200)
+  const reader = res.body!.getReader()
+  await reader.read()
+
+  const t0 = Date.now()
+  await server.close()
+  expect(Date.now() - t0).toBeLessThan(1000)
+  // 客户端看到流结束（随后会按 retry 重连到新进程）
+  let done = false
+  while (!done) done = (await reader.read()).done
+})

deploy/agent-server.service

@@ -21,6 +21,10 @@ Environment=AGENT_SERVER_TRUST_PROXY=100.96.0.2
 ExecStart=/usr/local/bin/node dist/main.js
 Restart=on-failure
 RestartSec=3
+# 停止时只给主进程发 SIGTERM，由它先关闭连接再退出；剩下的子进程（引擎 CLI、codex app-server）随后被结束。
+# 默认会同时给所有子进程发 SIGTERM，子进程先退出会让主进程在关闭过程中写入已关闭的管道
+KillMode=mixed
+TimeoutStopSec=30
 
 # 文件系统：整盘只读，home 不可见，只有服务数据可写；程序与运行环境由 root 所有、只读
 ProtectSystem=strict