rootandClaude Opus 5.5
2ce1841628
Co-Authored-By: Claude Opus 5.5 <[email protected]> Claude-Session: https://claude.ai/code/session_01Cxwgnn2pbzQ7rXv3rq7fei
594 lines
27 KiB
TypeScript
594 lines
27 KiB
TypeScript
import { mkdtempSync, rmSync } from "node:fs";
|
|
import { tmpdir } from "node:os";
|
|
import { join } from "node:path";
|
|
import type { DiskUsage } from "@umt/adapters";
|
|
import { SystemHealth, SystemHealthHistory, SystemResources } from "@umt/contracts";
|
|
import { sql } from "@umt/db";
|
|
import { afterAll, beforeAll, describe, expect, it } from "vitest";
|
|
import { inferTriggeredBy, triggeredByFromActor } from "../../src/modules/auth-settings/engine-business.ts";
|
|
import { scheduleRegistry } from "../../src/modules/schedules/registry.ts";
|
|
import { diskLevel, processLevel, splitProcesses, worst } from "../../src/modules/system/health.ts";
|
|
import {
|
|
HEARTBEAT_STALE_MS,
|
|
markHeartbeatStopped,
|
|
pruneHeartbeats,
|
|
startHeartbeatWriter,
|
|
writeHeartbeat,
|
|
} from "../../src/modules/system/heartbeat.ts";
|
|
import { createSystemModule } from "../../src/modules/system/index.ts";
|
|
import { createResourceSampler } from "../../src/modules/system/resources.ts";
|
|
import type { ServerContext } from "../../src/modules/types.ts";
|
|
import { bearer, createHarness, type Harness, pgAvailable } from "../auth-settings/harness.ts";
|
|
import { DEV_KEY, json } from "../helpers.ts";
|
|
|
|
// W9-I 进程与健康 + 引擎 runs 业务字段(真实 PG;磁盘 / 哨兵检查注入假实现)。
|
|
|
|
const hasPg = await pgAvailable();
|
|
|
|
describe("纯函数", () => {
|
|
it("档位合并忽略 unknown;磁盘阈值", () => {
|
|
expect(worst(["ok", "unknown", "warn"])).toBe("warn");
|
|
expect(worst(["unknown"])).toBe("ok");
|
|
expect(worst(["warn", "critical", "ok"])).toBe("critical");
|
|
const t = { warnPercent: 85, criticalPercent: 90 };
|
|
expect(diskLevel(0.5, t)).toBe("ok");
|
|
expect(diskLevel(0.86, t)).toBe("warn");
|
|
expect(diskLevel(0.9, t)).toBe("critical");
|
|
});
|
|
|
|
it("进程档位:stale / 无 worker → warn;延迟 ≥5s → critical", () => {
|
|
const p = (kind: "api" | "worker", o: { stale?: boolean; lag?: number } = {}) =>
|
|
({
|
|
instanceId: `${kind}:h:1`,
|
|
processKind: kind,
|
|
pid: 1,
|
|
host: "h",
|
|
version: "t",
|
|
startedAt: new Date().toISOString(),
|
|
lastBeatAt: new Date().toISOString(),
|
|
beatAgeSec: 0,
|
|
stale: o.stale ?? false,
|
|
stoppedAt: null,
|
|
eventLoopLagMs: 0,
|
|
eventLoopLagMaxMs: o.lag ?? 0,
|
|
concurrency: {},
|
|
lanes: [] as never[],
|
|
runningRuns: null,
|
|
sseConnections: null,
|
|
extra: {},
|
|
}) as const;
|
|
expect(processLevel([p("api"), p("worker")])).toBe("ok");
|
|
expect(processLevel([p("api")])).toBe("warn");
|
|
expect(processLevel([p("api"), p("worker", { stale: true })])).toBe("warn");
|
|
expect(processLevel([p("api"), p("worker", { lag: 1200 })])).toBe("warn");
|
|
expect(processLevel([p("api"), p("worker", { lag: 6000 })])).toBe("critical");
|
|
});
|
|
|
|
it("当前实例 / 已下线实例:host 10 分钟无心跳、已退出、被同 host 新进程替换的都算下线", () => {
|
|
const now = new Date("2026-09-28T12:00:00Z");
|
|
const ago = (sec: number) => new Date(now.getTime() - sec * 1000).toISOString();
|
|
const p = (id: string, kind: "api" | "worker" | "worker:agent", host: string, beatAgo: number, o = {}) => ({
|
|
instanceId: id,
|
|
processKind: kind,
|
|
pid: 1,
|
|
host,
|
|
version: "t",
|
|
startedAt: ago(beatAgo + 600),
|
|
lastBeatAt: ago(beatAgo),
|
|
beatAgeSec: beatAgo,
|
|
stale: beatAgo > 45,
|
|
stoppedAt: null as string | null,
|
|
eventLoopLagMs: 0,
|
|
eventLoopLagMaxMs: 0,
|
|
concurrency: {},
|
|
lanes: [] as never[],
|
|
runningRuns: null,
|
|
sseConnections: null,
|
|
extra: {},
|
|
...o,
|
|
});
|
|
const rows = [
|
|
// 当前容器 3 个进程,心跳正常
|
|
p("api:new-a:1", "api", "new-a", 5),
|
|
p("worker:new-w:1", "worker", "new-w", 5),
|
|
p("worker:agent:new-g:1", "worker:agent", "new-g", 5),
|
|
// 容器重建前的 3 行,最后心跳 40 分钟前
|
|
p("api:old-a:1", "api", "old-a", 2400),
|
|
p("worker:old-w:1", "worker", "old-w", 2400),
|
|
p("worker:agent:old-g:1", "worker:agent", "old-g", 2400),
|
|
// 同容器进程重启:旧 pid 行 2 分钟没心跳、启动更早
|
|
p("worker:new-w:0", "worker", "new-w", 120, { startedAt: ago(7200) }),
|
|
// 正常退出
|
|
p("api:new-a:0", "api", "new-a", 20, { startedAt: ago(9000), stoppedAt: ago(20) }),
|
|
];
|
|
const r = splitProcesses(rows, now);
|
|
expect(r.processes.map((x) => x.instanceId).sort()).toEqual([
|
|
"api:new-a:1",
|
|
"worker:agent:new-g:1",
|
|
"worker:new-w:1",
|
|
]);
|
|
expect(r.retiredProcesses.map((x) => x.instanceId)).toEqual([
|
|
"api:new-a:0",
|
|
"worker:new-w:0",
|
|
"api:old-a:1",
|
|
"worker:old-w:1",
|
|
"worker:agent:old-g:1",
|
|
]);
|
|
expect(processLevel(r.processes)).toBe("ok");
|
|
|
|
// 当前容器里的 worker 停了心跳(1 分钟):仍是当前实例,stale 告警
|
|
const hung = splitProcesses([p("api:a:1", "api", "a", 5), p("worker:w:1", "worker", "w", 60)], now);
|
|
expect(hung.processes).toHaveLength(2);
|
|
expect(processLevel(hung.processes)).toBe("warn");
|
|
// 退出后立即不算活进程
|
|
const gone = splitProcesses(
|
|
[p("api:a:1", "api", "a", 5), p("worker:w:1", "worker", "w", 3, { stoppedAt: ago(1) })],
|
|
now,
|
|
);
|
|
expect(gone.processes.map((x) => x.instanceId)).toEqual(["api:a:1"]);
|
|
expect(gone.retiredProcesses.map((x) => x.instanceId)).toEqual(["worker:w:1"]);
|
|
});
|
|
|
|
it("system.heartbeat_cleanup:1 小时,维护组", async () => {
|
|
const ctx = {} as ServerContext;
|
|
await createSystemModule({ apiHeartbeat: false }).start?.(ctx);
|
|
expect(scheduleRegistry(ctx).get("system.heartbeat_cleanup")?.defaults).toMatchObject({
|
|
group: "maintain",
|
|
schedule: { type: "interval", everyMs: 3_600_000 },
|
|
});
|
|
const off = {} as ServerContext;
|
|
await createSystemModule({ apiHeartbeat: false, cleanupIntervalMs: 0 }).start?.(off);
|
|
expect(scheduleRegistry(off).get("system.heartbeat_cleanup")).toBeUndefined();
|
|
});
|
|
|
|
it("触发来源推断", () => {
|
|
expect(triggeredByFromActor("api_key:abc")).toBe("api_key");
|
|
expect(triggeredByFromActor("admin_session:u1")).toBe("user");
|
|
expect(triggeredByFromActor("system")).toBeNull();
|
|
const base = {
|
|
kind: "download" as const,
|
|
idempotencyKey: null,
|
|
taskTriggeredBy: null,
|
|
agentTrigger: null,
|
|
agentCreatedBy: null,
|
|
fromOutbox: false,
|
|
};
|
|
expect(inferTriggeredBy({ ...base, kind: "agent", agentTrigger: "subagent" })).toBe("agent");
|
|
expect(inferTriggeredBy({ ...base, kind: "agent", agentTrigger: "user", agentCreatedBy: "api_key:k" })).toBe(
|
|
"api_key",
|
|
);
|
|
expect(inferTriggeredBy({ ...base, taskTriggeredBy: "agent" })).toBe("agent");
|
|
expect(inferTriggeredBy({ ...base, idempotencyKey: "retry:r1:1" })).toBe("user");
|
|
expect(inferTriggeredBy({ ...base, idempotencyKey: "ingestFeed:manual:f:1" })).toBe("user");
|
|
expect(inferTriggeredBy({ ...base, idempotencyKey: "release.observed:x", fromOutbox: true })).toBe("event");
|
|
expect(inferTriggeredBy({ ...base, idempotencyKey: "snapshotSite:s1:123" })).toBe("schedule");
|
|
expect(inferTriggeredBy({ ...base, kind: "schedule", idempotencyKey: "purgeTrash:1" })).toBe("schedule");
|
|
expect(inferTriggeredBy(base)).toBeNull();
|
|
});
|
|
});
|
|
|
|
describe.skipIf(!hasPg)("W9-I 进程与健康(真实 PG)", () => {
|
|
let h: Harness;
|
|
let admin: ReturnType<typeof bearer>;
|
|
const dir = mkdtempSync(join(tmpdir(), "umt-w9i-"));
|
|
const libOk = join(dir, "lib-ok");
|
|
const usage = (path: string): Promise<DiskUsage> =>
|
|
Promise.resolve(
|
|
path.includes("full")
|
|
? {
|
|
status: "ok",
|
|
path,
|
|
totalBytes: 1000,
|
|
freeBytes: 20,
|
|
availableBytes: 20,
|
|
usedBytes: 980,
|
|
usedRatio: 0.98,
|
|
}
|
|
: path.includes("hang")
|
|
? new Promise<DiskUsage>(() => {})
|
|
: {
|
|
status: "ok",
|
|
path,
|
|
totalBytes: 1000,
|
|
freeBytes: 500,
|
|
availableBytes: 500,
|
|
usedBytes: 500,
|
|
usedRatio: 0.5,
|
|
},
|
|
);
|
|
|
|
beforeAll(async () => {
|
|
h = await createHarness({
|
|
engine: true,
|
|
extraModules: [],
|
|
});
|
|
// harness 在 createHarness 时已建 app;这里再建一个挂了 system 模块的 app(同一 ctx / db)
|
|
const { createApp } = await import("../../src/http/app.ts");
|
|
const { createAuthSettingsModule } = await import("../../src/modules/auth-settings/index.ts");
|
|
h.app = createApp({
|
|
config: h.ctx.config,
|
|
logger: h.ctx.logger,
|
|
bus: h.bus,
|
|
db: h.db,
|
|
role: "all",
|
|
version: "0.0.0-test",
|
|
modules: [
|
|
createAuthSettingsModule({ setupTokenOnStart: false, engineSchema: h.engineSchema }),
|
|
createSystemModule({
|
|
engineSchema: h.engineSchema,
|
|
apiHeartbeat: false,
|
|
healthDeps: { dataDir: dir, diskUsage: usage, ioTimeoutMs: 200 },
|
|
// 资源采样:只给 meminfo 与 cgroup cpu.stat,其余读不到 → 降级为 null
|
|
resourceSampler: createResourceSampler({
|
|
autoStart: false,
|
|
read: (path) => {
|
|
if (path === "/proc/meminfo") return Promise.resolve("MemTotal: 1000 kB\nMemAvailable: 250 kB\n");
|
|
if (path === "/sys/fs/cgroup/cpu.stat")
|
|
return Promise.resolve(`usage_usec ${Math.round(performance.now() * 500)}\n`);
|
|
return Promise.resolve(null);
|
|
},
|
|
}),
|
|
}),
|
|
],
|
|
ctx: h.ctx,
|
|
}) as Harness["app"];
|
|
admin = bearer(h.app, DEV_KEY);
|
|
await Deno.mkdir(libOk, { recursive: true });
|
|
await Deno.writeTextFile(join(libOk, ".umt-root"), "x");
|
|
}, 60_000);
|
|
afterAll(async () => {
|
|
await h?.close();
|
|
rmSync(dir, { recursive: true, force: true });
|
|
});
|
|
|
|
it("心跳写入:upsert 同一 instance_id + 追加采样;超过 45s 标 stale", async () => {
|
|
const w = startHeartbeatWriter({
|
|
db: h.db,
|
|
logger: h.ctx.logger,
|
|
kind: "worker",
|
|
host: "test-host",
|
|
pid: 4242,
|
|
intervalMs: 60_000,
|
|
lag: { take: () => ({ lastMs: 3, windowMaxMs: 12, lifetimeMaxMs: 40 }), stop() {} },
|
|
sample: () => ({ concurrency: { io: 4 }, runningRuns: 2, extra: { group: "io" } }),
|
|
});
|
|
await w.beat();
|
|
await w.beat();
|
|
expect(
|
|
(await h.db.selectFrom("process_heartbeat").select("stopped_at").executeTakeFirstOrThrow()).stopped_at,
|
|
).toBeNull();
|
|
await w.stop();
|
|
expect(w.instanceId).toBe("worker:test-host:4242");
|
|
const rows = await h.db.selectFrom("process_heartbeat").selectAll().execute();
|
|
expect(rows).toHaveLength(1);
|
|
// 正常退出:标 stopped_at
|
|
expect(rows[0]!.stopped_at).toBeInstanceOf(Date);
|
|
expect(rows[0]).toMatchObject({
|
|
process_kind: "worker",
|
|
pid: 4242,
|
|
event_loop_lag_ms: 3,
|
|
event_loop_lag_max_ms: 12,
|
|
running_runs: 2,
|
|
concurrency: { io: 4 },
|
|
});
|
|
expect(rows[0]!.extra).toMatchObject({ group: "io", lifetimeMaxLagMs: 40 });
|
|
const samples = await h.db.selectFrom("process_heartbeat_sample").selectAll().execute();
|
|
expect(samples.length).toBeGreaterThanOrEqual(2);
|
|
|
|
// 一个 api 进程 60s 前的心跳 → stale
|
|
await writeHeartbeat(h.db, {
|
|
instanceId: "api:old-host:1",
|
|
kind: "api",
|
|
pid: 1,
|
|
host: "old-host",
|
|
version: "0.1.0",
|
|
startedAt: new Date(Date.now() - 3600_000),
|
|
lagMs: 1,
|
|
lagMaxMs: 2,
|
|
sample: { sseConnections: 3 },
|
|
at: new Date(Date.now() - HEARTBEAT_STALE_MS - 15_000),
|
|
});
|
|
|
|
const res = await admin("GET", "/system/health");
|
|
expect(res.status).toBe(200);
|
|
const d = SystemHealth.parse((await json(res)).data);
|
|
const byId = new Map(d.processes.map((p) => [p.instanceId, p]));
|
|
// 已退出的 worker 立即归入已下线实例,不再当活进程
|
|
expect(byId.has("worker:test-host:4242")).toBe(false);
|
|
expect(d.retiredProcesses.find((p) => p.instanceId === "worker:test-host:4242")).toMatchObject({
|
|
runningRuns: 2,
|
|
concurrency: { io: 4 },
|
|
stoppedAt: expect.any(String),
|
|
});
|
|
// 60s 前的心跳,host 仍在 10 分钟窗口内 → 当前实例,stale
|
|
expect(byId.get("api:old-host:1")).toMatchObject({ stale: true, sseConnections: 3, stoppedAt: null });
|
|
// stale 进程的 SSE 不计入
|
|
expect(d.sse.connections).toBe(0);
|
|
});
|
|
|
|
it("资源监控:GET /system/resources 给当前值与趋势,读不到的字段为 null", async () => {
|
|
const res = await admin("GET", "/system/resources?minutes=30");
|
|
expect(res.status).toBe(200);
|
|
const d = SystemResources.parse((await json(res)).data);
|
|
expect(d.memory).toMatchObject({ scope: "host", usedBytes: 750 * 1024, percent: 75 });
|
|
expect(d.cpu.scope).toBe("cgroup");
|
|
expect(d.cpu.percent).not.toBeNull();
|
|
expect(d.network).toMatchObject({ scope: "none", rxBps: null });
|
|
expect(d.diskIo.scope).toBe("none");
|
|
expect(d.samples.length).toBeGreaterThanOrEqual(1);
|
|
expect((await admin("GET", "/system/resources?minutes=1")).status).toBe(422);
|
|
});
|
|
|
|
it("心跳带上本进程 CPU 占用", async () => {
|
|
const w = startHeartbeatWriter({
|
|
db: h.db,
|
|
logger: h.ctx.logger,
|
|
kind: "worker:agent",
|
|
host: "cpu-host",
|
|
pid: 7,
|
|
intervalMs: 60_000,
|
|
lag: { take: () => ({ lastMs: 0, windowMaxMs: 0, lifetimeMaxMs: 0 }), stop() {} },
|
|
});
|
|
await w.beat();
|
|
await new Promise((r) => setTimeout(r, 20));
|
|
await w.beat();
|
|
const row = await h.db
|
|
.selectFrom("process_heartbeat")
|
|
.select("extra")
|
|
.where("instance_id", "=", w.instanceId)
|
|
.executeTakeFirstOrThrow();
|
|
expect(row.extra).toMatchObject({ cpuPercent: expect.any(Number), rssBytes: expect.any(Number) });
|
|
await w.stop();
|
|
await sql`delete from process_heartbeat where instance_id = ${w.instanceId}`.execute(h.db);
|
|
await sql`delete from process_heartbeat_sample where instance_id = ${w.instanceId}`.execute(h.db);
|
|
});
|
|
|
|
it("health 各段形状:DB / 磁盘阈值 / 挂载根哨兵 / 引擎 / 定时任务", async () => {
|
|
await sql`insert into library_root (name, local_path, sentinel_file) values
|
|
('电影', ${libOk}, '.umt-root'), ('剧集(掉线)', ${join(dir, "missing")}, null)`.execute(h.db);
|
|
await sql`insert into downloader_instance (name, base_url) values ('qb1', 'http://qb')`.execute(h.db);
|
|
await sql`insert into path_mapping (view_kind, downloader_instance_id, remote_prefix, local_prefix, priority)
|
|
select 'downloader', id, '/downloads/full', ${join(dir, "full")}, 0 from downloader_instance where name = 'qb1'`.execute(
|
|
h.db,
|
|
);
|
|
const t = (n: string) => sql.raw(`"${h.engineSchema}".${n}`);
|
|
const run = (id: string, status: string, name: string, input: unknown, at: Date, finished: Date | null = null) =>
|
|
sql`insert into ${t("workflow_runs")} (namespace_id, id, workflow_name, version, status, config, context, input, attempts,
|
|
worker_id, available_at, started_at, finished_at, created_at, updated_at)
|
|
values ('default', ${id}, ${name}, 'v1', ${status}, '{}'::jsonb, null, ${JSON.stringify(input)}::jsonb, 1,
|
|
${status === "running" ? "w1" : null}, ${at}, ${at}, ${finished}, ${at}, ${at})`.execute(h.db);
|
|
const now = Date.now();
|
|
await run("h1", "pending", "ingestFeed", { input: {} }, new Date(now - 600_000));
|
|
await run("h2", "failed", "ingestFeed", { input: {} }, new Date(now - 3600_000), new Date(now - 3500_000));
|
|
await run("h3", "completed", "ingestFeed", { input: {} }, new Date(now - 3600_000), new Date(now - 3500_000));
|
|
await sql`insert into scheduled_job (id, name, board_group, kind, enabled, default_enabled, schedule, default_schedule,
|
|
last_status, last_error, last_run_at)
|
|
values ('feeds.poll', 'Feed 抓取', 'collect', 'task', true, true, '{"type":"interval","everyMs":60000}'::jsonb,
|
|
'{"type":"interval","everyMs":60000}'::jsonb, 'failed', 'boom', now())`.execute(h.db);
|
|
await sql`insert into scheduled_job_run (job_id, trigger, status, error) values ('feeds.poll', 'schedule', 'failed', 'boom')`.execute(
|
|
h.db,
|
|
);
|
|
|
|
const d = SystemHealth.parse((await json(await admin("GET", "/system/health"))).data);
|
|
expect(d.db).toMatchObject({ level: expect.any(String), error: null });
|
|
expect(d.db.latencyMs).toBeGreaterThanOrEqual(0);
|
|
expect(d.db.sizeBytes).toBeGreaterThan(0);
|
|
expect(d.db.connections!.max).toBeGreaterThan(0);
|
|
|
|
expect(d.diskThresholds).toEqual({ warnPercent: 85, criticalPercent: 90 });
|
|
const disk = (scope: string, name?: string) => d.disks.find((x) => x.scope === scope && (!name || x.name === name));
|
|
expect(disk("data_dir")).toMatchObject({ path: dir, level: "ok", usedRatio: 0.5 });
|
|
expect(disk("download_dir")).toMatchObject({ name: "qb1 /downloads/full", level: "critical", usedRatio: 0.98 });
|
|
expect(disk("library_root", "电影")).toMatchObject({ level: "ok" });
|
|
|
|
const mount = (name: string) => d.mounts.find((m) => m.name === name);
|
|
expect(mount("电影")).toMatchObject({
|
|
kind: "library_root",
|
|
status: "online",
|
|
reason: null,
|
|
sentinelChecked: true,
|
|
});
|
|
expect(mount("剧集(掉线)")).toMatchObject({ status: "offline", reason: "root_missing" });
|
|
expect(mount("qb1 /downloads/full")).toMatchObject({
|
|
kind: "path_mapping",
|
|
status: "offline",
|
|
sentinelChecked: false,
|
|
});
|
|
|
|
expect(d.engine).toMatchObject({ pending: 1, finished24h: 2, failed24h: 1, failureRate24h: 0.5, level: "warn" });
|
|
expect(d.engine.oldestPendingAgeSec).toBeGreaterThanOrEqual(590);
|
|
expect(d.schedules).toMatchObject({ level: "warn", enabled: 1, failedRuns24h: 1 });
|
|
expect(d.schedules.failing).toEqual([
|
|
expect.objectContaining({ id: "feeds.poll", name: "Feed 抓取", lastError: "boom" }),
|
|
]);
|
|
expect(d.level).toBe("critical");
|
|
expect(d.api.version).toMatch(/\d+\.\d+\.\d+/);
|
|
});
|
|
|
|
it("磁盘 statfs 卡死 → timeout 标 critical,不拖住请求", async () => {
|
|
await sql`insert into library_root (name, local_path) values ('卡死', ${join(dir, "hang")})`.execute(h.db);
|
|
const t0 = Date.now();
|
|
const d = SystemHealth.parse((await json(await admin("GET", "/system/health"))).data);
|
|
expect(Date.now() - t0).toBeLessThan(3000);
|
|
expect(d.disks.find((x) => x.name === "卡死")).toMatchObject({ level: "critical", error: "timeout" });
|
|
await sql`delete from library_root where name = '卡死'`.execute(h.db);
|
|
});
|
|
|
|
it("history:按桶聚合的时间序列;hours 越界 422", async () => {
|
|
const res = await admin("GET", "/system/health/history?hours=1");
|
|
expect(res.status).toBe(200);
|
|
const d = SystemHealthHistory.parse((await json(res)).data);
|
|
expect(d.bucketSec).toBe(15);
|
|
const s = d.series.find((x) => x.instanceId === "worker:test-host:4242");
|
|
expect(s?.processKind).toBe("worker");
|
|
expect(s!.points[0]).toMatchObject({ lagMaxMs: 12, runningRuns: 2 });
|
|
expect((await admin("GET", "/system/health/history?hours=500")).status).toBe(422);
|
|
expect((await json(await admin("GET", "/system/health/history?hours=168"))).data.bucketSec).toBe(1680);
|
|
});
|
|
|
|
it("引擎 runs 补 business 字段 + businessKind / businessRefId 筛选", async () => {
|
|
const t = (n: string) => sql.raw(`"${h.engineSchema}".${n}`);
|
|
const at = new Date("2026-09-10T00:00:00Z");
|
|
const run = (id: string, name: string, input: unknown, sec: number) =>
|
|
sql`insert into ${t("workflow_runs")} (namespace_id, id, workflow_name, version, status, config, context, input, attempts,
|
|
available_at, created_at, updated_at)
|
|
values ('default', ${id}, ${name}, 'v1', 'completed', '{}'::jsonb, null, ${JSON.stringify(input)}::jsonb, 1,
|
|
${at}, ${new Date(at.getTime() + sec * 1000)}, ${at})`.execute(h.db);
|
|
const meta = (runId: string, name: string, key: string) =>
|
|
sql`insert into tasks_run_meta (run_id, task_name, task_version, idempotency_key) values (${runId}, ${name}, 1, ${key})`.execute(
|
|
h.db,
|
|
);
|
|
|
|
const feed = await sql<{
|
|
id: string;
|
|
}>`insert into feed_source (name, url_display) values ('HDSky RSS', 'https://x/rss')
|
|
returning id::text`.execute(h.db);
|
|
const feedId = feed.rows[0]!.id;
|
|
const ar = await sql<{ id: string }>`insert into agent_run (trigger, goal, created_by)
|
|
values ('schedule', '巡检:待处理', 'schedule') returning id::text`.execute(h.db);
|
|
const agentId = ar.rows[0]!.id;
|
|
|
|
// b1 事件链触发的 Feed 抓取(outbox 幂等键)
|
|
await run("b1", "ingestFeed", { input: { feedSourceId: feedId }, traceId: "t" }, 1);
|
|
await meta("b1", "ingestFeed", "feed.due:x:1");
|
|
await sql`insert into outbox_event (type, payload, idempotency_key) values ('feed.due', '{}'::jsonb, 'feed.due:x:1')`.execute(
|
|
h.db,
|
|
);
|
|
// b2 Agent 会话(定时巡检)
|
|
await run("b2", "runAgent", { input: { agentRunId: agentId } }, 2);
|
|
await meta("b2", "runAgent", `agent_run:${agentId}`);
|
|
// b3 API Key 发起的整理(只有业务任务行能关联)
|
|
await run("b3", "verifyAndImport", { input: { taskId: "00000000-0000-0000-0000-000000000001" } }, 3);
|
|
await meta("b3", "verifyAndImport", "verifyAndImport:files:p1");
|
|
await sql`insert into task (kind, idempotency_key, input, subject_type, subject_id, triggered_by, engine_run_id)
|
|
values ('verifyAndImport', 'files:p1', '{}'::jsonb, 'organization_record', '00000000-0000-0000-0000-0000000000aa', 'api_key:k1', 'b3')`.execute(
|
|
h.db,
|
|
);
|
|
// b4 定时清理回收站(无业务 id)
|
|
await run("b4", "purgeTrash", { input: {} }, 4);
|
|
await meta("b4", "purgeTrash", "purgeTrash:slot-1");
|
|
await sql`insert into scheduled_job (id, name, board_group, kind, enabled, default_enabled, schedule, default_schedule)
|
|
values ('trash.purge', '清理回收站', 'maintain', 'task', true, true, '{"type":"interval","everyMs":60000}'::jsonb,
|
|
'{"type":"interval","everyMs":60000}'::jsonb)`.execute(h.db);
|
|
// b5 手动立即执行的清理 → 不算 schedule,也没有业务对象
|
|
await run("b5", "purgeTrash", { input: {} }, 5);
|
|
await meta("b5", "purgeTrash", "purgeTrash:manual:r1");
|
|
|
|
const list = await json(await admin("GET", "/engine/runs?from=2026-09-10T00:00:00Z&to=2026-09-11T00:00:00Z"));
|
|
const by = new Map(list.data.items.map((r: { id: string; business: unknown }) => [r.id, r.business]));
|
|
expect(by.get("b1")).toEqual({ kind: "feed", refId: feedId, title: "HDSky RSS", triggeredBy: "event" });
|
|
expect(by.get("b2")).toEqual({
|
|
kind: "agent",
|
|
refId: agentId,
|
|
title: "巡检:待处理",
|
|
triggeredBy: "schedule",
|
|
// AG-3:Agent 会话带业务结果(任务页据此显示超出预算 / 失败,而不是引擎的「完成」)
|
|
outcome: "queued",
|
|
});
|
|
expect(by.get("b3")).toEqual({
|
|
kind: "organization",
|
|
refId: "00000000-0000-0000-0000-0000000000aa",
|
|
title: null,
|
|
triggeredBy: "api_key",
|
|
});
|
|
expect(by.get("b4")).toEqual({
|
|
kind: "schedule",
|
|
refId: "trash.purge",
|
|
title: "清理回收站",
|
|
triggeredBy: "schedule",
|
|
});
|
|
expect(by.get("b5")).toBeNull();
|
|
|
|
const f = await json(await admin("GET", "/engine/runs?businessKind=feed"));
|
|
expect(f.data.items.map((r: { id: string }) => r.id)).toEqual(["b1"]);
|
|
const g = await json(await admin("GET", `/engine/runs?businessRefId=${agentId}`));
|
|
expect(g.data.items.map((r: { id: string }) => r.id)).toEqual(["b2"]);
|
|
const s = await json(await admin("GET", "/engine/runs?businessKind=schedule&businessRefId=trash.purge"));
|
|
expect(s.data.items.map((r: { id: string }) => r.id)).toEqual(["b4"]);
|
|
expect((await admin("GET", "/engine/runs?businessKind=bogus")).status).toBe(422);
|
|
|
|
// triggeredBy:服务端逐页推断后筛选;多值逗号分隔;limit 截断时游标接着扫
|
|
const range = "from=2026-09-10T00:00:00Z&to=2026-09-11T00:00:00Z";
|
|
const ids = (r: { data: { items: { id: string }[] } }) => r.data.items.map((x) => x.id);
|
|
expect(ids(await json(await admin("GET", `/engine/runs?${range}&triggeredBy=schedule`)))).toEqual(["b4", "b2"]);
|
|
expect(ids(await json(await admin("GET", `/engine/runs?${range}&triggeredBy=event,api_key`)))).toEqual([
|
|
"b3",
|
|
"b1",
|
|
]);
|
|
const t1 = await json(await admin("GET", `/engine/runs?${range}&triggeredBy=schedule&limit=1`));
|
|
expect(ids(t1)).toEqual(["b4"]);
|
|
expect(t1.data.nextCursor).toBeTruthy();
|
|
const t2 = await json(
|
|
await admin("GET", `/engine/runs?${range}&triggeredBy=schedule&limit=1&cursor=${t1.data.nextCursor}`),
|
|
);
|
|
expect(ids(t2)).toEqual(["b2"]);
|
|
const t3 = await json(
|
|
await admin("GET", `/engine/runs?${range}&triggeredBy=schedule&limit=1&cursor=${t2.data.nextCursor}`),
|
|
);
|
|
expect(ids(t3)).toEqual([]);
|
|
expect(t3.data.nextCursor).toBeNull();
|
|
expect((await admin("GET", "/engine/runs?triggeredBy=bogus")).status).toBe(422);
|
|
|
|
const detail = (await json(await admin("GET", "/engine/runs/b2"))).data;
|
|
expect(detail.business).toMatchObject({ kind: "agent", refId: agentId });
|
|
expect(detail.stepAttempts).toEqual([]);
|
|
});
|
|
it("容器重建留下的旧心跳行不计入告警;24 小时的行由清理任务删除;重启后 upsert 清掉 stopped_at", async () => {
|
|
await sql`delete from process_heartbeat`.execute(h.db);
|
|
const now = Date.now();
|
|
const beat = (id: string, kind: "api" | "worker" | "worker:agent", host: string, agoMs: number) =>
|
|
writeHeartbeat(h.db, {
|
|
instanceId: id,
|
|
kind,
|
|
pid: 1,
|
|
host,
|
|
version: "0.2.0",
|
|
startedAt: new Date(now - agoMs - 3600_000),
|
|
lagMs: 0,
|
|
lagMaxMs: 0,
|
|
sample: {},
|
|
at: new Date(now - agoMs),
|
|
});
|
|
await beat("api:c-api:1", "api", "c-api", 2_000);
|
|
await beat("worker:c-io:1", "worker", "c-io", 2_000);
|
|
await beat("worker:agent:c-ag:1", "worker:agent", "c-ag", 2_000);
|
|
await beat("api:o-api:1", "api", "o-api", 40 * 60_000);
|
|
await beat("worker:o-io:1", "worker", "o-io", 40 * 60_000);
|
|
await beat("worker:agent:o-ag:1", "worker:agent", "o-ag", 40 * 60_000);
|
|
await beat("worker:ancient:1", "worker", "ancient", 25 * 3600_000);
|
|
|
|
const d = SystemHealth.parse((await json(await admin("GET", "/system/health"))).data);
|
|
expect(d.processes.map((p) => p.instanceId).sort()).toEqual([
|
|
"api:c-api:1",
|
|
"worker:agent:c-ag:1",
|
|
"worker:c-io:1",
|
|
]);
|
|
expect(d.processes.every((p) => !p.stale)).toBe(true);
|
|
// 超过 24 小时的行不再返回
|
|
expect(d.retiredProcesses.map((p) => p.instanceId).sort()).toEqual([
|
|
"api:o-api:1",
|
|
"worker:agent:o-ag:1",
|
|
"worker:o-io:1",
|
|
]);
|
|
|
|
// 清理任务:删 24 小时没心跳的行
|
|
const r = await pruneHeartbeats(h.db);
|
|
expect(r.rows).toBe(1);
|
|
const left = await h.db.selectFrom("process_heartbeat").select("instance_id").execute();
|
|
expect(left.map((x) => x.instance_id)).not.toContain("worker:ancient:1");
|
|
expect(left).toHaveLength(6);
|
|
|
|
// 退出标记后再次心跳(同 instance_id 重启):stopped_at 清回 null
|
|
await markHeartbeatStopped(h.db, "worker:c-io:1");
|
|
const d2 = SystemHealth.parse((await json(await admin("GET", "/system/health"))).data);
|
|
expect(d2.processes.map((p) => p.instanceId)).not.toContain("worker:c-io:1");
|
|
await beat("worker:c-io:1", "worker", "c-io", 0);
|
|
const row = await h.db
|
|
.selectFrom("process_heartbeat")
|
|
.select("stopped_at")
|
|
.where("instance_id", "=", "worker:c-io:1")
|
|
.executeTakeFirstOrThrow();
|
|
expect(row.stopped_at).toBeNull();
|
|
});
|
|
});
|