({
+ 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);
+ }
+ },
+ });
+}
diff --git a/apps/server/src/http/sse.ts b/apps/server/src/http/sse.ts
index bffed1d1..4919ecf4 100644
--- a/apps/server/src/http/sse.ts
+++ b/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(type: string, payload: P): EventEnvelope
;
+ /** 取 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();
@@ -42,7 +53,7 @@ export class EventBus {
}
export interface SseOptions {
- bus: EventBus;
+ bus: EventBusPort;
keepaliveMs?: number;
}
diff --git a/apps/server/src/main.ts b/apps/server/src/main.ts
index b1f421a3..c1c753a6 100644
--- a/apps/server/src/main.ts
+++ b/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> = [];
+// 装配 ServerContext:模块注册任务/能力,连接任务引擎(api 角色也要:入队、信号、取消)
+let booted: Awaited>;
+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);
diff --git a/apps/server/src/modules/acquisition-organize/index.ts b/apps/server/src/modules/acquisition-organize/index.ts
new file mode 100644
index 00000000..289d8cf4
--- /dev/null
+++ b/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() {},
+};
diff --git a/apps/server/src/modules/agent/capability-registry.ts b/apps/server/src/modules/agent/capability-registry.ts
new file mode 100644
index 00000000..df15c0cf
--- /dev/null
+++ b/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"),
+ });
+}
diff --git a/apps/server/src/modules/agent/index.ts b/apps/server/src/modules/agent/index.ts
new file mode 100644
index 00000000..64bfbc67
--- /dev/null
+++ b/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() {},
+};
diff --git a/apps/server/src/modules/auth-settings/index.ts b/apps/server/src/modules/auth-settings/index.ts
new file mode 100644
index 00000000..01abbeaf
--- /dev/null
+++ b/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() {},
+};
diff --git a/apps/server/src/modules/auth-settings/provided.ts b/apps/server/src/modules/auth-settings/provided.ts
new file mode 100644
index 00000000..b029e563
--- /dev/null
+++ b/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; masterKey: string }): SecretStore {
+ const notYet = async (): Promise => {
+ throw new Error("SecretStore 尚未实现(W2-A)");
+ };
+ return { resolve: notYet, create: notYet, replace: notYet, remove: notYet };
+}
+
+/** settings 表(key → jsonb)按分组合并缺省值;每次读库,不缓存(设置变更立即生效) */
+export function createSettingsReader(deps: { db: Kysely; logger: Logger }): SettingsReader {
+ const all = async (): Promise => {
+ 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 = {};
+ 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] };
+}
diff --git a/apps/server/src/modules/index.ts b/apps/server/src/modules/index.ts
new file mode 100644
index 00000000..c07e33b3
--- /dev/null
+++ b/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,
+];
diff --git a/apps/server/src/modules/sources-identity/index.ts b/apps/server/src/modules/sources-identity/index.ts
new file mode 100644
index 00000000..10b93630
--- /dev/null
+++ b/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() {},
+};
diff --git a/apps/server/src/modules/types.ts b/apps/server/src/modules/types.ts
new file mode 100644
index 00000000..9b4358f9
--- /dev/null
+++ b/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;
+
+/** 业务事务内写 outbox_event;中继(worker)再按 type 路由成任务 */
+export interface OutboxWriter {
+ /**
+ * payload 按 contracts EVENT_PAYLOADS 校验,不合法直接抛错。
+ * 传 idempotencyKey 时同键重复写入会被忽略(outbox_event.idempotency_key 唯一),也作为引擎入队幂等键。
+ */
+ emit(
+ executor: Kysely | Transaction,
+ type: T,
+ payload: EventPayload,
+ opts?: { idempotencyKey?: string },
+ ): Promise;
+}
+
+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;
+ create(input: { purpose: SecretPurpose; plaintext: string; label?: string }): Promise;
+ replace(secretRef: string, plaintext: string): Promise;
+ remove(secretRef: string): Promise;
+}
+
+/** 全局设置读取:settings 表按分组合并 DEFAULT_SETTINGS 后校验;表里值非法时回退缺省并记警告 */
+export interface SettingsReader {
+ get(key: K): Promise;
+ all(): Promise;
+}
+
+export interface ServerContext {
+ role: ServerRole;
+ db: Kysely;
+ 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;
+ /** 所有角色:用 contracts 的 capabilitySpec(id) 注册能力处理器 */
+ registerCapabilities?(registry: CapabilityRegistry, ctx: ServerContext): void;
+ /** worker / all 角色:启动轮询器、定时器等后台循环(引擎与中继已启动) */
+ start?(ctx: ServerContext): Promise | void;
+ /** 退出时按注册逆序调用 */
+ stop?(): Promise | void;
+}
diff --git a/apps/server/src/worker/index.ts b/apps/server/src/worker/index.ts
index ebcfda4c..81f41079 100644
--- a/apps/server/src/worker/index.ts
+++ b/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;
+}
+
+/**
+ * worker 角色:心跳 + 任务引擎 worker(执行全部已注册任务)+ OutboxRelay(按模块的 outboxRoutes 路由)
+ * + 各模块的后台循环(start)。stop 逆序停止;引擎连接本身由进程退出时关闭。
+ */
+export async function startWorker({ logger, heartbeatMs, bus, runtime }: WorkerDeps): Promise {
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");
},
};
diff --git a/apps/server/test/bootstrap.test.ts b/apps/server/test/bootstrap.test.ts
new file mode 100644
index 00000000..5135f468
--- /dev/null
+++ b/apps/server/test/bootstrap.test.ts
@@ -0,0 +1,138 @@
+import { EventType } from "@umt/contracts";
+import { type DB, migrateToLatest } from "@umt/db";
+import { defineTask } from "@umt/tasks";
+import { Kysely, PostgresDialect, sql } from "kysely";
+import pg from "pg";
+import { afterAll, beforeAll, describe, expect, it } from "vitest";
+import { z } from "zod";
+import { ConfigSchema } from "../src/config.ts";
+import { type Bootstrapped, bootstrap } from "../src/context.ts";
+import { EventBus } from "../src/http/sse.ts";
+import type { ServerModule } from "../src/modules/types.ts";
+import { type RunningWorker, startWorker } from "../src/worker/index.ts";
+import { memoryLogger } from "./helpers.ts";
+
+// 真实 PG:独立业务 schema + 独立引擎 schema,验证 bootstrap → worker → outbox → 模块任务 的整条装配。
+const url = process.env.DATABASE_URL ?? "postgres://umt:umt@localhost:55433/umt";
+const uniq = `${process.pid.toString(36)}${Math.random().toString(36).slice(2, 7)}`;
+const biz = `server_test_${uniq}`;
+const ow = `${biz}_ow`;
+
+async function pgAvailable() {
+ const c = new pg.Client({ connectionString: url, connectionTimeoutMillis: 2000 });
+ try {
+ await c.connect();
+ await c.end();
+ return true;
+ } catch {
+ return false;
+ }
+}
+const hasPg = process.env.UMT_SKIP_DB_TESTS !== "1" && (await pgAvailable());
+
+describe.skipIf(!hasPg)("bootstrap + worker(真实 PG)", () => {
+ let db: Kysely;
+ let booted: Bootstrapped;
+ let worker: RunningWorker;
+ const seen: string[] = [];
+
+ const probe: ServerModule = {
+ name: "probe",
+ registerRoutes() {},
+ registerTasks(registry, ctx) {
+ registry.register(
+ defineTask({
+ name: "evaluateAcquisition",
+ version: 1,
+ inputSchema: z.object({ sourceReleaseId: z.string().optional() }),
+ handler: async (t) => {
+ const s = await ctx.settings.get("cache");
+ seen.push(`${t.input.sourceReleaseId}:${s.hardThresholdPercent}`);
+ return null;
+ },
+ }),
+ );
+ },
+ outboxRoutes: () => ({
+ [EventType.releaseResolved]: (ev) => ({
+ task: "evaluateAcquisition",
+ input: { sourceReleaseId: (ev.payload as { sourceReleaseId: string }).sourceReleaseId },
+ }),
+ }),
+ };
+
+ beforeAll(async () => {
+ const admin = new pg.Client({ connectionString: url });
+ await admin.connect();
+ await admin.query(`create schema ${biz}`);
+ await admin.end();
+ db = new Kysely({
+ dialect: new PostgresDialect({
+ pool: new pg.Pool({ connectionString: url, max: 5, options: `-c search_path=${biz},public` }),
+ }),
+ });
+ await migrateToLatest(db, { migrationTableSchema: biz });
+ await db
+ .insertInto("settings")
+ .values({ key: "cache", value: JSON.stringify({ hardThresholdPercent: 95, resumePercent: 90 }) })
+ .execute();
+ const config = ConfigSchema.parse({ DATABASE_URL: url, SECRET_MASTER_KEY: "test-master-key-0123456789" });
+ const { logger } = memoryLogger();
+ booted = await bootstrap({
+ role: "all",
+ config,
+ logger,
+ db,
+ events: new EventBus(10),
+ modules: [probe],
+ engineSchema: ow,
+ });
+ worker = await startWorker({
+ logger,
+ heartbeatMs: 60_000,
+ runtime: { ctx: booted.ctx, modules: [probe], setRelay: booted.outbox.setRelay, outboxIntervalMs: 100 },
+ });
+ }, 60_000);
+
+ afterAll(async () => {
+ await worker?.stop();
+ await booted?.engine.stop();
+ await db?.destroy();
+ const admin = new pg.Client({ connectionString: url });
+ await admin.connect();
+ await admin.query(`drop schema if exists ${biz} cascade`);
+ await admin.query(`drop schema if exists ${ow} cascade`);
+ await admin.end();
+ }, 60_000);
+
+ it("settings 读取合并缺省值", async () => {
+ const cache = await booted.ctx.settings.get("cache");
+ expect(cache).toMatchObject({ hardThresholdPercent: 95, resumePercent: 90, checkIntervalSec: 60 });
+ expect((await booted.ctx.settings.all()).automation.enabled).toBe(true);
+ await expect(booted.ctx.secrets.resolve("x")).rejects.toThrow(/W2-A/);
+ });
+
+ it("业务事务内 emit → 中继路由 → worker 执行模块任务;同幂等键只投一次;非法 payload 拒写", async () => {
+ const payload = {
+ sourceReleaseId: "11111111-1111-4111-8111-111111111111",
+ observationId: null,
+ mediaEntityId: "22222222-2222-4222-8222-222222222222",
+ contentUnitIds: [],
+ evidenceLevel: "rule_confirmed" as const,
+ };
+ await db.transaction().execute(async (trx) => {
+ await booted.ctx.outbox.emit(trx, EventType.releaseResolved, payload, { idempotencyKey: "rr:1" });
+ await booted.ctx.outbox.emit(trx, EventType.releaseResolved, payload, { idempotencyKey: "rr:1" });
+ });
+ await expect(
+ booted.ctx.outbox.emit(db, EventType.releaseResolved, { ...payload, evidenceLevel: "bogus" } as never),
+ ).rejects.toThrow(/payload/);
+ expect(booted.ctx.outbox.relay).not.toBeNull();
+
+ const deadline = Date.now() + 30_000;
+ while (seen.length === 0 && Date.now() < deadline) await new Promise((r) => setTimeout(r, 200));
+ expect(seen).toEqual([`${payload.sourceReleaseId}:95`]);
+ const rows = await sql<{ status: string }>`select status from outbox_event`.execute(db);
+ expect(rows.rows).toEqual([{ status: "dispatched" }]);
+ }, 40_000);
+});
diff --git a/apps/server/test/mapping-compat.test.ts b/apps/server/test/mapping-compat.test.ts
new file mode 100644
index 00000000..ab62eb10
--- /dev/null
+++ b/apps/server/test/mapping-compat.test.ts
@@ -0,0 +1,19 @@
+import type { DownloaderClient, SiteRelease } from "@umt/adapters";
+import type { DomainDownloadStatus, SiteReleaseLike } from "@umt/mapping";
+import { toDomainDownloadStatus } from "@umt/mapping";
+import { describe, expect, it } from "vitest";
+
+// @umt/mapping 不依赖 adapters;server 同时依赖两者,在这里做类型层兼容校验。
+type Assert = T;
+export type _SiteReleaseCompatible = Assert;
+// 删数据接口必须带安全检查输入(B7)
+export type _DeleteRequiresSafety = Assert<
+ { deleteFiles: true } extends Parameters[1] ? false : true
+>;
+
+describe("mapping ↔ adapters 兼容", () => {
+ it("adapters 的 SiteRelease 可直接交给 mapping", () => {
+ const status: DomainDownloadStatus = toDomainDownloadStatus("paused", "threshold");
+ expect(status).toBe("paused_by_threshold");
+ });
+});
diff --git a/apps/server/test/modules.test.ts b/apps/server/test/modules.test.ts
new file mode 100644
index 00000000..bffa739b
--- /dev/null
+++ b/apps/server/test/modules.test.ts
@@ -0,0 +1,98 @@
+import { createRoute, z } from "@hono/zod-openapi";
+import { apiResponse, CAPABILITY_IDS, capabilitySpec } from "@umt/contracts";
+import { defineTask, TASK_NAMES } from "@umt/tasks";
+import { describe, expect, it } from "vitest";
+import { buildTaskRegistry, collectOutboxRoutes, DeferredTaskEngine } from "../src/context.ts";
+import { createApp } from "../src/http/app.ts";
+import { requireScope } from "../src/http/auth.ts";
+import { EventBus } from "../src/http/sse.ts";
+import { SERVER_MODULES } from "../src/modules/index.ts";
+import type { ServerContext, ServerModule } from "../src/modules/types.ts";
+import { DEV_KEY, json, memoryLogger } from "./helpers.ts";
+
+// register* 阶段模块只保存 ctx 引用,这里不需要真实依赖
+const fakeCtx = { role: "all" } as unknown as ServerContext;
+
+const pingRoute = createRoute({
+ method: "get",
+ path: "/test-ping",
+ tags: ["test"],
+ responses: {
+ 200: { description: "ok", content: { "application/json": { schema: apiResponse(z.object({ who: z.string() })) } } },
+ },
+});
+
+const pingModule: ServerModule = {
+ name: "test-ping",
+ registerRoutes(app) {
+ app.use("/test-ping", requireScope("admin"));
+ app.openapi(pingRoute, (c) => c.json({ ok: true as const, data: { who: c.get("principal").name } }, 200));
+ },
+ registerTasks(registry) {
+ registry.register(
+ defineTask({
+ name: "ingestFeed",
+ version: 1,
+ inputSchema: z.object({ feedSourceId: z.string() }),
+ handler: async () => "real",
+ }),
+ );
+ },
+ outboxRoutes: () => ({ "release.resolved": () => ({ task: "evaluateAcquisition", input: {} }) }),
+};
+
+describe("ServerModule 骨架", () => {
+ it("清单里是四个 W2 模块,名字唯一", () => {
+ expect(SERVER_MODULES.map((m) => m.name)).toEqual([
+ "auth-settings",
+ "sources-identity",
+ "acquisition-organize",
+ "agent",
+ ]);
+ });
+
+ it("模块路由挂在 /api/v1 下,经过鉴权与 scope 检查,并进入 OpenAPI 文档", async () => {
+ const { logger } = memoryLogger();
+ const app = createApp({
+ config: { NODE_ENV: "development", DEV_ADMIN_KEY: DEV_KEY },
+ logger,
+ bus: new EventBus(5),
+ role: "all",
+ version: "0.0.0-test",
+ modules: [pingModule],
+ ctx: fakeCtx,
+ });
+ expect((await app.request("/api/v1/test-ping")).status).toBe(401);
+ const ok = await app.request("/api/v1/test-ping", { headers: { authorization: `Bearer ${DEV_KEY}` } });
+ expect(ok.status).toBe(200);
+ expect((await json(ok)).data.who).toBe("dev-admin");
+ const doc = await json(await app.request("/api/openapi.json"));
+ expect(doc.paths["/api/v1/test-ping"]).toBeDefined();
+ // 未知路径仍是 JSON 404
+ expect((await app.request("/api/v1/nope", { headers: { authorization: `Bearer ${DEV_KEY}` } })).status).toBe(404);
+ });
+
+ it("任务注册表:模块任务优先,其余 §3.1-8 任务由空壳兜底;重复注册报错", async () => {
+ const reg = buildTaskRegistry([...SERVER_MODULES, pingModule], fakeCtx);
+ expect(reg.names()).toEqual([...TASK_NAMES].sort());
+ const ingest = reg.latest("ingestFeed");
+ expect(await ingest.handler({ input: {} } as never)).toBe("real");
+ expect(() => buildTaskRegistry([pingModule, { ...pingModule, name: "dup" }], fakeCtx)).toThrow(/重复注册/);
+ });
+
+ it("outbox 路由合并;同一事件被两个模块认领报错", () => {
+ expect(Object.keys(collectOutboxRoutes([...SERVER_MODULES, pingModule], fakeCtx))).toEqual(["release.resolved"]);
+ expect(() => collectOutboxRoutes([pingModule, { ...pingModule, name: "dup" }], fakeCtx)).toThrow(/重复认领/);
+ });
+
+ it("引擎就绪前调用给出明确错误", async () => {
+ const e = new DeferredTaskEngine();
+ expect(e.ready).toBe(false);
+ expect(() => e.get("x")).toThrow(/尚未就绪/);
+ await e.stop();
+ });
+
+ it("能力规格取自 contracts,可按模块注册", () => {
+ expect(CAPABILITY_IDS.map((id) => capabilitySpec(id).id)).toEqual([...CAPABILITY_IDS]);
+ });
+});
diff --git a/docs/runbook/dev.md b/docs/runbook/dev.md
index f31d053a..0d8b7841 100644
--- a/docs/runbook/dev.md
+++ b/docs/runbook/dev.md
@@ -48,7 +48,9 @@ curl -N -H "Authorization: Bearer umt-dev-admin-key" "localhost:4380/api/v1/even
`umt-server --role api|worker|all`(源码:`deno run ... apps/server/src/main.ts --role ...`)。`all` 下 worker 心跳会发布到进程内事件总线,SSE 可见;分进程部署时事件总线需改为 PG 投影(W2)。
-一次性子命令(容器入口用):`--role migrate`(迁移到最新,迁移文件已编进二进制)、`--role preflight`(启动自检:生产环境拒绝占位符密钥与 root、DB 可达、`WEB_ROOT/index.html` 存在)、`--role probe`(打本机 `/api/v1/health`,HEALTHCHECK 用)。
+启动时 `bootstrap()` 装配 `apps/server/src/modules/` 下的四个业务模块(见 apps/server/README.md),并连接任务引擎;`api` 角色也连接引擎(入队/信号/取消),`worker`/`all` 角色额外启动引擎 worker 与 OutboxRelay。数据库不可达时进程启动失败并退出。
+
+一次性子命令(容器入口用):`--role migrate`(迁移到最新,迁移文件已编进二进制;随后迁移任务引擎的 openworkflow schema)、`--role preflight`(启动自检:生产环境拒绝占位符密钥与 root、DB 可达、`WEB_ROOT/index.html` 存在)、`--role probe`(打本机 `/api/v1/health`,HEALTHCHECK 用)。
设置 `WEB_ROOT` 后 API 进程在同一端口托管 web 构建产物:存在的文件直接返回,其余非 `/api` 路径回退 `index.html`。
diff --git a/packages/tasks/README.md b/packages/tasks/README.md
index a9f35537..39a10c5e 100644
--- a/packages/tasks/README.md
+++ b/packages/tasks/README.md
@@ -15,6 +15,7 @@
| `OpenWorkflowEngine.connect({ databaseUrl, db, registry })` | `TaskEngine` 实现:`enqueue / signal / cancel / get / list / startWorker / stop` |
| `defineTask` / `TaskRegistry` / `createDefaultRegistry()` | 任务定义与注册表;默认注册表里有 §3.1-8 的 9 个任务(目前是空壳) |
| `OutboxRelay` | 用 `FOR UPDATE SKIP LOCKED` 认领 `outbox_event` 中 `status=pending` 且 `next_attempt_at` 已到的行,按 type 路由后 enqueue(幂等键取行的 `idempotency_key`,缺省 `outbox:`),成功标记 `dispatched` + `published_at`;失败 attempts+1 指数退避(上限 5 分钟),达 `maxAttempts` 置 `dead`;多进程并发时不会重复 |
+| `migrateEngineSchema(databaseUrl, schema?)` | 只跑引擎自己的迁移后断开(部署时 `--role migrate`) |
| `withAttemptToken(db, opKey, fn, { commit })` | 操作账本 fencing,表是 `tasks_op_ledger` |
| 表 | `tasks_run_meta`、`tasks_fact`、`tasks_op_ledger` 由 `packages/db` 迁移 `0013_tasks_tables` 建立(类型见 `@umt/db` 的 `DB`);本包不再自带迁移,只通过注入的 Kysely 访问 |
@@ -50,7 +51,7 @@
- 引擎幂等键只在 24h 内有效,而且不比较输入。长期去重靠业务表的唯一约束。
- `priority` 只记录在 `tasks_run_meta` 和 run 输入里,引擎仍按 `available_at` 先后认领,不支持真正的优先级调度。
- 取消只阻止后续步骤,不撤销已完成的外部动作。在已取消的 run 上,handler 抛出的错误会被引擎拒收,只在 worker 日志里出现。
-- `runMigrations` 默认 true(幂等)。等并入统一迁移后,启动时改成 false。
+- 引擎表(openworkflow schema)由 `migrateEngineSchema(databaseUrl)` 迁移:server 的 `--role migrate` 调用它;`--role all` 单进程连接时顺带迁移,`api`/`worker` 分进程以 `runMigrations:false` 连接。
## 测试
diff --git a/packages/tasks/src/index.ts b/packages/tasks/src/index.ts
index ada51add..fcc20859 100644
--- a/packages/tasks/src/index.ts
+++ b/packages/tasks/src/index.ts
@@ -36,6 +36,7 @@ export {
} from "./op-ledger.ts";
export {
ENGINE_SCHEMA,
+ migrateEngineSchema,
OpenWorkflowEngine,
type OpenWorkflowEngineOptions,
signalName,
diff --git a/packages/tasks/src/openworkflow-engine.ts b/packages/tasks/src/openworkflow-engine.ts
index 2b95300d..cee64ba8 100644
--- a/packages/tasks/src/openworkflow-engine.ts
+++ b/packages/tasks/src/openworkflow-engine.ts
@@ -41,7 +41,7 @@ export interface OpenWorkflowEngineOptions {
registry: TaskRegistry;
/** 固定 "openworkflow";只允许测试覆盖 */
schema?: string;
- /** 默认 true(幂等)。W2 并入统一迁移后启动时传 false */
+ /** 默认 true(幂等)。部署时由 migrateEngineSchema 预先迁移,多进程启动传 false */
runMigrations?: boolean;
logSink?: LogSink;
/** handler 执行期间轮询取消标志的间隔,默认 1000ms */
@@ -50,6 +50,15 @@ export interface OpenWorkflowEngineOptions {
export const signalName = (runId: string, name: string) => `umt:${runId}:${name}`;
+/**
+ * 只跑引擎自己的迁移(openworkflow schema)后断开。部署时由 `--role migrate` 调用,
+ * 之后 api / worker 进程以 runMigrations:false 连接,避免多进程同时迁移。
+ */
+export async function migrateEngineSchema(databaseUrl: string, schema = ENGINE_SCHEMA): Promise {
+ const backend = await BackendPostgres.connect(databaseUrl, { schema, runMigrations: true });
+ await backend.stop();
+}
+
export class OpenWorkflowEngine implements TaskEngine {
private readonly ow: OpenWorkflow;
private readonly workers: Worker[] = [];
diff --git a/packages/tasks/src/registry.ts b/packages/tasks/src/registry.ts
index c4a0b7ce..a2ecf09b 100644
--- a/packages/tasks/src/registry.ts
+++ b/packages/tasks/src/registry.ts
@@ -54,6 +54,11 @@ export class TaskRegistry {
return byVersion.get(Math.max(...byVersion.keys()))!;
}
+ /** 是否已注册该任务(任一版本) */
+ has(name: string): boolean {
+ return (this.defs.get(name)?.size ?? 0) > 0;
+ }
+
names(): string[] {
return [...this.defs.keys()].sort();
}
diff --git a/packages/tasks/test/registry.test.ts b/packages/tasks/test/registry.test.ts
index 38a8356a..9c23f37a 100644
--- a/packages/tasks/test/registry.test.ts
+++ b/packages/tasks/test/registry.test.ts
@@ -7,6 +7,8 @@ describe("TaskRegistry", () => {
const reg = createDefaultRegistry();
expect(reg.names()).toEqual([...TASK_NAMES].sort());
for (const n of TASK_NAMES) expect(reg.latest(n).version).toBe(1);
+ expect(reg.has("ingestFeed")).toBe(true);
+ expect(reg.has("nope")).toBe(false);
});
it("latest 取最高版本;select 按 name@version / @version 过滤", () => {