feat: SSE 放行 touch.workflow——参与人+admin 可见性（经任务服务归属端口）

I6 把 touch.workflow. 从可订阅集摘出去，理由是**当时没有规则**：形状一过就直接进订阅集，
而鉴权分支只覆盖 touch.task./stream.task.，任何登录用户订 touch.workflow.<任意 id> 都能收
别人流程的推送。N3 的引擎每次推进都发 touch.workflow.<instanceId>，规则也定了：参与人
（节点 ownerSub）／审批人（approverSub）／admin。规则到位，故放回来。

- workflow-access.ts：与 task-access 同形的 HTTP 口——服务账号 token、3s 超时、非 200
  一律 false 并告警（失败关闭）。打的是任务侧新加的 /workflow/:id/access。
- hub：SUBSCRIBABLE 收回 touch.workflow.，subscribe 补上对应鉴权分支——放形状与补鉴权
  必须同一次改动，只放形状就是把 I6 原样退回去。admin 旁路（全可见，也免掉白跑的往返）。
- 两口分开而不是复用一口：任务问的是单张单的归属，工作流问的是实例**定义里**的节点归属。
- app：两口共用同一个 token provider，缓存只此一份。

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

src/app/app.ts

@@ -7,6 +7,7 @@ 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';
 
 /**
  * hub 由外面传：server.ts 要拿同一个实例去 attachNats／shutdown。缺省自建一个，
@@ -23,9 +24,12 @@ export function buildApp(hub: SseHub = new SseHub()): Hono {
   app.route('/health', healthRoute()); // 公开：不过 requireAuth
   // 惰性构造：createKeycloakTokenProvider 只在闭包被调用时读 KEYCLOAK_CLIENT_SECRET，
   // 建 app 本身不要求环境齐备，否则任何 buildApp() 的测试都会当场炸。
-  const access = createHttpTaskAccess(createKeycloakTokenProvider());
+  // 两口共用同一个 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));
+  app.route('/api/stream', sseRoutes(hub, access, workflowAccess));
   app.route('/api/upload', uploadRoutes());
   app.route('/api', proxyRoutes());
   return app;

src/modules/sse/hub.ts

@@ -15,6 +15,11 @@ export interface Connection {
   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 整单驳回，
@@ -26,12 +31,16 @@ export type SubscribeResult =
   | { ok: false; reason: 'bad-shape' | 'no-conn' | 'denied'; message: string };
 
 /**
- * I6：可订阅集合＝**有授权规则**的集合，默认拒。此前这里收着 touch.workflow.，而
+ * 可订阅集合＝**有授权规则**的集合，默认拒。
- * subscribe 的鉴权分支只覆盖 touch.task./stream.task.——形状过了就直接进订阅集，
+ *
- * 于是任何登录用户订 touch.workflow.<任意 id> 都能收到别人流程的推送。工作流的
+ * touch.workflow. 在 I6 被摘出去过：那时形状一过就直接进订阅集，而 subscribe 的鉴权分支只
- * 可见性规则（发起人/参与节点/admin）要到 N3 才有，规则到位之前不放进来。
+ * 覆盖 touch.task./stream.task.，于是任何登录用户订 touch.workflow.<任意 id> 都能收别人流程
+ * 的推送——摘掉它是因为**当时没有规则**，不是因为工作流不该推。
+ * N3 规则到位，故放回来：可见性＝该实例定义里的参与人（节点 ownerSub）或审批人
+ * （approverSub），经 workflowAccess 口向任务服务问；admin 旁路。判据落在下面 subscribe 的
+ * 鉴权分支里——这两处必须同一次改动，只放形状白名单就是把 I6 原样退回去。
  */
