feat(tasks,server,web): bridge outbox events to the SSE bus (U3)

OutboxRelay gains a `mirror` option: listed types are claimed even without a
task route and, after the claim transaction commits, published in order to
the realtime bus (publish failures are logged, delivery state unaffected).
The worker mirrors feed.*, decision.*, download.*, organization.* and
subscription.* (feed.due excluded) with their business-id payloads, so the
home page, inbox, task stream and feed tables finally receive them. Feed
tables refresh on feed.fetched / feed.health_changed; the task stream also
listens to feed and decision events.

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

apps/server/src/worker/index.ts

@@ -6,6 +6,16 @@ import { type EventBusPort, sseConnectionCount } from "../http/sse.ts";
 import { type HeartbeatWriter, startHeartbeatWriter } from "../modules/system/heartbeat.ts";
 import type { ServerContext, ServerModule } from "../modules/types.ts";
 
+/**
+ * outbox → SSE 桥:这些前缀的 outbox 事件在中继投递成功后镜像到实时总线(payload 原样,带业务 id),
+ * 首页、收件箱、任务、来源与下载页据此失效刷新。feed.due 是调度器内部节拍,不推给浏览器。
+ */
+const SSE_MIRROR_PREFIXES = ["feed.", "decision.", "download.", "organization.", "subscription."] as const;
+const SSE_MIRROR_EXCLUDED = new Set<string>([EventType.feedDue]);
+export const SSE_MIRRORED_OUTBOX_TYPES: readonly string[] = Object.values(EventType).filter(
+  (t) => SSE_MIRROR_PREFIXES.some((p) => t.startsWith(p)) && !SSE_MIRROR_EXCLUDED.has(t),
+);
+
 /** agent 组任务:整段会话占一槽数分钟,与文件复制等 io 任务分池 */
 export const AGENT_TASKS = ["runAgent", "resolveDecision"] as const;
 
@@ -146,6 +156,10 @@ export async function startWorker({ logger, heartbeatMs, bus, runtime, lagProbe
       engine: ctx.engine,
       routes: collectOutboxRoutes(modules, ctx),
       logSink: tasksLogSink(logger),
+      mirror: {
+        types: SSE_MIRRORED_OUTBOX_TYPES,
+        publish: (ev) => ctx.events.publish(ev.type, ev.payload),
+      },
       ...(runtime.outboxIntervalMs ? { intervalMs: runtime.outboxIntervalMs } : {}),
     });
     runtime.setRelay?.(relay);

apps/server/test/bootstrap.test.ts

@@ -138,4 +138,40 @@ describe.skipIf(!hasPg)("bootstrap + worker(真实 PG)", () => {
     const rows = await sql<{ status: string }>`select status from outbox_event`.execute(db);
     expect(rows.rows).toEqual([{ status: "dispatched" }]);
   }, 40_000);
+
+  it("outbox → SSE 桥:feed / decision / download / organization 事件提交后镜像到总线,feed.due 不推", async () => {
+    const got: { type: string; payload: unknown }[] = [];
+    const off = booted.ctx.events.subscribe((e) => {
+      if (e.type !== EventType.heartbeat) got.push({ type: e.type, payload: e.payload });
+    });
+    const decisionId = "33333333-3333-4333-8333-333333333333";
+    const feedId = "44444444-4444-4444-8444-444444444444";
+    await db.transaction().execute(async (trx) => {
+      await booted.ctx.outbox.emit(trx, EventType.feedDue, { feedId, reason: "manual" });
+      await booted.ctx.outbox.emit(trx, EventType.feedFetched, {
+        feedId,
+        newItems: 2,
+        updatedItems: 0,
+        notModified: false,
+      });
+      await booted.ctx.outbox.emit(trx, EventType.decisionCreated, {
+        decisionId,
+        kind: "identity_unresolved",
+        severity: "info",
+        title: "待识别",
+      });
+    });
+    const deadline = Date.now() + 20_000;
+    while (got.length < 2 && Date.now() < deadline) await new Promise((r) => setTimeout(r, 100));
+    off();
+    // 同一事务内的事件 at 相同,相对顺序不保证
+    got.sort((a, b) => a.type.localeCompare(b.type));
+    expect(got).toEqual([
+      { type: EventType.decisionCreated, payload: expect.objectContaining({ decisionId }) },
+      { type: EventType.feedFetched, payload: expect.objectContaining({ feedId, newItems: 2 }) },
+    ]);
+    // feed.due 没有路由也不镜像,保持待认领
+    const due = await sql<{ status: string }>`select status from outbox_event where type = 'feed.due'`.execute(db);
+    expect(due.rows).toEqual([{ status: "pending" }]);
+  }, 30_000);
 });

