AgentServer/apps/server/src/http/sse.ts

// SSE：补发 seq > lastId 的事件，然后实时推送（docs/design/04-api.md §5）
import type { FastifyReply, FastifyRequest } from 'fastify'
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))
}

/** 通用的 SSE 推送：会话事件与工作组事件共用（都带递增的 seq） */
export function streamEvents<E extends { seq: number; type: string }>(
  req: FastifyRequest, reply: FastifyReply, lastId: number,
  list: (after: number) => E[], subscribe: (fn: (e: E) => void) => () => void,
) {
  reply.hijack()
  const res = reply.raw
  // hijack 后 reply.header 不再生效：带上前面钩子设置的头（CORS 等）
  res.writeHead(200, {
    ...(reply.getHeaders() as Record<string, string>),
    'Content-Type': 'text/event-stream; charset=utf-8',
    'Cache-Control': 'no-cache, no-transform',
    Connection: 'keep-alive',
    'X-Accel-Buffering': 'no',
  })
  res.write('retry: 2000\n\n')

  let sent = lastId
  const write = (e: E) => {
    if (e.seq <= sent) return
    sent = e.seq
    res.write(`id: ${e.seq}\nevent: ${e.type}\ndata: ${JSON.stringify(e)}\n\n`)
  }

  // 先订阅再补发，补发与实时之间的事件靠 seq 去重
  const pending: E[] = []
  let replaying = true
  const unsubscribe = subscribe(e => (replaying ? pending.push(e) : write(e)))
  for (const e of list(lastId)) write(e)
  replaying = false
  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)
  })
}