docs(env): keycloak-setup-log 背书改如实措辞(记有站点地址与 realm 名);测试纪律计数订正;runbook 补新人第一步

评审 Important + Minor 1 + Minor 3。

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

docs/产品需求.md

@@ -0,0 +1,151 @@
+# BPOS-Service-Gateway 产品需求（能力面 stub）
+
+> **本文是能力面 stub，字段级契约以协议册与 `src/contracts` 为准。**
+>
+> 上游《BPOS仓库结构》约定"各服务的产品需求文档随各自服务仓 `docs/` 维护"，本文是本仓那一份。
+
+---
+
+## 1. 一句话定位
+
+> **API 网关**：会话鉴权 · 按域路由（task/data/agent）· SSE 推送中心 · 上传转对象存储。
+> 前端唯一入口，8800，**无自有库**。
+
+（照录《BPOS仓库结构》`docs/proto-v13` 分支 §1 表格中本仓那一行；该行标注
+【提案 · 待架构负责人确认】——网关落仓在 §3"待定项"里尚未定稿。）
+
+---
+
+## 2. 能力面清单
+
+每一项都对得上仓内**已存在**的路由或模块。
+
+### 2.1 会话鉴权
+
+- `requireAuth` / `requireRealmRole` / `requireService`，JWKS 远端验签（与 Task/Data 同一份实现，同一性文件）。
+- 鉴权失败分类**只此一份**：token 坏 → 401；**JWKS 拉不动 → 503**（不是 401）。
+  SSE 是长连，客户端拿到 401 会清 token 去重登录，而重登录打的还是同一个 Keycloak
+  ——把"鉴权服务暂不可达"报成 401 会制造一场无解的重登录风暴。
+- `GET /health` **公开**（不过 `requireAuth`），汇总 `nats` 依赖状态；无自有库故无 `db` 探活。
+
+### 2.2 按域路由（代理）
+
+`/api/task/*` → `TASK_SERVICE_URL`、`/api/data/*` → `DATA_SERVICE_URL`、
+`/api/agent/*` → `AGENT_SERVICE_URL`。全方法透传（`r.all`）。
+
+**挂载顺序 load-bearing**：`/api/stream` 与 `/api/upload` 必须先于 `/api` 代理挂载，
+否则被代理的 catch-all 吞掉。
+
+三条安全约束（每条都是踩过的）：
+
+| 约束 | 做法 | 不这么做的后果 |
+|---|---|---|
+| **主机不可位移** | 在 `new URL(target)` 上**赋 `pathname`**，绝不 `new URL(路径, target)` | 剥前缀后若余下 `//evil.example.com/x` 这种协议相对路径，URL 解析只保留 scheme 而把 host 整个换掉——调用方的 Authorization 与请求体发去攻击者主机，响应还原样回传（SSRF ＋ 令牌外泄） |
+| **内部头剥离** | 转发前删掉 `x-user-id`、`x-user-roles`、`x-internal-token`、`host`、`content-length` | 外部调用方自带 `x-user-id` 就能冒充任意身份——下游若信这类头即越权 |
+| **响应头剥离** | 回传前删掉 `content-encoding`、`content-length`、`transfer-encoding` | undici 已按 `content-encoding` 解过压、body 是明文，原样回传就是自相矛盾的响应（客户端二次解压失败或按错长度截断） |
+
+**Authorization 原样带上**（下游自己再验一遍——纵深防御）。
+下游连不上 → 统一形状的 **502** 并点名是哪个下游。
+**traceparent**：调用方没带就在这里补发一条（网关是外部请求的第一跳），已有的**绝不覆盖**
+（那会把上游已开好的链从中间截断）。
+
+- 落点：`src/modules/proxy/`。
+
+### 2.3 SSE 推送中心
+
+`GET /api/stream`（建流）＋ `POST /api/stream/subscribe`（改订阅集）。
+
+**token 取法**：EventSource 带不了 `Authorization` 头，故 `?access_token=` 与头两收
+（降级轮询共用同一条口）。
+
+**帧序与编号**：
+
+1. `hello`（带 `connectionId`）**必须先于回放**——客户端拿到 id 之前收到的事件无从对账
+   （它还不知道该拿哪个 id 去 `/subscribe`）；
+2. 每条推送帧带 **`id: <seq>`**，`seq` 是 hub 内单调递增的全局序号；
+3. `event:` 是主题名，`data:` 是原始载荷。
+
+**Last-Event-ID 回放**：客户端重连带 `Last-Event-ID` 头（只认纯数字），
+hub 在 **512 条**的环形窗口里补发 `seq > lastEventId` 且**本连接可见**的事件。
+
+> **诚实的局限**：EventSource 重连不携带订阅状态，新连接的 `topics` 是空集，
+> 所以实际只补得到 `touch.system` 与 `touch.staff.<sub>`。任务域的补发要求客户端收到 `hello`
+> 后先重新 `/subscribe`——而那时这一跳已经过去。真正的任务域断点续传要把订阅集做成
+> 服务端持久态（按 `staffSub` 记住上次订阅），**尚未做**。
+
+**心跳**：每 **25s** 发一帧 `event: ping`，`data: {}`。
+流的寿命挂在一个可被两端 resolve 的 promise 上：客户端 abort、服务端 shutdown 都能收流
+（否则心跳定时器与整个闭包会一直漏着不回收）。
+
+**可订阅主题与授权**（默认拒；可订阅集 ＝ **有授权规则**的集合）：
+
+| 主题形状 | 授权判据 |
+|---|---|
+| `touch.task.<taskId>` | 问任务服务 `GET /task/:id/access?staff=<sub>`（负责人/审批人/admin） |
+| `stream.task.<taskId>` | 同上；额外**按需**建 NATS 订阅，最后一个订阅者退订即撤 |
+| `touch.workflow.<workflowId>` | 问任务服务 `GET /workflow/:id/access?staff=<sub>`——判据是该实例**定义里**的节点归属（参与人 `ownerSub` / 审批人 `approverSub`），与单张单的归属不是一回事，故是**另一口**而非复用 |
+| `touch.system` | 人人可见（无需订阅） |
+| `touch.staff.<sub>` | 只对本人可见（无需订阅） |
+
+`admin` 旁路两条 access 询问（既是规则，也免掉白跑的跨服务往返）。
+两次询问都走**服务账号 token**（不是转发用户 token），3s 超时，**非 200 一律当无权**
+——推送面失败关闭：宁可少订一条，不可把别人的任务事件扇给不该看的人；但非 200 会打日志，
+否则"人人被拒"无从诊断。
+
+`/subscribe` 的三条守则：
+
+1. **形状预检先于任何变更**：`add` 里有一个不可订阅的主题就整单 422、零副作用
+   （边订边查会把合法的那条真订上、还白跑一次跨服务往返，客户端以为全失败而服务端留了半截订阅）；
+2. **`add`/`remove` 必须是数组**，不是就 422（类型断言不是校验：字符串走到 `.find` 是 TypeError→500，
+   请求写错却报成服务端故障，而 500 在网关这层还会被当成可重试）；
+3. **只能操作自己名下的连接**：`hub.subscribe` 拿的是**连接的** `staffSub` 授权，
+   不校验归属的话任何登录用户只要知道别人的 `connectionId` 就能改人家的订阅集，
+   还能拿 `results` 当权限探测器。`admin` 例外。
+
+运行时状态（连接不存在／被拒）逐条落 `results` 回 200，客户端据此决定重连还是申请权限。
+
+- 落点：`src/modules/sse/`（`hub.ts`、`routes.ts`、`task-access.ts`、`workflow-access.ts`）。
+
+### 2.4 上传转对象存储
+
+- `POST /api/upload`（`user`/`admin`/`service`）：**8MB** 上限，
+  `bodyLimit` 先看 `content-length`、没有就边读边计数、一超限立刻断流
+  ——413 必须在收完 body 之前定，1GB 的 POST 不会先整个躺进内存再挨拒。
+  返回 `{ ref }`（201）。`x-filename` 可选；记 `ownerSub`（只落盘，读侧鉴权尚未做）。
+- `GET /api/upload/:ref`：回原始字节，**一律 `Content-Disposition: attachment`**（不内联）
+  ＋ `X-Content-Type-Options: nosniff`——`content-type` 是上传方自报的，原样回显＋内联渲染
+  就是网关源上的存储型 XSS。文件名走 **RFC 5987** `filename*=UTF-8''…`
+  （`filename="…"` 只认 ASCII，中文名塞 percent-escape 会原样显示成 `%E4%B8%AD`）。
+- 存储是 **FS 适配器**（`OBJECT_STORE_DIR`，缺省 `./.objectstore`）；
+  `objectstore.ts` / `fs-store.ts` 与 Task 仓 `channel/` 下那两份是同一性文件（逐字节相同）。
+- 落点：`src/modules/upload/`。
+
+---
+
+## 3. 明确不做（边界）
+
+- **不持久化**：无自有库、无 `db:migrate`、无 outbox、无 relay、无 consumer、无 durable。
+- **不做业务判断**：可见性一律**问任务服务**，网关不自己算归属。
+- **不建流、不发布业务事件**：只 core NATS 订阅 `touch.>` 与按需的 `stream.task.*`。
+- **上传读侧的归属鉴权**：`GET /api/upload/:ref` 现在只做验签＋realm 角色校验
+  （`user`/`admin`/`service`），**不校验对象归属**——拿到 `ref` 的任何登录用户都读得到。
+  `ownerSub` 已随上传落盘，归属规则待后续。
+- **SSE 任务域断点续传**：见 §2.3 的局限说明，待订阅集服务端持久化。
+- **不做本地设施兜底**：见《开发环境册》§2.1。
+
+---
+
+## 4. 依据链
+
+| 依据 | 位置 | 用来定什么 |
+|---|---|---|
+| 《系统架构设计.html》v2 §07「两端与实时推送」 | `BPOS-Document-Share/系统架构设计.html` | 网关的会话鉴权、按域路由、SSE 扇出 |
+| 《系统架构设计.html》v2 §06「消息总线：四类消息，五项保障」 | 同上 | 总线的能力要求（BUS-*）——本仓只涉及 BUS-08 尽力而为推送 |
+| 《系统架构设计.html》v2 §10「持久层与文档体系」 | 同上 | 文档与契约的家：谁维护什么、放在哪 |
+| 《BPOS服务消息协议》SVC 分节 | `BPOS-Document-Share/BPOS服务消息协议.md` | `/task/:id/access`、`/workflow/:id/access` 等服务间同步接口 |
+| 《BPOS仓库结构》§1 / §3 | `BPOS-Document-Share/BPOS仓库结构.md`（`docs/proto-v13` 分支） | 本仓定位一行（提案待确认）与"网关落仓"待定项 |
+| 本仓 `src/contracts/envelope.ts` | 本仓 | 信封结构（同一性文件，真源在 BPOS-Document-Share） |
+
+仓内既有记录：`docs/keycloak-setup-log.md`。
+验收演练的一手记录在姊妹仓：`BPOS-Service-Task/tools/n2-acceptance.md`、`n3-acceptance.md`、
+`BPOS-Service-Data/tools/n1-acceptance.md`。