apps/web/app/components/tasks/task-stream.tsx

@@ -29,12 +29,16 @@ import {
 
 export type StreamTab = "all" | "active" | "failed";
 
-/** 引擎本身不发事件;业务事件足以代表任务在动,2 秒节流刷新 */
+/** 引擎本身不发事件;业务事件(outbox → SSE 桥)足以代表任务在动,2 秒节流刷新 */
 const TASK_EVENTS = [
   "task.created",
   "task.status_changed",
   "download.status_changed",
+  "download.threshold",
   "organization.changed",
+  "feed.fetched",
+  "decision.created",
+  "decision.updated",
   "agent.run_status",
   "schedule.changed",
 ];

apps/web/app/components/workbench/feed-table.tsx

@@ -1,8 +1,9 @@
-import type { FeedHealth, FeedSource } from "@umt/contracts";
+import { useQueryClient } from "@tanstack/react-query";
+import { EventType, type FeedHealth, type FeedSource } from "@umt/contracts";
 import { RotateCcw, Rss, Zap } from "lucide-react";
-import { useState } from "react";
+import { useEffect, useRef, useState } from "react";
 import { toast } from "sonner";
-import { feedGapHint, useFeedAction, useFeeds, useFetchFeedNow } from "~/api/feeds.ts";
+import { feedGapHint, feedKeys, useFeedAction, useFeeds, useFetchFeedNow } from "~/api/feeds.ts";
 import { useSiteAccounts, useSites } from "~/api/sites.ts";
 import { type DataColumn, DataTable } from "~/components/console/data-table.tsx";
 import { DetailSection, DetailSheet } from "~/components/console/detail-sheet.tsx";
@@ -12,6 +13,7 @@ import { StatusBadge, type StatusTone } from "~/components/console/status-badge.
 import { Button } from "~/components/ui/button.tsx";
 import { Switch } from "~/components/ui/switch.tsx";
 import { errorMessage } from "~/lib/api.ts";
+import { useSse } from "~/lib/sse.ts";
 import { formatDateTime } from "~/lib/utils.ts";
 import { FeedFormDialog, PURPOSE_LABEL } from "./feed-form.tsx";
 import { relTime } from "./release-fields.ts";
@@ -78,6 +80,25 @@ function FeedActions({ f }: { f: FeedSource }) {
 /**
  * Feed 列表 + 详情抽屉 + 新建/编辑对话框。来源健康页列全部,站点详情按 siteId 过滤。
  */
+/** 抓取完成 / 健康变化(outbox → SSE 桥)到达时节流刷新 Feed 列表 */
+function useLiveFeeds() {
+  const qc = useQueryClient();
+  const timer = useRef<ReturnType<typeof setTimeout> | null>(null);
+  useSse([EventType.feedFetched, EventType.feedHealthChanged], () => {
+    if (timer.current) return;
+    timer.current = setTimeout(() => {
+      timer.current = null;
+      void qc.invalidateQueries({ queryKey: feedKeys.all });
+    }, 800);
+  });
+  useEffect(
+    () => () => {
+      if (timer.current) clearTimeout(timer.current);
+    },
+    [],
+  );
+}
+
 export function FeedTable({
   siteId,
   health,
@@ -93,6 +114,7 @@ export function FeedTable({
   filtered?: boolean;
 }) {
   const feeds = useFeeds({ health, siteId });
+  useLiveFeeds();
   const sites = useSites();
   const accounts = useSiteAccounts();
   const [editingFeed, setEditingFeed] = useState<FeedSource | null>(null);

packages/tasks/src/outbox-relay.ts

@@ -34,6 +34,12 @@ export interface OutboxRelayOptions {
   /** 失败达到该次数后置 status='dead'(默认 10);其间按 2^attempts 秒退避,上限 5 分钟 */
   maxAttempts?: number;
   logSink?: LogSink;
+  /**
+   * 镜像到实时总线(SSE):列出的 type 在投递成功、认领事务提交后逐条交给 publish(按 at, id 顺序)。
+   * 只在 mirror 里、没有路由的 type 同样会被认领并直接标记已投递。publish 失败只记日志,不影响投递状态
+   * (实时推送是尽力而为,前端断线或错过时靠重新拉快照)。
+   */
+  mirror?: { types: readonly string[]; publish: (event: OutboxEvent) => unknown };
 }
 
 /**
@@ -60,10 +66,12 @@ export class OutboxRelay {
 
   /** 处理一批;返回本批已投递与失败数 */
   async pollOnce(): Promise<{ published: number; failed: number }> {
-    const types = Object.keys(this.opts.routes);
+    const mirrorTypes = new Set(this.opts.mirror?.types ?? []);
+    const types = [...new Set([...Object.keys(this.opts.routes), ...mirrorTypes])];
     if (types.length === 0) return { published: 0, failed: 0 };
     let published = 0;
     let failed = 0;
+    const toMirror: OutboxEvent[] = [];
     await this.opts.db.transaction().execute(async (trx) => {
       const rows = await sql<OutboxEvent & { idempotency_key: string | null; attempts: number }>`
         SELECT id, type, payload, at, idempotency_key, attempts FROM outbox_event
@@ -73,7 +81,8 @@ export class OutboxRelay {
         FOR UPDATE SKIP LOCKED`.execute(trx);
       for (const ev of rows.rows) {
         try {
-          const route = this.opts.routes[ev.type]!(ev);
+          const routeFn = this.opts.routes[ev.type];
+          const route = routeFn ? routeFn(ev) : null;
           if (route) {
             const payloadTrace = (ev.payload as { traceId?: unknown } | null)?.traceId;
             await this.opts.engine.enqueue(route.task, route.input, {
@@ -85,6 +94,7 @@ export class OutboxRelay {
           await sql`UPDATE outbox_event SET status = 'dispatched', published_at = now(), attempts = attempts + 1,
             last_error = NULL WHERE id = ${ev.id}`.execute(trx);
           published++;
+          if (mirrorTypes.has(ev.type)) toMirror.push({ id: ev.id, type: ev.type, payload: ev.payload, at: ev.at });
         } catch (err) {
           failed++;
           // 退避后重试;超过上限置 dead(需人工处理)
@@ -99,6 +109,14 @@ export class OutboxRelay {
         }
       }
     });
+    // 提交后才推送:回滚的事件不会先被浏览器看到
+    for (const ev of toMirror) {
+      try {
+        await this.opts.mirror!.publish(ev);
+      } catch (err) {
+        this.log.error("outbox mirror publish failed", { eventId: ev.id, type: ev.type, err });
+      }
+    }
     return { published, failed };
   }
 

packages/tasks/test/outbox.test.ts

@@ -127,4 +127,40 @@ describe.skipIf(!hasPg)("OutboxRelay", () => {
     }>`SELECT status FROM outbox_event WHERE type = 'echo.requested' AND published_at IS NULL`.execute(db);
     expect(dead.rows).toEqual([{ status: "dead" }]);
   });
+
+  it("mirror:投递成功后按顺序镜像到实时总线;只镜像不路由的 type 也认领,失败的不镜像", async () => {
+    await sql`DELETE FROM outbox_event`.execute(db);
+    await sql`INSERT INTO outbox_event (type, payload, at) VALUES
+      ('mirror.only', '{"recordId":"r1"}'::jsonb, now() - interval '3 second'),
+      ('echo.requested', '{"n":1001}'::jsonb, now() - interval '2 second'),
+      ('echo.requested', '{"n":"bad"}'::jsonb, now() - interval '1 second'),
+      ('not.mirrored', '{}'::jsonb, now())`.execute(db);
+    const seen: { type: string; payload: unknown }[] = [];
+    const relay = new OutboxRelay({
+      db,
+      engine,
+      routes: { "echo.requested": (ev) => ({ task: "echo", input: ev.payload }) },
+      logSink: () => {},
+      mirror: {
+        types: ["mirror.only", "echo.requested"],
+        publish: (ev) => {
+          seen.push({ type: ev.type, payload: ev.payload });
+          if (ev.type === "mirror.only") throw new Error("bus down"); // 推送失败不影响投递
+        },
+      },
+    });
+    expect(await relay.pollOnce()).toEqual({ published: 2, failed: 1 });
+    expect(seen).toEqual([
+      { type: "mirror.only", payload: { recordId: "r1" } },
+      { type: "echo.requested", payload: { n: 1001 } },
+    ]);
+    const status = await sql<{ type: string; status: string }>`
+      SELECT type, status FROM outbox_event ORDER BY at`.execute(db);
+    expect(status.rows.map((r) => `${r.type}:${r.status}`)).toEqual([
+      "mirror.only:dispatched",
+      "echo.requested:dispatched",
+      "echo.requested:pending",
+      "not.mirrored:pending",
+    ]);
+  });
 });