From 86af723b2a3fd2d8a7d092f08d07aa2c40e0e5c6 Mon Sep 17 00:00:00 2001 From: root Date: Sun, 27 Sep 2026 22:35:03 +0800 Subject: [PATCH] feat(agent): W4-E2 in-grant auto recovery for decisions with verified outcome MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - settings.agent.autoInvestigateMode investigate|recover, investigateMaxAttempts, recoverVerifySec - new capability download.recheck (execute, replay never): qB force recheck / resume, no source swap/delete - autoInvestigate moves to agent/investigate.ts: recover mode runs execute with a recovery allowlist only when an active grant (published policy if bound) covers the recovery actions; otherwise degrades to read-only investigation and leaves suggestion + recheck option on the decision - runAgent verifies recovery against business facts (download_task/task/source_release) before finalize; verified -> decision resolved by system:auto-recover, run summary '已恢复(系统核实)'; otherwise stays awaiting_user; per-root-cause attempt cap - e2e E6: missingFiles -> download_failed -> recheck -> verified; no grant degrades; swap rejected; dedupe; attempts exhausted; budget exceeded keeps attention Co-Authored-By: Claude Opus 5.5 --- apps/server/src/modules/agent/authorize.ts | 1 + .../modules/agent/capabilities-download.ts | 100 ++++ .../src/modules/agent/capability-registry.ts | 6 +- apps/server/src/modules/agent/index.ts | 2 + apps/server/src/modules/agent/investigate.ts | 449 ++++++++++++++++++ apps/server/src/modules/agent/runner.ts | 20 +- apps/server/src/modules/agent/tasks.ts | 44 +- .../server/test/e2e/e2-agent-delegate.test.ts | 4 +- apps/server/test/e2e/e6-auto-recover.test.ts | 360 ++++++++++++++ docs/runbook/agent.md | 49 +- packages/contracts/src/capabilities-io.ts | 21 +- packages/contracts/src/capability.ts | 12 + packages/contracts/src/settings.ts | 19 +- 13 files changed, 1072 insertions(+), 15 deletions(-) create mode 100644 apps/server/src/modules/agent/capabilities-download.ts create mode 100644 apps/server/src/modules/agent/investigate.ts create mode 100644 apps/server/test/e2e/e6-auto-recover.test.ts diff --git a/apps/server/src/modules/agent/authorize.ts b/apps/server/src/modules/agent/authorize.ts index 7ebf67a8..fbff2884 100644 --- a/apps/server/src/modules/agent/authorize.ts +++ b/apps/server/src/modules/agent/authorize.ts @@ -16,6 +16,7 @@ export const GRANT_ACTIONS: Record = { "identity.resolve": ["identity.resolve", "policy.identity"], "download.submit": ["download.submit", "policy.acquisition"], "organize.execute": ["organize.execute", "policy.acquisition", "policy.organize"], + "download.recheck": ["download.recheck", "policy.automation"], "task.retry": ["task.retry", "task.*"], "task.cancel": ["task.cancel", "task.*"], "policy.apply": ["policy.apply"], diff --git a/apps/server/src/modules/agent/capabilities-download.ts b/apps/server/src/modules/agent/capabilities-download.ts new file mode 100644 index 00000000..178f0b79 --- /dev/null +++ b/apps/server/src/modules/agent/capabilities-download.ts @@ -0,0 +1,100 @@ +import { createQbClient, type DownloaderClient } from "@umt/adapters"; +import { CapabilityError, type CapabilityRegistry } from "@umt/agent"; +import { capabilitySpec, type DownloadStatus } from "@umt/contracts"; +import type { DownloaderInstanceRow } from "@umt/db"; +import type { ServerContext } from "../types.ts"; +import type { AgentModuleOptions } from "./runner.ts"; + +// download.recheck(REQ §8.4.2 / §10.7 授权内恢复):对已纳管的下载任务发起 qB 强制重新校验或恢复暂停。 +// 只动下载器里的这一个种子:不换源、不删除、不改保存路径 / 分类 / 标签;下载任务状态仍由后台同步按 qB 实际状态更新, +// 是否"已恢复"由 resolveDecision 的核实步骤看 download_task 状态决定,本能力只返回"已发出请求"。 + +/** recheck 允许的起点:文件缺失 / 出错 / 无进展 / 暂停(非阈值、非人工) */ +const RECHECK_FROM: DownloadStatus[] = ["missing", "failed", "stalled", "paused"]; +/** resume 允许的起点 */ +const RESUME_FROM: DownloadStatus[] = ["paused", "stalled", "failed"]; + +async function defaultClient(ctx: ServerContext, inst: DownloaderInstanceRow): Promise { + const settings = (inst.settings ?? {}) as Record; + const secret = inst.secret_ref ? await ctx.secrets.resolve(inst.secret_ref) : undefined; + return createQbClient({ + instanceId: inst.id, + baseUrl: inst.base_url, + ...(inst.username ? { username: inst.username } : {}), + ...(secret ? (settings.authKind === "api_key" ? { apiKey: secret } : { password: secret }) : {}), + }); +} + +/** 持续授权的对象范围:按下载任务 → 计划 → Slot → 作品核对(authorize 对非计划类能力拿不到目标作品) */ +async function checkGrantScope(ctx: ServerContext, grantId: string, planId: string | null) { + const g = await ctx.db.selectFrom("automation_grant").select("scope").where("id", "=", grantId).executeTakeFirst(); + const scope = (g?.scope ?? {}) as { mediaEntityIds?: string[]; mediaKinds?: string[] }; + if (!scope.mediaEntityIds?.length && !scope.mediaKinds?.length) return; + const e = planId + ? await ctx.db + .selectFrom("acquisition_plan as p") + .innerJoin("acquisition_slot as s", "s.id", "p.slot_id") + .innerJoin("media_entity as e", "e.id", "s.media_entity_id") + .select(["e.id", "e.kind"]) + .where("p.id", "=", planId) + .executeTakeFirst() + : undefined; + if (!e) throw new CapabilityError("unauthorized", `授权 ${grantId} 限定了对象范围,但下载任务没有关联作品`); + if (scope.mediaEntityIds?.length && !scope.mediaEntityIds.includes(e.id)) + throw new CapabilityError("unauthorized", `授权 ${grantId} 的对象范围不含目标作品`); + if (scope.mediaKinds?.length) { + const ok = + scope.mediaKinds.includes(e.kind) || + (e.kind !== "movie" && e.kind !== "variety" && scope.mediaKinds.includes("series")); + if (!ok) throw new CapabilityError("unauthorized", `授权 ${grantId} 的媒体类型范围不含 ${e.kind}`); + } +} + +export function registerDownloadRecoveryCapabilities( + registry: CapabilityRegistry, + ctx: ServerContext, + opts: AgentModuleOptions, +): void { + registry.register(capabilitySpec("download.recheck"), async (input, c) => { + const dt = await ctx.db + .selectFrom("download_task") + .selectAll() + .where("id", "=", input.downloadTaskId) + .executeTakeFirst(); + if (!dt) throw new CapabilityError("not_found", "下载任务不存在"); + if (!dt.managed || dt.removed_at || dt.status === "removed") + throw new CapabilityError("precondition", "未纳管或已移除的下载任务不做自动恢复"); + const status = dt.status as DownloadStatus; + if (input.action === "recheck" && !RECHECK_FROM.includes(status)) + throw new CapabilityError("precondition", `当前状态 ${status} 不需要重新校验`); + if (input.action === "resume" && !RESUME_FROM.includes(status)) + throw new CapabilityError("precondition", `当前状态 ${status} 不需要恢复`); + // 阈值暂停不被绕过;人工暂停只由人恢复 + if (status === "paused" && (dt.pause_reason === "threshold" || dt.pause_reason === "user")) + throw new CapabilityError( + "precondition", + `${dt.pause_reason === "threshold" ? "缓存阈值" : "人工"}暂停不做自动恢复`, + ); + if (c.authorization?.grantId) await checkGrantScope(ctx, c.authorization.grantId, dt.acquisition_plan_id); + const inst = await ctx.db + .selectFrom("downloader_instance") + .selectAll() + .where("id", "=", dt.instance_id) + .executeTakeFirst(); + if (!inst) throw new CapabilityError("not_found", "下载器实例不存在"); + const client = opts.downloader ? await opts.downloader(inst) : await defaultClient(ctx, inst); + try { + if (input.action === "recheck") await client.recheck([dt.torrent_hash]); + else await client.resume([dt.torrent_hash]); + } catch (err) { + throw new CapabilityError("internal", `下载器请求失败:${(err as Error).message}`); + } + return { + downloadTaskId: dt.id, + action: input.action, + statusBefore: status, + rawStateBefore: dt.raw_state, + requested: true, + }; + }); +} diff --git a/apps/server/src/modules/agent/capability-registry.ts b/apps/server/src/modules/agent/capability-registry.ts index 1cfd31b8..fb9d4467 100644 --- a/apps/server/src/modules/agent/capability-registry.ts +++ b/apps/server/src/modules/agent/capability-registry.ts @@ -7,7 +7,7 @@ import { type IdempotencyKeyGenerator, stableStringify, } from "@umt/agent"; -import type { ActorType, CapabilityKind, CapabilitySpec } from "@umt/contracts"; +import { type ActorType, CAPABILITY_IDS, type CapabilityKind, type CapabilitySpec } from "@umt/contracts"; import { type DB, type Kysely, sql } from "@umt/db"; import type { Logger } from "pino"; import type { z } from "zod"; @@ -20,7 +20,7 @@ import { createAuthorizer } from "./authorize.ts"; const SIDE_EFFECT: readonly CapabilityKind[] = ["propose", "approve", "execute"]; /** 首批能力总数:启动时断言注册数与此一致(漏挂处理器即启动失败) */ -export const EXPECTED_CAPABILITY_COUNT = 14; +export const EXPECTED_CAPABILITY_COUNT = CAPABILITY_IDS.length; /** * 幂等键:能力 ID + 版本 + 所属运行(agent run / 任务)+ 规范化输入的 sha256。 @@ -187,7 +187,7 @@ export function createCapabilityRegistry(deps: CapabilityRegistryDeps): Capabili }); } -/** 启动断言:首批 14 个能力全部挂上处理器 */ +/** 启动断言:首批能力全部挂上处理器 */ export function assertCapabilityCount(registry: CapabilityRegistry): void { const n = registry.list().length; if (n !== EXPECTED_CAPABILITY_COUNT) { diff --git a/apps/server/src/modules/agent/index.ts b/apps/server/src/modules/agent/index.ts index 4611e1d1..c266cdc4 100644 --- a/apps/server/src/modules/agent/index.ts +++ b/apps/server/src/modules/agent/index.ts @@ -1,5 +1,6 @@ import type { ServerContext, ServerModule } from "../types.ts"; import { registerTaskCapabilities } from "./capabilities.ts"; +import { registerDownloadRecoveryCapabilities } from "./capabilities-download.ts"; import { registerAgentRoutes } from "./routes.ts"; import type { AgentEnv, AgentModuleOptions } from "./runner.ts"; import { agentOutboxRoutes, registerAgentTasks } from "./tasks.ts"; @@ -34,6 +35,7 @@ export function createAgentModule(opts: AgentModuleOptions = {}): ServerModule & outboxRoutes: () => agentOutboxRoutes, registerCapabilities(registry, ctx) { registerTaskCapabilities(registry, ctx); + registerDownloadRecoveryCapabilities(registry, ctx, opts); }, }; } diff --git a/apps/server/src/modules/agent/investigate.ts b/apps/server/src/modules/agent/investigate.ts new file mode 100644 index 00000000..aefa411d --- /dev/null +++ b/apps/server/src/modules/agent/investigate.ts @@ -0,0 +1,449 @@ +import type { DecisionOption, DownloadStatus } from "@umt/contracts"; +import { type DecisionRow, sql } from "@umt/db"; +import type { ServerContext } from "../types.ts"; +import { GRANT_ACTIONS } from "./authorize.ts"; +import { type DriveResult, INVESTIGATE_CAPABILITIES, RECOVER_CAPABILITIES, RECOVERY_ACTIONS } from "./runner.ts"; +import { startRun } from "./service.ts"; + +// 后台自动调查与授权内恢复(REQ §8.4.2、§10.7、AT68 / AT71): +// decision.created → autoInvestigate:按 settings.agent.autoInvestigateMode 决定只读调查或授权内恢复, +// recover 需要有覆盖恢复动作的有效持续授权(策略版本须为 published),否则降级为只读调查并在事项上留建议; +// 同一 decision 只一个 run(dedupe_key=investigate:),同一根因(decision 去重键)的恢复次数有上限。 +// runAgent 结束 → recoveryActions / checkRecovery / applyRecoveryOutcome: +// 按业务事实(download_task / task / source_release 状态)核实恢复结果,核实成功才关闭事项; +// 失败 / 超时 / 预算超限一律保持待处理(finalize 把事项转为 awaiting_user),不自动再试。 + +const KINDS = ["identity_ambiguous", "download_failed", "download_stalled"] as const; +export const AUTO_ACTOR = "system:auto-investigate"; +export const RECOVER_ACTOR = "system:auto-recover"; + +/** 各类事项在 recover 模式下需要被授权覆盖的恢复动作 */ +const ACTIONS_FOR: Record = { + download_failed: ["download.recheck", "task.retry"], + download_stalled: ["download.recheck", "task.retry"], + identity_ambiguous: ["identity.resolve"], +}; + +/** 下载任务处于这些状态即视为已恢复(qB 重新开始下载或已经完成) */ +const DL_RECOVERED: DownloadStatus[] = ["downloading", "queued", "completed", "importing", "imported"]; + +export interface AutoRecoverEvidence { + mode: "investigate" | "recover"; + requestedMode: "investigate" | "recover"; + /** 本根因第几次自动恢复(只在 recover 时计数) */ + attempt: number; + maxAttempts: number; + recoverAttempted: boolean; + degraded?: { reason: string } | undefined; + skipped?: string | undefined; + runId?: string | null; + outcome?: RecoveryOutcomeKind; + actions?: RecoveryAction[]; + denied?: { capabilityId: string; reason: string | null }[]; + facts?: Record; + suggestion?: string; + verifiedAt?: string; +} + +type RecoveryOutcomeKind = "recovered" | "failed" | "unverified" | "not_attempted" | "budget_exceeded" | "cancelled"; + +export interface RecoveryAction { + capabilityId: string; + input: Record; + status: string; +} + +export interface RecoveryCheck { + state: "recovered" | "failed" | "pending"; + facts: Record; +} + +export interface RecoveryOutcome { + decisionId: string; + outcome: RecoveryOutcomeKind; + summary: string; +} + +const decisionIdOf = (dedupeKey: string | null | undefined) => + dedupeKey?.startsWith("investigate:") ? dedupeKey.slice("investigate:".length) : null; + +export const isInvestigationRun = (run: { dedupe_key: string | null }) => decisionIdOf(run.dedupe_key) !== null; + +async function mergeEvidence(ctx: ServerContext, decisionId: string, patch: Partial) { + await ctx.db + .updateTable("decision") + .set({ + evidence: sql`evidence || jsonb_build_object('autoRecover', + coalesce(evidence->'autoRecover', '{}'::jsonb) || ${JSON.stringify(patch)}::jsonb)`, + }) + .where("id", "=", decisionId) + .execute(); +} + +/** 有没有覆盖这些动作的有效持续授权(未撤销、未过期;绑定策略时策略须仍是 published) */ +export async function recoveryGrants(ctx: ServerContext, capabilityIds: readonly string[]): Promise { + const actions = [...new Set([...capabilityIds.flatMap((c) => GRANT_ACTIONS[c] ?? [c]), "*"])]; + const rows = await ctx.db + .selectFrom("automation_grant as g") + .leftJoin("policy_revision as p", "p.id", "g.policy_revision_id") + .select(["g.id", "g.scope"]) + .where("g.status", "=", "active") + .where("g.action", "in", actions) + .where((eb) => eb.or([eb("g.expires_at", "is", null), eb("g.expires_at", ">", new Date())])) + .where((eb) => eb.or([eb("g.policy_revision_id", "is", null), eb("p.status", "=", "published")])) + .execute(); + return rows + .filter((g) => { + const caps = ((g.scope ?? {}) as { capabilities?: string[] }).capabilities; + return !caps?.length || caps.some((c) => capabilityIds.includes(c)); + }) + .map((g) => g.id); +} + +/** 下载类事项关联的业务任务(提交下载 / 核验入库等),给模型 task.inspect / task.retry 用 */ +async function relatedTask(ctx: ServerContext, d: DecisionRow) { + if (d.task_id) return d.task_id; + if (d.subject_type !== "download_task" || !d.subject_id) return null; + const dt = await ctx.db + .selectFrom("download_task") + .select(["acquisition_plan_id"]) + .where("id", "=", d.subject_id) + .executeTakeFirst(); + const t = await ctx.db + .selectFrom("task") + .select("id") + .where((eb) => + eb.or([ + eb.and([eb("subject_type", "=", "download_task"), eb("subject_id", "=", d.subject_id!)]), + ...(dt?.acquisition_plan_id + ? [eb.and([eb("subject_type", "=", "acquisition_plan"), eb("subject_id", "=", dt.acquisition_plan_id)])] + : []), + ]), + ) + .orderBy("created_at", "desc") + .executeTakeFirst(); + return t?.id ?? null; +} + +function suggestionFor(d: DecisionRow, reason: string): string { + if (d.kind === "identity_ambiguous") + return `${reason}。建议在待处理中心人工选择身份,或发布识别策略授权 identity.resolve。`; + return ( + `${reason}。建议:在待处理中心选择"强制重新校验"(download.recheck),或为自动恢复发放持续授权` + + `(action=download.recheck / task.retry,可绑定已发布的 automation 策略);换源、删除仍需人工决定。` + ); +} + +/** 事项上补一个可直接执行的"强制重新校验"选项(人工点选仍经授权 / 审计 / 账本) */ +async function addRecheckOption(ctx: ServerContext, d: DecisionRow) { + if (d.subject_type !== "download_task" || !d.subject_id) return; + const opts = (d.options ?? []) as DecisionOption[]; + if (opts.some((o) => o.key === "recheck")) return; + const next: DecisionOption[] = [ + ...opts, + { + key: "recheck", + label: "强制重新校验", + description: "让下载器重新校验已有文件(不换源、不删除)", + capabilityId: "download.recheck", + input: { downloadTaskId: d.subject_id, action: "recheck" }, + }, + ]; + await ctx.db + .updateTable("decision") + .set({ options: JSON.stringify(next) }) + .where("id", "=", d.id) + .execute(); +} + +/** + * decision.created → 有界调查 / 授权内恢复。同一 decision 只一个 run; + * 同一根因(去重键)已自动恢复过 maxAttempts 次则不再唤醒模型,直接交还人工。 + */ +export async function autoInvestigate( + ctx: ServerContext, + decisionId: string, +): Promise<{ runId: string | null; created: boolean; mode?: "investigate" | "recover"; skipped?: string }> { + const d = await ctx.db.selectFrom("decision").selectAll().where("id", "=", decisionId).executeTakeFirst(); + if (d?.status !== "open") return { runId: null, created: false }; + if (!(KINDS as readonly string[]).includes(d.kind)) return { runId: null, created: false }; + // Agent 自己引起的事项不再触发 Agent(防止递归自触发) + if (d.subject_type === "agent_run") return { runId: null, created: false }; + const s = await ctx.settings.get("agent"); + if (!s.autoInvestigate) return { runId: null, created: false }; + + const requested = s.autoInvestigateMode; + let mode: "investigate" | "recover" = requested; + let degraded: { reason: string } | undefined; + let attempt = 0; + if (requested === "recover") { + // 同根因的历史恢复次数(包括已关闭的事项) + const prior = await sql<{ n: number }>` + select count(*)::int as n from decision + where dedupe_key = ${d.dedupe_key} and id <> ${d.id} + and (evidence->'autoRecover'->>'recoverAttempted')::boolean is true`.execute(ctx.db); + attempt = (prior.rows[0]?.n ?? 0) + 1; + if (attempt > s.investigateMaxAttempts) { + const reason = `同一根因已自动恢复 ${attempt - 1} 次(上限 ${s.investigateMaxAttempts}),不再自动处理`; + await mergeEvidence(ctx, d.id, { + mode: "investigate", + requestedMode: requested, + attempt: attempt - 1, + maxAttempts: s.investigateMaxAttempts, + recoverAttempted: false, + skipped: "attempts_exhausted", + suggestion: suggestionFor(d, reason), + }); + await addRecheckOption(ctx, d); + const r = await ctx.db + .updateTable("decision") + .set({ status: "awaiting_user" }) + .where("id", "=", d.id) + .where("status", "=", "open") + .returning("id") + .executeTakeFirst(); + if (r) + await ctx.outbox.emit( + ctx.db, + "decision.updated", + { decisionId: d.id, status: "awaiting_user" }, + { idempotencyKey: `decision.updated:${d.id}:attempts_exhausted` }, + ); + return { runId: null, created: false, skipped: "attempts_exhausted" }; + } + const grants = await recoveryGrants(ctx, ACTIONS_FOR[d.kind] ?? []); + if (grants.length === 0) { + mode = "investigate"; + degraded = { reason: `没有覆盖 ${(ACTIONS_FOR[d.kind] ?? []).join(" / ")} 的有效持续授权` }; + } + } + + const dt = + d.subject_type === "download_task" && d.subject_id + ? await ctx.db + .selectFrom("download_task") + .select(["id", "name", "status", "raw_state", "torrent_hash"]) + .where("id", "=", d.subject_id) + .executeTakeFirst() + : undefined; + const taskId = await relatedTask(ctx, d); + const goal = [ + `调查待处理事项 ${d.id}(${d.kind}):${d.title}。`, + d.summary ? `摘要:${d.summary}。` : "", + d.subject_type && d.subject_id ? `对象:${d.subject_type} ${d.subject_id}。` : "", + dt ? `下载任务 ${dt.id}「${dt.name ?? ""}」当前 ${dt.status}(qB 状态 ${dt.raw_state ?? "未知"})。` : "", + taskId ? `关联任务 ${taskId}。` : "", + mode === "recover" + ? "在已授权范围内做有界恢复:先用 task.inspect 等只读能力查证,再按需调用 download.recheck(强制重新校验 / 恢复暂停)、" + + "task.retry(重试失败任务)或 identity.resolve,每种最多一次;不换源、不删除、不清理;恢复结果由系统核实。" + : "只做有界的只读调查:查证据、给出结论与处理建议;不要提交下载或变更文件。", + ].join(""); + const { run, created } = await startRun(ctx, { + goal, + mode: mode === "recover" ? "execute" : "plan_only", + capabilityAllowlist: mode === "recover" ? [...RECOVER_CAPABILITIES] : [...INVESTIGATE_CAPABILITIES], + trigger: "event", + createdBy: mode === "recover" ? RECOVER_ACTOR : AUTO_ACTOR, + dedupeKey: `investigate:${d.id}`, + budget: { maxToolCalls: s.investigateMaxToolCalls, maxDurationSec: Math.min(600, s.maxDurationSec) }, + }); + if (created) { + const r = await ctx.db + .updateTable("decision") + .set({ status: "investigating", agent_run_id: run.id }) + .where("id", "=", d.id) + .where("status", "=", "open") + .returning("id") + .executeTakeFirst(); + if (r) { + await mergeEvidence(ctx, d.id, { + mode, + requestedMode: requested, + attempt, + maxAttempts: s.investigateMaxAttempts, + recoverAttempted: mode === "recover", + ...(degraded ? { degraded, suggestion: suggestionFor(d, degraded.reason) } : {}), + runId: run.id, + }); + if (degraded) await addRecheckOption(ctx, d); + } + } + return { runId: run.id, created, mode }; +} + +/** 本 run 调用过的恢复动作(账本)与被拒绝的调用(审计) */ +export async function recoveryActions( + ctx: ServerContext, + runId: string, +): Promise<{ actions: RecoveryAction[]; denied: { capabilityId: string; reason: string | null }[] }> { + const rows = await ctx.db + .selectFrom("capability_invocation") + .select(["capability_id", "input", "status"]) + .where("agent_run_id", "=", runId) + .where("capability_id", "in", [...RECOVERY_ACTIONS]) + .orderBy("created_at") + .execute(); + const denied = await ctx.db + .selectFrom("audit_event") + .select(["target_id", sql`payload->>'errorMessage'`.as("reason")]) + .where("action", "=", "capability.denied") + .where("trace_id", "=", runId) + .where(sql`payload->>'agentRunId'`, "=", runId) + .orderBy("at") + .execute(); + return { + actions: rows.map((r) => ({ + capabilityId: r.capability_id, + input: (r.input ?? {}) as Record, + status: r.status, + })), + denied: denied.map((x) => ({ capabilityId: x.target_id ?? "", reason: x.reason })), + }; +} + +/** 按业务事实核实恢复动作:全部已恢复 → recovered;任一明确失败 → failed;否则 pending(继续等) */ +export async function checkRecovery(ctx: ServerContext, actions: RecoveryAction[]): Promise { + const facts: Record = {}; + const states: RecoveryCheck["state"][] = []; + for (const a of actions.filter((x) => x.status === "succeeded")) { + if (a.capabilityId === "download.recheck") { + const id = String(a.input.downloadTaskId ?? ""); + const dt = await ctx.db + .selectFrom("download_task") + .select(["status", "raw_state", "progress"]) + .where("id", "=", id) + .executeTakeFirst(); + facts[`download_task:${id}`] = dt ? { status: dt.status, rawState: dt.raw_state } : null; + states.push( + !dt || dt.status === "removed" + ? "failed" + : DL_RECOVERED.includes(dt.status as DownloadStatus) + ? "recovered" + : "pending", + ); + } else if (a.capabilityId === "task.retry") { + const id = String(a.input.taskId ?? ""); + const t = await ctx.db.selectFrom("task").select(["status"]).where("id", "=", id).executeTakeFirst(); + facts[`task:${id}`] = t ? { status: t.status } : null; + states.push( + t?.status === "succeeded" + ? "recovered" + : !t || ["failed", "dead", "cancelled"].includes(t.status) + ? "failed" + : "pending", + ); + } else if (a.capabilityId === "identity.resolve") { + const id = String(a.input.sourceReleaseId ?? ""); + const r = id + ? await ctx.db + .selectFrom("source_release") + .select(["resolution_status", "media_entity_id"]) + .where("id", "=", id) + .executeTakeFirst() + : undefined; + facts[`source_release:${id}`] = r ? { resolutionStatus: r.resolution_status } : null; + states.push(r?.resolution_status === "resolved" && r.media_entity_id ? "recovered" : "pending"); + } + } + const state = states.includes("failed") ? "failed" : states.every((x) => x === "recovered") ? "recovered" : "pending"; + return { state, facts }; +} + +/** + * 核实结论写回:recovered → 事项 resolved(resolution.execution 记恢复动作,executeResolution 不会重复执行); + * 其余 → 只写证据,事项保持 investigating,由 finalize 转 awaiting_user(交还人工,不自动再试)。 + */ +export async function applyRecoveryOutcome( + ctx: ServerContext, + runId: string, + res: DriveResult, + acts: { actions: RecoveryAction[]; denied: { capabilityId: string; reason: string | null }[] }, + check: RecoveryCheck | null, +): Promise { + const run = await ctx.db + .selectFrom("agent_run") + .select(["dedupe_key", "cancel_requested_at"]) + .where("id", "=", runId) + .executeTakeFirstOrThrow(); + const decisionId = decisionIdOf(run.dedupe_key); + if (!decisionId) return null; + const d = await ctx.db.selectFrom("decision").selectAll().where("id", "=", decisionId).executeTakeFirst(); + if (!d) return null; + const done = acts.actions.filter((a) => a.status === "succeeded"); + let outcome: RecoveryOutcomeKind; + if (res.status === "cancelled" || run.cancel_requested_at) outcome = "cancelled"; + else if (check?.state === "recovered") outcome = "recovered"; + else if (check?.state === "failed") outcome = "failed"; + else if (done.length > 0) outcome = "unverified"; + else if (res.reason === "budget") outcome = "budget_exceeded"; + else outcome = "not_attempted"; + const what = done.map((a) => a.capabilityId).join(" + "); + const summary = + outcome === "recovered" + ? `已恢复(系统核实):执行 ${what} 后业务状态已恢复。` + : outcome === "failed" + ? `未恢复:执行 ${what} 后业务状态仍为失败,已交还待处理中心。` + : outcome === "unverified" + ? `未恢复:执行 ${what} 后在核实时限内未见业务状态恢复,已交还待处理中心。` + : outcome === "budget_exceeded" + ? `未恢复:调查超出预算(${res.message ?? ""}),未执行恢复动作,已交还待处理中心。` + : outcome === "cancelled" + ? "未恢复:运行已取消。" + : `未恢复:没有执行恢复动作${acts.denied.length ? `(被拒绝:${acts.denied.map((x) => x.capabilityId).join(", ")})` : ""},已交还待处理中心。`; + const evidence: Partial = { + runId, + outcome, + actions: acts.actions, + denied: acts.denied, + facts: check?.facts ?? {}, + verifiedAt: new Date().toISOString(), + ...(outcome === "recovered" ? {} : { suggestion: suggestionFor(d, summary.replace(/。$/, "")) }), + }; + await mergeEvidence(ctx, decisionId, evidence); + if (outcome === "recovered") { + await ctx.db.transaction().execute(async (trx) => { + const r = await trx + .updateTable("decision") + .set({ + status: "resolved", + resolved_by: RECOVER_ACTOR, + resolved_at: new Date(), + resolution: JSON.stringify({ + optionKey: null, + by: RECOVER_ACTOR, + note: summary, + execution: { kind: "agent_recover", runId, actions: done, verified: true, facts: check?.facts ?? {} }, + }), + }) + .where("id", "=", decisionId) + .where("status", "in", ["investigating", "open", "awaiting_user"]) + .returning("id") + .executeTakeFirst(); + if (r) + await ctx.outbox.emit( + trx, + "decision.updated", + { decisionId, status: "resolved" }, + { idempotencyKey: `decision.updated:${decisionId}:recovered:${runId}` }, + ); + }); + } else { + const fresh = await ctx.db.selectFrom("decision").selectAll().where("id", "=", decisionId).executeTakeFirst(); + if (fresh) await addRecheckOption(ctx, fresh); + } + return { decisionId, outcome, summary }; +} + +/** finalize 之后:把系统核实结论放在 run 汇总最前面(模型文本仍标注"未经核查") */ +export async function annotateRunSummary(ctx: ServerContext, runId: string, rec: RecoveryOutcome) { + const run = await ctx.db + .selectFrom("agent_run") + .select(["summary", "status"]) + .where("id", "=", runId) + .executeTakeFirstOrThrow(); + if (run.summary?.startsWith(rec.summary)) return { status: run.status, summary: run.summary }; + const summary = `${rec.summary}\n${run.summary ?? ""}`.slice(0, 4000); + await ctx.db.updateTable("agent_run").set({ summary }).where("id", "=", runId).execute(); + void ctx.events.publish("agent.run_status", { runId, status: run.status, summary: summary.slice(0, 2000) }); + return { status: run.status, summary }; +} diff --git a/apps/server/src/modules/agent/runner.ts b/apps/server/src/modules/agent/runner.ts index 7e6b01e9..8d8e9be4 100644 --- a/apps/server/src/modules/agent/runner.ts +++ b/apps/server/src/modules/agent/runner.ts @@ -1,4 +1,5 @@ import { join, resolve } from "node:path"; +import type { DownloaderClient } from "@umt/adapters"; import { type AgentRunHandle, type AgentSettings, @@ -21,7 +22,7 @@ import { type GlobalSettings, KnownCapabilityId, } from "@umt/contracts"; -import { type AgentRunRow, sql } from "@umt/db"; +import { type AgentRunRow, type DownloaderInstanceRow, sql } from "@umt/db"; import type { ServerContext } from "../types.ts"; import { DEMO_SCRIPTS, parseFakeScripts } from "./fake-scripts.ts"; import { buildGoalReport } from "./report.ts"; @@ -36,6 +37,8 @@ export interface AgentModuleOptions { fakeScripts?: FakeScript[]; /** 覆盖 settings.agent.goalPollSec(测试用小值) */ goalPollSec?: number; + /** download.recheck 用的下载器客户端;缺省按实例配置 + secret 现场创建 qB 客户端 */ + downloader?: (inst: DownloaderInstanceRow) => Promise; } export interface AgentEnv { @@ -54,6 +57,14 @@ export const INVESTIGATE_CAPABILITIES = [ "task.inspect", "identity.resolve", ] as const; +/** + * 授权内恢复(settings.agent.autoInvestigateMode = recover)的能力白名单:只读调查 + 恢复类 + * (重试失败任务、强制重新校验 / 恢复暂停)。换源(download.plan / download.submit)、删除 / 清理、 + * 取消任务、整理执行、策略发布一律不在白名单内。实际能否执行仍由 authorize 按持续授权判定。 + */ +export const RECOVER_CAPABILITIES = [...INVESTIGATE_CAPABILITIES, "task.retry", "download.recheck"] as const; +/** 被视为"恢复动作"的能力(结果需系统核实) */ +export const RECOVERY_ACTIONS = ["task.retry", "download.recheck", "identity.resolve"] as const; export interface RunScope { capabilities: string[]; @@ -243,7 +254,12 @@ function systemPromptFor(run: AgentRunRow, scope: RunScope): string { ]; if (run.mode === "plan_only") constraints.push("本次为“先给方案”:只调查与规划,不提交下载、不做文件变更、不发布策略。"); - if (run.trigger === "event") + if (run.trigger === "event" && run.mode === "execute") + constraints.push( + "本任务由后台事件触发,是有界的授权内恢复:先查证据,再只在授权范围内执行恢复(task.retry / download.recheck / identity.resolve)," + + "每种恢复动作最多一次;不换源、不删除、不清理;恢复是否成功由系统按业务状态核实,不要自行宣称已恢复。", + ); + else if (run.trigger === "event") constraints.push("本任务由后台事件触发,是有界的只读调查:收集证据、给出结论与建议,不要反复重试同一操作。"); return buildSystemPrompt({ role: run.trigger === "event" ? "媒体库异常调查员" : "媒体库自动化助手", diff --git a/apps/server/src/modules/agent/tasks.ts b/apps/server/src/modules/agent/tasks.ts index 40435d5a..602b9382 100644 --- a/apps/server/src/modules/agent/tasks.ts +++ b/apps/server/src/modules/agent/tasks.ts @@ -4,9 +4,19 @@ import { defineTask, type OutboxRouteFn, type TaskRegistry } from "@umt/tasks"; import { z } from "zod"; import type { ServerContext } from "../types.ts"; import { retryTask } from "./capabilities.ts"; +import { + annotateRunSummary, + applyRecoveryOutcome, + autoInvestigate, + checkRecovery, + isInvestigationRun, + type RecoveryCheck, + type RecoveryOutcome, + recoveryActions, +} from "./investigate.ts"; import { buildGoalReport } from "./report.ts"; import { type AgentEnv, type DriveResult, drive, goalSettled } from "./runner.ts"; -import { autoInvestigate, finalize } from "./service.ts"; +import { finalize } from "./service.ts"; // 持久任务:runAgent(驱动一次有界 Agent 会话 → 等目标关联的下载/入库落定 → 以业务表汇报) // 与 resolveDecision(decision.created → 自动调查;decision.updated(resolved)→ 执行所选处理)。 @@ -37,10 +47,31 @@ export function registerAgentTasks(registry: TaskRegistry, env: () => AgentEnv) let settled: AgentGoalReport | null = null; const run = await e.ctx.db .selectFrom("agent_run") - .select(["mode"]) + .select(["mode", "dedupe_key"]) .where("id", "=", runId) .executeTakeFirstOrThrow(); - if (res.status === "completed" && run.mode === "execute") { + // 后台授权内恢复:核实业务事实(下载 / 任务 / 身份状态)后再收尾;不等"目标入库" + const recovering = run.mode === "execute" && isInvestigationRun(run); + let recovery: RecoveryOutcome | null = null; + if (recovering) { + const acts = await t.step.run("recover-actions", () => recoveryActions(e.ctx, runId)); + let check: RecoveryCheck | null = null; + if (res.status === "completed" && acts.actions.some((a) => a.status === "succeeded")) { + const s = await e.ctx.settings.get("agent"); + const pollSec = Math.min(e.opts.goalPollSec ?? s.goalPollSec, 30); + check = await t.step.waitForFact("recover-verified", { + check: async () => { + const c = await checkRecovery(e.ctx, acts.actions); + return c.state === "pending" ? null : c; + }, + timeout: `${pollSec}s`, + maxRounds: Math.max(1, Math.min(900, Math.ceil(s.recoverVerifySec / pollSec))), + }); + // 超时:用最后一次观测作为证据(state 仍是 pending → unverified) + check ??= await t.step.run("recover-last-check", () => checkRecovery(e.ctx, acts.actions)); + } + recovery = await t.step.run("recover-outcome", () => applyRecoveryOutcome(e.ctx, runId, res, acts, check)); + } else if (res.status === "completed" && run.mode === "execute") { const s = await e.ctx.settings.get("agent"); const pollSec = e.opts.goalPollSec ?? s.goalPollSec; const rounds = Math.max(1, Math.min(900, Math.ceil((s.goalWaitHours * 3600) / pollSec))); @@ -59,7 +90,12 @@ export function registerAgentTasks(registry: TaskRegistry, env: () => AgentEnv) maxRounds: rounds, }); } - return t.step.run("finalize", () => finalize(e, runId, res, settled)); + const out = await t.step.run("finalize", () => finalize(e, runId, res, settled)); + if (recovery) { + const rec = recovery; + return t.step.run("recover-summary", () => annotateRunSummary(e.ctx, runId, rec)); + } + return out; }, }), defineTask({ diff --git a/apps/server/test/e2e/e2-agent-delegate.test.ts b/apps/server/test/e2e/e2-agent-delegate.test.ts index fa11e148..6296402c 100644 --- a/apps/server/test/e2e/e2-agent-delegate.test.ts +++ b/apps/server/test/e2e/e2-agent-delegate.test.ts @@ -42,8 +42,8 @@ describe.skipIf(!hasPg)("E2 Agent 委托闭环(fake 模型 + 真实能力/授权 return r.body.data.id as string; }; - it("启动断言:14 个能力全部注册,并在能力目录里可见", async () => { - expect(e.booted.ctx.capabilities.list()).toHaveLength(14); + it("启动断言:15 个能力全部注册,并在能力目录里可见", async () => { + expect(e.booted.ctx.capabilities.list()).toHaveLength(15); const r = await e.api("GET", "/capabilities"); expect(r.body.data.items.map((c: { id: string }) => c.id).sort()).toEqual( e.booted.ctx.capabilities diff --git a/apps/server/test/e2e/e6-auto-recover.test.ts b/apps/server/test/e2e/e6-auto-recover.test.ts new file mode 100644 index 00000000..4bae8572 --- /dev/null +++ b/apps/server/test/e2e/e6-auto-recover.test.ts @@ -0,0 +1,360 @@ +import { createHash } from "node:crypto"; +import type { FakeScript } from "@umt/agent"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { RECOVER_CAPABILITIES } from "../../src/modules/agent/runner.ts"; +import { createE2E, type E2E, pgAvailable, runEvents, seedBase, waitFor } from "./harness.ts"; + +// E6 授权内异常调查与恢复(REQ §8.4.2、§10.7、§15 M1、AT68 / AT71): +// qB 任务 missingFiles → 同步写 download_task=missing + download.status_changed → decision.created(download_failed) +// → resolveDecision → autoInvestigate(mode=recover): +// - 没有覆盖 download.recheck 的持续授权:降级为只读调查,不碰 qB(无 recheck 调用),事项留建议 + "强制重新校验"选项; +// - 有绑定已发布策略的授权:有界 run 调用 task.inspect → download.recheck(qB 收到 recheck 并恢复下载), +// 换源类能力(download.submit)不在白名单 → tool_call_rejected;系统按 download_task 状态核实后关闭事项, +// run 汇报"已恢复(系统核实)";同一 decision 重复触发只一个 run;同根因再次出错超过尝试上限不再唤醒模型; +// - 预算超限:run budget_exceeded,事项保持待处理(awaiting_user),不重试。 + +const hasPg = await pgAvailable(); +const ZERO = "00000000-0000-4000-8000-000000000000"; + +/** 当前场景的对象(fake 脚本的输入从这里取;场景串行执行) */ +const cur: { taskId?: string; downloadTaskId?: string } = {}; + +const recoverScript: FakeScript = { + name: "e6-recover", + match: /调查待处理事项/, + steps: [ + { + text: "先看关联任务与下载时间线。", + toolCalls: [{ capability: "task.inspect", input: () => ({ taskId: cur.taskId ?? ZERO }) }], + }, + { + text: "qB 报告文件缺失,强制重新校验。", + toolCalls: [ + { + capability: "download.recheck", + input: () => ({ downloadTaskId: cur.downloadTaskId ?? ZERO, action: "recheck" }), + }, + ], + }, + // 换源类动作:不在恢复白名单内,必须被拒绝 + { text: "顺手换个源重新下载。", toolCalls: [{ capability: "download.submit", input: { planId: ZERO } }] }, + ], + finalText: "已恢复下载。", +}; + +describe.skipIf(!hasPg)("E6 授权内异常调查与恢复", () => { + let e: E2E; + let instanceId: string; + let grantId: string; + + const decisionOf = (dtId: string, status?: string) => + waitFor( + async () => { + let q = e.db + .selectFrom("decision") + .selectAll() + .where("kind", "=", "download_failed") + .where("subject_id", "=", dtId); + if (status) q = q.where("status", "=", status as never); + return q.orderBy("created_at", "desc").executeTakeFirst(); + }, + 60_000, + `download_failed decision(${dtId})`, + ); + + const runOf = (decisionId: string) => + waitFor( + async () => + e.db + .selectFrom("agent_run") + .selectAll() + .where("dedupe_key", "=", `investigate:${decisionId}`) + .executeTakeFirst(), + 60_000, + "自动调查 run", + ); + + const finished = (runId: string) => + waitFor( + async () => { + const r = await e.db.selectFrom("agent_run").selectAll().where("id", "=", runId).executeTakeFirstOrThrow(); + return ["succeeded", "failed", "budget_exceeded", "cancelled"].includes(r.status) && r.finished_at ? r : null; + }, + 90_000, + "run 结束", + ).catch(async (err) => { + const r = await e.db.selectFrom("agent_run").selectAll().where("id", "=", runId).executeTakeFirst(); + const runs = await e.booted.ctx.engine.list({ taskName: "runAgent" }); + console.log("DEBUG", JSON.stringify({ r, runs }, null, 1).slice(0, 6000)); + throw err; + }); + + /** 纳管下载中的任务 + 关联业务任务;随后 qB 报 missingFiles */ + async function failingDownload(n: number) { + const hash = createHash("sha1").update(`e6-${n}`).digest("hex"); + const name = `E6.Sample.${n}.2024.1080p.WEB-DL`; + e.qb.seed({ hash, name, size: 10_000_000, progress: 0.3, tags: ["umt"], save_path: e.cache.root }); + const dt = await e.db + .insertInto("download_task") + .values({ + instance_id: instanceId, + torrent_hash: hash, + name, + save_path: e.cache.root, + tags: ["umt"], + managed: true, + status: "downloading", + raw_state: "downloading", + progress: 0.3, + downloaded_bytes: 3_000_000, + last_progress_at: new Date(), + }) + .returning("id") + .executeTakeFirstOrThrow(); + const task = await e.db + .insertInto("task") + .values({ + kind: "submitDownload", + idempotency_key: `e6:submit:${dt.id}`, + input: JSON.stringify({ downloadTaskId: dt.id }), + subject_type: "download_task", + subject_id: dt.id, + status: "succeeded", + triggered_by: "system:e6", + }) + .returning("id") + .executeTakeFirstOrThrow(); + cur.taskId = task.id; + cur.downloadTaskId = dt.id; + e.qb.setMissingFiles(hash); + return { hash, dtId: dt.id, taskId: task.id }; + } + + const toolStarts = async (runId: string) => + (await runEvents(e, runId)) + .filter((x) => x.type === "tool_call_start") + .map((x) => (x.payload as { capabilityId: string }).capabilityId); + const toolRejected = async (runId: string) => + (await runEvents(e, runId)) + .filter((x) => x.type === "tool_call_rejected") + .map((x) => (x.payload as { toolName: string }).toolName); + + beforeAll(async () => { + e = await createE2E("e6", { fakeScripts: [recoverScript] }); + instanceId = (await seedBase(e)).instanceId; + // download.recheck 走生产路径:按实例用户名 + secret 现场建 qB 客户端 + const ref = await e.booted.ctx.secrets.create({ purpose: "downloader_password", plaintext: "adminadmin" }); + await e.db.updateTable("downloader_instance").set({ secret_ref: ref }).where("id", "=", instanceId).execute(); + await e.settings({ + agent: { + autoInvestigate: true, + autoInvestigateMode: "recover", + investigateMaxToolCalls: 6, + investigateMaxAttempts: 1, + recoverVerifySec: 60, + }, + }); + }, 120_000); + + afterAll(async () => { + await e?.close(); + }, 60_000); + + it("没有持续授权:降级为只读调查,不调用 qB recheck,事项留建议与可执行选项", async () => { + const f = await failingDownload(1); + const d = await decisionOf(f.dtId); + const run = await runOf(d.id); + expect(run).toMatchObject({ trigger: "event", mode: "plan_only", created_by: "system:auto-investigate" }); + const done = await finished(run.id); + expect(done.status).toBe("succeeded"); + expect(await toolStarts(run.id)).toEqual(["task.inspect"]); + // 恢复与换源能力都不在只读白名单里 + expect(await toolRejected(run.id)).toEqual(["download__recheck", "download__submit"]); + expect(e.qb.callsTo("torrents/recheck")).toHaveLength(0); + + const d2 = await waitFor( + async () => { + const r = await e.db.selectFrom("decision").selectAll().where("id", "=", d.id).executeTakeFirstOrThrow(); + return r.status === "awaiting_user" ? r : null; + }, + 30_000, + "事项交还人工", + ); + const ar = (d2.evidence as { autoRecover: Record }).autoRecover; + expect(ar).toMatchObject({ requestedMode: "recover", mode: "investigate", recoverAttempted: false }); + expect((ar.degraded as { reason: string }).reason).toContain("download.recheck"); + expect(ar.suggestion).toMatch(/强制重新校验/); + expect(d2.options).toEqual( + expect.arrayContaining([ + expect.objectContaining({ + key: "recheck", + capabilityId: "download.recheck", + input: { downloadTaskId: f.dtId, action: "recheck" }, + }), + ]), + ); + const row = await e.db.selectFrom("download_task").select("status").where("id", "=", f.dtId).executeTakeFirst(); + expect(row?.status).toBe("missing"); + }, 180_000); + + it("有绑定已发布策略的授权:task.inspect → download.recheck → 系统核实已恢复;换源被拒;重复触发只一个 run", async () => { + const policy = await e.db + .insertInto("policy_revision") + .values({ + policy_key: "e6-auto-recover", + revision: 1, + kind: "automation", + status: "published", + title: "下载异常自动恢复", + published_at: new Date(), + published_by: "admin:e6", + }) + .returning("id") + .executeTakeFirstOrThrow(); + const g = await e.db + .insertInto("automation_grant") + .values({ + action: "download.recheck", + policy_revision_id: policy.id, + granted_by: "admin:e6", + reason: "e6:自动恢复下载异常", + }) + .returning("id") + .executeTakeFirstOrThrow(); + grantId = g.id; + + const f = await failingDownload(2); + const d = await decisionOf(f.dtId); + const run = await runOf(d.id); + expect(run).toMatchObject({ trigger: "event", mode: "execute", created_by: "system:auto-recover" }); + expect(run.goal).toContain(f.dtId); + expect(run.goal).toContain(f.taskId); + expect(run.scope).toMatchObject({ capabilities: [...RECOVER_CAPABILITIES] }); + expect(run.budget).toMatchObject({ maxToolCalls: 6 }); + + // qB 收到强制重新校验 → 校验完成后恢复下载 + const calls = await waitFor( + async () => { + const c = e.qb.callsTo("torrents/recheck"); + return c.length ? c : null; + }, + 60_000, + "qB recheck 调用", + ); + expect(calls[0]!.form.hashes).toBe(f.hash); + e.qb.setState(f.hash, "downloading"); + + const done = await finished(run.id); + expect(done.status).toBe("succeeded"); + expect(done.summary).toMatch(/^已恢复\(系统核实\)/); + expect(done.summary).toContain("模型汇报(未经核查)"); + expect(await toolStarts(run.id)).toEqual(["task.inspect", "download.recheck"]); + expect(await toolRejected(run.id)).toEqual(["download__submit"]); + const inv = await e.db + .selectFrom("capability_invocation") + .selectAll() + .where("agent_run_id", "=", run.id) + .where("capability_id", "=", "download.recheck") + .executeTakeFirstOrThrow(); + expect(inv).toMatchObject({ status: "succeeded", grant_id: grantId }); + expect(e.qb.callsTo("torrents/recheck")).toHaveLength(1); + + // 业务事实:下载恢复,事项由系统核实后关闭 + const dt = await e.db.selectFrom("download_task").select("status").where("id", "=", f.dtId).executeTakeFirst(); + expect(dt?.status).toBe("downloading"); + const d2 = await e.db.selectFrom("decision").selectAll().where("id", "=", d.id).executeTakeFirstOrThrow(); + expect(d2).toMatchObject({ status: "resolved", resolved_by: "system:auto-recover" }); + expect((d2.resolution as { execution: Record }).execution).toMatchObject({ + kind: "agent_recover", + runId: run.id, + verified: true, + }); + expect((d2.evidence as { autoRecover: Record }).autoRecover).toMatchObject({ + mode: "recover", + attempt: 1, + recoverAttempted: true, + outcome: "recovered", + }); + + // 同一 decision 的重复事件:仍只有一个 run + for (const k of ["dup-1", "dup-2"]) + await e.booted.ctx.outbox.emit( + e.db, + "decision.created", + { decisionId: d.id, kind: "download_failed", severity: "warning", title: d.title }, + { idempotencyKey: `e6:${k}` }, + ); + await waitFor( + async () => { + const rows = await e.db + .selectFrom("outbox_event") + .select("status") + .where("idempotency_key", "in", ["e6:dup-1", "e6:dup-2"]) + .execute(); + return rows.length === 2 && rows.every((r) => r.status === "dispatched"); + }, + 30_000, + "重复事件已投递", + ); + await waitFor( + async () => { + const runs = await e.booted.ctx.engine.list({ taskName: "resolveDecision" }); + return runs.length > 0 && runs.every((r) => r.status === "completed") ? runs : null; + }, + 30_000, + "resolveDecision 完成", + ); + expect( + await e.db.selectFrom("agent_run").select("id").where("dedupe_key", "=", `investigate:${d.id}`).execute(), + ).toHaveLength(1); + + // 同根因再次出错:已达自动恢复上限(1 次)→ 不再唤醒模型,直接交还人工 + const runsBefore = (await e.db.selectFrom("agent_run").select("id").execute()).length; + e.qb.setMissingFiles(f.hash); + const again = await waitFor( + async () => + e.db + .selectFrom("decision") + .selectAll() + .where("dedupe_key", "=", `download_failed:${f.dtId}`) + .where("id", "<>", d.id) + .where("status", "=", "awaiting_user") + .executeTakeFirst(), + 60_000, + "同根因新事项交还人工", + ); + expect((again.evidence as { autoRecover: Record }).autoRecover).toMatchObject({ + skipped: "attempts_exhausted", + recoverAttempted: false, + }); + expect((await e.db.selectFrom("agent_run").select("id").execute()).length).toBe(runsBefore); + expect(e.qb.callsTo("torrents/recheck")).toHaveLength(1); + }, 240_000); + + it("预算超限:run budget_exceeded,未执行恢复,事项保持待处理", async () => { + await e.settings({ agent: { investigateMaxToolCalls: 1 } }); + const f = await failingDownload(3); + const d = await decisionOf(f.dtId); + const run = await runOf(d.id); + expect(run.mode).toBe("execute"); + const done = await finished(run.id); + expect(done.status).toBe("budget_exceeded"); + expect(done.summary).toMatch(/^未恢复:调查超出预算/); + expect(await toolStarts(run.id)).toEqual(["task.inspect"]); + expect(e.qb.callsTo("torrents/recheck")).toHaveLength(1); + const d2 = await waitFor( + async () => { + const r = await e.db.selectFrom("decision").selectAll().where("id", "=", d.id).executeTakeFirstOrThrow(); + return r.status === "awaiting_user" ? r : null; + }, + 30_000, + "事项保持待处理", + ); + expect((d2.evidence as { autoRecover: Record }).autoRecover).toMatchObject({ + outcome: "budget_exceeded", + recoverAttempted: true, + }); + const dt = await e.db.selectFrom("download_task").select("status").where("id", "=", f.dtId).executeTakeFirst(); + expect(dt?.status).toBe("missing"); + }, 180_000); +}); diff --git a/docs/runbook/agent.md b/docs/runbook/agent.md index 6c3a81a5..07ced5ca 100644 --- a/docs/runbook/agent.md +++ b/docs/runbook/agent.md @@ -6,7 +6,7 @@ ## 能力 -启动时装配 14 个能力(`catalog.lookup`、`release.search`、`release.inspect`、`identity.resolve`、`download.plan`、`download.submit`、`organize.plan`、`organize.execute`、`task.inspect`、`task.retry`、`task.cancel`、`policy.preview`、`policy.apply`、`inventory.query`)。注册数不等于 14 时进程启动失败。Web(`POST /capabilities/{id}/invoke`)、Pi 工具和后台任务都走 `ctx.capabilities.invoke` 这一条路径。 +启动时装配 15 个能力(`catalog.lookup`、`release.search`、`release.inspect`、`identity.resolve`、`download.plan`、`download.submit`、`download.recheck`、`organize.plan`、`organize.execute`、`task.inspect`、`task.retry`、`task.cancel`、`policy.preview`、`policy.apply`、`inventory.query`)。注册数不等于 `CAPABILITY_IDS.length` 时进程启动失败。Web(`POST /capabilities/{id}/invoke`)、Pi 工具和后台任务都走 `ctx.capabilities.invoke` 这一条路径。 - **授权**(`agent/authorize.ts`):propose / approve / execute 类必须满足以下之一: - 单次审批 `approval`:状态为 approved,绑定计划类型 + 计划 ID + payload hash + revision,且未过期。执行成功后审批标记为 consumed。 @@ -57,7 +57,7 @@ worker 崩溃后,新 worker 在引擎租约到期后(约 30s)接手,`drive` 步 ## 后台触发与 resolveDecision -- `decision.created` 路由到 `resolveDecision(trigger=created)`。满足以下全部条件时,会发起一次有界调查 run:`agent.autoInvestigate=true`、kind 为 `identity_ambiguous` / `download_failed` / `download_stalled`、事项仍为 open、事项不是由 Agent 自己引起的。这个 run 的参数:`trigger=event`、`mode=plan_only`,白名单为只读能力加 `identity.resolve`,工具调用上限为 `agent.investigateMaxToolCalls`,`dedupe_key=investigate:`。同一事项只会有一个 run,重复事件或重新打开都不会再建。调查结束后,事项转为 `awaiting_user`,`evidence.agentInvestigation` 里带上结论。 +- `decision.created` 路由到 `resolveDecision(trigger=created)`,由 `agent/investigate.ts` 的 `autoInvestigate` 处理。满足以下全部条件时,会发起一次有界 run:`agent.autoInvestigate=true`、kind 为 `identity_ambiguous` / `download_failed` / `download_stalled`、事项仍为 open、事项不是由 Agent 自己引起的。`trigger=event`,工具调用上限为 `agent.investigateMaxToolCalls`,时长上限 `min(600, agent.maxDurationSec)` 秒,`dedupe_key=investigate:`。同一事项只会有一个 run,重复事件或重新打开都不会再建。模式见下一节。调查结束后,没有被系统核实恢复的事项转为 `awaiting_user`,`evidence.agentInvestigation` 里带上结论。 - `decision.updated(status=resolved)` 路由到 `resolveDecision(trigger=updated)`,执行用户选中的处理: - 选项带 `capabilityId` 时经注册中心调用(照样校验授权); - `retry` 重试关联任务; @@ -66,6 +66,48 @@ worker 崩溃后,新 worker 在引擎租约到期后(约 30s)接手,`drive` 步 执行结果写进 `decision.resolution.execution`。身份歧义类由 sources-identity 在处理端点里直接执行。 +## 自动调查与恢复 + +依据 REQ §8.4.2(下载无进展默认不换源)、§10.7(有界任务、同根因去重、有限重试)、AT68 / AT71。 + +**模式**(`agent.autoInvestigateMode`): + +| 模式 | run 参数 | 能做什么 | +|---|---|---| +| `investigate`(缺省) | `mode=plan_only`,白名单 = 只读能力 + `identity.resolve` | 查证据、给结论与建议,事项交还人工 | +| `recover` | `mode=execute`,白名单 = 上面 + `task.retry` + `download.recheck`,`created_by=system:auto-recover` | 在持续授权范围内执行恢复动作,系统核实后自动关闭事项 | + +**恢复范围**: +- `download.recheck`(execute,replay=never):对已纳管、未移除的下载任务调 qB 强制重新校验(`action=recheck`,起点 missing / failed / stalled / paused),或恢复暂停(`action=resume`)。缓存阈值暂停和人工暂停不自动恢复。只动这一个种子,不换源、不删除、不改路径 / 分类 / 标签。 +- `task.retry`:重试失败 / 死信的业务任务,副作用由各处理器的操作账本去重。 +- `identity.resolve`:按已有证据等级自动接受;证据不够时仍进待处理。 + +**明确排除**:换源(`download.plan` / `download.submit`)、删除与缓存清理、`task.cancel`、`organize.execute`、`policy.apply`。这些能力不在恢复白名单里,模型调用会得到 `tool_call_rejected`。 + +**授权**:每次恢复调用仍经 `authorize` 校验持续授权,模型声明无效。可用的 grant: +- `action=download.recheck`,或发布 kind=automation 的策略时自动生成的 `policy.automation`; +- `action=task.retry` / `task.*`;`identity.resolve` / `policy.identity`;或 `*`。 + +grant 必须 active、未过期,绑定策略版本时该版本必须仍是 published。`scope.mediaKinds` / `scope.mediaEntityIds` 由 `download.recheck` 处理器按下载任务 → 计划 → Slot → 作品核对。注意:authorize 对非计划类能力拿不到目标作品,所以带 `mediaEntityIds` 的 grant 目前一律不覆盖 `download.recheck`,是保守拒绝。建 run 前先查有没有覆盖该类事项恢复动作的 grant(下载类看 `download.recheck` / `task.retry`,身份类看 `identity.resolve`)。没有就**降级为 investigate**,并在事项上写 `evidence.autoRecover.degraded` 和 `suggestion`,再追加一个可直接点选的"强制重新校验"选项(`capabilityId=download.recheck`,人工点选同样经授权、审计、账本)。 + +授权示例(把恢复限定在某个已发布策略下): + +```sh +curl -X POST localhost:4380/api/v1/automation-grants -H "Authorization: Bearer " -H 'content-type: application/json' \ + -d '{"action":"download.recheck","policyRevisionId":"","reason":"下载异常自动重新校验"}' +``` + +**核实与收尾**(`runAgent` 里的 `recover-*` 步骤): +1. 从账本(`capability_invocation`)取出本 run 成功的恢复动作,从审计取出被拒绝的调用。 +2. 按业务事实轮询核实,间隔 `min(goalPollSec, 30)` 秒,最长 `agent.recoverVerifySec`: + - `download_task` 进入 downloading / queued / completed / importing / imported 算恢复,removed 算失败; + - 重试的 `task` 为 succeeded 算恢复,failed / dead / cancelled 算失败; + - `source_release` 已 resolved 算恢复。 +3. 核实已恢复:事项转 `resolved`,`resolved_by=system:auto-recover`,`resolution.execution={kind:"agent_recover", verified:true, facts}`,发 `decision.updated`(executeResolution 看到 execution 已存在,不会重复执行)。run 汇总以"已恢复(系统核实)"开头,模型原话附在后面,标注"未经核查"。 +4. 失败、超时(unverified)、预算超限、没有执行恢复动作:只写 `evidence.autoRecover.outcome` 和建议,事项由 finalize 转 `awaiting_user`,不会自动再试。run 汇总以"未恢复:…"开头。 + +**有界尝试**:`evidence.autoRecover.attempt` 记录该根因的第几次自动恢复。同一去重键(如 `download_failed:`)下,历史上 `recoverAttempted=true` 的事项数达到 `agent.investigateMaxAttempts`(缺省 2)后,同根因的新事项不再唤醒模型,直接 `awaiting_user`,并写 `skipped=attempts_exhausted`。 + ## 配置上游 在设置页或 `PATCH /settings` 里配置: @@ -85,6 +127,8 @@ worker 崩溃后,新 worker 在引擎租约到期后(约 30s)接手,`drive` 步 "maxToolCalls": 40, "maxDurationSec": 900, // 单次运行预算 "maxTokensPerRun": null, "maxCostUsdPerRun": null, "autoInvestigate": false, "investigateMaxToolCalls": 8, + "autoInvestigateMode": "investigate", // 或 "recover"(授权内恢复,见上一节) + "investigateMaxAttempts": 2, "recoverVerifySec": 600, "goalPollSec": 60, "goalWaitHours": 72 }, "budget": { "agentMaxRunsPerDay": 50, "agentMaxCostUsdPerDay": 5 } @@ -137,6 +181,7 @@ curl -X POST localhost:4380/api/v1/agent/runs -H "Authorization: Bearer umt-dev- | `e3-background-continues` | 关闭聊天后台照样推进 | | `e4-crash-recovery` | worker 子进程在 download.submit 时 kill -9,新 worker resume,qB 只收到一次 add | | `e5-auto-investigate` | 识别歧义 → 有界调查,去重 | +| `e6-auto-recover` | qB missingFiles → download_failed → 授权内恢复(task.inspect → download.recheck → 系统核实关闭事项);无授权时降级只调查;换源被拒;去重;同根因尝试上限;预算超限保持待处理 | ## 已知限制 diff --git a/packages/contracts/src/capabilities-io.ts b/packages/contracts/src/capabilities-io.ts index a77bfeb5..fe0fdda0 100644 --- a/packages/contracts/src/capabilities-io.ts +++ b/packages/contracts/src/capabilities-io.ts @@ -3,7 +3,7 @@ import { MediaAsset } from "./assets.ts"; import { CAPABILITY_CATALOG, type CapabilitySpec, type KnownCapabilityId } from "./capability.ts"; import { Id, JsonObject, NonNegInt } from "./common.ts"; import { MediaEntity, MediaKind } from "./content.ts"; -import { AcquisitionPlan, DownloadTask, Task, TaskStatus } from "./execution.ts"; +import { AcquisitionPlan, DownloadStatus, DownloadTask, Task, TaskStatus } from "./execution.ts"; import { ExternalEntityType, ExternalIdentity } from "./identity.ts"; import { ServerMapping } from "./ops.ts"; import { OrganizationAction, OrganizationRecord } from "./organization.ts"; @@ -121,6 +121,22 @@ export const OrganizePlanOutput = z.object({ export const OrganizeExecuteInput = z.object({ planId: Id, payloadHash: z.string().min(1), approvalId: Id.optional() }); export const OrganizeExecuteOutput = Accepted.extend({ record: OrganizationRecord.nullable() }); +/** 授权内恢复:强制重新校验(missingFiles / error 之后)或恢复暂停;换源、删除、清理不在此能力内 */ +export const DownloadRecheckAction = z.enum(["recheck", "resume"]); +export const DownloadRecheckInput = z.object({ + downloadTaskId: Id, + action: DownloadRecheckAction.default("recheck"), + reason: z.string().max(500).optional(), +}); +export const DownloadRecheckOutput = z.object({ + downloadTaskId: Id, + action: DownloadRecheckAction, + statusBefore: DownloadStatus, + rawStateBefore: z.string().nullable(), + /** 已向下载器发出请求;是否恢复以后续同步到的 download_task 状态为准 */ + requested: z.boolean(), +}); + export const TaskInspectInput = z.object({ taskId: Id }); export const TaskInspectOutput = z.object({ task: Task, @@ -178,6 +194,7 @@ export const CAPABILITY_IO = { "identity.resolve": { input: IdentityResolveInput, output: IdentityResolveOutput }, "download.plan": { input: DownloadPlanInput, output: DownloadPlanOutput }, "download.submit": { input: DownloadSubmitInput, output: DownloadSubmitOutput }, + "download.recheck": { input: DownloadRecheckInput, output: DownloadRecheckOutput }, "organize.plan": { input: OrganizePlanInput, output: OrganizePlanOutput }, "organize.execute": { input: OrganizeExecuteInput, output: OrganizeExecuteOutput }, "task.inspect": { input: TaskInspectInput, output: TaskInspectOutput }, @@ -223,6 +240,8 @@ export type OrganizeExecuteInput = z.infer; export type OrganizeExecuteOutput = z.infer; export type TaskInspectInput = z.infer; export type TaskInspectOutput = z.infer; +export type DownloadRecheckInput = z.infer; +export type DownloadRecheckOutput = z.infer; export type TaskRetryInput = z.infer; export type TaskRetryOutput = z.infer; export type TaskCancelInput = z.infer; diff --git a/packages/contracts/src/capability.ts b/packages/contracts/src/capability.ts index 3729038c..29f1901e 100644 --- a/packages/contracts/src/capability.ts +++ b/packages/contracts/src/capability.ts @@ -57,6 +57,7 @@ export const CAPABILITY_IDS = [ "identity.resolve", "download.plan", "download.submit", + "download.recheck", "organize.plan", "organize.execute", "task.inspect", @@ -171,6 +172,17 @@ export const CAPABILITY_CATALOG: Record true, false, ), + "download.recheck": d( + "download.recheck", + "重新校验/恢复下载", + "对已纳管的下载任务发起强制重新校验(action=recheck)或恢复非阈值、非人工的暂停(action=resume);不换源、不删除、不改保存路径", + "execute", + "DownloadRecheck", + "organize.execute", + "medium", + true, + false, + ), "organize.plan": d( "organize.plan", "生成整理计划", diff --git a/packages/contracts/src/settings.ts b/packages/contracts/src/settings.ts index a387c938..bde3aa41 100644 --- a/packages/contracts/src/settings.ts +++ b/packages/contracts/src/settings.ts @@ -73,8 +73,22 @@ export type PiSettings = z.infer; * 上游/模型/思考档位在 pi 分组;pi.provider = "fake" 时使用脚本化假模型(不联网,演示与 e2e 用)。 */ export const AgentRuntimeSettings = z.object({ - /** 识别歧义 / 下载失败诊断等待处理事项创建时,自动发起有界只读调查 */ + /** 识别歧义 / 下载失败诊断等待处理事项创建时,自动发起有界调查 */ autoInvestigate: z.boolean(), + /** + * 自动调查的模式:investigate = 只读调查,结论交还人工; + * recover = 在已发布策略 / 持续授权范围内可执行恢复类能力(task.retry、download.recheck、identity.resolve), + * 换源、删除、清理始终排除;没有覆盖的授权时自动降级为 investigate。 + */ + autoInvestigateMode: z.enum(["investigate", "recover"]), + /** 同一根因(decision 去重键)最多自动恢复几次;用尽后只交还人工,不再唤醒模型 */ + investigateMaxAttempts: z.number().int().min(1).max(10), + /** 恢复动作后等待业务事实(下载状态 / 任务状态)核实的最长时间(秒) */ + recoverVerifySec: z + .number() + .int() + .min(5) + .max(24 * 3600), /** 单次运行:工具调用次数上限 */ maxToolCalls: z.number().int().min(1).max(500), /** 单次运行:时长上限(秒) */ @@ -165,6 +179,9 @@ export const DEFAULT_SETTINGS: GlobalSettings = { }, agent: { autoInvestigate: false, + autoInvestigateMode: "investigate", + investigateMaxAttempts: 2, + recoverVerifySec: 600, maxToolCalls: 40, maxDurationSec: 900, maxTokensPerRun: null,