-const SUBSCRIBABLE = /^(touch\.task\.|stream\.task\.)[^.]+$/;
+const SUBSCRIBABLE = /^(touch\.task\.|touch\.workflow\.|stream\.task\.)[^.]+$/;
 
 /** 主题形状唯一判据。导出是为了让路由能在**动任何状态之前**预检，见 badShapeMessage */
 export function isSubscribable(topic: string): boolean {
@@ -40,7 +49,7 @@ export function isSubscribable(topic: string): boolean {
 
 /** 形状错的文案也只有这一份——路由预检与 hub.subscribe 报的必须是同一句 */
 export function badShapeMessage(topic: string): string {
-  return `主题 ${topic} 不可订阅——只接受 touch.task.* 与 stream.task.*`;
+  return `主题 ${topic} 不可订阅——只接受 touch.task.*、touch.workflow.* 与 stream.task.*`;
 }
 
 export class SseHub {
@@ -99,17 +108,33 @@ export class SseHub {
     if (conn) for (const t of conn.topics) this.maybeDropStreamSub(t);
   }
 
-  async subscribe(connId: string, topic: string, access: TaskAccessPort): Promise<SubscribeResult> {
+  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 taskId = topic.split('.').pop() ?? '';
+    const subjectId = topic.split('.').pop() ?? '';
     if (!conn.isAdmin && (topic.startsWith('stream.task.') || topic.startsWith('touch.task.'))) {
-      if (!(await access.allowed(taskId, conn.staffSub))) {
+      if (!(await access.allowed(subjectId, conn.staffSub))) {
-        return { ok: false, reason: 'denied', message: `无权订阅任务 ${taskId}——需要负责人/审批人/admin` };
+        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);

src/modules/sse/routes.ts

@@ -3,9 +3,9 @@ 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 } from './hub';
+import type { Connection, SseHub, TaskAccessPort, WorkflowAccessPort } from './hub';
 
-export function sseRoutes(hub: SseHub, access: TaskAccessPort): Hono {
+export function sseRoutes(hub: SseHub, access: TaskAccessPort, workflowAccess: WorkflowAccessPort): Hono {
   const r = new Hono();
 
   r.get('/', async (c) => {
@@ -97,7 +97,7 @@ export function sseRoutes(hub: SseHub, access: TaskAccessPort): Hono {
 
     const results: Record<string, string> = {};
     for (const topic of add) {
-      const res = await hub.subscribe(connectionId, topic, access);
+      const res = await hub.subscribe(connectionId, topic, access, workflowAccess);
       // 形状已由上面的预检挡掉，这里只是不变量兜底：真漏进来也绝不降级成 200。
       // 连接不存在／被拒是运行时状态，逐条落 results 回 200（客户端据此决定重连还是申请权限）。
       if (!res.ok && res.reason === 'bad-shape') {

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;
+    },
+  };
+}

test/integration/sse.test.ts

@@ -4,6 +4,7 @@ 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';
 
@@ -12,6 +13,10 @@ function conn(id: string, staffSub: string, topics: string[] = []): { c: Connect
   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);
@@ -36,48 +41,85 @@ describe('SseHub 路由', () => {
     const hub = new SseHub();
     const a = conn('a', 'sub-a');
     hub.register(a.c);
-    const denyAccess = { allowed: async () => false };
+    const res = await hub.subscribe('a', 'stream.task.t9', DENY, ALLOW);
-    const res = await hub.subscribe('a', 'stream.task.t9', denyAccess);
     expect(res.ok).toBe(false);
     expect(a.c.topics.has('stream.task.t9')).toBe(false);
   });
 
   it('D3：形状校验排在连接查找之前，失败带判别式 reason 供路由分流', async () => {
     const hub = new SseHub();
-    const denyAccess = { allowed: async () => false };
     // 连接不存在 + 形状也不对 → 必须报形状（否则 HTTP 面永远回不出 422）
-    expect(await hub.subscribe('nope', 'touch.staff.hack', denyAccess)).toMatchObject({
+    expect(await hub.subscribe('nope', 'touch.staff.hack', DENY, DENY)).toMatchObject({
       ok: false,
       reason: 'bad-shape',
     });
-    expect(await hub.subscribe('nope', 'touch.task.t1', denyAccess)).toMatchObject({
+    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', denyAccess)).toMatchObject({
+    expect(await hub.subscribe('a', 'stream.task.t9', DENY, DENY)).toMatchObject({
       ok: false,
       reason: 'denied',
     });
   });
 
-  it('I6：touch.workflow.* 默认拒——没有授权规则的主题不进可订阅集', async () => {
+  it('N3（原 I6 反转）：touch.workflow.* 放回可订阅集——规则到位了才收得进来', async () => {
-    // 形状白名单曾经收着 touch.workflow.，而 subscribe 的鉴权分支只覆盖 touch.task./stream.task.：
+    // I6 当初把它摘出去，是因为**当时没有规则**：形状一过就直接进订阅集，而鉴权分支只覆盖
-    // 形状一过就直接进订阅集，任何登录用户订 touch.workflow.<任意 id> 都能收别人流程的推送。
+    // touch.task./stream.task.，任何登录用户订 touch.workflow.<任意 id> 都能收别人流程的推送。
-    // N3 给工作流补上可见性规则之后再放回来。
+    // N3 补上归属口（参与人/审批人，admin 旁路）之后放回来——放形状与补鉴权是同一次改动，
-    expect(isSubscribable('touch.workflow.w1')).toBe(false);
+    // 下面两条用例就是"规则确实生效"的那一半，缺了它这条断言等于把 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);
-    // 放行一切的 access 口：证明拒绝来自形状白名单本身，而不是碰巧被授权挡住了
+    // 任务口一律放行：这样"被拒"只可能来自**工作流**那一口，而不是碰巧被任务授权挡住
-    const res = await hub.subscribe('a', 'touch.workflow.w1', { allowed: async () => true });
+    const denied = await hub.subscribe('a', 'touch.workflow.w1', ALLOW, DENY);
-    expect(res).toMatchObject({ ok: false, reason: 'bad-shape' });
+    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([]);
+    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 把手并摘除连接', () => {
@@ -123,7 +165,7 @@ describe('SSE HTTP 面', () => {
         return true;
       },
     };
-    const app = new Hono().route('/api/stream', sseRoutes(hub, access));
+    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',
@@ -143,7 +185,7 @@ describe('SSE HTTP 面', () => {
     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, { allowed: async () => true }));
+    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', {
@@ -218,6 +260,35 @@ describe('SSE HTTP 面', () => {
     }
   });
 
+  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);