feat(server): ServerModule skeleton with shared ServerContext, worker runs engine + outbox relay (B6)

modules/{auth-settings,sources-identity,acquisition-organize,agent} placeholders listed in
SERVER_MODULES; ServerModule registers routes (under /api/v1, authenticated), tasks,
outbox routes, capabilities and background loops. bootstrap() builds ServerContext (db,
config, logger, events, engine, outbox writer, secrets placeholder, settings reader,
capability registry), fills unregistered tasks with stubs and rejects duplicate task or
outbox ownership. worker starts the OpenWorkflow worker, OutboxRelay and module start().
--role migrate also migrates the engine schema; api/worker connect without migrating.
tasks: TaskRegistry.has, migrateEngineSchema.

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

apps/server/src/context.ts

@@ -0,0 +1,167 @@
+import { type EventType, parseEventPayload } from "@umt/contracts";
+import { type DB, type Kysely, sql } from "@umt/db";
+import {
+  createDefaultRegistry,
+  type LogSink,
+  OpenWorkflowEngine,
+  type OutboxRelay,
+  type OutboxRouteFn,
+  type TaskEngine,
+  TaskRegistry,
+} from "@umt/tasks";
+import type { Logger } from "pino";
+import type { Config } from "./config.ts";
+import type { EventBusPort } from "./http/sse.ts";
+import { createCapabilityRegistry } from "./modules/agent/capability-registry.ts";
+import { createSecretStore, createSettingsReader } from "./modules/auth-settings/provided.ts";
+import type { OutboxPort, ServerContext, ServerModule, ServerRole } from "./modules/types.ts";
+
+// 进程装配:ServerContext、任务注册表、outbox 路由。api / worker / all 三种角色共用。
+
+/** 引擎在任务注册表建好后才能连接;先给模块一个会转发的句柄 */
+export class DeferredTaskEngine implements TaskEngine {
+  private inner: TaskEngine | null = null;
+
+  attach(engine: TaskEngine) {
+    this.inner = engine;
+  }
+
+  get ready(): boolean {
+    return this.inner !== null;
+  }
+
+  private get e(): TaskEngine {
+    if (!this.inner) throw new Error("任务引擎尚未就绪(模块不要在 register* 阶段调用引擎)");
+    return this.inner;
+  }
+
+  enqueue: TaskEngine["enqueue"] = (...a) => this.e.enqueue(...a);
+  signal: TaskEngine["signal"] = (...a) => this.e.signal(...a);
+  cancel: TaskEngine["cancel"] = (...a) => this.e.cancel(...a);
+  get: TaskEngine["get"] = (...a) => this.e.get(...a);
+  list: TaskEngine["list"] = (...a) => this.e.list(...a);
+  startWorker: TaskEngine["startWorker"] = (...a) => this.e.startWorker(...a);
+  stop: TaskEngine["stop"] = async () => {
+    await this.inner?.stop();
+  };
+}
+
+export function createOutboxPort(): OutboxPort & { setRelay(r: OutboxRelay): void } {
+  let relay: OutboxRelay | null = null;
+  return {
+    get relay() {
+      return relay;
+    },
+    setRelay(r) {
+      relay = r;
+    },
+    async emit(executor, type, payload, opts = {}) {
+      if (parseEventPayload(type as EventType, payload) === null) {
+        throw new Error(`outbox 事件 ${type} 的 payload 不符合 contracts 定义`);
+      }
+      await sql`insert into outbox_event (type, payload, idempotency_key)
+        values (${type}, ${JSON.stringify(payload)}::jsonb, ${opts.idempotencyKey ?? null})
+        on conflict (idempotency_key) where idempotency_key is not null do nothing`.execute(executor);
+    },
+  };
+}
+
+/** 把 pino logger 接到 @umt/tasks 的 LogSink(tasks 自己已脱敏,这里再经 pino 的脱敏) */
+export function tasksLogSink(logger: Logger): LogSink {
+  // r.fields 里已带 component(tasks.engine / tasks.outbox)
+  return (r) => logger[r.level](r.fields, r.msg);
+}
+
+/**
+ * 任务注册表:各模块注册自己的任务;没人注册的 §3.1-8 任务用 @umt/tasks 的空壳兜底
+ * (outbox 路由到它们时不至于报"未知任务")。同名任务被两个模块注册会在这里报错。
+ */
+export function buildTaskRegistry(modules: readonly ServerModule[], ctx: ServerContext): TaskRegistry {
+  const registry = new TaskRegistry();
+  const owner = new Map<string, string>();
+  for (const m of modules) {
+    const own = new TaskRegistry();
+    m.registerTasks(own, ctx);
+    for (const name of own.names()) {
+      const prev = owner.get(name);
+      if (prev) throw new Error(`任务 ${name} 被模块 ${prev} 与 ${m.name} 重复注册`);
+      owner.set(name, m.name);
+    }
+    registry.register(...own.all());
+  }
+  for (const stub of createDefaultRegistry().all()) if (!registry.has(stub.name)) registry.register(stub);
+  return registry;
+}
+
+/** 合并各模块的 outbox 路由;同一事件 type 只能由一个模块认领 */
+export function collectOutboxRoutes(
+  modules: readonly ServerModule[],
+  ctx: ServerContext,
+): Record<string, OutboxRouteFn> {
+  const routes: Record<string, OutboxRouteFn> = {};
+  const owner = new Map<string, string>();
+  for (const m of modules) {
+    for (const [type, fn] of Object.entries(m.outboxRoutes?.(ctx) ?? {})) {
+      const prev = owner.get(type);
+      if (prev) throw new Error(`outbox 事件 ${type} 被模块 ${prev} 与 ${m.name} 重复认领`);
+      owner.set(type, m.name);
+      routes[type] = fn;
+    }
+  }
+  return routes;
+}
+
+export interface BootstrapOptions {
+  role: ServerRole;
+  config: Config;
+  logger: Logger;
+  db: Kysely<DB>;
+  events: EventBusPort;
+  modules: readonly ServerModule[];
+  /** 连接引擎时是否跑引擎迁移;默认只有 all 角色跑(api/worker 分进程时由 --role migrate 预先迁移) */
+  engineMigrations?: boolean;
+  /** 仅测试覆盖:引擎表 schema(默认 openworkflow) */
+  engineSchema?: string;
+  /** 仅测试覆盖:不连接真实引擎 */
+  engine?: TaskEngine;
+}
+
+export interface Bootstrapped {
+  ctx: ServerContext;
+  tasks: TaskRegistry;
+  outbox: ReturnType<typeof createOutboxPort>;
+  engine: DeferredTaskEngine;
+}
+
+/** 建 ServerContext → 模块注册任务与能力 → 连接引擎。路由与 worker 由调用方按角色启动 */
+export async function bootstrap(opts: BootstrapOptions): Promise<Bootstrapped> {
+  const { config, logger, db } = opts;
+  const engine = new DeferredTaskEngine();
+  const outbox = createOutboxPort();
+  const ctx: ServerContext = {
+    role: opts.role,
+    db,
+    config,
+    logger,
+    events: opts.events,
+    engine,
+    outbox,
+    secrets: createSecretStore({ db, masterKey: config.SECRET_MASTER_KEY }),
+    settings: createSettingsReader({ db, logger }),
+    capabilities: createCapabilityRegistry({ logger }),
+  };
+  const tasks = buildTaskRegistry(opts.modules, ctx);
+  for (const m of opts.modules) m.registerCapabilities?.(ctx.capabilities, ctx);
+  engine.attach(
+    opts.engine ??
+      (await OpenWorkflowEngine.connect({
+        databaseUrl: config.DATABASE_URL,
+        db,
+        registry: tasks,
+        logSink: tasksLogSink(logger),
+        runMigrations: opts.engineMigrations ?? opts.role === "all",
+        ...(opts.engineSchema ? { schema: opts.engineSchema } : {}),
+      })),
+  );
+  return { ctx, tasks, outbox, engine };
+}

