feat(server): W2-A login gate, secrets, settings, instances, audit, engine panel API and PG-backed SSE bus

- Admin login gate (docs/design/auth-reference.md): admin_user/admin_session/login_attempt
  (migration 0015), one-time setup token logged once + conditional UPDATE, argon2id via
  hash-wasm (WASM; compiled binary has no --allow-ffi), sha256-stored sessions in HttpOnly
  SameSite=Lax cookie umt_session (fixed 12h TTL, soft revoke), HMAC-derived double-submit
  CSRF, IP (5) + account (15) failure limits per 15 min, password change revokes other
  sessions, optional TOTP with replay counter, AuditAction enum.
- Unified auth middleware: cookie session or Bearer API key -> c.var.actor
  {type: admin|api_key, id, name, scopes}; requireScope works for both; requireAdminSession.
- ctx.secrets: AES-256-GCM, per-secret HKDF-SHA256 key from SECRET_MASTER_KEY, AAD bound
  to (id, purpose); ctx.settings: 30s cache invalidated by settings.changed events.
- Endpoints: /auth/*, /settings, /api-keys, /secrets, /audit-events, library-roots,
  downloader/media-server instances (+test), path-mappings (+verify),
  PUT /sites/{id}/credentials, /engine/runs|stats|cancel (read-only openworkflow schema).
- SSE: PgEventBus (sse_event table, migration 0016, LISTEN/NOTIFY umt_events, catch-up by
  seq, pruning, resync); EventBusPort is now async.
- Tests: login/rate limit/revocation/CSRF/TOTP, secrets round trip and no log leakage,
  admin APIs against mock qB/Emby, SSE across two Deno subprocesses.

Co-Authored-By: Claude Opus 5.5 <[email protected]>

apps/server/src/http/auth.ts

@@ -1,11 +1,16 @@
-import { createHash, timingSafeEqual } from "node:crypto";
-import type { ApiScope } from "@umt/contracts";
+import { createHash, createHmac, timingSafeEqual } from "node:crypto";
+import { type Actor, type ApiScope, ApiScope as ApiScopeSchema, CSRF_HEADER, SESSION_COOKIE } from "@umt/contracts";
 import type { DB, Kysely } from "@umt/db";
 import type { MiddlewareHandler } from "hono";
+import { getCookie } from "hono/cookie";
 import { ApiException } from "./errors.ts";
 
+// 鉴权中间件:控制台 cookie 会话(umt_session)与外部 API Key(Bearer)二选一,
+// 统一写入 c.var.actor({type:'admin'|'api_key', id, name, scopes});requireScope 对两者一视同仁。
+// 旧字段 c.var.principal 保留兼容(kind 区分 api_key / dev_admin / admin_session)。
+
 export interface Principal {
-  kind: "api_key" | "dev_admin";
+  kind: "api_key" | "dev_admin" | "admin_session";
   id: string;
   name: string;
   scopes: string[];
@@ -18,13 +23,32 @@ export interface ApiKeyRecord {
   revokedAt: Date | null;
 }
 
-/** API Key 查询接口,便于测试替换 */
+export interface SessionRecord {
+  id: string;
+  adminUserId: string;
+  username: string;
+  expiresAt: Date;
+  revokedAt: Date | null;
+  lastSeenAt: Date;
+}
+
+/** API Key / 会话查询接口,便于测试替换(会话方法可选:测试内存仓库可以只实现 API Key) */
 export interface ApiKeyStore {
   findByHash(hash: string): Promise<ApiKeyRecord | null>;
   touch(id: string): Promise<void>;
+  /** 按 sha256(token) 查会话(未过滤过期/吊销,由中间件判断) */
+  findSession?(hash: string): Promise<SessionRecord | null>;
+  /** 节流更新 last_seen_at */
+  touchSession?(id: string): Promise<void>;
 }
 
 export const hashApiKey = (key: string) => createHash("sha256").update(key, "utf8").digest("hex");
