diff --git a/apps/server/src/modules/agent/invoke-limits.ts b/apps/server/src/modules/agent/invoke-limits.ts index 6f45f79f..68630263 100644 --- a/apps/server/src/modules/agent/invoke-limits.ts +++ b/apps/server/src/modules/agent/invoke-limits.ts @@ -137,3 +137,13 @@ export function createInvokeLimiter( }, }; } + +let shared: InvokeLimiter | null = null; +/** + * 进程内共享的限流器:同一 API 进程里所有 HTTP 路径上的外部 IO 调用(能力调用端点、/files/identify 等) + * 共用同一组信号量,TMDB/站点并发上限才是真正的进程级上限(验收 07 S-6)。 + */ +export function sharedInvokeLimiter(): InvokeLimiter { + shared ??= createInvokeLimiter(); + return shared; +} diff --git a/apps/server/src/modules/agent/routes.ts b/apps/server/src/modules/agent/routes.ts index 9fd10adf..81e29859 100644 --- a/apps/server/src/modules/agent/routes.ts +++ b/apps/server/src/modules/agent/routes.ts @@ -14,7 +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 { type InvokeLimiter, sharedInvokeLimiter, UpstreamBusyError, UpstreamTimeoutError } from "./invoke-limits.ts"; import { AgentRunError, abortRun, appendMessage, startRun } from "./service.ts"; // Agent 会话 API、持续授权 / 单次审批、能力目录与调用、业务任务。 @@ -89,7 +89,7 @@ const decode = (c?: string) => { export function registerAgentRoutes( app: ModuleApp, ctx: ServerContext, - limiter: InvokeLimiter = createInvokeLimiter(), + limiter: InvokeLimiter = sharedInvokeLimiter(), ) { const db = () => ctx.db; diff --git a/apps/server/src/modules/file-manager/identify.ts b/apps/server/src/modules/file-manager/identify.ts index 3aa32b16..a7b1d3c0 100644 --- a/apps/server/src/modules/file-manager/identify.ts +++ b/apps/server/src/modules/file-manager/identify.ts @@ -11,12 +11,16 @@ import { sql } from "@umt/db"; import { type IdentityCandidate, type ParsedRelease, parseReleasePath, queryTitles, rankCandidates } from "@umt/domain"; import { localIdentityCandidate } from "@umt/mapping"; import type { Principal } from "../../http/auth.ts"; +import { EXTERNAL_INVOKE_TIMEOUT_MS, sharedInvokeLimiter } from "../agent/invoke-limits.ts"; import { actorOfP, errMessage, FilesError, type FmDeps, resolveIn } from "./common.ts"; import { kindOfName } from "./list.ts"; // 识别(只读,不落库):parseReleasePath(文件 + 父/祖父目录上下文)→ 本地召回(别名 pg_trgm + 外部身份) // → 远端候选(catalog.lookup 能力,走 TMDB 缓存;失败只降级为提示)→ rankCandidates,每项前 5。 +/** 单次 /files/identify 请求内去重后的远端(TMDB)查询上限 */ +export const REMOTE_QUERY_MAX = 20; + const TRGM = 0.45; const LEVEL: Record = { rule_confirmed: "high", @@ -140,6 +144,7 @@ export async function identify( return out; }; + let remoteQueries = 0; const remoteFor = async (p: ParsedRelease) => { const q = p.titleCn ?? p.title ?? p.titleEn; if (!q) return { list: [], warning: null }; @@ -147,17 +152,32 @@ export async function identify( const hit = remoteCache.get(key); if (hit) return hit; let res: { list: IdentityCandidate[]; warning: string | null }; + // 单次请求内去重后的远端查询设上限,超出的项只做本地召回(验收 07 S-6) + if (remoteQueries >= REMOTE_QUERY_MAX) { + res = { + list: [], + warning: `本次识别的远端查询已达上限 ${REMOTE_QUERY_MAX} 个,该项仅本地召回;可缩小范围后再识别`, + }; + remoteCache.set(key, res); + return res; + } + remoteQueries += 1; try { - const out = (await d.ctx.capabilities.invoke( - "catalog.lookup", - { query: q, includeRemote: true, limit: 10, ...(body.hints?.year ? { year: body.hints.year } : {}) }, - { + const input = { + query: q, + includeRemote: true, + limit: 10, + ...(body.hints?.year ? { year: body.hints.year } : {}), + }; + // 经进程共享的 TMDB 信号量(并发 4)+ 15s 超时,与能力调用端点同一个限流器 + const out = (await sharedInvokeLimiter().run("catalog.lookup", input, undefined, (signal) => + d.ctx.capabilities.invoke("catalog.lookup", input, { actorId: actorOfP(principal), taskId: `files.identify:${randomUUID()}`, scopeIds: [], - signal: AbortSignal.timeout(15_000), + signal: signal ?? AbortSignal.timeout(EXTERNAL_INVOKE_TIMEOUT_MS), source: "web", - }, + }), )) as { remote?: { provider: string; entityType: string; externalId: string; title: string; year: number | null }[]; }; diff --git a/apps/server/test/file-manager/identify-organize.test.ts b/apps/server/test/file-manager/identify-organize.test.ts index 5eae5b0b..9426e793 100644 --- a/apps/server/test/file-manager/identify-organize.test.ts +++ b/apps/server/test/file-manager/identify-organize.test.ts @@ -78,6 +78,17 @@ describe.skipIf(skip)("files 识别 / 整理 / 探测 / 刮削", () => { expect(c[0]).toMatchObject({ provider: "tmdb", externalId: "999", mediaEntityId: null, kind: "movie" }); }); + it("identify:去重后的远端查询单次上限 20,超出的项只本地召回并给 warning(验收 07 S-6)", async () => { + await mkdir(join(h.dl, "many"), { recursive: true }); + const names = Array.from({ length: 23 }, (_, i) => `Distinct Title ${String.fromCharCode(65 + i)}.2019.1080p.mkv`); + for (const n of names) await writeFile(join(h.dl, "many", n), "x"); + const r = await h.api("POST", "/files/identify", { items: [{ root: h.dlRoot, path: "many" }] }, READ_KEY); + expect(r.status).toBe(200); + expect(r.body.data.items).toHaveLength(23); + const capped = r.body.data.items.filter((i: { warnings: string[] }) => i.warnings.some((w) => w.includes("上限"))); + expect(capped).toHaveLength(3); + }); + it("organize:dryRun 返回 planId 与前后路径;源变更 → 该组 plan_stale;重新预览后执行入队 verifyAndImport", async () => { const items = [{ root: h.dlRoot, path: "Blue.Film.2021.2160p.BluRay.x265-GRP.mkv", mediaEntityId: film }]; const dry = await h.api("POST", "/files/organize", { idempotencyKey: key(), items, action: "copy", dryRun: true });