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 <[email protected]>

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(),
   /**

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 <file.json> [--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);
 }

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<string, unknown>;
+  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<void>((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<T>(
+    capabilityId: string,
+    input: unknown,
+    parent: AbortSignal | undefined,
+    fn: (signal: AbortSignal | undefined) => Promise<T>,
+  ): Promise<T>;
+  inUse(upstream: Upstream): number;
+}
+
+export function createInvokeLimiter(
+  opts: { timeoutMs?: number; limits?: Partial<Record<Upstream, number>> } = {},
+): 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<never>((_, 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);
+      }
+    },
+  };
+}

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, {
+      // 外部 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: c.req.raw.signal,
+          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"

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 可用)。

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<void>;
+  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<RunningWorker> {
-  const log = logger.child({ component: "worker" });
+export async function startWorker({ logger, heartbeatMs, bus, runtime, lagProbe }: WorkerDeps): Promise<RunningWorker> {
+  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<typeof workerPlan> = [];
   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?.();

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);
+  });
+});