feat(identity): lazy TMDB lookup and bounded identity pool (DEC-1)

RSS/list releases that match no subscription or local work no longer hit
TMDB or open a pending item; they go to the identity pool as `unmatched`.
TMDB is queried only on a local alias hit, a published new-title rule whose
prefilter passes, manual/Agent retry, search results, or when the new
setting identity.eagerTmdbLookup (default off) is on. Mirrors MoviePilot:
torrents are recognised only when matched to a subscription or searched,
and recognition failures are logged, not alerted.

- Pending items only for related releases (ambiguous after lookup, or
  related to a local work but unresolved); same source + normalized title
  merge into one decision ("来源「X」的 N 条资源待识别"), resolve/dismiss
  applies to the whole group, members detach as they get resolved.
- Pool governance: unmatched reviews capped (daily, 3x), title_alias
  insert trigger wakes similar entries, stale unmatched entries archived
  after identity.poolArchiveDays (30); review idempotency keys include the
  due time.
- Migration 0047_identity_pool: reason/outcome `unmatched`,
  identity_candidate.normalized_title (+trgm index), wake trigger, and
  cleanup of open identity_unresolved items with no related work
  (dismissed, resolution=moved_to_pool).
- Inbox: group by type + work/source with collapse, hide info-only items
  by default (minSeverity=warning), badge counts warning+ only; pool sheet
  gets manual TMDB/local-work assign and dismiss; settings card for the
  new identity settings.
- Runbook docs/runbook/identity.md.

Requires rebuild/restart (migration runs on start).

Co-Authored-By: Claude Opus 5.5 <[email protected]>

apps/server/src/modules/sources-identity/capabilities.ts

@@ -166,7 +166,8 @@ export function registerSourcesCapabilities(registry: CapabilityRegistry, rt: ()
         decisionId: null,
       };
     }
-    const res = await resolveRelease(r, { sourceReleaseId: srId, force: true });
+    // Agent / 用户显式要求识别:按手动触发查 TMDB(惰性识别不拦)
+    const res = await resolveRelease(r, { sourceReleaseId: srId, force: true, trigger: "manual" });
     const status: "resolved" | "proposed" | "ambiguous" | "unidentifiable" =
       res.outcome === "auto" || res.outcome === "skipped" || res.outcome === "skipped_locked"
         ? res.mediaEntityId

apps/server/src/modules/sources-identity/defs.ts

@@ -96,6 +96,8 @@ const TE = ["media-entities"];
 const TI = ["identity"];
 const TD = ["decisions"];
 