+/** 会话 token 与 API Key 一样只存 sha256(高熵随机串不需要慢哈希) */
+export const hashSessionToken = hashApiKey;
+
+/** CSRF 双提交 token:HMAC-SHA256(key=会话原始 token, "umt/csrf/v1") */
+export const deriveCsrfToken = (sessionToken: string) =>
+  createHmac("sha256", sessionToken).update("umt/csrf/v1").digest("base64url");
 
 export function kyselyApiKeyStore(db: Kysely<DB>): ApiKeyStore {
   return {
@@ -39,6 +63,26 @@ export function kyselyApiKeyStore(db: Kysely<DB>): ApiKeyStore {
     async touch(id) {
       await db.updateTable("api_key").set({ last_used_at: new Date() }).where("id", "=", id).execute();
     },
+    async findSession(hash) {
+      const row = await db
+        .selectFrom("admin_session as s")
+        .innerJoin("admin_user as u", "u.id", "s.admin_user_id")
+        .select(["s.id", "s.admin_user_id", "u.username", "s.expires_at", "s.revoked_at", "s.last_seen_at"])
+        .where("s.session_hash", "=", hash)
+        .executeTakeFirst();
+      if (!row?.username) return null;
+      return {
+        id: row.id,
+        adminUserId: row.admin_user_id,
+        username: row.username,
+        expiresAt: row.expires_at,
+        revokedAt: row.revoked_at,
+        lastSeenAt: row.last_seen_at,
+      };
+    },
+    async touchSession(id) {
+      await db.updateTable("admin_session").set({ last_seen_at: new Date() }).where("id", "=", id).execute();
+    },
   };
 }
 
@@ -48,7 +92,28 @@ export interface AuthOptions {
   devAdminKey?: string | undefined;
 }
 
-export type AuthEnv = { Variables: { principal: Principal } };
+export interface SessionVar {
+  id: string;
+  /** cookie 里的原始 token(用于派生 CSRF、改密时保留当前会话);不要写日志 */
+  token: string;
+  adminUserId: string;
+  expiresAt: Date;
+}
+
+export type AuthEnv = {
+  Variables: {
+    principal: Principal;
+    actor: Actor;
+    /** 仅 cookie 会话调用时存在 */
+    session: SessionVar | undefined;
+  };
+};
+
+/** 免鉴权路径(相对 /api/v1):登录前就要能访问;带有效 cookie 时仍会挂上 actor */
+export const PUBLIC_AUTH_PATHS = ["/auth/setup", "/auth/login", "/auth/session"] as const;
+const WRITE_METHODS = new Set(["POST", "PUT", "PATCH", "DELETE"]);
+/** last_seen_at 节流写库间隔 */
+const TOUCH_INTERVAL_MS = 60_000;
 
 function safeEqual(a: string, b: string) {
   const ab = Buffer.from(a);
@@ -56,25 +121,77 @@ function safeEqual(a: string, b: string) {
   return ab.length === bb.length && timingSafeEqual(ab, bb);
 }
 
-/** Bearer API Key 鉴权:成功后在 c.var.principal 写入调用者 */
-export function apiKeyAuth(opts: AuthOptions): MiddlewareHandler<AuthEnv> {
-  return async (c, next) => {
-    const header = c.req.header("authorization") ?? "";
-    const m = /^Bearer\s+(\S+)$/i.exec(header);
-    if (!m?.[1]) throw new ApiException("unauthorized", "缺少 Bearer API Key");
-    const key = m[1];
+const toScopes = (s: readonly string[]): ApiScope[] =>
+  s.filter((x): x is ApiScope => ApiScopeSchema.safeParse(x).success);
 
-    if (opts.devAdminKey && safeEqual(key, opts.devAdminKey)) {
-      c.set("principal", { kind: "dev_admin", id: "dev-admin", name: "dev-admin", scopes: ["admin"] });
+/**
+ * /api/v1 鉴权:
+ * - `Authorization: Bearer <key>`:API Key(开发模式另认 DEV_ADMIN_KEY);
+ * - 否则 cookie `umt_session`:管理员会话(未过期、未吊销),写方法还要校验 `x-csrf-token` == HMAC 派生值;
+ * - PUBLIC_AUTH_PATHS 免鉴权(有有效会话时照样挂 actor)。
+ */
+export function apiKeyAuth(opts: AuthOptions): MiddlewareHandler<AuthEnv> {
+  const setActor = (
+    c: Parameters<MiddlewareHandler<AuthEnv>>[0],
+    p: Principal,
+    type: Actor["type"],
+    session?: SessionVar,
+  ) => {
+    c.set("principal", p);
+    c.set("actor", { type, id: p.id, name: p.name, scopes: toScopes(p.scopes) });
+    c.set("session", session);
+  };
+
+  return async (c, next) => {
+    const rel = c.req.path.replace(/^\/api\/v1/, "");
+    const isPublic = (PUBLIC_AUTH_PATHS as readonly string[]).includes(rel);
+    const header = c.req.header("authorization") ?? "";
+
+    if (header) {
+      const m = /^Bearer\s+(\S+)$/i.exec(header);
+      if (!m?.[1]) {
+        if (isPublic) return next();
+        throw new ApiException("unauthorized", "Authorization 头格式应为 Bearer <key>");
+      }
+      const key = m[1];
+      if (opts.devAdminKey && safeEqual(key, opts.devAdminKey)) {
+        setActor(c, { kind: "dev_admin", id: "dev-admin", name: "dev-admin", scopes: ["admin"] }, "admin");
+        return next();
+      }
+      const rec = opts.store ? await opts.store.findByHash(hashApiKey(key)) : null;
+      if (!rec || rec.revokedAt) {
+        if (isPublic) return next();
+        throw new ApiException("unauthorized", "API Key 无效或已撤销");
+      }
+      setActor(c, { kind: "api_key", id: rec.id, name: rec.name, scopes: rec.scopes }, "api_key");
+      // last_used_at 尽力更新,失败不影响请求
+      opts.store?.touch(rec.id).catch(() => {});
       return next();
     }
-    if (!opts.store) throw new ApiException("unauthorized", "API Key 无效");
-    const rec = await opts.store.findByHash(hashApiKey(key));
-    if (!rec || rec.revokedAt) throw new ApiException("unauthorized", "API Key 无效或已撤销");
-    c.set("principal", { kind: "api_key", id: rec.id, name: rec.name, scopes: rec.scopes });
-    // last_used_at 尽力更新,失败不影响请求
-    opts.store.touch(rec.id).catch(() => {});
-    return next();
+
+    const token = getCookie(c, SESSION_COOKIE);
+    if (token && opts.store?.findSession) {
+      const s = await opts.store.findSession(hashSessionToken(token));
+      const now = Date.now();
+      if (s && !s.revokedAt && s.expiresAt.getTime() > now) {
+        if (!isPublic && WRITE_METHODS.has(c.req.method)) {
+          const provided = c.req.header(CSRF_HEADER);
+          if (!provided || !safeEqual(provided, deriveCsrfToken(token))) {
+            throw new ApiException("forbidden", "CSRF 校验失败,请刷新页面后重试");
+          }
+        }
+        setActor(c, { kind: "admin_session", id: s.adminUserId, name: s.username, scopes: ["admin"] }, "admin", {
+          id: s.id,
+          token,
+          adminUserId: s.adminUserId,
+          expiresAt: s.expiresAt,
+        });
+        if (now - s.lastSeenAt.getTime() > TOUCH_INTERVAL_MS) opts.store.touchSession?.(s.id).catch(() => {});
+        return next();
+      }
+    }
+    if (isPublic) return next();
+    throw new ApiException("unauthorized", token ? "会话已过期或已注销,请重新登录" : "需要登录或 Bearer API Key");
   };
 }
 
@@ -86,7 +203,7 @@ export function hasScope(granted: readonly string[], required: ApiScope): boolea
   return false;
 }
 
-/** 要求调用者至少具备其中一个 scope */
+/** 要求调用者至少具备其中一个 scope(cookie 会话与 API Key 同样适用) */
 export function requireScope(...anyOf: ApiScope[]): MiddlewareHandler<AuthEnv> {
   return async (c, next) => {
     const p = c.get("principal");
@@ -97,3 +214,13 @@ export function requireScope(...anyOf: ApiScope[]): MiddlewareHandler<AuthEnv> {
     return next();
   };
 }
+
+/** 只允许控制台管理员会话(或开发模式 DEV_ADMIN_KEY),拒绝外部 API Key —— 用于改密、TOTP 等账号操作 */
+export function requireAdminSession(): MiddlewareHandler<AuthEnv> {
+  return async (c, next) => {
+    const p = c.get("principal");
+    if (!p) throw new ApiException("unauthorized", "未鉴权");
+    if (p.kind !== "admin_session") throw new ApiException("forbidden", "该操作只能在控制台登录后进行");
+    return next();
+  };
+}

apps/server/src/http/sse-pg.ts

@@ -0,0 +1,217 @@
+import type { EventEnvelope } from "@umt/contracts";
+import { type DB, type Kysely, sql } from "@umt/db";
+import pg from "pg";
+import type { Logger } from "pino";
+import type { BusListener, EventBusPort, SinceResult } from "./sse.ts";
+
+/** 默认通知通道 */
+export const EVENTS_CHANNEL = "umt_events";
+/** 发布时串行化 seq 分配与提交的咨询锁键('umte') */
+const PUBLISH_LOCK_KEY = 0x756d7465;
+
+export interface PgEventBusOptions {
+  db: Kysely<DB>;
+  /** LISTEN 专用连接(不走连接池) */
+  connectionString: string;
+  logger: Logger;
+  channel?: string;
+  /** sse_event 保留最近多少条供 ?after= 补读,默认 10000 */
+  retain?: number;
+  /** 每发布多少条裁剪一次,默认 200 */
+  pruneEvery?: number;
+  /** LISTEN 断线重连退避上限,默认 10s */
+  maxReconnectMs?: number;
+  /** 仅测试:LISTEN 连接的 search_path 等 pg 启动参数 */
+  pgOptions?: string;
+}
+
+type Row = { seq: number; type: string; at: Date; payload: unknown };
+const toEnvelope = (r: Row): EventEnvelope => ({
+  seq: Number(r.seq),
+  type: r.type,
+  at: r.at.toISOString(),
+  payload: r.payload,
+});
+
+/**
+ * 多进程 SSE 总线:发布 = 在 sse_event 追加一行(持咨询锁,保证 seq 顺序即提交顺序)+ pg_notify(channel, seq);
+ * 每个进程一条 LISTEN 连接,收到通知后按 seq 从表里补读并分发给本进程订阅者(包括自己发布的)。
+ * LISTEN 断线期间的事件在重连后按 lastDispatched 补读,不丢。
+ */
+export class PgEventBus implements EventBusPort {
+  private readonly listeners = new Set<BusListener>();
+  private readonly channel: string;
+  private readonly retain: number;
+  private readonly pruneEvery: number;
+  private client: pg.Client | null = null;
+  private lastDispatched = 0;
+  private draining: Promise<void> | null = null;
+  private drainAgain = false;
+  private published = 0;
+  private stopped = false;
+  private reconnectTimer: ReturnType<typeof setTimeout> | null = null;
+
+  private constructor(private readonly opts: PgEventBusOptions) {
+    this.channel = opts.channel ?? EVENTS_CHANNEL;
+    if (!/^[a-z_][a-z0-9_]*$/.test(this.channel)) throw new Error(`非法通道名 ${this.channel}`);
+    this.retain = opts.retain ?? 10_000;
+    this.pruneEvery = opts.pruneEvery ?? 200;
+  }
+
+  /** 建总线并开始 LISTEN;从当前最大 seq 之后开始分发 */
+  static async start(opts: PgEventBusOptions): Promise<PgEventBus> {
+    const bus = new PgEventBus(opts);
+    bus.lastDispatched = await bus.maxSeq();
+    await bus.connect();
+    return bus;
+  }
+
+  private get log() {
+    return this.opts.logger;
+  }
+
+  private async maxSeq(): Promise<number> {
+    const r = await sql<{ m: number | null }>`select max(seq) as m from sse_event`.execute(this.opts.db);
+    return Number(r.rows[0]?.m ?? 0);
+  }
+
+  private async connect(): Promise<void> {
+    const client = new pg.Client({
+      connectionString: this.opts.connectionString,
+      ...(this.opts.pgOptions ? { options: this.opts.pgOptions } : {}),
+    });
+    client.on("notification", (msg) => {
+      if (msg.channel === this.channel) this.scheduleDrain();
+    });
+    const onLost = (err?: Error) => {
+      if (this.client !== client) return;
+      this.client = null;
+      client.removeAllListeners("notification");
+      client.end().catch(() => {});
+      if (!this.stopped) {
+        this.log.warn({ err }, "SSE 总线 LISTEN 连接断开,准备重连");
+        this.reconnect(500);
+      }
+    };
+    client.on("error", onLost);
+    client.on("end", () => onLost());
+    await client.connect();
+    await client.query(`listen ${this.channel}`);
+    this.client = client;
+    // 断线期间可能漏掉通知:连上后补读一次
+    this.scheduleDrain();
+  }
+
+  private reconnect(delayMs: number) {
+    if (this.stopped || this.reconnectTimer) return;
+    this.reconnectTimer = setTimeout(async () => {
+      this.reconnectTimer = null;
+      try {
+        await this.connect();
+        this.log.info("SSE 总线 LISTEN 已重连");
+      } catch (err) {
+        this.log.warn({ err }, "SSE 总线重连失败");
+        this.reconnect(Math.min(delayMs * 2, this.opts.maxReconnectMs ?? 10_000));
+      }
+    }, delayMs);
+  }
+
+  private scheduleDrain() {
+    if (this.draining) {
+      this.drainAgain = true;
+      return;
+    }
+    this.draining = (async () => {
+      try {
+        do {
+          this.drainAgain = false;
+          await this.drainOnce();
+        } while (this.drainAgain && !this.stopped);
+      } catch (err) {
+        this.log.warn({ err }, "SSE 总线补读失败");
+      } finally {
+        this.draining = null;
+      }
+    })();
+  }
+
+  private async drainOnce() {
+    for (;;) {
+      const rows = await sql<Row>`select seq, type, at, payload from sse_event
+        where seq > ${this.lastDispatched} order by seq limit 500`.execute(this.opts.db);
+      for (const r of rows.rows) {
+        const e = toEnvelope(r);
+        this.lastDispatched = e.seq;
+        for (const l of this.listeners) {
+          try {
+            l(e);
+          } catch (err) {
+            this.log.warn({ err, type: e.type }, "SSE 订阅者异常");
+          }
+        }
+      }
+      if (rows.rows.length < 500) return;
+    }
+  }
+
+  async publish<P>(type: string, payload: P): Promise<EventEnvelope<P> | null> {
+    try {
+      const row = await this.opts.db.transaction().execute(async (trx) => {
+        await sql`select pg_advisory_xact_lock(${PUBLISH_LOCK_KEY})`.execute(trx);
+        const r = await sql<Row>`insert into sse_event (type, payload)
+          values (${type}, ${JSON.stringify(payload ?? null)}::jsonb) returning seq, type, at, payload`.execute(trx);
+        const inserted = r.rows[0]!;
+        // NOTIFY 在提交时才投递
+        await sql`select pg_notify(${this.channel}, ${String(inserted.seq)})`.execute(trx);
+        return inserted;
+      });
+      this.published += 1;
+      if (this.published % this.pruneEvery === 0) void this.prune();
+      return toEnvelope(row) as EventEnvelope<P>;
+    } catch (err) {
+      this.log.error({ err, type }, "SSE 事件发布失败");
+      return null;
+    }
+  }
+
+  /** 只保留最近 retain 条 */
+  async prune(): Promise<void> {
+    try {
+      await sql`delete from sse_event where seq <= (select max(seq) from sse_event) - ${this.retain}`.execute(
+        this.opts.db,
+      );
+    } catch (err) {
+      this.log.warn({ err }, "sse_event 裁剪失败");
+    }
+  }
+
+  async since(after: number, limit = 1000): Promise<SinceResult> {
+    const [rows, bounds] = await Promise.all([
+      sql<Row>`select seq, type, at, payload from sse_event where seq > ${after} order by seq limit ${limit}`.execute(
+        this.opts.db,
+      ),
+      sql<{ lo: number | null; hi: number | null }>`select min(seq) as lo, max(seq) as hi from sse_event`.execute(
+        this.opts.db,
+      ),
+    ]);
+    const lo = bounds.rows[0]?.lo;
+    const hi = Number(bounds.rows[0]?.hi ?? 0);
+    // 裁剪过:请求起点早于最老保留条目
+    const gap = after > 0 && lo != null && after < Number(lo) - 1;
+    return { events: rows.rows.map(toEnvelope), gap, latestSeq: Math.max(hi, after) };
+  }
+
+  subscribe(l: BusListener): () => void {
+    this.listeners.add(l);
+    return () => this.listeners.delete(l);
+  }
+
+  async stop(): Promise<void> {
+    this.stopped = true;
+    if (this.reconnectTimer) clearTimeout(this.reconnectTimer);
+    const c = this.client;
+    this.client = null;
+    await c?.end().catch(() => {});
+    await this.draining;
+  }
+}

apps/server/src/http/sse.ts

@@ -3,27 +3,39 @@ import { Hono } from "hono";
 import { streamSSE } from "hono/streaming";
 import { ApiException } from "./errors.ts";
 
-type Listener = (e: EventEnvelope) => void;
+export type BusListener = (e: EventEnvelope) => void;
 
-/**
- * SSE 总线端口。模块与 SSE 路由只依赖这个接口;W2-A 把实现换成 PG NOTIFY + 持久化 seq 时接口不变。
- */
-export interface EventBusPort {
-  readonly latestSeq: number;
-  publish<P>(type: string, payload: P): EventEnvelope<P>;
-  /** 取 seq > after 的事件;gap=true 表示起点已不可补读,客户端需重新拉快照 */
-  since(after: number): { events: EventEnvelope[]; gap: boolean };
-  subscribe(l: Listener): () => void;
+/** since() 结果:gap=true 表示起点已不可补读(被裁剪),客户端需重新拉快照 */
+export interface SinceResult {
+  events: EventEnvelope[];
+  gap: boolean;
+  latestSeq: number;
 }
 
 /**
- * 进程内事件总线:序号自增,保留最近 capacity 条供断线补读。
- * 多进程部署时需换成 PG LISTEN/NOTIFY 或 outbox 投影(W2)。
+ * SSE 总线端口。模块与 SSE 路由只依赖这个接口。
+ * 生产实现是 PgEventBus(sse-pg.ts:sse_event 表 + LISTEN/NOTIFY,多进程可见);
+ * EventBus 是进程内实现,只给单元测试与无库场景用。
+ */
+export interface EventBusPort {
+  /**
+   * 发布事件并广播给所有进程的订阅者;resolve 为带 seq 的信封。
+   * 永不 reject:持久化失败时记日志并返回 null(调用方可以不 await)。
+   */
+  publish<P>(type: string, payload: P): Promise<EventEnvelope<P> | null>;
+  /** 取 seq > after 的事件(按 seq 升序,最多 limit 条) */
+  since(after: number, limit?: number): Promise<SinceResult>;
+  /** 订阅本进程收到的事件(含其它进程发布的);返回退订函数 */
+  subscribe(l: BusListener): () => void;
+}
+
+/**
+ * 进程内事件总线:序号自增,保留最近 capacity 条供断线补读。仅测试/单进程无库场景使用。
  */
 export class EventBus implements EventBusPort {
   private seq = 0;
   private readonly buffer: EventEnvelope[] = [];
-  private readonly listeners = new Set<Listener>();
+  private readonly listeners = new Set<BusListener>();
 
   constructor(private readonly capacity = 1000) {}
 
@@ -31,7 +43,8 @@ export class EventBus implements EventBusPort {
     return this.seq;
   }
 
-  publish<P>(type: string, payload: P): EventEnvelope<P> {
+  /** 同步发布(测试便利);publish 是它的异步包装 */
+  publishSync<P>(type: string, payload: P): EventEnvelope<P> {
     const e: EventEnvelope<P> = { seq: ++this.seq, type, at: new Date().toISOString(), payload };
     this.buffer.push(e);
     if (this.buffer.length > this.capacity) this.buffer.shift();
@@ -39,19 +52,28 @@ export class EventBus implements EventBusPort {
     return e;
   }
 
-  /** 取 seq > after 的缓冲事件;gap=true 表示请求的起点已被淘汰,客户端需重新拉快照 */
-  since(after: number): { events: EventEnvelope[]; gap: boolean } {
-    const oldest = this.buffer[0]?.seq ?? this.seq + 1;
-    const gap = after > 0 && after < oldest - 1;
-    return { events: this.buffer.filter((e) => e.seq > after), gap };
+  publish<P>(type: string, payload: P): Promise<EventEnvelope<P> | null> {
+    return Promise.resolve(this.publishSync(type, payload));
   }
 
-  subscribe(l: Listener): () => void {
+  sinceSync(after: number, limit = 1000): SinceResult {
+    const oldest = this.buffer[0]?.seq ?? this.seq + 1;
+    const gap = after > 0 && after < oldest - 1;
+    return { events: this.buffer.filter((e) => e.seq > after).slice(0, limit), gap, latestSeq: this.seq };
+  }
+
+  since(after: number, limit?: number): Promise<SinceResult> {
+    return Promise.resolve(this.sinceSync(after, limit));
+  }
+
+  subscribe(l: BusListener): () => void {
     this.listeners.add(l);
     return () => this.listeners.delete(l);
   }
 }
 
+const SINCE_PAGE = 500;
+
 export interface SseOptions {
   bus: EventBusPort;
   keepaliveMs?: number;
@@ -79,10 +101,10 @@ export function sseRoutes({ bus, keepaliveMs = 15_000 }: SseOptions) {
         wake?.();
       });
 
-      const { events, gap } = bus.since(after);
+      const { events, gap, latestSeq } = await bus.since(after, SINCE_PAGE);
       let last = after;
       if (gap) {
-        await stream.writeSSE({ event: EventType.resync, data: JSON.stringify({ latestSeq: bus.latestSeq }) });
+        await stream.writeSSE({ event: EventType.resync, data: JSON.stringify({ latestSeq }) });
       }
       const write = async (e: EventEnvelope) => {
         if (e.seq <= last) return;
@@ -90,6 +112,12 @@ export function sseRoutes({ bus, keepaliveMs = 15_000 }: SseOptions) {
         await stream.writeSSE({ id: String(e.seq), event: e.type, data: JSON.stringify(e) });
       };
       for (const e of events) await write(e);
+      // 积压超过一页时继续翻页补读(实时事件已在 queue 里排队,write 按 seq 去重)
+      let page = events;
+      while (page.length >= SINCE_PAGE && !closed) {
+        page = (await bus.since(last, SINCE_PAGE)).events;
+        for (const e of page) await write(e);
+      }
 
       const timer = setInterval(() => {
         void stream.write(": keepalive\n\n");