feat(agent): W4-E2 in-grant auto recovery for decisions with verified outcome

- 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 <[email protected]>
This commit is contained in:
rootandClaude Opus 5.5 committed 2026-09-27 22:35:03 +08:00
1 parent 23309042dc
commit 86af723b2a
13 files changed
+1072 -15

No files matched your search

@@ -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:<id>),同一根因(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<string, readonly string[]> = {
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<string, unknown>;
suggestion?: string;
verifiedAt?: string;
}
type RecoveryOutcomeKind = "recovered" | "failed" | "unverified" | "not_attempted" | "budget_exceeded" | "cancelled";
export interface RecoveryAction {
capabilityId: string;
input: Record<string, unknown>;
status: string;
}
export interface RecoveryCheck {
state: "recovered" | "failed" | "pending";
facts: Record<string, unknown>;
}
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<AutoRecoverEvidence>) {
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<string[]> {
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<string | null>`payload->>'errorMessage'`.as("reason")])
.where("action", "=", "capability.denied")
.where("trace_id", "=", runId)
.where(sql<string>`payload->>'agentRunId'`, "=", runId)
.orderBy("at")
.execute();
return {
actions: rows.map((r) => ({
capabilityId: r.capability_id,
input: (r.input ?? {}) as Record<string, unknown>,
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<RecoveryCheck> {
const facts: Record<string, unknown> = {};
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<RecoveryOutcome | null> {
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<AutoRecoverEvidence> = {
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 };
}