apps/server/src/http/app.ts

@@ -8,17 +8,18 @@ import type { Logger } from "pino";
 import type { Config } from "../config.ts";
 import { isDev } from "../config.ts";
 import { redactString } from "../logger.ts";
+import type { ServerContext, ServerModule } from "../modules/types.ts";
 import { type ApiKeyStore, type AuthEnv, apiKeyAuth, kyselyApiKeyStore, requireScope } from "./auth.ts";
 import { ApiException, errorBody } from "./errors.ts";
 import { healthRoutes } from "./routes/health.ts";
 import { mountOpenApi } from "./routes/openapi.ts";
-import { type EventBus, sseRoutes } from "./sse.ts";
+import { type EventBusPort, sseRoutes } from "./sse.ts";
 import { mountWebRoot } from "./static.ts";
 
 export interface AppDeps {
   config: Pick<Config, "NODE_ENV" | "DEV_ADMIN_KEY">;
   logger: Logger;
-  bus: EventBus;
+  bus: EventBusPort;
   version: string;
   role: "api" | "worker" | "all";
   startedAt?: number;
@@ -27,6 +28,9 @@ export interface AppDeps {
   apiKeyStore?: ApiKeyStore;
   /** 单容器模式:托管 web 静态产物的目录 */
   webRoot?: string;
+  /** 业务模块(modules/index.ts 的 SERVER_MODULES);给了 ctx 才挂载 */
+  modules?: readonly ServerModule[];
+  ctx?: ServerContext;
 }
 
 export const API_PREFIX = "/api/v1";
@@ -35,15 +39,7 @@ export function createApp(deps: AppDeps) {
   const { config, logger, bus, db } = deps;
   const dev = isDev(config);
 
-  const app = new OpenAPIHono<AuthEnv>({
-    // zod 校验失败统一成 validation_failed 信封
-    defaultHook: (result, c) => {
-      if (!result.success) {
-        const msg = result.error.issues.map((i) => `${i.path.join(".") || "(root)"}: ${i.message}`).join("; ");
-        return c.json(errorBody("validation_failed", msg), 422);
-      }
-    },
-  });
+  const app = newApiApp();
 
   app.use("*", requestId());
   // 访问日志:路径中的 passkey/token 参数脱敏
@@ -112,8 +108,27 @@ export function createApp(deps: AppDeps) {
     return c.json({ ok: true, data: { id: p.id, name: p.name, kind: p.kind, scopes: p.scopes } });
   });
 
+  // ── 业务模块:各自挂在 /api/v1 下(已鉴权),按清单顺序注册 ──
+  if (deps.ctx && deps.modules?.length) {
+    const api = newApiApp();
+    for (const m of deps.modules) m.registerRoutes(api, deps.ctx);
+    app.route(API_PREFIX, api);
+  }
+
   if (deps.webRoot) mountWebRoot(app, deps.webRoot);
 
   return app;
 }
 export type App = ReturnType<typeof createApp>;