+export const DecisionMinSeverity = z.enum(["info", "warning", "critical"]);
+
 export const D = {
   sites: {
     list: {
@@ -297,7 +299,12 @@ export const D = {
       "/decisions",
       "待处理事项（缺省只列未关闭）",
       Decision,
-      { status: DecisionStatus.optional(), kind: DecisionKind.optional() },
+      {
+        status: DecisionStatus.optional(),
+        kind: DecisionKind.optional(),
+        /** 只看不低于此级别的事项;收件箱缺省传 warning 隐藏"仅信息"类 */
+        minSeverity: DecisionMinSeverity.optional(),
+      },
       TD,
     ),
     get: get("/decisions/{id}", "待处理事项详情", Decision, TD),

apps/server/src/modules/sources-identity/routes/decisions.ts

@@ -4,12 +4,13 @@ import { sql } from "@umt/db";
 import { z } from "zod";
 import { ApiException } from "../../../http/errors.ts";
 import type { ModuleApp } from "../../types.ts";
-import { D } from "../defs.ts";
+import { D, DecisionMinSeverity } from "../defs.ts";
 import { route } from "../http.ts";
 import { decodeCursor, jsonb, page, toApi, toApiList } from "../rows.ts";
 import type { SourcesRuntime } from "../runtime.ts";
 import { audit } from "../service/common.ts";
 import { assignReleaseIdentity, dismissPoolEntry } from "../service/identity-admin.ts";
+import { groupMemberIds } from "../service/identity-decisions.ts";
 
 // 待处理中心数据源:decision 列表/详情/聚合/处理。
 // 身份类(歧义 / 待识别)由本模块直接执行所选候选或手动指定;其它类型只记录处理结果并发 decision.updated,
@@ -17,6 +18,11 @@ import { assignReleaseIdentity, dismissPoolEntry } from "../service/identity-adm
 
 const OPEN = ["open", "investigating", "awaiting_user"] as const;
 const T = ["decisions"];
+const SEV_AT_LEAST = {
+  info: ["info", "warning", "critical"],
+  warning: ["warning", "critical"],
+  critical: ["critical"],
+} as const;
 
 const SummaryGroup = z.object({
   kind: DecisionKind,
@@ -36,6 +42,7 @@ export function decisionRoutes(app: ModuleApp, rt: () => SourcesRuntime) {
     let q = db().selectFrom("decision").selectAll();
     q = query.status ? q.where("status", "=", query.status) : q.where("status", "in", [...OPEN]);
     if (query.kind) q = q.where("kind", "=", query.kind);
+    if (query.minSeverity) q = q.where("severity", "in", [...SEV_AT_LEAST[query.minSeverity]]);
     const rows = await q
       .orderBy(sql`case severity when 'critical' then 0 when 'warning' then 1 else 2 end`)
       .orderBy("created_at", "desc")
@@ -55,11 +62,12 @@ export function decisionRoutes(app: ModuleApp, rt: () => SourcesRuntime) {
       scope: "read",
       summary: "未关闭待处理事项按作品与问题类型聚合",
       tags: T,
-      query: z.object({ status: DecisionStatus.optional() }),
+      query: z.object({ status: DecisionStatus.optional(), minSeverity: DecisionMinSeverity.optional() }),
       response: z.object({ total: z.number().int(), groups: z.array(SummaryGroup) }),
     },
     async ({ query }) => {
       const statuses = query.status ? [query.status] : [...OPEN];
+      const severities: string[] = [...SEV_AT_LEAST[query.minSeverity ?? "info"]];
       // 作品归属:evidence.mediaEntityId,或 subject 为 media_entity,或 subject 为已识别的 source_release
       const rows = await sql<{
         kind: z.infer<typeof DecisionKind>;
@@ -75,7 +83,7 @@ export function decisionRoutes(app: ModuleApp, rt: () => SourcesRuntime) {
             case when d.subject_type = 'media_entity' then d.subject_id end,
             (select r.media_entity_id from source_release r where d.subject_type = 'source_release' and r.id = d.subject_id)
           ) as entity_id
-          from decision d where d.status = any(${statuses})
+          from decision d where d.status = any(${statuses}) and d.severity = any(${severities})
         )
         select d.kind, d.entity_id, e.title as entity_title, count(*)::int as count,
                max(case d.severity when 'critical' then 2 when 'warning' then 1 else 0 end)::int as max_sev,
@@ -104,6 +112,44 @@ export function decisionRoutes(app: ModuleApp, rt: () => SourcesRuntime) {
     return toApi(row);
   });
 
+  /** 组内其它资源:忽略 → 各自池条目 dismissed;指定 → 关联到代表资源已确定的作品(同样加人工锁)。单条失败不影响其它 */
+  async function applyToGroup(
+    decisionId: string,
+    subjectId: string,
+    others: string[],
+    dismissing: boolean,
+    note: string | null,
+    actor: Parameters<typeof dismissPoolEntry>[2],
+  ) {
+    const subject = await db()
+      .selectFrom("source_release")
+      .select(["media_entity_id"])
+      .where("id", "=", subjectId)
+      .executeTakeFirst();
+    if (!dismissing && !subject?.media_entity_id) return;
+    for (const srId of others) {
+      try {
+        if (dismissing) {
+          const pool = await db()
+            .selectFrom("identity_candidate")
+            .select(["id"])
+            .where("source_release_id", "=", srId)
+            .executeTakeFirst();
+          if (pool) await dismissPoolEntry(rt(), pool.id, actor, note);
+        } else {
+          await assignReleaseIdentity(
+            rt(),
+            srId,
+            { mediaEntityId: subject!.media_entity_id!, lock: true, reason: note ?? `decision ${decisionId}` },
+            actor,
+          );
+        }
+      } catch (e) {
+        rt().ctx.logger.warn({ err: e, srId, decisionId }, "合并事项组内资源处理失败");
+      }
+    }
+  }
+
   route(app, D.decisions.resolve, async ({ params, body, actor }) => {
     const d = await db().selectFrom("decision").selectAll().where("id", "=", params.id).executeTakeFirst();
     if (!d) throw new ApiException("not_found", "待处理事项不存在");
@@ -118,7 +164,10 @@ export function decisionRoutes(app: ModuleApp, rt: () => SourcesRuntime) {
     const identityKind = d.kind === "identity_ambiguous" || d.kind === "identity_unresolved";
     if (identityKind && d.subject_type === "source_release" && d.subject_id) {
       const requires = opt?.input?.requires;
-      if (body.dismiss || opt?.key === "dismiss") {
+      // 合并事项(同来源同标题):先处理代表资源,再把结果套到组内其它资源
+      const others = d.dedupe_key.startsWith("identity-group:") ? await groupMemberIds(db(), d.id, d.subject_id) : [];
+      const dismissing = body.dismiss || opt?.key === "dismiss";
+      if (dismissing) {
         const pool = await db()
           .selectFrom("identity_candidate")
           .select(["id"])
@@ -163,6 +212,7 @@ export function decisionRoutes(app: ModuleApp, rt: () => SourcesRuntime) {
           actor,
         );
       }
+      if (others.length) await applyToGroup(d.id, d.subject_id, others, dismissing, body.note ?? null, actor);
     }
     // 站点登录需人工验证:"重试登录" = 入队刷新登录(结果另行更新账号状态与待处理)
     if (d.dedupe_key?.startsWith("site-login:") && opt?.key === "retry") {

apps/server/src/modules/sources-identity/routes/identity.ts

@@ -628,7 +628,8 @@ export function identityRoutes(app: ModuleApp, rt: () => SourcesRuntime) {
       if (!pool) throw new ApiException("not_found", "待识别条目不存在");
       const r = await rt().ctx.engine.enqueue(
         "resolveObservation",
-        { sourceReleaseId: pool.source_release_id, force: true },
+        // 用户点"重新识别" = 手动触发:即使不相关也查 TMDB(惰性识别的 c 路径)
+        { sourceReleaseId: pool.source_release_id, force: true, trigger: "manual" },
         { idempotencyKey: `resolve:retry:${params.id}:${Math.floor(Date.now() / 10_000)}`, traceId: randomUUID() },
       );
       return { taskId: r.runId };

apps/server/src/modules/sources-identity/scheduler.ts

@@ -1,6 +1,7 @@
 import { randomUUID } from "node:crypto";
 import { sql } from "@umt/db";
 import type { SourcesRuntime } from "./runtime.ts";
+import { archiveStalePool } from "./service/identity-decisions.ts";
 
 // worker 后台调度:按 Feed 轮询策略发 feed.due(outbox → ingestFeed)、按站点快照周期入队 snapshotSite、
 // 按待识别池 next_review_at 入队复查。幂等键按时间槽生成,多 worker 并发也不会重复。
@@ -86,13 +87,17 @@ export async function tickSnapshots(rt: SourcesRuntime): Promise<number> {
   return sites.rows.length;
 }
 
-/** 待识别池复查:到期的先顺延一个租约,避免下个 tick 重复入队 */
+/**
+ * 待识别池复查:到期的先顺延一个租约,避免下个 tick 重复入队;顺带(每小时至多一次)归档过期的 unmatched 条目。
+ * 幂等键带上到期时间:别名触发器唤醒 / 证据版本变化后计数归 1 时,不会撞上旧的复查键而被吞掉。
+ */
 export async function tickReviews(rt: SourcesRuntime): Promise<number> {
   const { ctx } = rt;
   const now = rt.now();
+  await archiveStalePool(rt);
   const reviews = await ctx.db.transaction().execute(async (trx) => {
-    const rows = await sql<{ id: string; source_release_id: string; attempts: number }>`
-      select id, source_release_id, attempts from identity_candidate
+    const rows = await sql<{ id: string; source_release_id: string; attempts: number; due: Date }>`
+      select id, source_release_id, attempts, next_review_at as due from identity_candidate
        where status = 'pending' and next_review_at is not null and next_review_at <= ${now}
        order by next_review_at limit 50 for update skip locked`.execute(trx);
     for (const r of rows.rows) {
@@ -108,7 +113,7 @@ export async function tickReviews(rt: SourcesRuntime): Promise<number> {
     await ctx.engine.enqueue(
       "resolveObservation",
       { sourceReleaseId: r.source_release_id, force: true },
-      { idempotencyKey: `review:${r.id}:${r.attempts}`, traceId: randomUUID() },
+      { idempotencyKey: `review:${r.id}:${r.attempts}:${new Date(r.due).getTime()}`, traceId: randomUUID() },
     );
   }
   return reviews.length;

apps/server/src/modules/sources-identity/service/identity-admin.ts

@@ -12,7 +12,7 @@ import { toParsedColumns } from "@umt/mapping";
 import { ApiException } from "../../../http/errors.ts";
 import { jsonb } from "../rows.ts";
 import type { SourcesRuntime } from "../runtime.ts";
-import { type Actor, audit, closeDecisions, type Exec } from "./common.ts";
+import { type Actor, audit, type Exec } from "./common.ts";
 import {
   acceptIdentity,
   createEntityFromTmdb,
@@ -22,6 +22,7 @@ import {
   looksLikeVariety,
   resolveEntityId,
 } from "./entities.ts";
+import { detachFromIdentityDecisions } from "./identity-decisions.ts";
 import { recordAttempt } from "./resolve.ts";
 import { backfillRecognizeCache } from "./tmdb-recognize-cache.ts";
 
@@ -149,10 +150,10 @@ export async function assignReleaseIdentity(
           })
           .where("source_release_id", "=", srId)
           .execute();
-        await closeDecisions(
+        await detachFromIdentityDecisions(
           ctx,
           trx,
-          `identity:${srId}`,
+          srId,
           "resolved",
           { mediaEntityId: entityId, by: "manual" },
           actorName(actor),
@@ -468,7 +469,7 @@ export async function dismissPoolEntry(rt: SourcesRuntime, poolId: string, actor
       .where("id", "=", row.source_release_id)
       .where("resolution_locked", "=", false)
       .execute();
-    await closeDecisions(ctx, trx, `identity:${row.source_release_id}`, "dismissed", { note }, actorName(actor));
+    await detachFromIdentityDecisions(ctx, trx, row.source_release_id, "dismissed", { note }, actorName(actor));
     await audit(trx, actor, "identity.pool_dismissed", { type: "identity_candidate", id: poolId }, { note });
     return row;
   });

apps/server/src/modules/sources-identity/service/identity-decisions.ts

@@ -0,0 +1,228 @@
+import type { DecisionOption, PoolCandidate } from "@umt/contracts";
+import { sql } from "@umt/db";
+import type { ServerContext } from "../../types.ts";
+import { jsonb } from "../rows.ts";
+import type { SourcesRuntime } from "../runtime.ts";
+import { closeDecisions, type Exec, openDecision } from "./common.ts";
+
+// DEC-1 身份类待处理的合并与池治理(REQ §7.5、§13):
+// - 只有与订阅 / 本地作品相关(本地别名命中)或已查过 TMDB 仍有歧义的资源才开事项;
+// - 同一来源 + 同一归一化标题合并为一条事项(dedupeKey identity-group:<siteId>:<normalized>),成员 = 池里
+//   decision_id 指向它且仍 pending 的条目;事项的 subject 与候选选项取自代表资源(最近一次进来的那条);
+// - 成员被识别 / 指定 / 忽略时从组里摘除,组空了才关闭事项;
+// - unmatched 池条目超过 identity.poolArchiveDays 没有新证据就归档。
+
+const OPEN = ["open", "investigating", "awaiting_user"] as const;
+/** 事项证据里最多列多少条成员资源(完整列表在待识别池按标题检索) */
+export const GROUP_LIST_LIMIT = 50;
+
+export const groupDedupeKey = (siteId: string, groupKey: string) => `identity-group:${siteId}:${groupKey}`;
+
+/** 身份类事项的手动指定选项;resolve 时按 input.requires 取 DecisionResolveInput.identity 的对应字段 */
+function manualIdentityOptions(sourceReleaseId: string): DecisionOption[] {
+  return [
+    {
+      key: "manual_tmdb",
+      label: "手动指定 TMDB ID",
+      description: "按 TMDB 电影或剧集 ID 建档并关联，加人工锁",
+      input: { sourceReleaseId, requires: "tmdb" },
+    },
+    {
+      key: "local_entity",
+      label: "选本地作品",
+      description: "关联到已有的本地作品，加人工锁",
+      input: { sourceReleaseId, requires: "mediaEntityId" },
+    },
+  ];
+}
+
+export function identityOptions(sourceReleaseId: string, candidates: readonly PoolCandidate[]): DecisionOption[] {
+  return [
+    ...candidates.slice(0, 5).map((c, i) => ({
+      key: `c${i}`,
+      label: `${c.title}${c.year ? ` (${c.year})` : ""} · ${c.localEntityId ? "本地作品" : `${c.provider}:${c.entityType}/${c.externalId}`}`,
+      input: { sourceReleaseId, candidateIndex: i },
+    })),
+    ...manualIdentityOptions(sourceReleaseId),
+    { key: "dismiss", label: "忽略此资源" },
+  ];
+}
+
+interface Member {
+  source_release_id: string;
+  title: string;
+  reason: string;
+  candidates: unknown;
+  updated_at: Date;
+}
+
+async function members(exec: Exec, decisionId: string, exclude: string | null): Promise<Member[]> {
+  const rows = await sql<Member>`
+    select p.source_release_id, r.title, p.reason, p.candidates, p.updated_at
+      from identity_candidate p join source_release r on r.id = p.source_release_id
+     where p.decision_id = ${decisionId} and p.status = 'pending'
+       and (${exclude}::uuid is null or p.source_release_id <> ${exclude}::uuid)
+     order by p.updated_at desc, p.source_release_id`.execute(exec);
+  return rows.rows;
+}
+
+/** 合并事项的成员资源 ID(不含 subject 本身时传 exclude) */
+export async function groupMemberIds(exec: Exec, decisionId: string, exclude: string | null = null) {
+  return (await members(exec, decisionId, exclude)).map((m) => m.source_release_id);
+}
+
+function titleFor(
+  kind: "identity_ambiguous" | "identity_unresolved",
+  siteName: string,
+  n: number,
+  one: string,
+  work: string | null,
+) {
+  const what = kind === "identity_ambiguous" ? "待确认身份" : "待识别";
+  if (n <= 1) return `${what}：${one}`;
+  return `来源「${siteName}」的 ${n} 条资源${what}${work ? `：${work}` : ""}`;
+}
+
+/**
+ * 按当前成员重写合并事项的标题 / 摘要 / 成员清单;代表资源 = representative(缺省取最近更新的成员)。
+ * 组内没有成员时关闭事项并返回 false。
+ */
+export async function refreshGroupDecision(
+  ctx: ServerContext,
+  trx: Exec,
+  decisionId: string,
+  opts: {
+    representative?: string | null;
+    exclude?: string | null;
+    closeAs?: { status: "resolved" | "dismissed"; resolution: Record<string, unknown>; by: string };
+  } = {},
+): Promise<boolean> {
+  const d = await trx
+    .selectFrom("decision")
+    .selectAll()
+    .where("id", "=", decisionId)
+    .where("status", "in", [...OPEN])
+    .executeTakeFirst();
+  if (!d) return false;
+  const list = await members(trx, decisionId, opts.exclude ?? null);
+  if (!list.length) {
+    const close = opts.closeAs ?? { status: "resolved" as const, resolution: { by: "group_empty" }, by: "system" };
+    await closeDecisions(ctx, trx, d.dedupe_key, close.status, close.resolution, close.by);
+    return false;
+  }
+  const rep = list.find((m) => m.source_release_id === opts.representative) ?? list[0]!;
+  const ev = (d.evidence ?? {}) as Record<string, unknown>;
+  const group = (ev.group ?? {}) as { siteName?: string; title?: string | null };
+  const kind = (d.kind === "identity_ambiguous" ? "identity_ambiguous" : "identity_unresolved") as
+    | "identity_ambiguous"
+    | "identity_unresolved";
+  const candidates = (Array.isArray(rep.candidates) ? rep.candidates : []) as PoolCandidate[];
+  const title = titleFor(kind, group.siteName ?? "未知来源", list.length, rep.title, group.title ?? null);
+  await trx
+    .updateTable("decision")
+    .set({
+      title,
+      subject_type: "source_release",
+      subject_id: rep.source_release_id,
+      options: jsonb(identityOptions(rep.source_release_id, candidates)),
+      evidence: jsonb({
+        ...ev,
+        sourceReleaseId: rep.source_release_id,
+        count: list.length,
+        releases: list
+          .slice(0, GROUP_LIST_LIMIT)
+          .map((m) => ({ sourceReleaseId: m.source_release_id, title: m.title })),
+      }),
+    })
+    .where("id", "=", decisionId)
+    .execute();
+  return true;
+}
+
+export interface GroupDecisionInput {
+  kind: "identity_ambiguous" | "identity_unresolved";
+  siteId: string;
+  /** 分组键:归一化标题;解析不出标题时用资源 ID(不合并) */
+  groupKey: string;
+  /** 分组展示用的作品标题(解析标题) */
+  workTitle: string | null;
+  sourceReleaseId: string;
+  releaseTitle: string;
+  summary: string;
+  candidates: PoolCandidate[];
+  evidence: Record<string, unknown>;
+  nextReviewAt: Date | null;
+}
+
+/** 开 / 并入合并事项;返回事项 ID(调用方随后把池条目 decision_id 指向它,再 refreshGroupDecision) */
+export async function openGroupDecision(ctx: ServerContext, trx: Exec, g: GroupDecisionInput): Promise<string> {
+  const site = await trx.selectFrom("site").select(["name"]).where("id", "=", g.siteId).executeTakeFirst();
+  const d = await openDecision(ctx, trx, {
+    kind: g.kind,
+    // 与订阅 / 本地作品相关的身份问题会卡住追更,不是"仅信息"
+    severity: "warning",
+    title: titleFor(g.kind, site?.name ?? "未知来源", 1, g.releaseTitle, g.workTitle),
+    summary: g.summary,
+    subjectType: "source_release",
+    subjectId: g.sourceReleaseId,
+    dedupeKey: groupDedupeKey(g.siteId, g.groupKey),
+    evidence: {
+      ...g.evidence,
+      group: { siteId: g.siteId, siteName: site?.name ?? null, key: g.groupKey, title: g.workTitle },
+    },
+    options: identityOptions(g.sourceReleaseId, g.candidates),
+    nextReviewAt: g.nextReviewAt,
+  });
+  return d.id;
+}
+
+/**
+ * 资源已识别 / 已指定 / 已忽略:关闭它的旧式单条事项(identity:<id>),并从合并事项里摘除;组空了才关闭合并事项。
+ */
+export async function detachFromIdentityDecisions(
+  ctx: ServerContext,
+  trx: Exec,
+  srId: string,
+  status: "resolved" | "dismissed",
+  resolution: Record<string, unknown>,
+  by: string,
+): Promise<void> {
+  await closeDecisions(ctx, trx, `identity:${srId}`, status, resolution, by);
+  const pool = await trx
+    .selectFrom("identity_candidate")
+    .select(["decision_id"])
+    .where("source_release_id", "=", srId)
+    .executeTakeFirst();
+  if (!pool?.decision_id) return;
+  const d = await trx
+    .selectFrom("decision")
+    .select(["id", "dedupe_key"])
+    .where("id", "=", pool.decision_id)
+    .where("status", "in", [...OPEN])
+    .executeTakeFirst();
+  if (!d?.dedupe_key.startsWith("identity-group:")) return;
+  await refreshGroupDecision(ctx, trx, d.id, { exclude: srId, closeAs: { status, resolution, by } });
+}
+
+const ARCHIVE_EVERY_MS = 3600_000;
+let lastArchiveAt = 0;
+
+/**
+ * 过期归档:unmatched 池条目超过 identity.poolArchiveDays 没有新证据 → archived(仍可按状态筛出、可手动指定;
+ * 同一资源再被观察到或新增相近别名时回到 pending)。每进程每小时最多执行一次。
+ */
+export async function archiveStalePool(rt: SourcesRuntime, opts: { force?: boolean } = {}): Promise<number> {
+  const now = rt.now();
+  if (!opts.force && now.getTime() - lastArchiveAt < ARCHIVE_EVERY_MS) return 0;
+  lastArchiveAt = now.getTime();
+  const { poolArchiveDays } = await rt.ctx.settings.get("identity");
+  const cutoff = new Date(now.getTime() - poolArchiveDays * 86_400_000);
+  const r = await rt.ctx.db
+    .updateTable("identity_candidate")
+    .set({ status: "archived", next_review_at: null })
+    .where("status", "=", "pending")
+    .where("reason", "=", "unmatched")
+    .where((eb) => eb(eb.fn.coalesce("last_attempt_at", "created_at"), "<", cutoff))
+    .executeTakeFirst();
+  return Number(r.numUpdatedRows ?? 0);
+}

apps/server/src/modules/sources-identity/service/resolve.ts

@@ -22,7 +22,7 @@ import {
 } from "@umt/mapping";
 import { jsonb } from "../rows.ts";
 import type { SourcesRuntime } from "../runtime.ts";
-import { type Actor, closeDecisions, type Exec, openDecision, SYSTEM } from "./common.ts";
+import { type Actor, closeDecisions, type Exec, SYSTEM } from "./common.ts";
 import {
   addAliasIfAbsent,
   createEntityFromTmdb,
@@ -32,12 +32,21 @@ import {
   looksLikeVariety,
   resolveEntityId,
 } from "./entities.ts";
+import { detachFromIdentityDecisions, openGroupDecision, refreshGroupDecision } from "./identity-decisions.ts";
 import { loadRecognizeCache, type RecognizeMeta, recognizeMetaOf, saveRecognizeCache } from "./tmdb-recognize-cache.ts";
 
-// resolveObservation:parseRelease → 本地召回(别名 pg_trgm + 已有外部身份 + 人工锁)→ TMDB 候选 → domain.resolveIdentity。
+// resolveObservation:parseRelease → 本地召回(别名 pg_trgm + 已有外部身份 + 人工锁)→ [按需] TMDB 候选 → domain.resolveIdentity。
 // auto 才建/关联正式作品;ambiguous / no_match / 查询失败进待识别池(候选只存 JSON),有界退避复查。
+//
+// DEC-1 惰性识别(对照 MP:RSS / 站点资源只在匹配订阅或用户搜索时识别,识别失败不告警):TMDB 只在 lookupTrigger
+// 给出理由时查询——命中本地作品别名(订阅 / 追踪 / 已登记的作品)、已发布的新作品收录规则前置条件通过、用户手动识别、
+// 搜索结果、池条目此前已查过 TMDB(复查延续),或设置 identity.eagerTmdbLookup 打开。其余只进池(reason=unmatched),
+// 不开待处理。待处理只收"相关"的:与本地作品相关却无法识别,或查过 TMDB 仍有歧义;同来源同标题合并为一条。
 
 const MAX_REVIEW_ATTEMPTS = 8;
+/** unmatched(只做了本地召回)的周期复查:每天一次、最多 3 次;新增相近别名时由触发器立即唤醒,不受此上限 */
+const UNMATCHED_REVIEW_MS = 24 * 3600_000;
+const MAX_UNMATCHED_REVIEWS = 3;
 const TRGM_THRESHOLD = 0.45;
 
 export interface ResolveInput {
@@ -45,10 +54,26 @@ export interface ResolveInput {
   sourceReleaseId?: string | undefined;
   /** 复查/手动重试:忽略"已识别且证据未变"的短路 */
   force?: boolean | undefined;
+  /**
+   * 谁触发的识别,决定是否查 TMDB(惰性识别):observed = 新观察到的资源(缺省);review = 池周期复查(force 缺省);
+   * manual = 用户在资源列表 / 待识别池点开或重新识别、Agent 调 identity.resolve;search = 搜索结果。
+   * manual / search 总是查 TMDB;observed / review 只在命中本地别名、新作品收录规则或设置打开时查。
+   */
+  trigger?: ResolveTrigger | undefined;
 }
 
+export type ResolveTrigger = "observed" | "review" | "manual" | "search";
+
 export interface ResolveResult {
-  outcome: "auto" | "proposed" | "ambiguous" | "no_match" | "lookup_failed" | "skipped_locked" | "skipped";
+  outcome:
+    | "auto"
+    | "proposed"
+    | "ambiguous"
+    | "no_match"
+    | "lookup_failed"
+    | "unmatched"
+    | "skipped_locked"
+    | "skipped";
   sourceReleaseId: string | null;
   mediaEntityId: string | null;
   contentUnitIds: string[];
@@ -251,12 +276,22 @@ export async function resolveRelease(rt: SourcesRuntime, input: ResolveInput): P
   }
   const parsed = parseRelease(release.title, release.subtitle ? { subtitle: release.subtitle } : {});
   const cols = toParsedColumns(parsed);
+  const trigger: ResolveTrigger = input.trigger ?? (input.force ? "review" : "observed");
   if (!parsed.title || parsed.auxiliaryOnly) {
-    return toPool(rt, release, observationId, parsed, null, [], "no_match", "unparseable_title", null);
+    return toPool(rt, release, observationId, parsed, null, [], "unmatched", "unparseable_title", null, {
+      tracked: false,
+      lookup: null,
+    });
   }
 
-  // 召回:本地优先,再 TMDB;TMDB 失败不等于无结果
+  // 召回:本地优先;TMDB 只在有理由时查(惰性识别),查询失败不等于无结果
   const local = await localCandidates(ctx.db, parsed, release);
+  const lookup = await lookupReason(rt, parsed, trigger, local.length > 0, pool?.reason ?? null);
+  const scope = { tracked: local.length > 0, lookup };
+  if (!lookup) {
+    // 不相关的资源:只进池(可检索、可手动指定),不查 TMDB、不开待处理
+    return toPool(rt, release, observationId, parsed, null, [], "unmatched", "no_local_match", null, scope);
+  }
   let remote: MetadataCandidate[] = [];
   let lookupError: string | null = null;
   let recognize: { cacheHit: boolean; meta: RecognizeMeta | null } = { cacheHit: false, meta: null };
@@ -321,7 +356,7 @@ export async function resolveRelease(rt: SourcesRuntime, input: ResolveInput): P
         }
       }
     }
-    return bindAuto(rt, release, observationId, parsed, cols, resolution, version, detailed ?? null);
+    return bindAuto(rt, release, observationId, parsed, cols, resolution, version, detailed ?? null, lookup);
   }
 
   if (resolution.status === "auto" && !autoAccept) {
@@ -335,6 +370,7 @@ export async function resolveRelease(rt: SourcesRuntime, input: ResolveInput): P
       "proposed",
       "automation_off",
       version,
+      scope,
     );
   }
   if (lookupError && resolution.status !== "ambiguous") {
@@ -348,6 +384,7 @@ export async function resolveRelease(rt: SourcesRuntime, input: ResolveInput): P
       "lookup_failed",
       lookupError,
       version,
+      scope,
     );
   }
   return toPool(
@@ -360,14 +397,60 @@ export async function resolveRelease(rt: SourcesRuntime, input: ResolveInput): P
     resolution.status === "ambiguous" ? "ambiguous" : "no_match",
     resolution.reason,
     version,
+    scope,
   );
 }
 
+/** 为什么要查 TMDB;null = 不查(惰性识别) */
+export type LookupReason = "tracked_alias" | "manual" | "search" | "previously_looked_up" | "eager" | "new_title_rule";
+
+async function lookupReason(
+  rt: SourcesRuntime,
+  parsed: ParsedRelease,
+  trigger: ResolveTrigger,
+  tracked: boolean,
+  poolReason: string | null,
+): Promise<LookupReason | null> {
+  if (trigger === "manual") return "manual";
+  if (trigger === "search") return "search";
+  if (tracked) return "tracked_alias";
+  // 池条目此前已查过 TMDB(旧行为 / 手动 / 规则触发):复查延续,仍受识别缓存与有界退避约束
+  if (trigger === "review" && poolReason && poolReason !== "unmatched") return "previously_looked_up";
+  if ((await rt.ctx.settings.get("identity")).eagerTmdbLookup) return "eager";
+  if (await passesNewTitlePrefilter(rt, parsed)) return "new_title_rule";
+  return null;
+}
+
+/**
+ * 新作品收录规则的前置硬条件:有已发布且开启 newTitle 的获取策略(并有有效持续授权),资源的类型落在规则的
+ * mediaKinds 内,标题年份(若有)不早于窗口能覆盖的年份。通过才值得查 TMDB 拿首播 / 上映日期交给评估判断。
+ */
+async function passesNewTitlePrefilter(rt: SourcesRuntime, parsed: ParsedRelease): Promise<boolean> {
+  const rows = await sql<{ rules: unknown }>`
+    select p.rules from policy_revision p
+     where p.kind = 'acquisition' and p.status = 'published'
+       and coalesce((p.rules->'newTitle'->>'enabled')::boolean, false)
+       and exists (select 1 from automation_grant g where g.policy_revision_id = p.id and g.status = 'active')`.execute(
+    rt.ctx.db,
+  );
+  if (!rows.rows.length) return false;
+  const tvLike = parsed.mediaTypeHint === "tv" || parsed.season !== undefined || parsed.episodes !== undefined;
+  const nowYear = rt.now().getUTCFullYear();
+  return rows.rows.some((r) => {
+    const nt = ((r.rules ?? {}) as { newTitle?: { windowDays?: number; mediaKinds?: string[] } }).newTitle ?? {};
+    const kinds = nt.mediaKinds ?? [];
+    if (kinds.length && !kinds.some((k) => (tvLike ? k === "series" || k === "variety" : k === "movie"))) return false;
+    // 与评估同一缺省(acquisition-organize ports.newTitleWindowDays = 30)
+    const windowYears = Math.ceil((nt.windowDays ?? 30) / 365);
+    return parsed.year === undefined || parsed.year >= nowYear - windowYears;
+  });
+}
+
 async function recordAttempt(
   exec: Exec,
   srId: string,
   observationId: string | null,
-  outcome: "auto" | "proposed" | "ambiguous" | "no_match" | "lookup_failed" | "manual" | "skipped_locked",
+  outcome: "auto" | "proposed" | "ambiguous" | "no_match" | "lookup_failed" | "unmatched" | "manual" | "skipped_locked",
   reason: string | null,
   version: string,
   count: number,
@@ -407,6 +490,7 @@ async function bindAuto(
   resolution: IdentityResolution,
   version: string,
   tmdbDetail: MetadataCandidate | null,
+  lookup: LookupReason,
 ): Promise<ResolveResult> {
   const { ctx } = rt;
   const best = resolution.best!;
@@ -458,6 +542,7 @@ async function bindAuto(
           evidenceVersion: version,
           best: rankedSummary(best),
           candidateCount: resolution.ranked.length,
+          lookup,
         };
         await trx
           .updateTable("source_release")
@@ -484,10 +569,10 @@ async function bindAuto(
           .where("source_release_id", "=", release.id)
           .where("status", "=", "pending")
           .execute();
-        await closeDecisions(
+        await detachFromIdentityDecisions(
           ctx,
           trx,
-          `identity:${release.id}`,
+          release.id,
           "resolved",
           { mediaEntityId: entityId, by: "auto" },
           "system",
@@ -532,29 +617,12 @@ async function bindAuto(
 }
 
 function reviewDelayMs(reason: string, attempts: number): number | null {
+  if (reason === "unmatched") return attempts >= MAX_UNMATCHED_REVIEWS ? null : UNMATCHED_REVIEW_MS;
   if (attempts >= MAX_REVIEW_ATTEMPTS) return null;
   const base = reason === "lookup_failed" ? 15 * 60_000 : 3600_000;
   return Math.min(base * 2 ** Math.max(0, attempts - 1), 7 * 24 * 3600_000);
 }
 
-/** 身份类事项的手动指定选项;resolve 时按 input.requires 取 DecisionResolveInput.identity 的对应字段 */
-function manualIdentityOptions(sourceReleaseId: string) {
-  return [
-    {
-      key: "manual_tmdb",
-      label: "手动指定 TMDB ID",
-      description: "按 TMDB 电影或剧集 ID 建档并关联，加人工锁",
-      input: { sourceReleaseId, requires: "tmdb" },
-    },
-    {
-      key: "local_entity",
-      label: "选本地作品",
-      description: "关联到已有的本地作品，加人工锁",
-      input: { sourceReleaseId, requires: "mediaEntityId" },
-    },
-  ];
-}
-
 /** 事项摘要里给人看的原因;原始代码仍记在 resolution_evidence */
 const DETAIL_TEXT: Record<string, string> = {
   no_viable_candidate: "没有可信候选",
@@ -562,9 +630,18 @@ const DETAIL_TEXT: Record<string, string> = {
   weak_evidence: "证据不足",
   multiple_candidates_same_level: "多个候选同级",
   automation_off: "自动化已关闭",
+  no_local_match: "没有命中订阅或本地作品，未查询 TMDB",
 };
 const detailText = (d: string) => DETAIL_TEXT[d] ?? d;
 
+type PoolReason = "ambiguous" | "no_match" | "proposed" | "lookup_failed" | "unmatched";
+
+/** 这次识别的相关性:tracked = 本地别名命中(订阅 / 追踪 / 已登记作品);lookup = 查 TMDB 的理由(null = 未查) */
+interface ResolveScope {
+  tracked: boolean;
+  lookup: LookupReason | null;
+}
+
 async function toPool(
   rt: SourcesRuntime,
   release: SourceReleaseRow,
@@ -572,9 +649,10 @@ async function toPool(
   parsed: ParsedRelease,
   resolution: IdentityResolution | null,
   ranked: RankedCandidate[],
-  reason: "ambiguous" | "no_match" | "proposed" | "lookup_failed",
+  reason: PoolReason,
   detail: string,
   versionIn: string | null,
+  scope: ResolveScope,
 ): Promise<ResolveResult> {
   const { ctx } = rt;
   const version = versionIn ?? evidenceVersion(parsed, ranked);
@@ -584,14 +662,16 @@ async function toPool(
     .slice(0, 10)
     .map(poolCandidate);
   const now = rt.now();
+  const normalized = parsed.title ? normalizeTitle(parsed.title) || null : null;
   const query = {
     title: parsed.title ?? null,
     titleCn: parsed.titleCn ?? null,
     titleEn: parsed.titleEn ?? null,
-    normalized: parsed.title ? normalizeTitle(parsed.title) : null,
+    normalized,
     year: parsed.year ?? null,
     season: parsed.season ?? null,
     episodes: parsed.episodes ?? null,
+    lookup: scope.lookup,
   };
   const contractStatus =
     reason === "proposed"
@@ -603,7 +683,7 @@ async function toPool(
   return ctx.db.transaction().execute(async (trx) => {
     const prev = await trx
       .selectFrom("identity_candidate")
-      .select(["id", "attempts", "evidence_version"])
+      .select(["id", "attempts", "evidence_version", "decision_id"])
       .where("source_release_id", "=", release.id)
       .executeTakeFirst();
     // 证据版本变化 = 新证据:从头计数;同一证据反复复查按退避递增,达上限停止自动复查(仍可检索)
@@ -611,7 +691,6 @@ async function toPool(
     const delay = reviewDelayMs(reason, attempts);
     const nextReviewAt = delay === null ? null : new Date(now.getTime() + delay);
 
-    let decisionId: string | null = null;
     const evidence = {
       sourceReleaseId: release.id,
       siteId: release.site_id,
@@ -620,64 +699,54 @@ async function toPool(
       evidenceVersion: version,
       query,
     };
-    if ((reason === "ambiguous" || reason === "proposed") && candidates.length > 0) {
-      const d = await openDecision(ctx, trx, {
-        kind: "identity_ambiguous",
-        severity: "info",
-        title: `待确认身份：${release.title}`,
-        summary: `${candidates.length} 个候选证据等级相近(${detailText(detail)})。选择一个或忽略`,
-        subjectType: "source_release",
-        subjectId: release.id,
-        dedupeKey: `identity:${release.id}`,
+    // 只收相关的:查过 TMDB 仍有歧义(有候选可选),或与本地作品相关却无结果 / 查询失败且复查已停
+    const wantsDecision =
+      reason === "ambiguous" || reason === "proposed"
+        ? candidates.length > 0
+        : scope.tracked && (reason === "no_match" || (reason === "lookup_failed" && delay === null));
+    let decisionId: string | null = null;
+    if (wantsDecision) {
+      const ambiguous = (reason === "ambiguous" || reason === "proposed") && candidates.length > 0;
+      decisionId = await openGroupDecision(ctx, trx, {
+        kind: ambiguous ? "identity_ambiguous" : "identity_unresolved",
+        siteId: release.site_id,
+        groupKey: normalized ?? release.id,
+        workTitle: parsed.title ?? null,
+        sourceReleaseId: release.id,
+        releaseTitle: release.title,
+        summary: ambiguous
+          ? `${candidates.length} 个候选证据等级相近(${detailText(detail)})。选择一个或忽略；同组资源一并处理`
+          : reason === "lookup_failed"
+            ? `元数据查询多次失败(${detail})，已停止自动复查。手动指定作品或忽略；同组资源一并处理`
+            : `标题与本地作品相关，但没有匹配的作品(${detailText(detail)})。手动指定 TMDB ID、选本地作品或忽略；同组资源一并处理`,
+        candidates,
         evidence,
-        options: [
-          ...candidates.slice(0, 5).map((c, i) => ({
-            key: `c${i}`,
-            label: `${c.title}${c.year ? ` (${c.year})` : ""} · ${c.localEntityId ? "本地作品" : `${c.provider}:${c.entityType}/${c.externalId}`}`,
-            input: { sourceReleaseId: release.id, candidateIndex: i },
-          })),
-          ...manualIdentityOptions(release.id),
-          { key: "dismiss", label: "忽略此资源" },
-        ],
         nextReviewAt,
       });
-      decisionId = d.id;
-    } else if (reason !== "lookup_failed" || delay === null) {
-      // 没有可选候选(无结果 / 无法解析 / 查询失败且自动复查已停):只能人工指定,同样进待处理
-      const d = await openDecision(ctx, trx, {
-        kind: "identity_unresolved",
-        severity: "info",
-        title: `待识别：${release.title}`,
-        summary:
-          reason === "lookup_failed"
-            ? `元数据查询多次失败(${detail})，已停止自动复查。手动指定作品或忽略`
-            : `没有匹配的作品(${detailText(detail)})。手动指定 TMDB ID、选本地作品或忽略`,
-        subjectType: "source_release",
-        subjectId: release.id,
-        dedupeKey: `identity:${release.id}`,
-        evidence,
-        options: [
-          ...manualIdentityOptions(release.id),
-          {
-            key: "dismiss",
-            label: "忽略此资源",
-          },
-        ],
-        nextReviewAt,
-      });
-      decisionId = d.id;
+    } else if (prev?.decision_id) {
+      // 不再需要人工处理(例如别名删除后变得不相关):从原合并事项里摘除
+      await detachFromIdentityDecisions(ctx, trx, release.id, "resolved", { by: "no_longer_relevant" }, "system");
     }
 
     await sql`
-      insert into identity_candidate (source_release_id, status, reason, query, candidates, evidence_version, attempts,
-                                      last_attempt_at, next_review_at, decision_id)
-      values (${release.id}, 'pending', ${reason}, ${jsonb(query)}::jsonb, ${jsonb(candidates)}::jsonb, ${version}, ${attempts},
-              ${now}, ${nextReviewAt}, ${decisionId})
+      insert into identity_candidate (source_release_id, status, reason, query, normalized_title, candidates, evidence_version,
+                                      attempts, last_attempt_at, next_review_at, decision_id)
+      values (${release.id}, 'pending', ${reason}, ${jsonb(query)}::jsonb, ${normalized}, ${jsonb(candidates)}::jsonb, ${version},
+              ${attempts}, ${now}, ${nextReviewAt}, ${decisionId})
       on conflict (source_release_id) do update set
-        status = 'pending', reason = excluded.reason, query = excluded.query, candidates = excluded.candidates,
-        evidence_version = excluded.evidence_version, attempts = excluded.attempts, last_attempt_at = excluded.last_attempt_at,
-        next_review_at = excluded.next_review_at, decision_id = coalesce(excluded.decision_id, identity_candidate.decision_id),
+        status = 'pending', reason = excluded.reason, query = excluded.query, normalized_title = excluded.normalized_title,
+        candidates = excluded.candidates, evidence_version = excluded.evidence_version, attempts = excluded.attempts,
+        last_attempt_at = excluded.last_attempt_at, next_review_at = excluded.next_review_at,
+        decision_id = case when ${wantsDecision} then excluded.decision_id else null end,
         resolved_media_entity_id = null, resolved_by = null, resolved_at = null`.execute(trx);
+    if (decisionId) {
+      await refreshGroupDecision(ctx, trx, decisionId, { representative: release.id });
+      // 旧式单条事项(identity:<id>)并入合并事项;换了组(标题更新)的,原组重算成员
+      await closeDecisions(ctx, trx, `identity:${release.id}`, "resolved", { mergedInto: decisionId }, "system");
+      if (prev?.decision_id && prev.decision_id !== decisionId) {
+        await refreshGroupDecision(ctx, trx, prev.decision_id);
+      }
+    }
 
     await trx
       .updateTable("source_release")
@@ -685,7 +754,13 @@ async function toPool(
         parser_version: cols.parserVersion,
         parsed: jsonb(cols.parsed),
         resolution_status: toReleaseResolutionStatus(contractStatus),
-        resolution_evidence: jsonb({ reason, detail, evidenceVersion: version, candidateCount: candidates.length }),
+        resolution_evidence: jsonb({
+          reason,
+          detail,
+          evidenceVersion: version,
+          candidateCount: candidates.length,
+          lookup: scope.lookup,
+        }),
       })
       .where("id", "=", release.id)
       .where("resolution_locked", "=", false)

apps/server/src/modules/sources-identity/tasks.ts

@@ -36,7 +36,13 @@ export function registerSourcesTasks(registry: TaskRegistry, rt: () => SourcesRu
       name: "resolveObservation",
       version: 1,
       inputSchema: z
-        .object({ observationId: id.nullish(), sourceReleaseId: id.nullish(), force: z.boolean().optional() })
+        .object({
+          observationId: id.nullish(),
+          sourceReleaseId: id.nullish(),
+          force: z.boolean().optional(),
+          // 惰性识别(DEC-1):manual / search 总是查 TMDB;缺省按 force 视为 review / observed
+          trigger: z.enum(["observed", "review", "manual", "search"]).optional(),
+        })
         .refine((v) => !!v.observationId || !!v.sourceReleaseId, { message: "需要 observationId 或 sourceReleaseId" }),
       retryPolicy: NETWORK_RETRY,
       handler: (t) =>
@@ -45,6 +51,7 @@ export function registerSourcesTasks(registry: TaskRegistry, rt: () => SourcesRu
             observationId: t.input.observationId ?? undefined,
             sourceReleaseId: t.input.sourceReleaseId ?? undefined,
             force: t.input.force,
+            trigger: t.input.trigger,
           }),
         ),
     }),