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/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, {
-        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"

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