docs/开发环境册.md

@@ -0,0 +1,132 @@
+# BPOS-Service-Gateway 开发环境册
+
+> 简版。与姊妹仓《BPOS-Service-Task/docs/开发环境册.md》共享设施与纪律段落，本册只写本仓差异与要点。
+>
+> **凭证纪律**：本册只写环境变量的**键名**，绝不落主机、IP、账号、口令、密钥值。
+> 真值去共享设施的凭证发放渠道要，写进本机 `.env`（该文件永不入 git）。
+
+---
+
+## 1. 本仓是什么
+
+API 网关（`bpos-service-gateway`）：前端唯一入口。
+会话鉴权 · 按域路由（task/data/agent）· SSE 推送中心 · 上传转对象存储。
+
+**8800，无自有库**——不持久化，故无 `db:migrate`、无 outbox、无 relay、无 consumer。
+NATS 只用连通性（订阅 `touch.>` 与按需的 `stream.task.*`）。
+
+> 落仓状态：上游《BPOS仓库结构》原六仓拓扑未含网关，§3"待定项"记为待定；
+> `docs/proto-v13` 分支上已有【提案】"已建独立仓 BPOS-Service-Gateway"，**待架构负责人确认**。
+
+---
+
+## 2. 共享设施与配置
+
+### 2.1 一条硬性纪律：不许自建本地设施
+
+**严禁在本机启动 Docker / 容器 / 任何本地替代基础设施。** NATS 在共享开发机上，
+Keycloak 是共享站点。设施不可达时如实降级或停下，**不起本地兜底**：
+NATS 不可达 → SSE 推送停摆、`/health` 照实报 fail，但代理/上传/健康检查照常服务
+（`src/app/server.ts` 明确不因此拖垮 HTTP 面）。交付如实标注 `SKIPPED-INFRA-DOWN` 或报 BLOCKED。
+
+### 2.2 环境变量（只写键名）
+
+`.env.example` 是权威清单。
+
+| 键名 | 作用 | 缺省与后果 |
+|---|---|---|
+| `PORT` | HTTP 监听端口 | 缺省 8800 |
+| `NATS_URL` | 共享 NATS | `src/shared/config.ts` 有开发期缺省，部署一律显式配 |
+| `KEYCLOAK_URL` / `KEYCLOAK_REALM` | 验签用的站点与 realm | 用户 token 与服务 token 同一 realm |
+| `KEYCLOAK_CLIENT_ID` | 本仓 client，缺省 `api-gateway` | |
+| `KEYCLOAK_CLIENT_SECRET` | **须为 `api-gateway` 服务账号的密钥** | 只从环境读、无缺省。配错/配成别的 client 的密钥＝换来别人的服务账号 token（或直接被拒），SSE 的订阅授权全线失败并"人人被拒"。**值绝不写进任何文档或提交**，从 Keycloak 该 client 的 Credentials 页取 |
+| `TASK_SERVICE_URL` | 代理与订阅授权的目标 | 缺省本机 8801 |
+| `DATA_SERVICE_URL` | 代理目标 | 缺省本机 8802 |
+| `AGENT_SERVICE_URL` | 代理目标 | 缺省本机 8901（开发期即 Task 仓的 fake-agent 桩） |
+| `OBJECT_STORE_MODE` / `OBJECT_STORE_DIR` | 上传落盘（FS 适配器） | 缺省目录 `./.objectstore`，已在 `.gitignore` |
+
+**为什么这里必须是 `api-gateway` 自己的密钥**：SSE 订阅授权要拿服务账号 token 去问任务服务
+`GET /task/:id/access` 与 `GET /workflow/:id/access`，后者在任务侧是
+`requireService(['api-gateway'])`——只认网关。`src/shared/service-token.ts` 用
+`config.keycloak.clientId` ＋ `KEYCLOAK_CLIENT_SECRET` 换 `client_credentials` token，
+两者必须是**同一个 client** 的。相关的 realm 角色与 audience mapper 配置记录见
+`docs/keycloak-setup-log.md`——**该文不含任何密钥值**（它自己在开头也声明"不含任何 secret、
+密码或完整 token"），**但记有 Keycloak 站点地址与 realm 名**，引用或外发时注意范围。
+
+`.env` / `.env.*` 已在 `.gitignore`（`!.env.example` 例外）。收尾核验里
+`git log --all --oneline --diff-filter=A -- .env | wc -l` 必须为 `0`。
+
+---
+
+## 3. 端口与运行
+
+**新人第一步**：本仓 `package.json` 的 `engines` 声明 **Node ≥ 22**；包管理器用 **pnpm**
+（仓内有 `pnpm-lock.yaml` 与 `pnpm-workspace.yaml`，别用 npm/yarn 装）。
+
+```bash
+pnpm i               # 装依赖
+cp .env.example .env # 照 §2.2 逐项填；真值向共享设施的凭证发放渠道索取，不要向别人的 .env 复制
+pnpm dev             # tsx watch，读 .env，起在 :8800
+pnpm gates           # typecheck + lint + test
+```
+
+**无 `db:migrate`**（本仓无自有库）。
+
+启动顺序：建 SseHub → 建 app → 起 HTTP → 连 NATS（通则 `hub.attachNats`，不通只告警）。
+`SIGINT`/`SIGTERM`：**先收 SSE 长连**再 `server.close()`——那些响应永远不结束，
+不主动关就会一直等；3s 兜底强退。
+
+跨仓端口速查：Gateway 8800（无库）、Task 8801（`jx_dispatch`）、Data 8802（`jx_business`）、
+fake-agent 桩 8901（在 Task 仓 `tools/fake-agent.ts`，Activity 仓交付真桩后退役）。
+
+---
+
+## 4. 总线用法：只订阅，不建流
+
+本仓**不建流、不发布业务事件、无 outbox、无 relay、无 durable**。
+只用 core NATS 订阅两类推送主题（BUS-08 尽力而为，本就不落 JetStream）：
+
+- `touch.>`：连上就订，进程内按连接可见性扇出；
+- `stream.task.<id>`：**按需**订——有连接订阅了才建 NATS 订阅，最后一个连接退订就撤掉。
+
+三条 JetStream 流（`BPOS_EVT` / `BPOS_CMD` / `BPOS_DLQ`）由 Task / Data 两仓引导，与本仓无关。
+
+---
+
+## 5. 测试纪律
+
+本仓无库，共享库那套 `zz_test_` 前缀 / `afterAll` 谓词清理 / LIKE 下划线转义 / 生产扫描排除
+的纪律**在本仓无直接落点**，但三条相关的仍适用：
+
+1. **集成用例打的是共享 Keycloak 与共享 NATS**，不许起本地替身；不可达就如实跳过或失败。
+2. **`test/support/token.ts` 是三仓同一性文件**（逐字节相同）——改它要同步三仓并重跑同一性脚本。
+3. 用例造出来的 taskId / workflowId 一律用 `zz_test_` 前缀：它们会经服务账号 token 打到
+   任务服务的 `/access` 端点上，落在**共用**的 `jx_dispatch` 库前面，前缀是那边扫描排除的抓手。
+
+---
+
+## 6. 门禁与验收
+
+```bash
+pnpm typecheck        # tsc --noEmit
+pnpm lint             # eslint src test
+pnpm vitest run       # 全量
+pnpm gates            # 上面三条
+```
+
+跨仓同一性由 Task 仓的脚本统一跑（三仓需为同级目录）：
+
+```bash
+node ../BPOS-Service-Task/scripts/check-identity.mjs    # 期望 16/16
+```
+
+本仓参与其中 11 条：`keycloak-auth.ts`、`bus/nats.ts`、`contracts/envelope.ts`、`app/health.ts`、
+`test/support/token.ts`、`vitest.config.ts`（三仓）＋ `shared/errors.ts`、`shared/service-token.ts`、
+`shared/trace.ts`（与 Task 两仓）＋ `upload/objectstore.ts`、`upload/fs-store.ts`
+（与 Task 的 `channel/objectstore.ts`、`channel/fs-store.ts` 同内容、不同路径）。
+**改这些文件必须同步复制到相关仓，再重跑脚本。**
+
+仓内既有记录：`docs/keycloak-setup-log.md`（realm 角色 `service`、两个服务账号的 audience mapper、
+共享 client 影响面、解码真 token 的验证结果）。
+验收演练的一手记录在姊妹仓：`BPOS-Service-Task/tools/n2-acceptance.md`（touch 经 SSE 到浏览器）、
+`n3-acceptance.md`（编排全链）、`BPOS-Service-Data/tools/n1-acceptance.md`（含网关这条腿的 aud 边界）。

