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

// SSE：补发 seq > lastId 的事件，然后实时推送（docs/design/04-api.md §5）
import type { FastifyReply, FastifyRequest } from 'fastify'
import type { AgentEvent } from '@agent-server/shared'
import type { App } from '../app'

const PING_MS = 15_000

export function streamSessionEvents(app: App, req: FastifyRequest, reply: FastifyReply, sessionId: string, lastId: number) {
  reply.hijack()
  const res = reply.raw
  res.writeHead(200, {
    '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: AgentEvent) => {
    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: AgentEvent[] = []
  let replaying = true
  const unsubscribe = app.events.subscribe(sessionId, e => (replaying ? pending.push(e) : write(e)))
  for (const e of app.events.list(sessionId, lastId)) write(e)
  replaying = false
  for (const e of pending) write(e)

  const ping = setInterval(() => res.write(': ping\n\n'), PING_MS)
  req.raw.on('close', () => {
    clearInterval(ping)
    unsubscribe()
  })
}