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