src/app/app.ts

@@ -1,14 +1,35 @@
 import { Hono } from 'hono';
 import { healthRoute, registerHealthCheck } from './health';
 import { checkNats } from '../shared/bus/nats';
+import { createKeycloakTokenProvider } from '../shared/service-token';
 import { proxyRoutes } from '../modules/proxy/routes';
 import { uploadRoutes } from '../modules/upload/routes';
+import { SseHub } from '../modules/sse/hub';
+import { sseRoutes } from '../modules/sse/routes';
+import { createHttpTaskAccess } from '../modules/sse/task-access';
+import { createHttpWorkflowAccess } from '../modules/sse/workflow-access';
 
-export function buildApp(): Hono {
+/**
+ * hub 由外面传：server.ts 要拿同一个实例去 attachNats／shutdown。缺省自建一个，
+ * 让只关心 HTTP 面的测试照旧 buildApp() 就能用。
+ */
+export function buildApp(hub: SseHub = new SseHub()): Hono {
   registerHealthCheck('nats', checkNats);
   const app = new Hono();
+  // 统一错误形状：任何未捕获异常都收敛成 {error,message} 的 500，不漏 Hono 默认裸文本／栈
+  app.onError((err, c) => {
+    console.error(err);
+    return c.json({ error: 'internal_error', message: '服务内部错误——已记录，稍后重试或联系维护' }, 500);
+  });
   app.route('/health', healthRoute()); // 公开：不过 requireAuth
-  // 顺序要紧：/api/upload 必须先于 /api 代理挂载，否则被代理的 catch-all 吞掉
+  // 惰性构造：createKeycloakTokenProvider 只在闭包被调用时读 KEYCLOAK_CLIENT_SECRET，
+  // 建 app 本身不要求环境齐备，否则任何 buildApp() 的测试都会当场炸。
+  // 两口共用同一个 token provider：换来的是同一个服务账号 token，缓存也只此一份
+  const serviceToken = createKeycloakTokenProvider();
+  const access = createHttpTaskAccess(serviceToken);
+  const workflowAccess = createHttpWorkflowAccess(serviceToken);
+  // 顺序要紧：/api/stream 与 /api/upload 必须先于 /api 代理挂载，否则被代理的 catch-all 吞掉
+  app.route('/api/stream', sseRoutes(hub, access, workflowAccess));
   app.route('/api/upload', uploadRoutes());
   app.route('/api', proxyRoutes());
   return app;

src/app/server.ts

@@ -1,7 +1,35 @@
 import { serve } from '@hono/node-server';
 import { buildApp } from './app';
 import { config } from '../shared/config';
+import { closeNats, getNats } from '../shared/bus/nats';
+import { SseHub } from '../modules/sse/hub';
 
-serve({ fetch: buildApp().fetch, port: config.port }, (info) => {
+const hub = new SseHub();
+const app = buildApp(hub);
+
+const server = serve({ fetch: app.fetch, port: config.port }, (info) => {
   console.log(`${config.serviceName} listening on :${info.port}`);
 });
+
+// NATS 不可达不该拖垮 HTTP 面：代理／上传／健康检查照常，只是没有推送（/health 会照实报 fail）
+const nc = await getNats().catch(() => null);
+if (nc) hub.attachNats(nc);
+else console.warn(`NATS ${config.natsUrl} 不可达——SSE 推送停摆，HTTP 面照常服务`);
+
+/**
+ * D12：网关此前没有信号处理（N1 靠默认 exit 143）。现在挂着长连 SSE，那些响应永远不结束，
+ * server.close() 会一直等它们——必须先把连接收掉。3s 兜底强退：任何一环卡住也不赖着不走。
+ */
+let closing = false;
+async function shutdown(signal: string): Promise<void> {
+  if (closing) return;
+  closing = true;
+  console.log(`${signal} 收到——关闭 SSE 连接后退出`);
+  setTimeout(() => process.exit(0), 3000).unref();
+  hub.shutdown();
+  server.close();
+  if (nc) await closeNats();
+  process.exit(0);
+}
+process.on('SIGINT', () => void shutdown('SIGINT'));
+process.on('SIGTERM', () => void shutdown('SIGTERM'));

src/modules/proxy/forward.ts

@@ -1,4 +1,5 @@
 import type { Context } from 'hono';
+import { newTraceparent } from '../../shared/trace';
 
 const STRIP_HEADERS = ['x-user-id', 'x-user-roles', 'x-internal-token', 'host', 'content-length'];
 
@@ -11,10 +12,19 @@ const STRIP_RESPONSE_HEADERS = ['content-encoding', 'content-length', 'transfer-
 
 /** 验签后转发：Authorization 原样带上（下游自己再验一遍——纵深防御，指南 5.6） */
 export async function forward(c: Context, target: string, stripPrefix: string): Promise<Response> {
-  const search = new URL(c.req.url).search;
-  const path = c.req.path.slice(stripPrefix.length) || '/';
+  // 主机不可位移（C1）：绝不能 new URL(路径, target)——剥前缀后若余下 `//evil.example.com/x`
+  // 这种协议相对路径，URL 解析只保留 target 的 scheme 并把 host 整个换掉，调用方的
+  // Authorization 与请求体就发去攻击者指定的主机，响应还原样回传（SSRF＋令牌外泄）。
+  // 改成在 target 上赋 pathname：赋 pathname 只动路径，永远动不了 host/port。
+  const url = new URL(target);
+  const sub = c.req.path.slice(stripPrefix.length) || '/';
+  url.pathname = url.pathname.replace(/\/$/, '') + (sub.startsWith('/') ? sub : `/${sub}`);
+  url.search = new URL(c.req.url).search;
   const headers = new Headers(c.req.raw.headers);
   for (const h of STRIP_HEADERS) headers.delete(h);
+  // 网关是外部请求的第一跳：调用方没带 traceparent 就在这里补发一条，让下游一路 childOf 续链。
+  // 已有的绝不覆盖——那会把上游（浏览器/上级网关）已经开好的链从中间截断。
+  if (!headers.has('traceparent')) headers.set('traceparent', newTraceparent());
   const method = c.req.method;
   const init: RequestInit & { duplex?: 'half' } = { method, headers };
   if (method !== 'GET' && method !== 'HEAD') {
@@ -23,7 +33,7 @@ export async function forward(c: Context, target: string, stripPrefix: string):
   }
   let res: Response;
   try {
-    res = await fetch(new URL(path + search, target), init);
+    res = await fetch(url, init);
   } catch {
     // 下游没起/连不上：给统一错误形状的 502，并点名是哪个下游
     return c.json(

src/modules/sse/hub.ts

@@ -0,0 +1,191 @@
+import type { NatsConnection, Subscription } from 'nats';
+
+export interface SseEvent { seq: number; topic: string; data: string }
+export interface Connection {
+  id: string;
+  staffSub: string;
+  isAdmin: boolean;
+  topics: Set<string>;
+  send(e: SseEvent): void;
+  /**
+   * D12：从 streamSSE 回调之外关掉这条流的把手。流的生命周期挂在回调里的一个
+   * 挂起 promise 上，只有 resolve 它才能收尾——所以 hub 得拿到一个由路由实现的
+   * 闭包（清心跳＋resolve）。单测手搓的连接不需要，故可选。
+   */
+  close?(): void;
+}
+export interface TaskAccessPort { allowed(taskId: string, staffSub: string): Promise<boolean> }
+/**
+ * 工作流可见性口。与 TaskAccessPort 分开而不是复用同一口：判据不同源——任务问的是单张单的
+ * 归属，工作流问的是**实例定义里**的节点归属（参与人/审批人），任务服务上也是两条端点。
+ */
+export interface WorkflowAccessPort { allowed(workflowId: string, staffSub: string): Promise<boolean> }
+
+/**
+ * 订阅失败带判别式：路由要按失败**种类**分流 HTTP 码（形状错＝422 整单驳回，
+ * 连接不存在／被拒＝200 里逐条报）。靠 message.includes('不可订阅') 猜种类，
+ * 改一个字文案就会把 422 悄悄变成 200。
+ */
+export type SubscribeResult =
+  | { ok: true }
+  | { ok: false; reason: 'bad-shape' | 'no-conn' | 'denied'; message: string };
+
+/**
+ * 可订阅集合＝**有授权规则**的集合，默认拒。
+ *
+ * touch.workflow. 在 I6 被摘出去过：那时形状一过就直接进订阅集，而 subscribe 的鉴权分支只
+ * 覆盖 touch.task./stream.task.，于是任何登录用户订 touch.workflow.<任意 id> 都能收别人流程
+ * 的推送——摘掉它是因为**当时没有规则**，不是因为工作流不该推。
+ * N3 规则到位，故放回来：可见性＝该实例定义里的参与人（节点 ownerSub）或审批人
+ * （approverSub），经 workflowAccess 口向任务服务问；admin 旁路。判据落在下面 subscribe 的
+ * 鉴权分支里——这两处必须同一次改动，只放形状白名单就是把 I6 原样退回去。
+ */
+const SUBSCRIBABLE = /^(touch\.task\.|touch\.workflow\.|stream\.task\.)[^.]+$/;
+
+/** 主题形状唯一判据。导出是为了让路由能在**动任何状态之前**预检，见 badShapeMessage */
+export function isSubscribable(topic: string): boolean {
+  return SUBSCRIBABLE.test(topic);
+}
+
+/** 形状错的文案也只有这一份——路由预检与 hub.subscribe 报的必须是同一句 */
+export function badShapeMessage(topic: string): string {
+  return `主题 ${topic} 不可订阅——只接受 touch.task.*、touch.workflow.* 与 stream.task.*`;
+}
+
+export class SseHub {
+  private conns = new Map<string, Connection>();
+  private ring: SseEvent[] = [];
+  private seq = 0;
+  private streamSubs = new Map<string, Subscription>();
+  private nc: NatsConnection | null = null;
+
+  constructor(private replayWindow = 512) {}
+
+  attachNats(nc: NatsConnection): void {
+    this.nc = nc;
+    const sub = nc.subscribe('touch.>', {
+      callback: (_err, msg) => this.deliver(msg.subject, new TextDecoder().decode(msg.data)),
+    });
+    this.streamSubs.set('touch.>', sub);
+  }
+
+  /**
+   * D11：只登记，**不**回放。回放必须排在 hello 帧之后——客户端拿到 connectionId
+   * 之前收到的事件无从对账（它还不知道该拿哪个 id 去 /subscribe）。
+   */
+  register(conn: Connection): void {
+    this.conns.set(conn.id, conn);
+  }
+
+  /**
+   * 断线重连补发：按全局 seq 补窗口内错过的、且本连接可见的事件。
+   *
+   * 诚实的局限（D11）：EventSource 重连不带订阅状态，路由只能以空 topics 重新登记，
+   * 于是这里实际只补得到 `touch.system` 与 `touch.staff.<sub>`。任务域的补发要求客户端
+   * 收到 hello 后先重新 /subscribe——但那时 lastEventId 这一跳已经过去了。真正的
+   * 任务域断点续传得等 N3 把订阅集做成服务端持久态（按 staffSub 记住上次订阅）。
+   */
+  replay(conn: Connection, lastEventId?: number): void {
+    if (lastEventId === undefined) return;
+    for (const e of this.ring) {
+      if (e.seq > lastEventId && this.visible(conn, e.topic)) conn.send(e);
+    }
+  }
+
+  /**
+   * F2：连接归属。订阅面的授权判据是**连接的** staffSub，不是调用方的——
+   * 路由必须先确认调用方就是连接主人，否则任何拿到别人 connectionId 的登录用户
+   * 都能改人家的订阅集，还能拿 results 当权限探测器。
+   * 连接不存在返回 null（交回路由走原本的 no-conn 路径，不额外泄漏存在性）。
+   */
+  ownerOf(connId: string): string | null {
+    return this.conns.get(connId)?.staffSub ?? null;
+  }
+
+  unregister(connId: string): void {
+    const conn = this.conns.get(connId);
+    this.conns.delete(connId);
+    if (conn) for (const t of conn.topics) this.maybeDropStreamSub(t);
+  }
+
+  async subscribe(
+    connId: string,
+    topic: string,
+    access: TaskAccessPort,
+    workflowAccess: WorkflowAccessPort,
+  ): Promise<SubscribeResult> {
+    // D3：形状校验排在连接查找**之前**。请求体里写了个根本不可订阅的主题，是这个请求
+    // 自身不合法（422），跟连接在不在无关；反过来先查连接的话，一个失效 connectionId
+    // 会把形状错整个遮掉，HTTP 面只能回 200。
+    if (!isSubscribable(topic)) return { ok: false, reason: 'bad-shape', message: badShapeMessage(topic) };
+    const conn = this.conns.get(connId);
+    if (!conn) return { ok: false, reason: 'no-conn', message: `连接 ${connId} 不存在——先建立 /api/stream` };
+    const subjectId = topic.split('.').pop() ?? '';
+    if (!conn.isAdmin && (topic.startsWith('stream.task.') || topic.startsWith('touch.task.'))) {
+      if (!(await access.allowed(subjectId, conn.staffSub))) {
+        return { ok: false, reason: 'denied', message: `无权订阅任务 ${subjectId}——需要负责人/审批人/admin` };
+      }
+    }
+    // 工作流另走一口：判据是实例**定义里**的节点归属，与单张单的归属不是一回事。
+    // admin 同样旁路——既是规则（admin 全可见），也免掉一次白跑的跨服务往返。
+    if (!conn.isAdmin && topic.startsWith('touch.workflow.')) {
+      if (!(await workflowAccess.allowed(subjectId, conn.staffSub))) {
+        return {
+          ok: false,
+          reason: 'denied',
+          message: `无权订阅工作流 ${subjectId}——需要该流程的参与人/审批人或 admin`,
+        };
+      }
+    }
+    conn.topics.add(topic);
+    if (topic.startsWith('stream.task.') && this.nc && !this.streamSubs.has(topic)) {
+      const sub = this.nc.subscribe(topic, {
+        callback: (_err, msg) => this.deliver(msg.subject, new TextDecoder().decode(msg.data)),
+      });
+      this.streamSubs.set(topic, sub);
+    }
+    return { ok: true };
+  }
+
+  unsubscribe(connId: string, topic: string): void {
+    this.conns.get(connId)?.topics.delete(topic);
+    this.maybeDropStreamSub(topic);
+  }
+
+  deliver(topic: string, data: string): void {
+    const e: SseEvent = { seq: ++this.seq, topic, data };
+    this.ring.push(e);
+    if (this.ring.length > this.replayWindow) this.ring.shift();
+    for (const conn of this.conns.values()) {
+      if (this.visible(conn, topic)) conn.send(e);
+    }
+  }
+
+  /**
+   * D12：优雅退出。网关此前没有 SIGTERM 处理（N1 靠默认行为 exit 143），现在挂着
+   * 长连 SSE——不主动关，server.close() 会一直等这些永不结束的响应。
+   * 先快照再遍历：close() 可能同步触发 unregister，边删边迭代会漏。
+   */
+  shutdown(): void {
+    for (const conn of [...this.conns.values()]) conn.close?.();
+    this.conns.clear();
+    for (const sub of this.streamSubs.values()) sub.unsubscribe();
+    this.streamSubs.clear();
+    this.nc = null;
+  }
+
+  private visible(conn: Connection, topic: string): boolean {
+    if (topic === 'touch.system') return true;
+    if (topic.startsWith('touch.staff.')) return topic === `touch.staff.${conn.staffSub}`;
+    return conn.topics.has(topic);
+  }
+
+  private maybeDropStreamSub(topic: string): void {
+    if (!topic.startsWith('stream.task.')) return;
+    const stillUsed = [...this.conns.values()].some((c) => c.topics.has(topic));
+    if (!stillUsed) {
+      this.streamSubs.get(topic)?.unsubscribe();
+      this.streamSubs.delete(topic);
+    }
+  }
+}

src/modules/sse/routes.ts

@@ -0,0 +1,113 @@
+import { Hono } from 'hono';
+import { randomUUID } from 'node:crypto';
+import { streamSSE } from 'hono/streaming';
+import { classifyAuthError, requireAuth, toAuthUser, verifyAccessToken } from '../../shared/keycloak-auth';
+import { badShapeMessage, isSubscribable } from './hub';
+import type { Connection, SseHub, TaskAccessPort, WorkflowAccessPort } from './hub';
+
+export function sseRoutes(hub: SseHub, access: TaskAccessPort, workflowAccess: WorkflowAccessPort): Hono {
+  const r = new Hono();
+
+  r.get('/', async (c) => {
+    // EventSource 带不了 Authorization 头：接受 ?access_token=，也接受头（降级轮询共用）
+    const token = c.req.query('access_token') ?? (c.req.header('authorization') ?? '').replace(/^bearer /i, '');
+    if (!token) return c.json({ error: 'unauthorized', message: '缺 token（?access_token= 或 Authorization 头）' }, 401);
+    let user;
+    try {
+      user = toAuthUser(await verifyAccessToken(token));
+    } catch (err) {
+      // I4：这里不能自己另判一套。SSE 是长连，客户端拿到 401 会清 token 去重登录，
+      // 而 JWKS 拉不动的时候重登录打的还是同一个 Keycloak——分类判据只此一份（requireAuth 同源）。
+      if (classifyAuthError(err) === 503) {
+        return c.json({ error: 'auth_unavailable', message: '鉴权服务暂不可达——稍后重试' }, 503);
+      }
+      return c.json({ error: 'invalid_token', message: 'token 校验失败' }, 401);
+    }
+    const lastRaw = c.req.header('last-event-id');
+    const lastEventId = lastRaw !== undefined && /^\d+$/.test(lastRaw) ? Number(lastRaw) : undefined;
+    const connectionId = randomUUID();
+    return streamSSE(c, async (stream) => {
+      // D12：流的寿命＝这个 promise 的寿命。旧写法 `new Promise(() => {})` 永不 settle——
+      // 客户端断开后回调仍卡在 await 上，心跳定时器与整个闭包（连接对象、user、stream）
+      // 一起漏着不回收。resolve 把手要两个来源都够得着：客户端 abort、服务端 shutdown。
+      let endStream!: () => void;
+      const closed = new Promise<void>((resolve) => {
+        endStream = resolve;
+      });
+      const heartbeat = setInterval(() => void stream.writeSSE({ event: 'ping', data: '{}' }), 25_000);
+      const end = (): void => {
+        clearInterval(heartbeat);
+        endStream();
+      };
+      const conn: Connection = {
+        id: connectionId,
+        staffSub: user.id,
+        isAdmin: user.roles.includes('admin'),
+        topics: new Set(),
+        send: (e) => void stream.writeSSE({ id: String(e.seq), event: e.topic, data: e.data }),
+        close: end, // D12：hub.shutdown() 借它从外面收流
+      };
+      // abort 先挂上：hello 还没写完客户端就跑了也要收得住
+      stream.onAbort(() => {
+        hub.unregister(connectionId);
+        end();
+      });
+      hub.register(conn);
+      // D11：hello 必须先于回放——客户端要先拿到 connectionId 才谈得上 /subscribe 与对账。
+      await stream.writeSSE({ event: 'hello', data: JSON.stringify({ connectionId }) });
+      // 局限：新连接的 topics 是空集（EventSource 重连不携带订阅状态），所以这里实际只补
+      // touch.system 与 touch.staff.<sub>；任务域的补发要等客户端重新 /subscribe 之后才可能，
+      // 而那时 Last-Event-ID 这一跳已经过去。详见 hub.replay 的注释。
+      hub.replay(conn, lastEventId);
+      await closed;
+    });
+  });
+
+  r.post('/subscribe', requireAuth(), async (c) => {
+    const body = (await c.req.json()) as { connectionId?: string; add?: unknown; remove?: unknown };
+    const connectionId = body.connectionId;
+    if (!connectionId) return c.json({ error: 'validation_failed', message: '缺 connectionId（hello 事件里给了）' }, 422);
+    // I7：类型断言不是校验。`{"add":"touch.task.t1"}` 过去被 as string[] 一路带到 .find，
+    // 字符串上没有 find 就是 TypeError → 500：请求写错了却报成服务端故障，且 500 在网关
+    // 这一层还会被当成可重试。形状不对一律 422，且必须排在任何使用之前。
+    if (body.add !== undefined && !Array.isArray(body.add)) {
+      return c.json({ error: 'validation_failed', message: 'add 必须是主题数组' }, 422);
+    }
+    if (body.remove !== undefined && !Array.isArray(body.remove)) {
+      return c.json({ error: 'validation_failed', message: 'remove 必须是主题数组' }, 422);
+    }
+    const add = (body.add ?? []) as string[];
+
+    // F1：形状预检必须**先于**任何变更。此前是边订边查，`add:['touch.task.t1','bogus']`
+    // 会先把 t1 真订上（还白跑一次 task-access 往返）再整单 422——客户端以为全失败，
+    // 服务端却留了半截订阅，两边模型就此错开。422 是"请求写错了"，应当零副作用。
+    const badShape = add.find((t) => !isSubscribable(t));
+    if (badShape !== undefined) {
+      return c.json({ error: 'validation_failed', message: badShapeMessage(badShape) }, 422);
+    }
+
+    // F2：只能操作自己名下的连接。hub.subscribe 是拿**连接的** staffSub 去授权的，
+    // 不校验归属的话，任何登录用户只要知道别人的 connectionId 就能改人家的订阅集
+    // （remove 更是连 subscribe 都不经过，此前完全没有任何校验）。
+    const user = c.get('user');
+    const owner = hub.ownerOf(connectionId);
+    if (owner !== null && owner !== user.id && !user.roles.includes('admin')) {
+      return c.json({ error: 'forbidden', message: '不能操作他人连接的订阅' }, 403);
+    }
+
+    const results: Record<string, string> = {};
+    for (const topic of add) {
+      const res = await hub.subscribe(connectionId, topic, access, workflowAccess);
+      // 形状已由上面的预检挡掉，这里只是不变量兜底：真漏进来也绝不降级成 200。
+      // 连接不存在／被拒是运行时状态，逐条落 results 回 200（客户端据此决定重连还是申请权限）。
+      if (!res.ok && res.reason === 'bad-shape') {
+        return c.json({ error: 'validation_failed', message: res.message }, 422);
+      }
+      results[topic] = res.ok ? 'ok' : `denied: ${res.message}`;
+    }
+    for (const topic of (body.remove ?? []) as string[]) hub.unsubscribe(connectionId, topic);
+    return c.json({ results });
+  });
+
+  return r;
+}

src/modules/sse/task-access.ts

@@ -0,0 +1,29 @@
+import { config } from '../../shared/config';
+import type { TokenProvider } from '../../shared/service-token';
+import type { TaskAccessPort } from './hub';
+
+/**
+ * 订阅任务域主题前问任务服务一句：`GET {TASK_SERVICE_URL}/task/:id/access?staff=<sub>`。
+ * 走服务账号 token（该端点在任务侧是 requireService(['api-gateway', ...])），不是转发用户 token。
+ * 非 200 一律当"无权"——推送面失败关闭：宁可少订一条，不可把别人的任务事件扇给不该看的人。
+ */
+export function createHttpTaskAccess(token: TokenProvider): TaskAccessPort {
+  return {
+    async allowed(taskId, staffSub) {
+      const url = `${config.taskServiceUrl}/task/${encodeURIComponent(taskId)}/access?staff=${encodeURIComponent(staffSub)}`;
+      const res = await fetch(url, {
+        headers: { authorization: `Bearer ${await token()}` },
+        // F3：/subscribe 的 add 循环是串行的——任务服务吊着不答就会把整个订阅请求钉死
+        signal: AbortSignal.timeout(3000),
+      });
+      if (res.status !== 200) {
+        // F3：失败关闭是对的，但"人人被拒"如果不出声就无从诊断——把下游状态码打出来。
+        // 注意只在非 200 时告警，不把网络异常吞成 false（下游宕机不该读作"没权限"）。
+        console.warn('[task-access] 非200', taskId, res.status);
+        return false;
+      }
+      const body = (await res.json()) as { allowed?: boolean };
+      return body.allowed === true;
+    },
+  };
+}

src/modules/sse/workflow-access.ts

@@ -0,0 +1,30 @@
+import { config } from '../../shared/config';
+import type { TokenProvider } from '../../shared/service-token';
+import type { WorkflowAccessPort } from './hub';
+
+/**
+ * 订阅工作流主题前问任务服务一句：`GET {TASK_SERVICE_URL}/workflow/:id/access?staff=<sub>`。
+ * 走服务账号 token（该端点在任务侧是 requireService(['api-gateway'])——只认网关，代理没有
+ * 理由问流程可见性），不是转发用户 token。
+ * 非 200 一律当"无权"——推送面失败关闭：宁可少订一条，不可把别人流程的推进扇给不该看的人。
+ */
+export function createHttpWorkflowAccess(token: TokenProvider): WorkflowAccessPort {
+  return {
+    async allowed(workflowId, staffSub) {
+      const url = `${config.taskServiceUrl}/workflow/${encodeURIComponent(workflowId)}/access?staff=${encodeURIComponent(staffSub)}`;
+      const res = await fetch(url, {
+        headers: { authorization: `Bearer ${await token()}` },
+        // F3 同理：/subscribe 的 add 循环是串行的——任务服务吊着不答就会把整个订阅请求钉死
+        signal: AbortSignal.timeout(3000),
+      });
+      if (res.status !== 200) {
+        // 失败关闭是对的，但"人人被拒"如果不出声就无从诊断——把下游状态码打出来。
+        // 注意只在非 200 时告警，不把网络异常吞成 false（下游宕机不该读作"没权限"）。
+        console.warn('[workflow-access] 非200', workflowId, res.status);
+        return false;
+      }
+      const body = (await res.json()) as { allowed?: boolean };
+      return body.allowed === true;
+    },
+  };
+}

src/shared/bus/nats.ts

@@ -3,11 +3,29 @@ import type { NatsConnection } from 'nats';
 import { config } from '../config';
 
 let nc: NatsConnection | null = null;
+/**
+ * 在途去重：nc 只在 connect() **resolve 之后**才被赋值，所以并发的首批调用者读到的
+ * 都还是 null，于是各自拨一条连接；最后一个赋值进 nc，前面那些谁也够不着——drain 不掉、
+ * 也不计入 closeNats，就那么挂在服务端的连接表上直到进程死。记住这次在途的 promise，
+ * 让并发调用者共享同一次拨号；成功与失败都清空，失败后下一次调用可以重新拨。
+ */
+let connecting: Promise<NatsConnection> | null = null;
 
 export async function getNats(): Promise<NatsConnection> {
   if (nc && !nc.isClosed()) return nc;
-  nc = await connect({ servers: config.natsUrl, name: config.serviceName, timeout: 5000 });
-  return nc;
+  if (connecting) return await connecting;
+  const dial = connect({ servers: config.natsUrl, name: config.serviceName, timeout: 5000 })
+    .then((c) => {
+      nc = c;
+      connecting = null;
+      return c;
+    })
+    .catch((err: unknown) => {
+      connecting = null;
+      throw err;
+    });
+  connecting = dial;
+  return await dial;
 }
 
 export async function checkNats(): Promise<'ok' | 'fail'> {

src/shared/errors.ts

@@ -0,0 +1,11 @@
+/** 跨模块的失败分类——放 shared/ 而非某个业务模块下，好让 service-token 这类通用件能整份复用到别的仓 */
+
+/**
+ * 永久失败：重试多少次都不会好（配置缺失、凭据无效、被拒绝的鉴权）。
+ * 与普通 Error 分开，是为了让 saga 能分流——永久失败落 blocked＋治理，
+ * 暂时失败原地等下一轮重试；两者混在一个 Error 里只能无限重试。
+ * name 显式覆盖：错误要以 String(err) 落库与进治理消息，"Error: xx" 会把永久性这层信息丢掉。
+ */
+export class PermanentFailure extends Error {
+  override name = 'PermanentFailure';
+}

src/shared/keycloak-auth.ts

@@ -88,11 +88,38 @@ export function requireAuth(): MiddlewareHandler {
       c.set('user', toAuthUser(await verifyAccessToken(h.slice(7).trim())));
       return await next();
     } catch (err) {
+      if (classifyAuthError(err) === 503) {
+        return c.json({ error: 'auth_unavailable', message: '鉴权服务暂不可达——稍后重试' }, 503);
+      }
       return c.json({ error: 'invalid_token', message: tokenErrorMessage(err) }, 401);
     }
   };
 }
 
+/**
+ * I4：401 与 503 是两件事。token 本身不对（过期／签名／aud／iss／根本不是 JWT）是调用方
+ * 的问题，401 让它换票；远端 JWKS 拉不动（DNS、连不上、超时、Keycloak 5xx）是**我们这边**
+ * 的基础设施故障，回 401 等于把停机说成"你的凭据无效"——客户端会去重新登录，而重新登录
+ * 打的还是同一个连不上的 Keycloak，并且监控上看到的是一片鉴权失败而非依赖不可用。
+ *
+ * 判据取白名单而非黑名单：只有 jose 明确的 token 类错误算 401，其余一律 503。这样
+ * 非 JOSEError（fetch 的 TypeError）、JWKSTimeout、JWKS 响应非 200 抛的 JOSEError 基类，
+ * 以及将来 jose 新增的任何基础设施类错误，默认都落在"稍后重试"这一侧——失败方向朝安全。
+ */
+export function classifyAuthError(err: unknown): 401 | 503 {
+  const tokenClass =
+    err instanceof joseErrors.JWTExpired ||
+    err instanceof joseErrors.JWTClaimValidationFailed ||
+    err instanceof joseErrors.JWTInvalid ||
+    err instanceof joseErrors.JWSInvalid ||
+    err instanceof joseErrors.JWSSignatureVerificationFailed ||
+    err instanceof joseErrors.JOSEAlgNotAllowed ||
+    err instanceof joseErrors.JOSENotSupported ||
+    err instanceof joseErrors.JWKSNoMatchingKey ||
+    err instanceof joseErrors.JWKSMultipleMatchingKeys;
+  return tokenClass ? 401 : 503;
+}
+
 function tokenErrorMessage(err: unknown): string {
   if (err instanceof joseErrors.JWTExpired) return 'token 已过期';
   if (err instanceof joseErrors.JWTClaimValidationFailed && err.claim === 'aud') {

src/shared/service-token.ts

@@ -0,0 +1,39 @@
+/** client_credentials 换服务账号 token（指南 8.2）；缓存到 exp-30s；secret 只从环境读 */
+
+import { config } from './config';
+import { PermanentFailure } from './errors';
+
+export type TokenProvider = () => Promise<string>;
+
+export function createKeycloakTokenProvider(): TokenProvider {
+  let cached: { token: string; expiresAt: number } | null = null;
+  return async () => {
+    if (cached && Date.now() < cached.expiresAt) return cached.token;
+    const url = process.env.KEYCLOAK_URL ?? 'https://keycloak.akhr-ai.com';
+    const realm = process.env.KEYCLOAK_REALM ?? 'corp';
+    // I3：这份文件三仓同源，写死 'task-dispatcher' 的兜底到了网关就是错的 client——
+    // 换 token 会拿到别人的服务账号（或直接被拒）。clientId 归 config（每仓一份，非同一性
+    // 文件）来定，本文件就能保持逐字节一致。URL/realm/secret 仍走 process.env：service-token
+    // 的用例在 beforeAll（模块求值之后）才改 KEYCLOAK_URL，搬进 config 会读到求值那一刻的旧值。
+    const clientId = config.keycloak.clientId;
+    const secret = process.env.KEYCLOAK_CLIENT_SECRET;
+    // 漏配的 secret 不会因为重试而长出来 → 永久失败
+    if (!secret) throw new PermanentFailure('缺 KEYCLOAK_CLIENT_SECRET——服务间调用需要 client_credentials，检查 .env');
+    const res = await fetch(`${url}/realms/${realm}/protocol/openid-connect/token`, {
+      method: 'POST',
+      headers: { 'content-type': 'application/x-www-form-urlencoded' },
+      body: new URLSearchParams({ grant_type: 'client_credentials', client_id: clientId, client_secret: secret }),
+    });
+    if (!res.ok) {
+      const text = await res.text();
+      // 400/401＝凭据本身错（client 不存在 / secret 失效），重试不会自愈；其余（5xx、网关）当 Keycloak 暂时不可达
+      if (res.status === 400 || res.status === 401) {
+        throw new PermanentFailure(`换 token 被拒 ${res.status}：${text}——client_credentials 配置有误，检查 KEYCLOAK_CLIENT_ID/KEYCLOAK_CLIENT_SECRET`);
+      }
+      throw new Error(`换 token 失败 ${res.status}：${text}`);
+    }
+    const body = (await res.json()) as { access_token: string; expires_in: number };
+    cached = { token: body.access_token, expiresAt: Date.now() + (body.expires_in - 30) * 1000 };
+    return cached.token;
+  };
+}

src/shared/trace.ts

@@ -0,0 +1,13 @@
+import { randomBytes } from 'node:crypto';
+
+const SHAPE = /^00-([0-9a-f]{32})-[0-9a-f]{16}-[0-9a-f]{2}$/;
+
+export function newTraceparent(): string {
+  return `00-${randomBytes(16).toString('hex')}-${randomBytes(8).toString('hex')}-01`;
+}
+
+export function childOf(parent?: string): string {
+  const m = parent ? SHAPE.exec(parent) : null;
+  if (!m) return newTraceparent();
+  return `00-${m[1]}-${randomBytes(8).toString('hex')}-01`;
+}

test/integration/proxy.test.ts

@@ -13,7 +13,11 @@ vi.hoisted(() => {
 });
 
 let downstream: ReturnType<typeof serve>;
-let captured: { auth?: string; fake?: string; path?: string; method?: string; body?: string } = {};
+let captured: {
+  auth?: string; fake?: string; path?: string; method?: string; body?: string; trace?: string;
+} = {};
+/** 主机保全见证：只有真打到"配置的下游"这台服务的请求才会记进来（C1 回归用） */
+const hits: string[] = [];
 
 describe('gateway proxy', () => {
   let token = '';
@@ -39,12 +43,18 @@ describe('gateway proxy', () => {
         path: c.req.path,
         method: c.req.method,
         body: await c.req.text(),
+        trace: c.req.header('traceparent') ?? '',
       };
       return c.json({ from: 'downstream' });
     });
     // D11：等 listening 回调再往下走，否则首个用例可能打在还没起来的端口上
+    // 在路由之前记账：不管 echo 内部匹不匹得上这条路径，只要连上了这台服务就算"主机没被换掉"
+    const witnessed = (req: Request): Response | Promise<Response> => {
+      hits.push(new URL(req.url).pathname);
+      return echo.fetch(req);
+    };
     downstream = await new Promise<ReturnType<typeof serve>>((resolve) => {
-      const s = serve({ fetch: echo.fetch, port: 18811 }, () => resolve(s));
+      const s = serve({ fetch: witnessed, port: 18811 }, () => resolve(s));
     });
   });
   afterAll(() => {
@@ -92,6 +102,40 @@ describe('gateway proxy', () => {
     expect(body.hint).toBeTruthy();
   });
 
+  // C1 回归：剥前缀后余下的路径若以 // 开头就是协议相对 URL，旧写法 new URL(path, target)
+  // 会让它顶掉 target 的 host——Authorization 与请求体直接外泄到攻击者选定的主机。
+  // 两个变体都选"必然连不上"的主机：位移一旦发生就是 502（端口 1 拒连 / 域名解析失败），
+  // 而 hits 里没有记录；主机保全则相反：不是 502，且配置的下游确实收到了这条路径。
+  it('路径双斜杠不得位移下游主机：Authorization 不外泄（C1）', async () => {
+    for (const injected of ['//127.0.0.1:1/x', '//evil.example.com/x']) {
+      hits.length = 0;
+      const res = await buildApp().request(`/api/task${injected}`, {
+        headers: { authorization: `Bearer ${token}` },
+      });
+      expect(res.status).not.toBe(502);           // 502＝网关真去连了被注入的主机
+      expect(hits).toEqual([injected]);           // 配置的下游收到了＝host 没被换掉
+    }
+  });
+
+  // 网关是外部请求的第一跳：缺 traceparent 补发一条，已有的绝不覆盖。
+  // 两条断言互为反向：去掉注入那行 → 第一条红；把 `!` 去掉（无条件 set）→ 第二条红。
+  it('traceparent：缺则补发合法形状，已有则原样透传不覆盖', async () => {
+    captured = {};
+    const injected = await buildApp().request('/api/task/task/t-1/view', {
+      headers: { authorization: `Bearer ${token}` },
+    });
+    expect(injected.status).toBe(200);
+    expect(captured.trace).toMatch(/^00-[0-9a-f]{32}-[0-9a-f]{16}-01$/);
+
+    captured = {};
+    const mine = '00-0123456789abcdef0123456789abcdef-89abcdef01234567-01';
+    const passed = await buildApp().request('/api/task/task/t-1/view', {
+      headers: { authorization: `Bearer ${token}`, traceparent: mine },
+    });
+    expect(passed.status).toBe(200);
+    expect(captured.trace).toBe(mine);   // 覆盖＝把上游已开好的链从中间截断
+  });
+
   it('下游 gzip 响应：解码后的 body 不得再带 content-encoding（D9）', async () => {
     const res = await buildApp().request('/api/task/gzip', {
       headers: { authorization: `Bearer ${token}` },

test/integration/sse.test.ts

@@ -0,0 +1,324 @@
+import { describe, expect, it, beforeAll, vi } from 'vitest';
+import { Hono } from 'hono';
+import { SseHub, isSubscribable } from '../../src/modules/sse/hub';
+import type { Connection, SseEvent } from '../../src/modules/sse/hub';
+import { sseRoutes } from '../../src/modules/sse/routes';
+import { createHttpTaskAccess } from '../../src/modules/sse/task-access';
+import { createHttpWorkflowAccess } from '../../src/modules/sse/workflow-access';
+import { buildApp } from '../../src/app/app';
+import { makeToken, setupTestJwks } from '../support/token';
+
+function conn(id: string, staffSub: string, topics: string[] = []): { c: Connection; got: SseEvent[] } {
+  const got: SseEvent[] = [];
+  return { c: { id, staffSub, isAdmin: false, topics: new Set(topics), send: (e) => got.push(e) }, got };
+}
+
+/** 一律放行/一律拒绝的假口：两口分开传，才证明得了"这一拒来自哪一口" */
+const ALLOW = { allowed: async (): Promise<boolean> => true };
+const DENY = { allowed: async (): Promise<boolean> => false };
+
+describe('SseHub 路由', () => {
+  it('staff 精确投递、system 广播、task 按订阅集、回放窗口', () => {
+    const hub = new SseHub(8);
+    const a = conn('a', 'sub-a', ['touch.task.t1']);
+    const b = conn('b', 'sub-b');
+    hub.register(a.c);
+    hub.register(b.c);
+    hub.deliver('touch.system', '{}');
+    hub.deliver('touch.staff.sub-a', '{}');
+    hub.deliver('touch.task.t1', '{}');
+    expect(a.got.map((e) => e.topic)).toEqual(['touch.system', 'touch.staff.sub-a', 'touch.task.t1']);
+    expect(b.got.map((e) => e.topic)).toEqual(['touch.system']);
+    // 断线重连回放：c 以 lastEventId=1 上线且订阅 t1 → 补 seq>1 的可见事件。
+    // D11：register 只登记、replay 单独一步——路由要把 hello 插在两者之间。
+    const c = conn('c', 'sub-c', ['touch.task.t1']);
+    hub.register(c.c);
+    hub.replay(c.c, 1);
+    expect(c.got.map((e) => e.topic)).toEqual(['touch.task.t1']);
+  });
+
+  it('subscribe 授权：access 拒 → 不入订阅集', async () => {
+    const hub = new SseHub();
+    const a = conn('a', 'sub-a');
+    hub.register(a.c);
+    const res = await hub.subscribe('a', 'stream.task.t9', DENY, ALLOW);
+    expect(res.ok).toBe(false);
+    expect(a.c.topics.has('stream.task.t9')).toBe(false);
+  });
+
+  it('D3：形状校验排在连接查找之前，失败带判别式 reason 供路由分流', async () => {
+    const hub = new SseHub();
+    // 连接不存在 + 形状也不对 → 必须报形状（否则 HTTP 面永远回不出 422）
+    expect(await hub.subscribe('nope', 'touch.staff.hack', DENY, DENY)).toMatchObject({
+      ok: false,
+      reason: 'bad-shape',
+    });
+    expect(await hub.subscribe('nope', 'touch.task.t1', DENY, DENY)).toMatchObject({
+      ok: false,
+      reason: 'no-conn',
+    });
+    const a = conn('a', 'sub-a');
+    hub.register(a.c);
+    expect(await hub.subscribe('a', 'stream.task.t9', DENY, DENY)).toMatchObject({
+      ok: false,
+      reason: 'denied',
+    });
+  });
+
+  it('N3（原 I6 反转）：touch.workflow.* 放回可订阅集——规则到位了才收得进来', async () => {
+    // I6 当初把它摘出去，是因为**当时没有规则**：形状一过就直接进订阅集，而鉴权分支只覆盖
+    // touch.task./stream.task.，任何登录用户订 touch.workflow.<任意 id> 都能收别人流程的推送。
+    // N3 补上归属口（参与人/审批人，admin 旁路）之后放回来——放形状与补鉴权是同一次改动，
+    // 下面两条用例就是"规则确实生效"的那一半，缺了它这条断言等于把 I6 原样退回去。
+    expect(isSubscribable('touch.workflow.w1')).toBe(true);
+    expect(isSubscribable('touch.task.t1')).toBe(true);
+    expect(isSubscribable('stream.task.t1')).toBe(true);
+    // 放回的只是这一节：多一节的主题、以及别的域照旧不可订阅
+    expect(isSubscribable('touch.workflow.w1.extra')).toBe(false);
+    expect(isSubscribable('touch.workflow.')).toBe(false);
+    expect(isSubscribable('touch.staff.hack')).toBe(false);
+  });
+
+  it('N3：工作流按归属口放行——拒则不入集不投递，准则入集且收得到', async () => {
+    const hub = new SseHub();
+    const a = conn('a', 'sub-a');
+    hub.register(a.c);
+    // 任务口一律放行：这样"被拒"只可能来自**工作流**那一口，而不是碰巧被任务授权挡住
+    const denied = await hub.subscribe('a', 'touch.workflow.w1', ALLOW, DENY);
+    expect(denied).toMatchObject({ ok: false, reason: 'denied' });
+    expect(denied.ok === false && denied.message).toContain('参与人/审批人或 admin');
+    expect(a.c.topics.size).toBe(0);
+    hub.deliver('touch.workflow.w1', '{}');
+    expect(a.got).toEqual([]); // 拒了就一条都不该扇过去
+
+    const asked: string[][] = [];
+    const ok = await hub.subscribe('a', 'touch.workflow.w1', ALLOW, {
+      allowed: async (id: string, sub: string): Promise<boolean> => {
+        asked.push([id, sub]);
+        return true;
+      },
+    });
+    expect(ok).toEqual({ ok: true });
+    expect(asked).toEqual([['w1', 'sub-a']]); // 问的是主题末节＋**连接的** staffSub，不是调用方的
+    hub.deliver('touch.workflow.w1', '{"n":1}');
+    expect(a.got.map((e) => e.topic)).toEqual(['touch.workflow.w1']);
+  });
+
+  it('N3：admin 订工作流不问归属口——admin 全可见，白跑一次跨服务往返没意义', async () => {
+    const hub = new SseHub();
+    const a = conn('adm', 'sub-adm');
+    hub.register({ ...a.c, isAdmin: true }); // topics/send 与 a.c 共用，断言照旧看 a
+    let calls = 0;
+    const counting = {
+      allowed: async (): Promise<boolean> => {
+        calls += 1;
+        return false; // 真被问到就会拒——订上了即证明没问
+      },
+    };
+    expect(await hub.subscribe('adm', 'touch.workflow.w9', counting, counting)).toEqual({ ok: true });
+    expect(calls).toBe(0);
+    expect(a.c.topics.has('touch.workflow.w9')).toBe(true);
+    hub.deliver('touch.workflow.w9', '{}');
+    expect(a.got.map((e) => e.topic)).toEqual(['touch.workflow.w9']);
+  });
+
+  it('D12：shutdown 逐条调 close 把手并摘除连接', () => {
+    const hub = new SseHub();
+    let closed = 0;
+    const a = conn('a', 'sub-a');
+    hub.register({ ...a.c, close: () => (closed += 1) });
+    hub.shutdown();
+    expect(closed).toBe(1);
+    hub.deliver('touch.system', '{}');
+    expect(a.got).toEqual([]); // 已摘除，不再收扇出
+  });
+});
+
+describe('SSE HTTP 面', () => {
+  beforeAll(async () => {
+    await setupTestJwks();
+  });
+
+  it('subscribe 主题形状校验 422；无 token 401', async () => {
+    const app = buildApp();
+    expect((await app.request('/api/stream/subscribe', { method: 'POST' })).status).toBe(401);
+    const token = await makeToken({ roles: ['user'], aud: 'api-gateway' });
+    const bad = await app.request('/api/stream/subscribe', {
+      method: 'POST',
+      headers: { authorization: `Bearer ${token}`, 'content-type': 'application/json' },
+      body: JSON.stringify({ connectionId: 'nope', add: ['touch.staff.hack'] }),
+    });
+    expect(bad.status).toBe(422);
+  });
+
+  it('F1：形状预检先于变更——整单 422 时一条订阅都不留', async () => {
+    const hub = new SseHub();
+    const a = conn('live-1', 'test-sub-1'); // token 的默认 sub，过得了 F2 的归属校验
+    hub.register(a.c);
+    // 直接挂 sseRoutes 并注入"一律放行"的授权口：走 buildApp 的话内建 access 会在
+    // await token() 处就抛（测试环境没有 KEYCLOAK_CLIENT_SECRET），t1 到不了订阅集，
+    // 这条用例也就证明不了"预检之前不会发生变更"这件事。
+    let accessCalls = 0;
+    const access = {
+      allowed: async (): Promise<boolean> => {
+        accessCalls += 1;
+        return true;
+      },
+    };
+    const app = new Hono().route('/api/stream', sseRoutes(hub, access, access));
+    const token = await makeToken({ roles: ['user'], aud: 'api-gateway' });
+    const res = await app.request('/api/stream/subscribe', {
+      method: 'POST',
+      headers: { authorization: `Bearer ${token}`, 'content-type': 'application/json' },
+      body: JSON.stringify({ connectionId: 'live-1', add: ['touch.task.t1', 'bogus'] }),
+    });
+    expect(res.status).toBe(422);
+    // 关键：合法的 t1 排在非法项**前面**，边订边查的写法会把它真订上
+    expect(a.c.topics.has('touch.task.t1')).toBe(false);
+    expect(a.c.topics.size).toBe(0);
+    expect(accessCalls).toBe(0); // 且一次授权往返都不该白跑
+  });
+
+  it('I7：add/remove 不是数组 → 422（不是 500）', async () => {
+    // `as string[]` 只是断言。字符串上没有 .find，此前直接 TypeError 冒到 500——
+    // 请求写错却报成服务端故障，而 500 在网关这一层还会被客户端/重试层当成可重试。
+    const hub = new SseHub();
+    const a = conn('live-7', 'test-sub-1');
+    hub.register(a.c);
+    const app = new Hono().route('/api/stream', sseRoutes(hub, ALLOW, ALLOW));
+    const token = await makeToken({ roles: ['user'], aud: 'api-gateway' });
+    const post = async (payload: unknown): Promise<Response> =>
+      app.request('/api/stream/subscribe', {
+        method: 'POST',
+        headers: { authorization: `Bearer ${token}`, 'content-type': 'application/json' },
+        body: JSON.stringify(payload),
+      });
+
+    const badAdd = await post({ connectionId: 'live-7', add: 'touch.task.t1' });
+    expect(badAdd.status).toBe(422);
+    expect(await badAdd.json()).toMatchObject({ error: 'validation_failed' });
+    const badRemove = await post({ connectionId: 'live-7', remove: { '0': 'touch.task.t1' } });
+    expect(badRemove.status).toBe(422);
+    expect(a.c.topics.size).toBe(0);                       // 422 零副作用
+    // 缺省（undefined）仍然合法：只订不退、只退不订都是正常用法
+    expect((await post({ connectionId: 'live-7', add: ['touch.task.t1'] })).status).toBe(200);
+    expect(a.c.topics.has('touch.task.t1')).toBe(true);
+  });
+
+  it('F2：连接归属校验——他人 connectionId 一律 403（add 与 remove 同守）', async () => {
+    const hub = new SseHub();
+    const app = buildApp(hub);
+    const owner = conn('own-1', 'test-sub-1');
+    owner.c.topics.add('touch.task.t1');
+    hub.register(owner.c);
+    const post = async (tok: string, payload: unknown): Promise<Response> =>
+      app.request('/api/stream/subscribe', {
+        method: 'POST',
+        headers: { authorization: `Bearer ${tok}`, 'content-type': 'application/json' },
+        body: JSON.stringify(payload),
+      });
+
+    // 闯入者：合法 token，但连接不是他的 —— remove 此前完全不过任何校验
+    const intruder = await makeToken({ sub: 'sub-intruder', roles: ['user'], aud: 'api-gateway' });
+    const denied = await post(intruder, { connectionId: 'own-1', remove: ['touch.task.t1'] });
+    expect(denied.status).toBe(403);
+    expect(await denied.json()).toMatchObject({ error: 'forbidden' });
+    expect(owner.c.topics.has('touch.task.t1')).toBe(true); // 没被别人退订掉
+
+    // 主人自己来则照常放行——校验不是一刀切拒绝
+    const self = await makeToken({ roles: ['user'], aud: 'api-gateway' }); // sub=test-sub-1
+    const ok = await post(self, { connectionId: 'own-1', remove: ['touch.task.t1'] });
+    expect(ok.status).toBe(200);
+    expect(owner.c.topics.has('touch.task.t1')).toBe(false);
+  });
+
+  it('F3：task-access 非 200 → false（失败关闭）且 id 都转义进 URL', async () => {
+    const warn = vi.spyOn(console, 'warn').mockImplementation(() => undefined);
+    const calls: string[] = [];
+    try {
+      vi.stubGlobal('fetch', (url: string | URL) => {
+        calls.push(String(url));
+        return Promise.resolve(new Response('down', { status: 503 }));
+      });
+      const access = createHttpTaskAccess(async () => 'stub-token');
+      expect(await access.allowed('t 1', 'sub/a')).toBe(false);
+      expect(calls[0]).toContain('/task/t%201/access?staff=sub%2Fa');
+      expect(warn).toHaveBeenCalled(); // 不出声的"人人被拒"无从诊断
+
+      vi.stubGlobal('fetch', () =>
+        Promise.resolve(
+          new Response(JSON.stringify({ allowed: true }), {
+            status: 200,
+            headers: { 'content-type': 'application/json' },
+          }),
+        ),
+      );
+      expect(await access.allowed('t1', 's1')).toBe(true);
+    } finally {
+      vi.unstubAllGlobals();
+      warn.mockRestore();
+    }
+  });
+
+  it('workflow-access 与 task-access 同形：非 200 → false（失败关闭），打的是 /workflow/:id/access', async () => {
+    const warn = vi.spyOn(console, 'warn').mockImplementation(() => undefined);
+    const calls: string[] = [];
+    try {
+      vi.stubGlobal('fetch', (url: string | URL) => {
+        calls.push(String(url));
+        return Promise.resolve(new Response('down', { status: 503 }));
+      });
+      const wf = createHttpWorkflowAccess(async () => 'stub-token');
+      expect(await wf.allowed('w 1', 'sub/a')).toBe(false);
+      // 路径与任务口分开：任务侧是两条端点，判据也不同源
+      expect(calls[0]).toContain('/workflow/w%201/access?staff=sub%2Fa');
+      expect(warn).toHaveBeenCalled(); // 不出声的"人人被拒"无从诊断
+
+      vi.stubGlobal('fetch', () =>
+        Promise.resolve(
+          new Response(JSON.stringify({ allowed: true }), {
+            status: 200,
+            headers: { 'content-type': 'application/json' },
+          }),
+        ),
+      );
+      expect(await wf.allowed('w1', 's1')).toBe(true);
+    } finally {
+      vi.unstubAllGlobals();
+      warn.mockRestore();
+    }
+  });
+
+  it('D11/D12：hello 先于回放；hub.shutdown() 能把在途长连收尾', async () => {
+    const hub = new SseHub(8);
+    const app = buildApp(hub);
+    hub.deliver('touch.system', '{"n":1}'); // seq 1
+    hub.deliver('touch.staff.test-sub-1', '{"n":2}'); // seq 2
+    const token = await makeToken({ roles: ['user'], aud: 'api-gateway' });
+    const res = await app.request(`/api/stream?access_token=${token}`, {
+      headers: { 'last-event-id': '1' },
+    });
+    expect(res.status).toBe(200);
+    const body = res.body;
+    if (!body) throw new Error('SSE 响应缺流体');
+    const reader = body.getReader();
+    const decoder = new TextDecoder();
+    let text = '';
+    while (!text.includes('event: touch.staff.test-sub-1')) {
+      const { value, done } = await reader.read();
+      if (done) break;
+      text += decoder.decode(value, { stream: true });
+    }
+    expect(text).toContain('event: hello');
+    expect(text.indexOf('event: hello')).toBeLessThan(text.indexOf('event: touch.staff.test-sub-1'));
+    expect(text).not.toContain('event: touch.system'); // seq 1 ≤ lastEventId，不补
+
+    // D12：回调里挂住的 promise 必须能被 close 把手 settle——settle 了 streamSSE 才会
+    // close 流，客户端这边才读得到 done。旧写法 `await new Promise(() => {})` 永不 settle：
+    // 下面这次 read 会一直吊着（用例超时），线上则是 SIGTERM 关不掉服务。
+    hub.shutdown();
+    let ended = false;
+    for (let i = 0; i < 4 && !ended; i += 1) ended = (await reader.read()).done;
+    expect(ended).toBe(true);
+  });
+});

test/unit/keycloak-auth.test.ts

@@ -1,7 +1,8 @@
 import { Hono } from 'hono';
+import { errors as joseErrors } from 'jose';
 import { beforeAll, describe, expect, it } from 'vitest';
 import { makeToken, setupTestJwks } from '../support/token';
-import { requireAuth, requireRealmRole, requireService } from '../../src/shared/keycloak-auth';
+import { classifyAuthError, requireAuth, requireRealmRole, requireService } from '../../src/shared/keycloak-auth';
 
 function appWith(...mw: Parameters<Hono['use']>[1][]): Hono {
   const app = new Hono();
@@ -66,6 +67,22 @@ describe('keycloak-auth', () => {
     expect(res.status).toBe(200);
   });
 
+  it('I4：token 类失败 → 401；JWKS/网络类失败 → 503（不把停机说成凭据无效）', () => {
+    // token 类：换张票才有意义
+    expect(classifyAuthError(new joseErrors.JWTExpired('exp', {}))).toBe(401);
+    expect(classifyAuthError(new joseErrors.JWTClaimValidationFailed('aud', {}, 'aud'))).toBe(401);
+    expect(classifyAuthError(new joseErrors.JWTInvalid('不是 JWT'))).toBe(401);
+    expect(classifyAuthError(new joseErrors.JWSSignatureVerificationFailed())).toBe(401);
+    expect(classifyAuthError(new joseErrors.JWKSNoMatchingKey())).toBe(401);
+    // 基础设施类：只有重试才有意义。回 401 会把客户端赶去重新登录，而登录打的是同一个
+    // 连不上的 Keycloak——排障时看到的也只是一片"凭据无效"，而非依赖不可用。
+    expect(classifyAuthError(new TypeError('fetch failed'))).toBe(503);
+    expect(classifyAuthError(new joseErrors.JWKSTimeout())).toBe(503);
+    expect(classifyAuthError(new joseErrors.JWKSInvalid('JWKS 结构不对'))).toBe(503);
+    expect(classifyAuthError(new joseErrors.JOSEError('Expected 200 OK from the JSON Web Key Set HTTP response'))).toBe(503);
+    expect(classifyAuthError('连字符串都不是')).toBe(503);
+  });
+
   it('requireService：service 角色＋azp 白名单', async () => {
     const svc = await makeToken({
       roles: ['service'], azp: 'task-dispatcher', username: 'service-account-task-dispatcher',