+
+/** 统一的 OpenAPIHono 实例:zod 校验失败统一成 validation_failed 信封 */
+function newApiApp() {
+  return new OpenAPIHono<AuthEnv>({
+    defaultHook: (result, c) => {
+      if (!result.success) {
+        const msg = result.error.issues.map((i) => `${i.path.join(".") || "(root)"}: ${i.message}`).join("; ");
+        return c.json(errorBody("validation_failed", msg), 422);
+      }
+    },
+  });
+}

apps/server/src/http/sse.ts

@@ -5,11 +5,22 @@ import { ApiException } from "./errors.ts";
 
 type Listener = (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;
+}
+
 /**
  * 进程内事件总线:序号自增,保留最近 capacity 条供断线补读。
  * 多进程部署时需换成 PG LISTEN/NOTIFY 或 outbox 投影(W2)。
  */
-export class EventBus {
+export class EventBus implements EventBusPort {
   private seq = 0;
   private readonly buffer: EventEnvelope[] = [];
   private readonly listeners = new Set<Listener>();
@@ -42,7 +53,7 @@ export class EventBus {
 }
 
 export interface SseOptions {
-  bus: EventBus;
+  bus: EventBusPort;
   keepaliveMs?: number;
 }
 

apps/server/src/main.ts

@@ -1,10 +1,13 @@
 import { parseArgs } from "node:util";
 import { createDb, migrateToLatest } from "@umt/db";
+import { migrateEngineSchema } from "@umt/tasks";
 import denoJson from "../deno.json" with { type: "json" };
 import { loadConfig } from "./config.ts";
+import { bootstrap } from "./context.ts";
 import { createApp } from "./http/app.ts";
 import { EventBus } from "./http/sse.ts";
 import { createLogger } from "./logger.ts";
+import { SERVER_MODULES } from "./modules/index.ts";
 import { runPreflight } from "./preflight.ts";
 import { startWorker } from "./worker/index.ts";
 
@@ -52,6 +55,9 @@ if (role === "migrate" || role === "preflight") {
       const { results } = await migrateToLatest(db);
       for (const r of results ?? []) logger.info({ migration: r.migrationName, status: r.status }, "migration");
       logger.info({ applied: results?.length ?? 0 }, "迁移完成");
+      // 任务引擎自己的表(openworkflow schema);api/worker 进程启动时不再迁移
+      await migrateEngineSchema(config.DATABASE_URL);
+      logger.info("任务引擎迁移完成");
     } else {
       const problems = await runPreflight({ config, db });
       for (const p of problems) logger.error({ check: p.check }, p.message);
@@ -70,6 +76,17 @@ if (role === "migrate" || role === "preflight") {
 const bus = new EventBus();
 const stops: Array<() => Promise<void>> = [];
 
+// 装配 ServerContext:模块注册任务/能力,连接任务引擎(api 角色也要:入队、信号、取消)
+let booted: Awaited<ReturnType<typeof bootstrap>>;
+try {
+  booted = await bootstrap({ role, config, logger, db, events: bus, modules: SERVER_MODULES });
+} catch (err) {
+  logger.error({ err }, "启动装配失败");
+  await db.destroy();
+  Deno.exit(1);
+}
+const { ctx, outbox, engine } = booted;
+
 if (role === "api" || role === "all") {
   const app = createApp({
     config,
@@ -79,6 +96,8 @@ if (role === "api" || role === "all") {
     role,
     version: denoJson.version,
     ...(config.WEB_ROOT ? { webRoot: config.WEB_ROOT } : {}),
+    modules: SERVER_MODULES,
+    ctx,
   });
   const ac = new AbortController();
   const server = Deno.serve(
@@ -101,7 +120,12 @@ if (role === "api" || role === "all") {
   });
 }
 if (role === "worker" || role === "all") {
-  const worker = startWorker({ logger, heartbeatMs: config.WORKER_HEARTBEAT_MS, bus });
+  const worker = await startWorker({
+    logger,
+    heartbeatMs: config.WORKER_HEARTBEAT_MS,
+    bus,
+    runtime: { ctx, modules: SERVER_MODULES, setRelay: outbox.setRelay },
+  });
   stops.push(worker.stop);
 }
 
@@ -113,6 +137,7 @@ async function shutdown(signal: Deno.Signal) {
   // 兜底:10 秒内没退干净就强退,避免容器 stop 超时被 SIGKILL
   const force = setTimeout(() => Deno.exit(1), 10_000);
   await Promise.allSettled(stops.map((s) => s()));
+  await engine.stop().catch((err) => logger.error({ err }, "任务引擎关闭失败"));
   await db.destroy();
   clearTimeout(force);
   Deno.exit(0);

apps/server/src/modules/acquisition-organize/index.ts

@@ -0,0 +1,13 @@
+import type { ServerModule } from "../types.ts";
+
+/**
+ * W2-C 规则/下载/入库/整理记录:subscription、acquisition_slot、policy_revision、preference_tag;
+ * evaluateAcquisition / submitDownload / verifyAndImport / cleanupDownloadCache / reconcileLibrary;
+ * qB 同步与缓存阈值、Copy/Move、path_binding、organization_record 与公开增量 API、媒体库与首页聚合。
+ * 消费 release.resolved(outboxRoutes);删种带数据一律经 adapters 的 assertSafeToDeleteData。
+ */
+export const acquisitionOrganizeModule: ServerModule = {
+  name: "acquisition-organize",
+  registerRoutes() {},
+  registerTasks() {},
+};

apps/server/src/modules/agent/capability-registry.ts

@@ -0,0 +1,13 @@
+import { CapabilityRegistry } from "@umt/agent";
+import type { Logger } from "pino";
+
+/**
+ * 能力注册中心工厂(W2-D 负责接 authorize:automation_grant / approval + 自动化总开关;audit:写 audit_event)。
+ * 当前未注入 authorize:read/preview 可调用,propose/approve/execute 一律拒绝(安全默认)。
+ */
+export function createCapabilityRegistry(deps: { logger: Logger }): CapabilityRegistry {
+  const log = deps.logger.child({ component: "capabilities" });
+  return new CapabilityRegistry({
+    audit: (r) => log.info({ capability: r.capabilityId, outcome: r.outcome, actor: r.actorId }, "capability.invoke"),
+  });
+}

apps/server/src/modules/agent/index.ts

@@ -0,0 +1,11 @@
+import type { ServerModule } from "../types.ts";
+
+/**
+ * W2-D Agent 装配:14 个能力的 authorize/audit 注入(./capability-registry.ts)、Agent 会话 API + SSE、
+ * agent_run_event 落库、resolveDecision 处理器、后台触发的有界 agent 任务。
+ */
+export const agentModule: ServerModule = {
+  name: "agent",
+  registerRoutes() {},
+  registerTasks() {},
+};

apps/server/src/modules/auth-settings/index.ts

@@ -0,0 +1,12 @@
+import type { ServerModule } from "../types.ts";
+
+/**
+ * W2-A 认证/设置/实例/引擎面板:管理员登录门禁、settings、api_key、secret 加密服务、
+ * downloader/media_server 实例与 library_root/path_mapping、审计查询、SSE 总线改 PG 通知、引擎面板只读 API、自动化总开关。
+ * secrets / settings 的实现放在 ./provided.ts(context 从这里取),SSE 总线在 http/sse.ts,鉴权在 http/auth.ts。
+ */
+export const authSettingsModule: ServerModule = {
+  name: "auth-settings",
+  registerRoutes() {},
+  registerTasks() {},
+};

apps/server/src/modules/auth-settings/provided.ts

@@ -0,0 +1,33 @@
+import { DEFAULT_SETTINGS, GlobalSettings, type SettingsKey } from "@umt/contracts";
+import type { DB, Kysely } from "@umt/db";
+import type { Logger } from "pino";
+import type { SecretStore, SettingsReader } from "../types.ts";
+
+// auth-settings 模块向 ServerContext 提供的基础服务。W2-A 在这里替换实现,其它模块只用 ctx 上的接口。
+
+/** 占位:W2-A 实现 AES-GCM 加密存取(secret 表,SECRET_MASTER_KEY) */
+export function createSecretStore(_deps: { db: Kysely<DB>; masterKey: string }): SecretStore {
+  const notYet = async (): Promise<never> => {
+    throw new Error("SecretStore 尚未实现(W2-A)");
+  };
+  return { resolve: notYet, create: notYet, replace: notYet, remove: notYet };
+}
+
+/** settings 表(key → jsonb)按分组合并缺省值;每次读库,不缓存(设置变更立即生效) */
+export function createSettingsReader(deps: { db: Kysely<DB>; logger: Logger }): SettingsReader {
+  const all = async (): Promise<GlobalSettings> => {
+    const rows = await deps.db.selectFrom("settings").select(["key", "value"]).execute();
+    const stored = new Map(rows.map((r) => [r.key, r.value]));
+    const out: Record<string, unknown> = {};
+    for (const key of Object.keys(DEFAULT_SETTINGS) as SettingsKey[]) {
+      const def = DEFAULT_SETTINGS[key];
+      const v = stored.get(key);
+      const merged = v && typeof v === "object" && !Array.isArray(v) ? { ...def, ...(v as object) } : def;
+      const parsed = GlobalSettings.shape[key].safeParse(merged);
+      if (!parsed.success) deps.logger.warn({ key }, "settings 值不合法,回退缺省");
+      out[key] = parsed.success ? parsed.data : def;
+    }
+    return out as GlobalSettings;
+  };
+  return { all, get: async (key) => (await all())[key] };
+}

apps/server/src/modules/index.ts

@@ -0,0 +1,18 @@
+import { acquisitionOrganizeModule } from "./acquisition-organize/index.ts";
+import { agentModule } from "./agent/index.ts";
+import { authSettingsModule } from "./auth-settings/index.ts";
+import { sourcesIdentityModule } from "./sources-identity/index.ts";
+import type { ServerModule } from "./types.ts";
+
+export type * from "./types.ts";
+
+/**
+ * 模块清单:http/app.ts(路由)与 worker(任务/中继/后台循环)都从这里取。
+ * 顺序即注册顺序;auth-settings 在前(其它模块的路由依赖它的登录门禁中间件)。
+ */
+export const SERVER_MODULES: readonly ServerModule[] = [
+  authSettingsModule,
+  sourcesIdentityModule,
+  acquisitionOrganizeModule,
+  agentModule,
+];

apps/server/src/modules/sources-identity/index.ts

@@ -0,0 +1,12 @@
+import type { ServerModule } from "../types.ts";
+
+/**
+ * W2-B 来源/身份/元数据:site、site_account、feed_source;ingestFeed / resolveObservation / refreshMetadata;
+ * observation/source_release 写入与去重;身份解析、待识别池与 decision;站点工作台 API。
+ * 发出 release.resolved、decision.created(经 ctx.outbox.emit)。
+ */
+export const sourcesIdentityModule: ServerModule = {
+  name: "sources-identity",
+  registerRoutes() {},
+  registerTasks() {},
+};

apps/server/src/modules/types.ts

@@ -0,0 +1,94 @@
+import type { OpenAPIHono } from "@hono/zod-openapi";
+import type { CapabilityRegistry } from "@umt/agent";
+import type { EventPayload, EventType, GlobalSettings, SecretPurpose, SettingsKey } from "@umt/contracts";
+import type { DB, Kysely, Transaction } from "@umt/db";
+import type { OutboxRelay, OutboxRouteFn, TaskEngine, TaskRegistry } from "@umt/tasks";
+import type { Logger } from "pino";
+import type { Config } from "../config.ts";
+import type { AuthEnv } from "../http/auth.ts";
+import type { EventBusPort } from "../http/sse.ts";
+
+// 服务端模块骨架(W1 审查 B6;任务书 §5)。
+// 每个 W2 子任务是一个 ServerModule,只改 modules/<自己的目录>/ 与 contracts/db 中本域的文件。
+// 跨模块只通过 DB 表 + outbox_event(事件类型在 contracts/events.ts)+ contracts 类型耦合,不互相 import。
+
+export type ServerRole = "api" | "worker" | "all";
+
+/**
+ * 模块路由挂载点:已挂在 /api/v1 下、已经过 API Key 鉴权(c.var.principal 可用)。
+ * 路径写相对 /api/v1 的部分(如 "/sites"),每个路由自己加 requireScope(...);
+ * 用 app.openapi(createRoute(...)) 注册的路由会进 /api/openapi.json。
+ */
+export type ModuleApp = OpenAPIHono<AuthEnv>;
+
+/** 业务事务内写 outbox_event;中继(worker)再按 type 路由成任务 */
+export interface OutboxWriter {
+  /**
+   * payload 按 contracts EVENT_PAYLOADS 校验,不合法直接抛错。
+   * 传 idempotencyKey 时同键重复写入会被忽略(outbox_event.idempotency_key 唯一),也作为引擎入队幂等键。
+   */
+  emit<T extends EventType>(
+    executor: Kysely<DB> | Transaction<DB>,
+    type: T,
+    payload: EventPayload<T>,
+    opts?: { idempotencyKey?: string },
+  ): Promise<void>;
+}
+
+export interface OutboxPort extends OutboxWriter {
+  /** 仅 worker 角色有中继实例;api 角色为 null */
+  readonly relay: OutboxRelay | null;
+}
+
+/**
+ * 凭证存取(secret 表 + SECRET_MASTER_KEY 加密)。W2-A 在 modules/auth-settings/provided.ts 实现,
+ * 当前占位实现一律抛"未实现"。其它模块只存/传 secret_ref,需要明文时现场 resolve,不缓存、不打日志。
+ */
+export interface SecretStore {
+  resolve(secretRef: string): Promise<string>;
+  create(input: { purpose: SecretPurpose; plaintext: string; label?: string }): Promise<string>;
+  replace(secretRef: string, plaintext: string): Promise<void>;
+  remove(secretRef: string): Promise<void>;
+}
+
+/** 全局设置读取:settings 表按分组合并 DEFAULT_SETTINGS 后校验;表里值非法时回退缺省并记警告 */
+export interface SettingsReader {
+  get<K extends SettingsKey>(key: K): Promise<GlobalSettings[K]>;
+  all(): Promise<GlobalSettings>;
+}
+
+export interface ServerContext {
+  role: ServerRole;
+  db: Kysely<DB>;
+  config: Config;
+  logger: Logger;
+  /** SSE 事件总线 */
+  events: EventBusPort;
+  /** 任务引擎:api 角色也可 enqueue/signal/cancel;只有 worker 角色执行任务 */
+  engine: TaskEngine;
+  outbox: OutboxPort;
+  secrets: SecretStore;
+  settings: SettingsReader;
+  /** 能力注册中心(Web/Pi/CLI/MCP 共用一条 invoke 路径) */
+  capabilities: CapabilityRegistry;
+}
+
+export interface ServerModule {
+  /** 与目录名一致,用于日志 */
+  name: string;
+  /** api / all 角色:挂 HTTP 路由 */
+  registerRoutes(app: ModuleApp, ctx: ServerContext): void;
+  /**
+   * 所有角色:注册持久任务处理器(api 角色也要注册,enqueue 时按定义校验输入)。
+   * 同名任务只能由一个模块注册;没人注册的 §3.1-8 任务由 @umt/tasks 的空壳兜底。
+   */
+  registerTasks(registry: TaskRegistry, ctx: ServerContext): void;
+  /** 所有角色:outbox 事件 type → 任务路由;同一 type 只能由一个模块认领 */
+  outboxRoutes?(ctx: ServerContext): Record<string, OutboxRouteFn>;
+  /** 所有角色:用 contracts 的 capabilitySpec(id) 注册能力处理器 */
+  registerCapabilities?(registry: CapabilityRegistry, ctx: ServerContext): void;
+  /** worker / all 角色:启动轮询器、定时器等后台循环(引擎与中继已启动) */
+  start?(ctx: ServerContext): Promise<void> | void;
+  /** 退出时按注册逆序调用 */
+  stop?(): Promise<void> | void;
+}

apps/server/src/worker/index.ts

@@ -1,16 +1,35 @@
 import { EventType } from "@umt/contracts";
+import { OutboxRelay } from "@umt/tasks";
 import type { Logger } from "pino";
-import type { EventBus } from "../http/sse.ts";
+import { collectOutboxRoutes, tasksLogSink } from "../context.ts";
+import type { EventBusPort } from "../http/sse.ts";
+import type { ServerContext, ServerModule } from "../modules/types.ts";
 
 export interface WorkerDeps {
   logger: Logger;
   heartbeatMs: number;
   /** role=all 时与 API 共享同一总线,心跳可在 SSE 中看到 */
-  bus?: EventBus;
+  bus?: EventBusPort;
+  /** 业务装配;省略时只跑心跳(测试用) */
+  runtime?: {
+    ctx: ServerContext;
+    modules: readonly ServerModule[];
+    /** 设置 outbox 中继实例到 ctx.outbox.relay */
+    setRelay?: (relay: OutboxRelay) => void;
+    outboxIntervalMs?: number;
+    concurrency?: number;
+  };
 }
 
-/** worker 占位:只打印心跳。持久任务引擎在 W1(packages/tasks)接入 */
-export function startWorker({ logger, heartbeatMs, bus }: WorkerDeps) {
+export interface RunningWorker {
+  stop(): Promise<void>;
+}
+
+/**
+ * worker 角色:心跳 + 任务引擎 worker(执行全部已注册任务)+ OutboxRelay(按模块的 outboxRoutes 路由)
+ * + 各模块的后台循环(start)。stop 逆序停止;引擎连接本身由进程退出时关闭。
+ */
+export async function startWorker({ logger, heartbeatMs, bus, runtime }: WorkerDeps): Promise<RunningWorker> {
   const log = logger.child({ component: "worker" });
   let n = 0;
   const beat = () => {
@@ -18,12 +37,41 @@ export function startWorker({ logger, heartbeatMs, bus }: WorkerDeps) {
     log.debug({ n }, "worker heartbeat");
     bus?.publish(EventType.heartbeat, { role: "worker", n });
   };
-  log.info({ heartbeatMs }, "worker started");
+
+  const started: ServerModule[] = [];
+  let relay: OutboxRelay | null = null;
+  if (runtime) {
+    const { ctx, modules } = runtime;
+    await ctx.engine.startWorker({ ...(runtime.concurrency ? { concurrency: runtime.concurrency } : {}) });
+    relay = new OutboxRelay({
+      db: ctx.db,
+      engine: ctx.engine,
+      routes: collectOutboxRoutes(modules, ctx),
+      logSink: tasksLogSink(logger),
+      ...(runtime.outboxIntervalMs ? { intervalMs: runtime.outboxIntervalMs } : {}),
+    });
+    runtime.setRelay?.(relay);
+    relay.start();
+    for (const m of modules) {
+      await m.start?.(ctx);
+      started.push(m);
+    }
+  }
+
+  log.info({ heartbeatMs, modules: started.map((m) => m.name) }, "worker started");
   beat();
   const timer = setInterval(beat, heartbeatMs);
   return {
     async stop() {
       clearInterval(timer);
+      for (const m of started.reverse()) {
+        try {
+          await m.stop?.();
+        } catch (err) {
+          log.error({ err, module: m.name }, "模块停止失败");
+        }
+      }
+      await relay?.stop();
       log.info("worker stopped");
     },
   };