From 473c5e94a50306da403e02d88fb919365e3125bc Mon Sep 17 00:00:00 2001 From: root Date: Sun, 27 Sep 2026 22:12:16 +0800 Subject: [PATCH] feat(server): io/agent worker groups, invoke limits for external IO capabilities, loop lag probe - WORKER_CONCURRENCY / WORKER_AGENT_CONCURRENCY; --role worker (io), worker:agent, all (both) - HTTP invoke of release.search/inspect, identity.resolve, catalog.lookup(remote): 15s timeout + per-upstream semaphore (site 2, TMDB 4) - worker event-loop lag probe: warn + count when drift > 5s Co-Authored-By: Claude Opus 5.5 --- apps/server/src/config.ts | 6 + apps/server/src/main.ts | 27 +++- .../server/src/modules/agent/invoke-limits.ts | 139 ++++++++++++++++++ apps/server/src/modules/agent/routes.ts | 26 +++- apps/server/src/modules/types.ts | 2 +- apps/server/src/worker/index.ts | 118 ++++++++++++++- apps/server/test/invoke-limits.test.ts | 100 +++++++++++++ 7 files changed, 399 insertions(+), 19 deletions(-) create mode 100644 apps/server/src/modules/agent/invoke-limits.ts create mode 100644 apps/server/test/invoke-limits.test.ts diff --git a/apps/server/src/config.ts b/apps/server/src/config.ts index 81248435..fa6f36f2 100644 --- a/apps/server/src/config.ts +++ b/apps/server/src/config.ts @@ -15,6 +15,12 @@ export const ConfigSchema = z.object({ DEV_ADMIN_KEY: z.string().min(8).optional(), /** worker 心跳间隔 */ WORKER_HEARTBEAT_MS: z.coerce.number().int().min(1000).default(30_000), + /** io 组 worker(除 runAgent/resolveDecision 外的全部任务)并发槽数 */ + WORKER_CONCURRENCY: z.coerce.number().int().min(1).max(64).default(4), + /** agent 组 worker(runAgent/resolveDecision:整段会话占一槽数分钟)并发槽数 */ + WORKER_AGENT_CONCURRENCY: z.coerce.number().int().min(1).max(16).default(2), + /** 手动整理 dryRun / 按路径整理时递归遍历的文件数上限(超出截断并标记 truncated) */ + ORGANIZE_WALK_MAX_FILES: z.coerce.number().int().min(1).max(1_000_000).default(5000), /** web 静态产物目录(单容器模式):设置后同端口托管 SPA,非 /api 路径回退 index.html */ WEB_ROOT: z.string().min(1).optional(), /** diff --git a/apps/server/src/main.ts b/apps/server/src/main.ts index f2d714bb..1cbd89af 100644 --- a/apps/server/src/main.ts +++ b/apps/server/src/main.ts @@ -12,11 +12,22 @@ import { diffCounts, enableReadOnlyReconcile, exportMappings, type MappingExport import { auditStorageWarnings, checkStorage, loadMountedPaths, runPreflight } from "./preflight.ts"; import { startWorker } from "./worker/index.ts"; -// api/worker/all 是常驻角色;其余是容器入口 / 备份恢复脚本用的一次性子命令: +// api / worker / worker:agent / all 是常驻角色(worker = io 组:除 runAgent/resolveDecision 外的任务 + outbox 中继 + +// 模块后台循环;worker:agent = 只跑 Agent 会话任务;all = api + 两组 worker 同进程);其余是容器入口 / 备份恢复脚本用的一次性子命令: // migrate | preflight | probe // export-mappings --out [--compare <基准清单.json>] 映射清单导出(可与基准比行数) // read-only-on [--reason <文本>] 打开 automation.readOnlyReconcile -const Roles = ["api", "worker", "all", "migrate", "preflight", "probe", "export-mappings", "read-only-on"] as const; +const Roles = [ + "api", + "worker", + "worker:agent", + "all", + "migrate", + "preflight", + "probe", + "export-mappings", + "read-only-on", +] as const; type Role = (typeof Roles)[number]; const OneShot = ["migrate", "preflight", "export-mappings", "read-only-on"] as const; type OneShotRole = (typeof OneShot)[number]; @@ -195,12 +206,20 @@ if (role === "api" || role === "all") { await server.finished; }); } -if (role === "worker" || role === "all") { +if (role === "worker" || role === "worker:agent" || role === "all") { const worker = await startWorker({ logger, heartbeatMs: config.WORKER_HEARTBEAT_MS, bus, - runtime: { ctx, modules: SERVER_MODULES, setRelay: outbox.setRelay }, + runtime: { + ctx, + modules: SERVER_MODULES, + setRelay: outbox.setRelay, + group: role === "all" ? "split" : role === "worker" ? "io" : "agent", + taskNames: booted.tasks.names(), + concurrency: config.WORKER_CONCURRENCY, + agentConcurrency: config.WORKER_AGENT_CONCURRENCY, + }, }); stops.push(worker.stop); } diff --git a/apps/server/src/modules/agent/invoke-limits.ts b/apps/server/src/modules/agent/invoke-limits.ts new file mode 100644 index 00000000..6f45f79f --- /dev/null +++ b/apps/server/src/modules/agent/invoke-limits.ts @@ -0,0 +1,139 @@ +// HTTP 路径上的能力调用分级(W4-D P1):外部 IO 型能力(远程站点 / TMDB)在请求内执行时 +// 强制 15s 超时 + 按上游限并发,避免慢站点把 API 进程的连接与事件循环拖住。 +// Agent 在 worker:agent 进程里调用同一能力不经过这里(由运行预算约束)。 + +/** 外部 IO 能力的 HTTP 超时 */ +export const EXTERNAL_INVOKE_TIMEOUT_MS = 15_000; + +/** 每个上游的并发上限:站点实时搜索/详情刷新、TMDB 查询 */ +export const UPSTREAM_LIMITS = { site: 2, tmdb: 4 } as const; +export type Upstream = keyof typeof UPSTREAM_LIMITS; + +/** + * 判断一次调用是否会打外部上游;返回上游名,null 表示纯本地读写(不限流、不加超时)。 + * 只看输入里确实会触发远程的开关,纯快照查询不受影响。 + */ +export function externalUpstream(capabilityId: string, input: unknown): Upstream | null { + const i = (input ?? {}) as Record; + switch (capabilityId) { + case "release.search": + return i.allowLiveSearch === true && typeof i.query === "string" && i.query.length > 0 ? "site" : null; + case "release.inspect": + return i.refreshDetail === true ? "site" : null; + case "identity.resolve": + return "tmdb"; + case "catalog.lookup": + return i.includeRemote === true ? "tmdb" : null; + default: + return null; + } +} + +export class UpstreamBusyError extends Error { + constructor(readonly upstream: Upstream) { + super(`${upstream === "tmdb" ? "TMDB" : "站点"}并发已满,请稍后重试`); + } +} + +export class UpstreamTimeoutError extends Error { + constructor(readonly upstream: Upstream) { + super(`调用${upstream === "tmdb" ? " TMDB" : "站点"}超时(${EXTERNAL_INVOKE_TIMEOUT_MS / 1000}s),已放弃等待`); + } +} + +/** 计数信号量:acquire 在 signal 触发前拿不到许可则抛 signal.reason */ +export class Semaphore { + private active = 0; + private readonly waiters: Array<() => void> = []; + + constructor(readonly limit: number) {} + + get inUse(): number { + return this.active; + } + + async acquire(signal: AbortSignal): Promise<() => void> { + if (this.active >= this.limit) { + await new Promise((resolve, reject) => { + const wake = () => { + signal.removeEventListener("abort", onAbort); + resolve(); + }; + const onAbort = () => { + const i = this.waiters.indexOf(wake); + if (i >= 0) this.waiters.splice(i, 1); + reject(signal.reason); + }; + if (signal.aborted) return reject(signal.reason); + this.waiters.push(wake); + signal.addEventListener("abort", onAbort, { once: true }); + }); + } else { + this.active += 1; + } + let released = false; + return () => { + if (released) return; + released = true; + const next = this.waiters.shift(); + // 许可直接转交给下一个等待者(active 不变) + if (next) next(); + else this.active -= 1; + }; + } +} + +export interface InvokeLimiter { + /** + * 包一层:外部 IO 型调用排队拿许可 + 超时。超时后 HTTP 立即返回,但许可要等底层调用真正结束才归还, + * 这样并发上限约束的是"实际在打上游的调用数"。 + */ + run( + capabilityId: string, + input: unknown, + parent: AbortSignal | undefined, + fn: (signal: AbortSignal | undefined) => Promise, + ): Promise; + inUse(upstream: Upstream): number; +} + +export function createInvokeLimiter( + opts: { timeoutMs?: number; limits?: Partial> } = {}, +): InvokeLimiter { + const timeoutMs = opts.timeoutMs ?? EXTERNAL_INVOKE_TIMEOUT_MS; + const sems = { + site: new Semaphore(opts.limits?.site ?? UPSTREAM_LIMITS.site), + tmdb: new Semaphore(opts.limits?.tmdb ?? UPSTREAM_LIMITS.tmdb), + }; + return { + inUse: (u) => sems[u].inUse, + async run(capabilityId, input, parent, fn) { + const upstream = externalUpstream(capabilityId, input); + if (!upstream) return fn(parent); + const timeout = AbortSignal.timeout(timeoutMs); + const signal = parent ? AbortSignal.any([parent, timeout]) : timeout; + let release: () => void; + try { + release = await sems[upstream].acquire(signal); + } catch (err) { + if (timeout.aborted) throw new UpstreamBusyError(upstream); + throw err; + } + const work = fn(signal).finally(release); + let onAbort: (() => void) | undefined; + const aborted = new Promise((_, reject) => { + onAbort = () => reject(timeout.aborted ? new UpstreamTimeoutError(upstream) : signal.reason); + if (signal.aborted) onAbort(); + else signal.addEventListener("abort", onAbort, { once: true }); + }); + // 超时后底层调用继续跑完(结果丢弃);两边都挂空 catch,避免未处理的 rejection + work.catch(() => {}); + aborted.catch(() => {}); + try { + return await Promise.race([work, aborted]); + } finally { + if (onAbort) signal.removeEventListener("abort", onAbort); + } + }, + }; +} diff --git a/apps/server/src/modules/agent/routes.ts b/apps/server/src/modules/agent/routes.ts index b095850c..a27be3ed 100644 --- a/apps/server/src/modules/agent/routes.ts +++ b/apps/server/src/modules/agent/routes.ts @@ -14,6 +14,7 @@ import type { ModuleApp, ServerContext } from "../types.ts"; import { planBinding } from "./authorize.ts"; import { cancelTask, retryTask, taskToApi } from "./capabilities.ts"; import { route } from "./http.ts"; +import { createInvokeLimiter, type InvokeLimiter, UpstreamBusyError, UpstreamTimeoutError } from "./invoke-limits.ts"; import { AgentRunError, abortRun, appendMessage, startRun } from "./service.ts"; // Agent 会话 API、持续授权 / 单次审批、能力目录与调用、业务任务。 @@ -85,7 +86,11 @@ const decode = (c?: string) => { return at && id ? { at, id } : null; }; -export function registerAgentRoutes(app: ModuleApp, ctx: ServerContext) { +export function registerAgentRoutes( + app: ModuleApp, + ctx: ServerContext, + limiter: InvokeLimiter = createInvokeLimiter(), +) { const db = () => ctx.db; // ── Agent 会话 ── @@ -328,13 +333,16 @@ export function registerAgentRoutes(app: ModuleApp, ctx: ServerContext) { if (body.preview && !entry.descriptor.supportsPreview) throw new ApiException("bad_request", `${entry.descriptor.id} 不支持预览`); try { - const output = await ctx.capabilities.invoke(params.capabilityId, body.input, { - actorId, - taskId: `http:${c.get("requestId") ?? "req"}`, - scopeIds: [], - signal: c.req.raw.signal, - source: "web", - }); + // 外部 IO 型能力(站点实时搜索/详情刷新、TMDB):15s 超时 + 按上游限并发 + const output = await limiter.run(params.capabilityId, body.input, c.req.raw.signal, (signal) => + ctx.capabilities.invoke(params.capabilityId, body.input, { + actorId, + taskId: `http:${c.get("requestId") ?? "req"}`, + scopeIds: [], + signal: signal ?? c.req.raw.signal, + source: "web", + }), + ); const o = (output ?? {}) as { taskId?: unknown; planId?: unknown; plan?: { id?: unknown } | null }; return { output, @@ -499,6 +507,8 @@ export function registerAgentRoutes(app: ModuleApp, ctx: ServerContext) { /** 能力错误 → HTTP 错误码 */ function capabilityHttpError(err: unknown): never { + if (err instanceof UpstreamBusyError) throw new ApiException("rate_limited", err.message); + if (err instanceof UpstreamTimeoutError) throw new ApiException("upstream_unavailable", err.message); if (!isCapabilityError(err)) throw err; const code = err.code === "unauthorized" diff --git a/apps/server/src/modules/types.ts b/apps/server/src/modules/types.ts index 9b4358f9..e3f60619 100644 --- a/apps/server/src/modules/types.ts +++ b/apps/server/src/modules/types.ts @@ -12,7 +12,7 @@ import type { EventBusPort } from "../http/sse.ts"; // 每个 W2 子任务是一个 ServerModule,只改 modules/<自己的目录>/ 与 contracts/db 中本域的文件。 // 跨模块只通过 DB 表 + outbox_event(事件类型在 contracts/events.ts)+ contracts 类型耦合,不互相 import。 -export type ServerRole = "api" | "worker" | "all"; +export type ServerRole = "api" | "worker" | "worker:agent" | "all"; /** * 模块路由挂载点:已挂在 /api/v1 下、已经过 API Key 鉴权(c.var.principal 可用)。 diff --git a/apps/server/src/worker/index.ts b/apps/server/src/worker/index.ts index 81f41079..4665d33a 100644 --- a/apps/server/src/worker/index.ts +++ b/apps/server/src/worker/index.ts @@ -5,6 +5,18 @@ import { collectOutboxRoutes, tasksLogSink } from "../context.ts"; import type { EventBusPort } from "../http/sse.ts"; import type { ServerContext, ServerModule } from "../modules/types.ts"; +/** agent 组任务:整段会话占一槽数分钟,与文件复制等 io 任务分池 */ +export const AGENT_TASKS = ["runAgent", "resolveDecision"] as const; + +/** + * worker 分组: + * all — 单个引擎 worker 执行全部任务(测试 / 省略分组时的兼容行为) + * io — 除 AGENT_TASKS 外的全部任务 + OutboxRelay + 模块后台循环 + * agent — 只执行 AGENT_TASKS;不跑 relay 与模块循环 + * split — 同一进程里 io + agent 两个引擎 worker(--role all) + */ +export type WorkerGroup = "all" | "io" | "agent" | "split"; + export interface WorkerDeps { logger: Logger; heartbeatMs: number; @@ -17,32 +29,117 @@ export interface WorkerDeps { /** 设置 outbox 中继实例到 ctx.outbox.relay */ setRelay?: (relay: OutboxRelay) => void; outboxIntervalMs?: number; + /** io 组(或 all 组)并发 */ concurrency?: number; + /** agent 组并发 */ + agentConcurrency?: number; + /** 缺省 all */ + group?: WorkerGroup; + /** 注册表里的全部任务名(io 组 = 全部 − AGENT_TASKS);group 为 io/split 时必填 */ + taskNames?: readonly string[]; }; + /** 事件循环延迟探针:采样间隔与告警阈值(测试可调小) */ + lagProbe?: { intervalMs?: number; warnMs?: number }; +} + +export interface WorkerStats { + /** 事件循环偏差超过阈值的次数 */ + loopLagWarnings: number; + /** 观测到的最大偏差(ms) */ + maxLoopLagMs: number; } export interface RunningWorker { stop(): Promise; + stats(): WorkerStats; +} + +/** 按分组拆出引擎 worker 的任务选择(versions 为任务名,匹配该任务全部版本) */ +export function workerPlan( + group: WorkerGroup, + taskNames: readonly string[], + concurrency: { io?: number; agent?: number }, +): { label: "all" | "io" | "agent"; versions?: string[]; concurrency?: number }[] { + const agentNames = taskNames.filter((n) => (AGENT_TASKS as readonly string[]).includes(n)); + const ioNames = taskNames.filter((n) => !(AGENT_TASKS as readonly string[]).includes(n)); + const io = { label: "io" as const, versions: ioNames, ...(concurrency.io ? { concurrency: concurrency.io } : {}) }; + const agent = { + label: "agent" as const, + versions: agentNames, + ...(concurrency.agent ? { concurrency: concurrency.agent } : {}), + }; + switch (group) { + case "all": + return [{ label: "all", ...(concurrency.io ? { concurrency: concurrency.io } : {}) }]; + case "io": + return ioNames.length ? [io] : []; + case "agent": + return agentNames.length ? [agent] : []; + case "split": + return [...(ioNames.length ? [io] : []), ...(agentNames.length ? [agent] : [])]; + } +} + +/** + * 事件循环延迟探针:setInterval 实际触发时刻与预期的偏差超过 warnMs 记一次 warn + 计数。 + * 偏差大说明有同步重活占住主线程(租约续期 15s、租约 30s,偏差逼近这个量级就有丢租约风险)。 + */ +export function startLoopLagProbe(log: Logger, opts: { intervalMs?: number; warnMs?: number } = {}) { + const intervalMs = opts.intervalMs ?? 1000; + const warnMs = opts.warnMs ?? 5000; + const stats: WorkerStats = { loopLagWarnings: 0, maxLoopLagMs: 0 }; + let expected = performance.now() + intervalMs; + const timer = setInterval(() => { + const now = performance.now(); + const lag = Math.max(0, now - expected); + expected = now + intervalMs; + if (lag > stats.maxLoopLagMs) stats.maxLoopLagMs = Math.round(lag); + if (lag > warnMs) { + stats.loopLagWarnings += 1; + log.warn({ lagMs: Math.round(lag), warnings: stats.loopLagWarnings }, "worker 事件循环延迟超过阈值"); + } + }, intervalMs); + return { + stats: () => ({ ...stats }), + stop: () => clearInterval(timer), + }; } /** * worker 角色:心跳 + 任务引擎 worker(执行全部已注册任务)+ OutboxRelay(按模块的 outboxRoutes 路由) * + 各模块的后台循环(start)。stop 逆序停止;引擎连接本身由进程退出时关闭。 */ -export async function startWorker({ logger, heartbeatMs, bus, runtime }: WorkerDeps): Promise { - const log = logger.child({ component: "worker" }); +export async function startWorker({ logger, heartbeatMs, bus, runtime, lagProbe }: WorkerDeps): Promise { + const group = runtime?.group ?? "all"; + const log = logger.child({ component: "worker", group }); + const probe = startLoopLagProbe(log, lagProbe); let n = 0; const beat = () => { n += 1; - log.debug({ n }, "worker heartbeat"); - bus?.publish(EventType.heartbeat, { role: "worker", n }); + const s = probe.stats(); + log.debug({ n, ...s }, "worker heartbeat"); + bus?.publish(EventType.heartbeat, { role: group === "agent" ? "worker:agent" : "worker", n, ...s }); }; const started: ServerModule[] = []; let relay: OutboxRelay | null = null; + const plan: ReturnType = []; if (runtime) { + const { ctx } = runtime; + if (group !== "all" && !runtime.taskNames) throw new Error(`worker 分组 ${group} 需要 taskNames`); + plan.push( + ...workerPlan(group, runtime.taskNames ?? [], { io: runtime.concurrency, agent: runtime.agentConcurrency }), + ); + for (const p of plan) { + await ctx.engine.startWorker({ + ...(p.versions ? { versions: p.versions } : {}), + ...(p.concurrency ? { concurrency: p.concurrency } : {}), + }); + } + } + // outbox 中继与模块后台循环只在 io 组(agent 组进程只执行会话任务) + if (runtime && group !== "agent") { const { ctx, modules } = runtime; - await ctx.engine.startWorker({ ...(runtime.concurrency ? { concurrency: runtime.concurrency } : {}) }); relay = new OutboxRelay({ db: ctx.db, engine: ctx.engine, @@ -58,12 +155,21 @@ export async function startWorker({ logger, heartbeatMs, bus, runtime }: WorkerD } } - log.info({ heartbeatMs, modules: started.map((m) => m.name) }, "worker started"); + log.info( + { + heartbeatMs, + modules: started.map((m) => m.name), + pools: plan.map((p) => ({ group: p.label, concurrency: p.concurrency ?? 4, tasks: p.versions?.length ?? "all" })), + }, + "worker started", + ); beat(); const timer = setInterval(beat, heartbeatMs); return { + stats: () => probe.stats(), async stop() { clearInterval(timer); + probe.stop(); for (const m of started.reverse()) { try { await m.stop?.(); diff --git a/apps/server/test/invoke-limits.test.ts b/apps/server/test/invoke-limits.test.ts new file mode 100644 index 00000000..eb0a70cb --- /dev/null +++ b/apps/server/test/invoke-limits.test.ts @@ -0,0 +1,100 @@ +import { describe, expect, it } from "vitest"; +import { createLogger } from "../src/logger.ts"; +import { + createInvokeLimiter, + externalUpstream, + Semaphore, + UpstreamBusyError, + UpstreamTimeoutError, +} from "../src/modules/agent/invoke-limits.ts"; +import { startLoopLagProbe, workerPlan } from "../src/worker/index.ts"; + +const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms)); + +describe("能力调用分级(HTTP 路径)", () => { + it("只把会打外部上游的调用归类", () => { + expect(externalUpstream("release.search", { query: "x", allowLiveSearch: true })).toBe("site"); + expect(externalUpstream("release.search", { query: "x", allowLiveSearch: false })).toBeNull(); + expect(externalUpstream("release.inspect", { refreshDetail: true })).toBe("site"); + expect(externalUpstream("release.inspect", {})).toBeNull(); + expect(externalUpstream("identity.resolve", {})).toBe("tmdb"); + expect(externalUpstream("catalog.lookup", { includeRemote: true })).toBe("tmdb"); + expect(externalUpstream("catalog.lookup", { includeRemote: false })).toBeNull(); + expect(externalUpstream("inventory.query", {})).toBeNull(); + }); + + it("超时:HTTP 立即放弃,许可等底层结束才归还", async () => { + const lim = createInvokeLimiter({ timeoutMs: 50, limits: { tmdb: 1 } }); + let finished = false; + const slow = lim.run("identity.resolve", {}, undefined, async () => { + await sleep(200); + finished = true; + return 1; + }); + await expect(slow).rejects.toBeInstanceOf(UpstreamTimeoutError); + expect(finished).toBe(false); + expect(lim.inUse("tmdb")).toBe(1); + // 许可仍被占:排队到超时 → busy + await expect(lim.run("identity.resolve", {}, undefined, async () => 2)).rejects.toBeInstanceOf(UpstreamBusyError); + await sleep(250); + expect(finished).toBe(true); + expect(lim.inUse("tmdb")).toBe(0); + await expect(lim.run("identity.resolve", {}, undefined, async () => 3)).resolves.toBe(3); + }); + + it("并发上限:同时在跑的外部调用不超过 limit;本地调用不受限", async () => { + const lim = createInvokeLimiter({ timeoutMs: 5000, limits: { site: 2 } }); + let running = 0; + let peak = 0; + const job = () => + lim.run("release.search", { query: "q", allowLiveSearch: true }, undefined, async () => { + running++; + peak = Math.max(peak, running); + await sleep(30); + running--; + }); + await Promise.all(Array.from({ length: 7 }, job)); + expect(peak).toBe(2); + await expect(lim.run("inventory.query", {}, undefined, async () => "local")).resolves.toBe("local"); + }); + + it("Semaphore:等待者被中止时移出队列", async () => { + const s = new Semaphore(1); + const r1 = await s.acquire(new AbortController().signal); + const ac = new AbortController(); + const p = s.acquire(ac.signal); + ac.abort(new Error("gone")); + await expect(p).rejects.toThrow("gone"); + r1(); + expect(s.inUse).toBe(0); + }); +}); + +describe("worker 分组与事件循环探针", () => { + const names = ["evaluateAcquisition", "resolveDecision", "runAgent", "verifyAndImport"]; + it("io / agent / split 的任务选择", () => { + expect(workerPlan("io", names, { io: 4 })).toEqual([ + { label: "io", versions: ["evaluateAcquisition", "verifyAndImport"], concurrency: 4 }, + ]); + expect(workerPlan("agent", names, { agent: 2 })).toEqual([ + { label: "agent", versions: ["resolveDecision", "runAgent"], concurrency: 2 }, + ]); + expect(workerPlan("split", names, { io: 3, agent: 1 }).map((p) => [p.label, p.concurrency])).toEqual([ + ["io", 3], + ["agent", 1], + ]); + expect(workerPlan("all", names, {})).toEqual([{ label: "all" }]); + }); + + it("主线程被同步占住超过阈值 → 计数", async () => { + const probe = startLoopLagProbe(createLogger({ level: "silent" }), { intervalMs: 20, warnMs: 100 }); + const until = performance.now() + 250; + while (performance.now() < until) { + // 同步忙等,模拟阻塞事件循环 + } + await sleep(60); + probe.stop(); + expect(probe.stats().loopLagWarnings).toBeGreaterThanOrEqual(1); + expect(probe.stats().maxLoopLagMs).toBeGreaterThan(100); + }); +});