feat(agent): subscription.list, decision.list, download.inspect; grant-scoped requiresApproval

- three read-only summary capabilities (38 registered); task.inspect points
  download task ids to download.inspect
- probe queries reject empty / single-letter / wildcard-only terms; externalId
  documented as string; every description says when to use it
- download.plan requiresApproval uses coveringGrant (shared with authorize)

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

apps/server/src/modules/acquisition-organize/capabilities.ts

@@ -2,6 +2,7 @@ import { CapabilityError, type CapabilityRegistry } from "@umt/agent";
 import { capabilitySpec } from "@umt/contracts";
 import { sql } from "@umt/db";
 import { normalizeTitle, transition } from "@umt/domain";
+import { coveringGrant, type PlanBinding, planBinding } from "../agent/authorize.ts";
 import { previewPolicy, publishRevision } from "./routes/rules.ts";
 import { evaluateAcquisition, publishedAcquisitionPolicies } from "./services/acquisition.ts";
 import { type AoDeps, enqueueTracked, ensureTaskRow, markTask } from "./services/common.ts";
@@ -76,6 +77,178 @@ export function registerAoCapabilities(registry: CapabilityRegistry, deps: () =>
     return { items } as never;
   });
 
+  // AG-3:订阅摘要(分组计数 + 条目),版本目标只给总数 / 已满足数,细节用 inventory.query
+  registry.register(capabilitySpec("subscription.list"), async (input) => {
+    const db = deps().ctx.db;
+    let q = db
+      .selectFrom("subscription as s")
+      .leftJoin("media_entity as e", "e.id", "s.media_entity_id")
+      .where((eb) => eb.or([eb("e.status", "is", null), eb("e.status", "<>", "merged")]));
+    if (input.status?.length) q = q.where("s.status", "in", input.status);
+    if (input.mediaKind) q = q.where("s.media_kind", "=", input.mediaKind);
+    if (input.query) {
+      const like = `%${input.query.replace(/[\\%_]/g, (m) => `\\${m}`)}%`;
+      q = q.where((eb) =>
+        eb.or([
+          eb("e.title", "ilike", like),
+          eb("e.original_title", "ilike", like),
+          eb("s.title_query", "ilike", like),
+        ]),
+      );
+    }
+    const groupCol = input.groupBy === "mediaKind" ? "s.media_kind" : "s.status";
+    const [groups, rows] = await Promise.all([
+      input.groupBy === "none"
+        ? q.select(sql<string>`'all'`.as("key")).select(sql<number>`count(*)::int`.as("count")).execute()
+        : q
+            .select(sql<string>`${sql.ref(groupCol)}::text`.as("key"))
+            .select(sql<number>`count(*)::int`.as("count"))
+            .groupBy(groupCol)
+            .orderBy("count", "desc")
+            .execute(),
+      q
+        .select([
+          "s.id",
+          "s.media_entity_id",
+          "s.title_query",
+          "s.media_kind",
+          "s.mode",
+          "s.status",
+          "s.priority",
+          "s.status_reason",
+          "s.last_searched_at",
+          "s.next_search_at",
+          "s.last_progress_at",
+          "e.title as entity_title",
+        ])
+        .select((eb) => [
+          eb
+            .selectFrom("acquisition_slot as a")
+            .select(sql<number>`count(*)::int`.as("n"))
+            .whereRef("a.subscription_id", "=", "s.id")
+            .as("slot_total"),
+          eb
+            .selectFrom("acquisition_slot as a")
+            .select(sql<number>`count(*)::int`.as("n"))
+            .whereRef("a.subscription_id", "=", "s.id")
+            .where("a.status", "=", "satisfied")
+            .as("slot_satisfied"),
+        ])
+        .orderBy("s.updated_at", "desc")
+        .limit(input.limit)
+        .execute(),
+    ]);
+    const iso = (d: Date | null) => (d ? d.toISOString() : null);
+    return {
+      total: groups.reduce((n, g) => n + Number(g.count), 0),
+      groups: groups.map((g) => ({ key: g.key, count: Number(g.count) })),
+      items: rows.map((r) => ({
+        id: r.id,
+        title: r.entity_title ?? r.title_query,
+        mediaEntityId: r.media_entity_id,
+        mediaKind: r.media_kind,
+        mode: r.mode,
+        status: r.status,
+        priority: r.priority,
+        statusReason: r.status_reason,
+        slots: { total: Number(r.slot_total ?? 0), satisfied: Number(r.slot_satisfied ?? 0) },
+        lastSearchedAt: iso(r.last_searched_at),
+        nextSearchAt: iso(r.next_search_at),
+        lastProgressAt: iso(r.last_progress_at),
+      })),
+    } as never;
+  });
+
+  // AG-3:下载任务详情(按 id 或 hash);关联计划 / 资源 / 业务任务 / 整理记录 / 待处理事项
+  registry.register(capabilitySpec("download.inspect"), async (input) => {
+    const db = deps().ctx.db;
+    const hash = input.hash?.toLowerCase();
+    const found = await db
+      .selectFrom("download_task")
+      .selectAll()
+      .$if(!!input.downloadTaskId, (qb) => qb.where("id", "=", input.downloadTaskId!))
+      .$if(!input.downloadTaskId && !!hash, (qb) =>
+        qb.where((eb) =>
+          eb.or([eb("torrent_hash", "=", hash!), eb("infohash_v1", "=", hash!), eb("infohash_v2", "=", hash!)]),
+        ),
+      )
+      .orderBy("updated_at", "desc")
+      .limit(10)
+      .execute();
+    const dt = found[0];
+    if (!dt) {
+      if (input.downloadTaskId) {
+        const t = await db.selectFrom("task").select("id").where("id", "=", input.downloadTaskId).executeTakeFirst();
+        if (t) throw new CapabilityError("not_found", "这是业务任务 ID，不是下载任务 ID；业务任务用 task.inspect 查看");
+      }
+      throw new CapabilityError("not_found", "下载任务不存在");
+    }
+    const plan = dt.acquisition_plan_id
+      ? await db
+          .selectFrom("acquisition_plan")
+          .select(["id", "status", "status_reason", "slot_id", "task_id"])
+          .where("id", "=", dt.acquisition_plan_id)
+          .executeTakeFirst()
+      : undefined;
+    const [release, tasks, organization, decisions] = await Promise.all([
+      dt.source_release_id
+        ? db
+            .selectFrom("source_release")
+            .select(["id", "title", "site_id"])
+            .where("id", "=", dt.source_release_id)
+            .executeTakeFirst()
+        : Promise.resolve(undefined),
+      db
+        .selectFrom("task")
+        .select(["id", "kind", "status", "last_error"])
+        .where((eb) =>
+          eb.or([
+            eb.and([eb("subject_type", "=", "download_task"), eb("subject_id", "=", dt.id)]),
+            ...(plan ? [eb.and([eb("subject_type", "=", "acquisition_plan"), eb("subject_id", "=", plan.id)])] : []),
+          ]),
+        )
+        .orderBy("created_at", "desc")
+        .limit(20)
+        .execute(),
+      db
+        .selectFrom("organization_record")
+        .select(["id", "status", "target_path"])
+        .where("download_task_id", "=", dt.id)
+        .orderBy("created_at", "desc")
+        .limit(20)
+        .execute(),
+      db
+        .selectFrom("decision")
+        .select(["id", "kind", "status", "title"])
+        .where("subject_id", "=", dt.id)
+        .orderBy("created_at", "desc")
+        .limit(20)
+        .execute(),
+    ]);
+    return {
+      downloadTask: toApi(dt) as never,
+      plan: plan
+        ? {
+            id: plan.id,
+            status: plan.status,
+            statusReason: plan.status_reason,
+            slotId: plan.slot_id,
+            taskId: plan.task_id,
+          }
+        : null,
+      release: release ? { id: release.id, title: release.title, siteId: release.site_id } : null,
+      tasks: tasks.map((t) => ({
+        id: t.id,
+        kind: t.kind,
+        status: t.status,
+        lastError: (t.last_error ?? null) as Record<string, unknown> | null,
+      })),
+      organization: organization.map((o) => ({ id: o.id, status: o.status, targetPath: o.target_path })),
+      decisions: decisions.map((x) => ({ id: x.id, kind: x.kind, status: x.status, title: x.title })),
+      otherMatches: found.slice(1).map((x) => x.id),
+    } as never;
+  });
+
   registry.register(capabilitySpec("download.plan"), async (input) => {
     const d = deps();
     const db = d.ctx.db;
@@ -119,17 +292,19 @@ export function registerAoCapabilities(registry: CapabilityRegistry, deps: () =>
       .selectAll()
       .where("id", "=", input.slotId)
       .executeTakeFirstOrThrow();
-    const grant = await db
-      .selectFrom("automation_grant")
-      .select("id")
-      .where("status", "=", "active")
-      .where("action", "in", ["download.submit", "policy.acquisition", "*"])
-      .executeTakeFirst();
+    // 按授权范围判定(AG-3):有覆盖本计划(作品 / 类型 / 每日预算)的持续授权 → 不需审批;否则需要
+    const target = plan ? await planBinding(db, "acquisition_plan", plan.id) : null;
+    const covered = await coveringGrant(
+      db,
+      "download.submit",
+      target ?? (await slotBinding(db, fresh.media_entity_id)),
+      new Date(),
+    );
     return {
       plan: plan ? (toApi(plan) as never) : null,
       slot: toApi(fresh) as never,
       rejected,
-      requiresApproval: !grant,
+      requiresApproval: !("grantId" in covered),
     };
   });
 
@@ -412,3 +587,13 @@ function planHash(request: object, frozen: FrozenPlan): string {
 }
 
 const isUuid = (s: string) => /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i.test(s);
+
+/** 没生成计划时按 Slot 的作品判定授权范围(字节预算按 0 计) */
+async function slotBinding(db: AoDeps["ctx"]["db"], mediaEntityId: string): Promise<PlanBinding> {
+  const e = await db
+    .selectFrom("media_entity")
+    .select(["id", "kind"])
+    .where("id", "=", mediaEntityId)
+    .executeTakeFirst();
+  return { payloadHash: "", revision: 0, mediaEntityId: e?.id ?? null, mediaKind: e?.kind ?? null, bytes: 0 };
+}

apps/server/src/modules/agent/authorize.ts

@@ -102,6 +102,76 @@ export async function planBinding(db: Kysely<DB>, planType: string, planId: stri
   return null;
 }
 
+/**
+ * 持续授权是否覆盖本能力(不含单次审批与会话内审批):action 覆盖、active 未过期、绑定策略仍发布、
+ * 对象范围与每日预算。download.plan 用它判定 requiresApproval,authorize 用它放行。
+ */
+export async function coveringGrant(
+  db: Kysely<DB>,
+  capabilityId: string,
+  target: PlanBinding | null,
+  now: Date,
+): Promise<{ grantId: string } | { reasons: string[] }> {
+  const d = { id: capabilityId };
+  const actions = [...(GRANT_ACTIONS[d.id] ?? [d.id]), "*"];
+  const grants = await db
+    .selectFrom("automation_grant as g")
+    .leftJoin("policy_revision as p", "p.id", "g.policy_revision_id")
+    .select(["g.id", "g.action", "g.scope", "g.budget", "g.policy_revision_id", "p.status as policy_status"])
+    .where("g.status", "=", "active")
+    .where("g.action", "in", actions)
+    .where((eb) => eb.or([eb("g.expires_at", "is", null), eb("g.expires_at", ">", now)]))
+    .orderBy("g.created_at", "desc")
+    .execute();
+  const reasons: string[] = [];
+  for (const g of grants) {
+    if (g.policy_revision_id && g.policy_status !== "published") {
+      reasons.push(`授权 ${g.id} 绑定的策略版本已不是发布状态`);
+      continue;
+    }
+    const scope = (g.scope ?? {}) as { mediaEntityIds?: string[]; mediaKinds?: string[]; capabilities?: string[] };
+    if (scope.capabilities?.length && !scope.capabilities.includes(d.id)) {
+      reasons.push(`授权 ${g.id} 不含能力 ${d.id}`);
+      continue;
+    }
+    if (
+      scope.mediaEntityIds?.length &&
+      (!target?.mediaEntityId || !scope.mediaEntityIds.includes(target.mediaEntityId))
+    ) {
+      reasons.push(`授权 ${g.id} 的对象范围不含目标作品`);
+      continue;
+    }
+    if (scope.mediaKinds?.length && target?.mediaKind && !kindCovered(scope.mediaKinds, target.mediaKind)) {
+      reasons.push(`授权 ${g.id} 的媒体类型范围不含 ${target.mediaKind}`);
+      continue;
+    }
+    const budget = (g.budget ?? {}) as { maxInvocationsPerDay?: number; maxBytesPerDay?: number };
+    if (budget.maxInvocationsPerDay !== undefined || budget.maxBytesPerDay !== undefined) {
+      const day = new Date(now);
+      day.setHours(0, 0, 0, 0);
+      const used = await sql<{ n: number }>`
+        select count(*)::int as n from capability_invocation
+         where grant_id = ${g.id} and status = 'succeeded' and created_at >= ${day}`.execute(db);
+      if (budget.maxInvocationsPerDay !== undefined && (used.rows[0]?.n ?? 0) >= budget.maxInvocationsPerDay) {
+        reasons.push(`授权 ${g.id} 今日次数预算已用尽`);
+        continue;
+      }
+      if (budget.maxBytesPerDay !== undefined && d.id === "download.submit") {
+        const bytes = await sql<{ b: string }>`
+          select coalesce(sum(p.reserved_bytes), 0)::bigint as b
+            from capability_invocation ci join acquisition_plan p on p.id::text = ci.input->>'planId'
+           where ci.grant_id = ${g.id} and ci.status = 'succeeded' and ci.created_at >= ${day}`.execute(db);
+        if (Number(bytes.rows[0]?.b ?? 0) + (target?.bytes ?? 0) > budget.maxBytesPerDay) {
+          reasons.push(`授权 ${g.id} 今日字节预算不足`);
+          continue;
+        }
+      }
+    }
+    return { grantId: g.id };
+  }
+  return { reasons };
+}
+
 export function createAuthorizer(deps: {
   db: Kysely<DB>;
   settings: SettingsReader;
@@ -168,62 +238,9 @@ export function createAuthorizer(deps: {
     }
 
     // 2) 持续授权
-    const actions = [...(GRANT_ACTIONS[d.id] ?? [d.id]), "*"];
-    const grants = await db
-      .selectFrom("automation_grant as g")
-      .leftJoin("policy_revision as p", "p.id", "g.policy_revision_id")
-      .select(["g.id", "g.action", "g.scope", "g.budget", "g.policy_revision_id", "p.status as policy_status"])
-      .where("g.status", "=", "active")
-      .where("g.action", "in", actions)
-      .where((eb) => eb.or([eb("g.expires_at", "is", null), eb("g.expires_at", ">", now())]))
-      .orderBy("g.created_at", "desc")
-      .execute();
-    const reasons: string[] = [];
-    for (const g of grants) {
-      if (g.policy_revision_id && g.policy_status !== "published") {
-        reasons.push(`授权 ${g.id} 绑定的策略版本已不是发布状态`);
-        continue;
-      }
-      const scope = (g.scope ?? {}) as { mediaEntityIds?: string[]; mediaKinds?: string[]; capabilities?: string[] };
-      if (scope.capabilities?.length && !scope.capabilities.includes(d.id)) {
-        reasons.push(`授权 ${g.id} 不含能力 ${d.id}`);
-        continue;
-      }
-      if (
-        scope.mediaEntityIds?.length &&
-        (!target?.mediaEntityId || !scope.mediaEntityIds.includes(target.mediaEntityId))
-      ) {
-        reasons.push(`授权 ${g.id} 的对象范围不含目标作品`);
-        continue;
-      }
-      if (scope.mediaKinds?.length && target?.mediaKind && !kindCovered(scope.mediaKinds, target.mediaKind)) {
-        reasons.push(`授权 ${g.id} 的媒体类型范围不含 ${target.mediaKind}`);
-        continue;
-      }
-      const budget = (g.budget ?? {}) as { maxInvocationsPerDay?: number; maxBytesPerDay?: number };
-      if (budget.maxInvocationsPerDay !== undefined || budget.maxBytesPerDay !== undefined) {
-        const day = new Date(now());
-        day.setHours(0, 0, 0, 0);
-        const used = await sql<{ n: number }>`
-          select count(*)::int as n from capability_invocation
-           where grant_id = ${g.id} and status = 'succeeded' and created_at >= ${day}`.execute(db);
-        if (budget.maxInvocationsPerDay !== undefined && (used.rows[0]?.n ?? 0) >= budget.maxInvocationsPerDay) {
-          reasons.push(`授权 ${g.id} 今日次数预算已用尽`);
-          continue;
-        }
-        if (budget.maxBytesPerDay !== undefined && d.id === "download.submit") {
-          const bytes = await sql<{ b: string }>`
-            select coalesce(sum(p.reserved_bytes), 0)::bigint as b
-              from capability_invocation ci join acquisition_plan p on p.id::text = ci.input->>'planId'
-             where ci.grant_id = ${g.id} and ci.status = 'succeeded' and ci.created_at >= ${day}`.execute(db);
-          if (Number(bytes.rows[0]?.b ?? 0) + (target?.bytes ?? 0) > budget.maxBytesPerDay) {
-            reasons.push(`授权 ${g.id} 今日字节预算不足`);
-            continue;
-          }
-        }
-      }
-      return { allowed: true, grantId: g.id };
-    }
+    const covered = await coveringGrant(db, d.id, target, now());
+    if ("grantId" in covered) return { allowed: true, grantId: covered.grantId };
+    const reasons = covered.reasons;
     // 3) 会话内审批:控制台管理员会话里,没有持续授权的计划类动作在步骤处发起审批并挂起会话
     if (ap && target && !approvalId) {
       const planId = inp[ap.field] as string;

apps/server/src/modules/agent/capabilities.ts

@@ -188,7 +188,20 @@ export async function cancelTask(ctx: ServerContext, taskId: string, reason: str
 export function registerTaskCapabilities(registry: CapabilityRegistry, ctx: ServerContext): void {
   registry.register(capabilitySpec("task.inspect"), async (input) => {
     const t = await ctx.db.selectFrom("task").selectAll().where("id", "=", input.taskId).executeTakeFirst();
-    if (!t) throw new CapabilityError("not_found", "任务不存在");
+    if (!t) {
+      // 模型常把下载任务 ID 当业务任务 ID 传进来:明确指出该用哪个能力
+      const dt = await ctx.db
+        .selectFrom("download_task")
+        .select("id")
+        .where("id", "=", input.taskId)
+        .executeTakeFirst();
+      if (dt)
+        throw new CapabilityError(
+          "invalid_input",
+          "这是下载任务 ID，不是业务任务 ID；下载任务用 download.inspect 查看（参数 downloadTaskId）",
+        );
+      throw new CapabilityError("not_found", "任务不存在");
+    }
     const children = await ctx.db
       .selectFrom("task")
       .selectAll()
@@ -203,6 +216,44 @@ export function registerTaskCapabilities(registry: CapabilityRegistry, ctx: Serv
     };
   });
 
+  // AG-3:待处理事项摘要(含可选处理 options);缺省只列未关闭的
+  registry.register(capabilitySpec("decision.list"), async (input) => {
+    const status = input.status?.length ? input.status : (["open", "investigating", "awaiting_user"] as const);
+    let q = ctx.db.selectFrom("decision").where("status", "in", [...status]);
+    if (input.kind?.length) q = q.where("kind", "in", input.kind);
+    const [byKind, rows] = await Promise.all([
+      q
+        .select(["kind", sql<number>`count(*)::int`.as("count")])
+        .groupBy("kind")
+        .orderBy("count", "desc")
+        .execute(),
+      q
+        .selectAll()
+        .orderBy(sql`case severity when 'critical' then 0 when 'warning' then 1 else 2 end`)
+        .orderBy("updated_at", "desc")
+        .limit(input.limit)
+        .execute(),
+    ]);
+    return {
+      total: byKind.reduce((n, k) => n + Number(k.count), 0),
+      byKind: byKind.map((k) => ({ kind: k.kind, count: Number(k.count) })),
+      items: rows.map((d) => ({
+        id: d.id,
+        kind: d.kind,
+        status: d.status,
+        severity: d.severity,
+        title: d.title,
+        summary: d.summary,
+        subjectType: d.subject_type,
+        subjectId: d.subject_id,
+        attempts: d.attempts,
+        options: (Array.isArray(d.options) ? d.options : []) as never,
+        createdAt: d.created_at.toISOString(),
+        updatedAt: d.updated_at.toISOString(),
+      })),
+    };
+  });
+
   registry.register(capabilitySpec("task.retry"), async (input, c) => {
     const row = await retryTask(ctx, input.taskId, c.actorId, input.reason);
     return { taskId: row.id, status: row.status as TaskStatus };