Files
UHD_Media_Tool/apps/server/test/e2e/e6-auto-recover.test.ts
T
rootandClaude Opus 5.5 86af723b2a 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]>
2026-09-27 22:35:03 +08:00

361 lines
14 KiB
TypeScript

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<string, unknown> }).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<string, unknown> }).execution).toMatchObject({
kind: "agent_recover",
runId: run.id,
verified: true,
});
expect((d2.evidence as { autoRecover: Record<string, unknown> }).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<string, unknown> }).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<string, unknown> }).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);
});