feat(ao): W4-E3 manual organize guard, series completion, engine retry, v0.1.0

- POST /organization-records: sourcePaths must resolve (realpath) under a
  registered library_root or downloader path-mapping host path; `..` and
  symlink escapes rejected (400), dryRun included. Move requires an admin
  session, or an API key with an automation_grant covering the entity or a
  matching one-time approval (consumed on use), else 403. Manual Move from a
  managed download_task with unsatisfied H&R returns 409 and opens a
  decision; never silently downgraded to Copy.
- Series slot completion: total episode count is unknown while units are
  provisional, so the slot is not satisfied and the subscription keeps
  tracking; coverage is registered per actual file (creating provisional
  episode units), and already_in_library files count as verified presence.
- POST /engine/runs/{id}/retry (admin): re-enqueue by original task name and
  raw input with idempotency key retry:<id>:<n>, audited; retry button on
  the run detail page.
- E7 regular series e2e, E8 AT24 tag removal / task disappearance e2e.
- All workspace members and root set to version 0.1.0; CHANGELOG.md.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
This commit is contained in:
rootandClaude Opus 5.5 committed 2026-09-27 22:47:11 +08:00
1 parent 591cc78c7d
commit ceff9cf51f
25 files changed
+1566 -57

No files matched your search

+308
View File
@@ -0,0 +1,308 @@
import { existsSync } from "node:fs";
import { copyFile, mkdir } from "node:fs/promises";
import { dirname, join } from "node:path";
import { createMockRssServer, type MockFeedItem, type MockRssServer } from "@umt/testkit";
import { afterAll, beforeAll, describe, expect, it } from "vitest";
import { TORRENT_SIZE, torrentFor } from "./fixtures/modules.ts";
import { completeDownload, createE2E, type E2E, pgAvailable, seedBase, waitFor } from "./harness.ts";
// E7 常规剧集纵切(验收 S5,REQ §7.1.1 / §8.4):全部经真实任务引擎 + outbox 中继。
// mock RSS 依次出 S01E01、S01E02 单集,最后出一个整季包;已发布收录策略(剧集 1080p Slot),订阅改为持续追踪。
// 断言:
// · 两集分别入库到同一个已绑定的季目录(季绑定只建一次,之后固定);
// · content_unit 覆盖按实际文件计算(E01→E01、E02→E02);
// · 整季包"声称"整季,但实际文件只有 E01/E02(与库内相同)→ 不新增覆盖,Slot 不因包名判满足;
// · 第三集从未出现:Slot 仍未 satisfied,订阅保持 active/continuous(总集数未知不结束)。
const hasPg = await pgAvailable();
const BASE = "https://audiences.example";
const E01 = "The.Bear.S01E01.1080p.WEB-DL.x264-GRP";
const E02 = "The.Bear.S01E02.1080p.WEB-DL.x264-GRP";
const PACK = "The.Bear.S01.1080p.WEB-DL.x264-GRP";
const item = (id: number, title: string): MockFeedItem => ({
guid: `aud-${id}`,
title,
link: `${BASE}/details.php?id=${id}&hit=1`,
enclosureUrl: `${BASE}/download.php?id=${id}&passkey=E7PASSKEY`,
size: TORRENT_SIZE,
pubDate: new Date(),
categories: ["剧集"],
});
describe.skipIf(!hasPg)("E7 常规剧集纵切", () => {
let e: E2E;
let rss: MockRssServer;
let base: Awaited<ReturnType<typeof seedBase>>;
let feedId: string;
let entityId: string;
let slotId: string;
let subId: string;
// 手动抓取按 10 秒时间槽去重(ingestFeed:manual:<feed>:<槽>):同一槽内的第二次请求会被合并,等到下一个槽再发
let lastSlot = -1;
const fetchFeed = async () => {
const slotOf = () => Math.floor(Date.now() / 10_000);
if (slotOf() === lastSlot) await new Promise((r) => setTimeout(r, (lastSlot + 1) * 10_000 - Date.now() + 50));
lastSlot = slotOf();
const r = await e.api("POST", `/feed-sources/${feedId}/fetch`);
expect(r.status).toBe(200);
};
const importedFor = async (title: string) => {
const hash = torrentFor(title).hash;
const dt = await waitFor(
async () => e.db.selectFrom("download_task").selectAll().where("torrent_hash", "=", hash).executeTakeFirst(),
60_000,
`${title} download_task`,
);
return waitFor(
async () =>
e.db
.selectFrom("organization_record")
.selectAll()
.where("download_task_id", "=", dt.id)
.where("status", "in", ["succeeded", "failed", "partial"])
.executeTakeFirst(),
90_000,
`${title} 整理记录结束`,
);
};
const slot = () => e.db.selectFrom("acquisition_slot").selectAll().where("id", "=", slotId).executeTakeFirstOrThrow();
const sub = () => e.db.selectFrom("subscription").selectAll().where("id", "=", subId).executeTakeFirstOrThrow();
const coverage = async () => {
const r = await e.db
.selectFrom("asset_unit_coverage as c")
.innerJoin("content_unit as u", "u.id", "c.content_unit_id")
.innerJoin("media_asset as a", "a.id", "c.media_asset_id")
.select(["a.id as asset", "u.season_number", "u.episode_number"])
.where("a.media_entity_id", "=", entityId)
.where("a.status", "=", "active")
.orderBy("u.episode_number")
.execute();
return r.map((x) => ({ asset: x.asset, unit: `S${x.season_number}E${x.episode_number}` }));
};
beforeAll(async () => {
e = await createE2E("e7");
base = await seedBase(e, { baseUrl: `${BASE}/` });
rss = await createMockRssServer({ passkey: "E7PASSKEY", items: [item(910001, E01)] });
const urlRef = await e.booted.ctx.secrets.create({ purpose: "feed_url", plaintext: rss.feedUrl });
const feed = await e.db
.insertInto("feed_source")
.values({
site_id: base.siteId,
name: "观众-剧集",
purpose: "series",
url_secret_ref: urlRef,
url_display: rss.feedUrl.replace(/passkey=[^&]+/, "passkey=[REDACTED]"),
poll_interval_sec: 3600,
})
.returning("id")
.executeTakeFirstOrThrow();
feedId = feed.id;
}, 120_000);
afterAll(async () => {
await rss?.close();
await e?.close();
}, 60_000);
it("发布剧集 1080p 收录策略 → 持续授权", async () => {
const created = await e.api("POST", "/policy-revisions", {
policyKey: "series",
kind: "acquisition",
title: "剧集 1080p",
scope: { mediaKinds: ["series"] },
rules: {
newTitle: {
enabled: true,
windowDays: 3650,
mediaKinds: ["series"],
slots: [{ name: "1080p", hardConditions: { resolutions: ["1080p"] } }],
},
},
});
expect(created.status).toBe(201);
const pub = await e.api("POST", `/policy-revisions/${created.body.data.id}/publish`, {
expectedRevision: created.body.data.revision,
});
expect(pub.status).toBe(200);
});
it("S01E01:识别为剧集 → 订阅改为持续追踪 → 下载 → 入库到季目录;总集数未知,Slot 不判满足", async () => {
await fetchFeed();
const rel = await waitFor(
async () =>
e.db
.selectFrom("source_release")
.selectAll()
.where("title", "=", E01)
.where("resolution_status", "=", "resolved")
.executeTakeFirst(),
60_000,
"S01E01 识别完成",
);
entityId = rel.media_entity_id!;
const entity = await e.db
.selectFrom("media_entity")
.selectAll()
.where("id", "=", entityId)
.executeTakeFirstOrThrow();
expect(entity.kind).toBe("series");
const s = await waitFor(
async () =>
e.db.selectFrom("subscription").selectAll().where("media_entity_id", "=", entityId).executeTakeFirst(),
30_000,
"新作品订阅",
);
subId = s.id;
const patched = await e.api("PATCH", `/subscriptions/${subId}`, { mode: "continuous" });
expect(patched.status).toBe(200);
slotId = (
await e.db
.selectFrom("acquisition_slot")
.select("id")
.where("media_entity_id", "=", entityId)
.executeTakeFirstOrThrow()
).id;
await completeDownload(e, E01);
const rec = await importedFor(E01);
expect(rec).toMatchObject({ status: "succeeded", action: "copy", trigger: "auto" });
expect((await coverage()).map((c) => c.unit)).toEqual(["S1E1"]);
expect((await slot()).status).toBe("empty");
expect(await sub()).toMatchObject({ status: "active", mode: "continuous" });
}, 180_000);
let seasonDir: string;
it("S01E02:第二集进同一个已绑定的季目录,覆盖逐集计算,Slot 仍未满足", async () => {
const bindings0 = await e.db
.selectFrom("path_binding")
.select(["binding_kind", "season_number", "relative_path"])
.where("media_entity_id", "=", entityId)
.orderBy("binding_kind")
.execute();
expect(bindings0.map((b) => `${b.binding_kind}:${b.season_number ?? "-"}`)).toEqual(["entity:-", "season:1"]);
seasonDir = bindings0.find((b) => b.binding_kind === "season")!.relative_path;
rss.addItem(item(910002, E02));
await fetchFeed();
await completeDownload(e, E02);
const rec = await importedFor(E02);
expect(rec.status).toBe("succeeded");
const files = await e.db
.selectFrom("organization_file_result as f")
.innerJoin("organization_record as r", "r.id", "f.organization_record_id")
.select(["f.target_path", "f.status"])
.where("r.media_entity_id", "=", entityId)
.where("f.role", "=", "main")
.orderBy("f.target_path")
.execute();
expect(files).toHaveLength(2);
for (const f of files) {
expect(f.status).toBe("succeeded");
expect(dirname(f.target_path!)).toBe(seasonDir);
expect(existsSync(join(e.lib.root, f.target_path!))).toBe(true);
}
expect(files.map((f) => f.target_path!)).toEqual([
expect.stringMatching(/S01E01\.mkv$/),
expect.stringMatching(/S01E02\.mkv$/),
]);
// 路径绑定固定:没有因为第二集新建任何绑定
const bindings1 = await e.db
.selectFrom("path_binding")
.select(["binding_kind", "season_number", "relative_path"])
.where("media_entity_id", "=", entityId)
.orderBy("binding_kind")
.execute();
expect(bindings1).toEqual(bindings0);
const cov = await coverage();
expect(cov.map((c) => c.unit)).toEqual(["S1E1", "S1E2"]);
expect(new Set(cov.map((c) => c.asset)).size).toBe(2);
expect((await slot()).status).toBe("empty");
expect(await sub()).toMatchObject({ status: "active", mode: "continuous" });
}, 180_000);
it("整季包:实际文件只有 E01/E02(与库内相同)→ 不新增覆盖,不因包名判满足;第三集未出现,订阅继续", async () => {
const libFiles = await e.db
.selectFrom("organization_file_result as f")
.innerJoin("organization_record as r", "r.id", "f.organization_record_id")
.select(["f.target_path"])
.where("r.media_entity_id", "=", entityId)
.where("f.role", "=", "main")
.where("f.status", "=", "succeeded")
.orderBy("f.target_path")
.execute();
const assetsBefore = await e.db
.selectFrom("media_asset")
.select("id")
.where("media_entity_id", "=", entityId)
.execute();
rss.addItem(item(910003, PACK));
await fetchFeed();
const packRel = await waitFor(
async () =>
e.db
.selectFrom("source_release")
.selectAll()
.where("title", "=", PACK)
.where("resolution_status", "=", "resolved")
.executeTakeFirst(),
60_000,
"整季包识别完成",
);
expect(packRel.media_entity_id).toBe(entityId);
const hash = torrentFor(PACK).hash;
await waitFor(async () => e.qb.get(hash), 60_000, "qB 收到整季包");
// 包内真实文件清单:只有 E01、E02,内容与已入库的相同(常见的"同组合集")
const pack = [E01, E02].map((t, i) => ({
path: `${PACK}/${t}.mkv`,
src: join(e.lib.root, libFiles[i]!.target_path!),
}));
for (const p of pack) {
const dst = join(e.cache.root, p.path);
await mkdir(dirname(dst), { recursive: true });
await copyFile(p.src, dst);
}
await e.qb.complete(hash, { files: pack.map((p) => ({ path: p.path, size: TORRENT_SIZE })), writeFiles: false });
const rec = await importedFor(PACK);
expect(rec.status).toBe("succeeded");
const packFiles = await e.db
.selectFrom("organization_file_result")
.select(["status", "skip_reason", "media_asset_id"])
.where("organization_record_id", "=", rec.id)
.where("role", "=", "main")
.execute();
expect(packFiles).toHaveLength(2);
for (const f of packFiles) {
expect(f).toMatchObject({ status: "skipped", skip_reason: "already_in_library" });
expect(assetsBefore.map((a) => a.id)).toContain(f.media_asset_id);
}
// 没有新资产、没有新覆盖;季单元的"整季"声明不算覆盖,也没有凭空出现第三集
expect(
await e.db.selectFrom("media_asset").select("id").where("media_entity_id", "=", entityId).execute(),
).toHaveLength(assetsBefore.length);
expect((await coverage()).map((c) => c.unit)).toEqual(["S1E1", "S1E2"]);
const eps = await e.db
.selectFrom("content_unit")
.select(["episode_number", "status"])
.where("media_entity_id", "=", entityId)
.where("kind", "=", "episode")
.orderBy("episode_number")
.execute();
expect(eps).toEqual([
{ episode_number: 1, status: "provisional" },
{ episode_number: 2, status: "provisional" },
]);
// 第三集没出现、总集数未知:Slot 不满足,订阅继续追踪
const s = await slot();
expect(s.status).not.toBe("satisfied");
expect(await sub()).toMatchObject({ status: "active", mode: "continuous" });
// 三个种子各提交一次
expect(e.qb.callsTo("torrents/add")).toHaveLength(3);
}, 180_000);
});
+259
View File
@@ -0,0 +1,259 @@
import { createQbClient } from "@umt/adapters";
import { afterAll, beforeAll, describe, expect, it } from "vitest";
import type { createAcquisitionOrganizeModule } from "../../src/modules/acquisition-organize/index.ts";
import { syncDownloads } from "../../src/modules/acquisition-organize/services/downloads.ts";
import { torrentFor } from "./fixtures/modules.ts";
import { createE2E, type E2E, pgAvailable, seedBase, waitFor } from "./harness.ts";
// E8 AT24(REQ §8.4):管理标签被移除 / qB 任务消失。真实任务引擎 + worker + 后台同步循环 + mock qB。
// A(已入库)、B(下载中)被用户在 qB 里去掉管理标签 → managed=false:B 之后完成也不入库,A 不再被缓存清理;
// 历史(download_task 行、状态事件)与资产保留。
// D(已入库)在 qB 里被外部删除 → download_task=removed,media_asset / asset_location 不变,Slot 仍 satisfied,不重新下载。
// C(已入库、仍纳管)作对照:同一轮缓存清理里被删,证明清理确实跑过。
// 断言以 mock qB 调用日志为准:add 只有最初 4 次;系统发出的 delete 只针对 C。
const hasPg = await pgAvailable();
const T = {
A: "Alpha.Tag.2020.1080p.BluRay.x264-GRP",
B: "Beta.Tag.2020.1080p.BluRay.x264-GRP",
C: "Gamma.Tag.2020.1080p.BluRay.x264-GRP",
D: "Delta.Tag.2020.1080p.BluRay.x264-GRP",
} as const;
type K = keyof typeof T;
describe.skipIf(!hasPg)("E8 AT24 标签移除 / 任务消失", () => {
let e: E2E;
let base: Awaited<ReturnType<typeof seedBase>>;
const ids = {} as Record<K, { entityId: string; slotId: string; releaseId: string; hash: string; dtId?: string }>;
const deps = () =>
(
e.modules.find((m) => m.name === "acquisition-organize") as ReturnType<typeof createAcquisitionOrganizeModule>
).deps();
const dt = (k: K) =>
e.db.selectFrom("download_task").selectAll().where("id", "=", ids[k].dtId!).executeTakeFirstOrThrow();
const hashesOf = (endpoint: string) =>
e.qb.callsTo(endpoint).flatMap((c) => (c.form.hashes ?? "").split("|").filter(Boolean));
beforeAll(async () => {
e = await createE2E("e8");
base = await seedBase(e);
let n = 0;
for (const k of Object.keys(T) as K[]) {
const ent = await e.db
.insertInto("media_entity")
.values({ kind: "movie", title: `${k} 标签片`, year: 2020, status: "active" })
.returning("id")
.executeTakeFirstOrThrow();
const sub = await e.db
.insertInto("subscription")
.values({ media_entity_id: ent.id, media_kind: "movie", mode: "once" })
.returning("id")
.executeTakeFirstOrThrow();
const slot = await e.db
.insertInto("acquisition_slot")
.values({
media_entity_id: ent.id,
subscription_id: sub.id,
name: "1080p",
hard_conditions: JSON.stringify({ resolutions: ["1080p"] }),
})
.returning("id")
.executeTakeFirstOrThrow();
const rel = await e.db
.insertInto("source_release")
.values({
site_id: base.siteId,
site_release_id: String(770000 + ++n),
title: T[k],
size_bytes: 300_000,
media_entity_id: ent.id,
resolution_status: "resolved",
raw: JSON.stringify({ _site: { hr: false } }),
})
.returning("id")
.executeTakeFirstOrThrow();
ids[k] = { entityId: ent.id, slotId: slot.id, releaseId: rel.id, hash: torrentFor(T[k]).hash };
}
}, 120_000);
afterAll(async () => {
await e?.close();
}, 60_000);
const submit = async (k: K) => {
const r = await e.api("POST", "/acquisition-plans", { sourceReleaseId: ids[k].releaseId });
expect(r.status, JSON.stringify(r.body)).toBe(200);
expect(r.body.data.submitted).toBe(true);
await waitFor(async () => e.qb.get(ids[k].hash), 60_000, `qB 收到 ${k}`);
const row = await waitFor(
async () =>
e.db.selectFrom("download_task").select("id").where("torrent_hash", "=", ids[k].hash).executeTakeFirst(),
30_000,
`${k} download_task`,
);
ids[k].dtId = row.id;
};
const imported = async (k: K) => {
await e.qb.complete(ids[k].hash, { writeFiles: true });
await waitFor(
async () =>
e.db
.selectFrom("organization_record")
.select("id")
.where("download_task_id", "=", ids[k].dtId!)
.where("status", "=", "succeeded")
.executeTakeFirst(),
90_000,
`${k} 入库`,
);
};
it("准备:A/C/D 下载完成并 Copy 入库,B 仍在下载", async () => {
for (const k of ["A", "C", "D"] as K[]) {
await submit(k);
await imported(k);
}
await submit("B");
expect((await dt("B")).status).not.toBe("completed");
for (const k of ["A", "C", "D"] as K[]) {
expect(await dt(k)).toMatchObject({ status: "imported", managed: true });
const slot = await e.db
.selectFrom("acquisition_slot")
.select("status")
.where("id", "=", ids[k].slotId)
.executeTakeFirstOrThrow();
expect(slot.status).toBe("satisfied");
}
expect(e.qb.callsTo("torrents/add")).toHaveLength(4);
}, 300_000);
it("移除管理标签 → managed=false;之后完成也不入库、不再被自动动作触碰,历史保留", async () => {
const tag = (
await e.db
.selectFrom("downloader_instance")
.select("managed_tag")
.where("id", "=", base.instanceId)
.executeTakeFirstOrThrow()
).managed_tag;
for (const k of ["A", "B"] as K[]) {
const t = e.qb.get(ids[k].hash)!;
expect(t.tags).toContain(tag);
t.tags = t.tags.filter((x) => x !== tag);
}
await syncDownloads(deps());
for (const k of ["A", "B"] as K[]) {
const row = await dt(k);
expect(row.managed).toBe(false);
expect(row.removed_at).toBeNull();
expect(row.tags).not.toContain(tag);
}
const bStatus = (await dt("B")).status;
const statusEventsA = (await e.db.selectFrom("outbox_event").select(["type", "payload"]).execute()).filter(
(x) =>
x.type === "download.status_changed" &&
(x.payload as { downloadTaskId?: string }).downloadTaskId === ids.A.dtId,
);
expect(statusEventsA.length).toBeGreaterThan(0);
// B 在 qB 里完成:不再同步状态、不发完成事件、不入库
await e.qb.complete(ids.B.hash, { writeFiles: true });
for (let i = 0; i < 3; i++) await syncDownloads(deps());
await new Promise((r) => setTimeout(r, 1_500));
expect((await dt("B")).status).toBe(bStatus);
expect(
await e.db.selectFrom("organization_record").select("id").where("download_task_id", "=", ids.B.dtId!).execute(),
).toHaveLength(0);
const bImports = await e.db
.selectFrom("task")
.select("id")
.where("kind", "=", "verifyAndImport")
.where("subject_id", "=", ids.B.dtId!)
.execute();
expect(bImports).toHaveLength(0);
// A 的历史与资产原样
const assetsA = await e.db
.selectFrom("media_asset as a")
.innerJoin("asset_location as l", "l.media_asset_id", "a.id")
.select(["a.status", "l.status as loc"])
.where("a.download_task_id", "=", ids.A.dtId!)
.execute();
expect(assetsA).toEqual([{ status: "active", loc: "available" }]);
const after = (await e.db.selectFrom("outbox_event").select(["type", "payload"]).execute()).filter(
(x) =>
x.type === "download.status_changed" &&
(x.payload as { downloadTaskId?: string }).downloadTaskId === ids.A.dtId,
);
expect(after.length).toBe(statusEventsA.length);
}, 120_000);
it("qB 任务消失 → removed;资产与位置不变,Slot 仍 satisfied,不重新下载", async () => {
const before = await e.db
.selectFrom("media_asset as a")
.innerJoin("asset_location as l", "l.media_asset_id", "a.id")
.select(["a.id", "a.status", "l.id as loc_id", "l.status as loc", "l.relative_path"])
.where("a.download_task_id", "=", ids.D.dtId!)
.execute();
expect(before).toHaveLength(1);
// 外部(用户在 qB 里)删掉 D;这一条 delete 是测试发出的,不是系统
const ext = createQbClient({
instanceId: base.instanceId,
baseUrl: e.qb.url,
username: "admin",
password: "adminadmin",
});
await ext.delete([ids.D.hash], { deleteFiles: false });
const externalDeletes = e.qb.callsTo("torrents/delete").length;
await syncDownloads(deps());
const row = await dt("D");
expect(row.status).toBe("removed");
expect(row.removed_at).not.toBeNull();
const after = await e.db
.selectFrom("media_asset as a")
.innerJoin("asset_location as l", "l.media_asset_id", "a.id")
.select(["a.id", "a.status", "l.id as loc_id", "l.status as loc", "l.relative_path"])
.where("a.download_task_id", "=", ids.D.dtId!)
.execute();
expect(after).toEqual(before);
// 几轮后台同步 + 编排之后:没有新计划、没有新 add
await new Promise((r) => setTimeout(r, 1_500));
const slot = await e.db
.selectFrom("acquisition_slot")
.select("status")
.where("id", "=", ids.D.slotId)
.executeTakeFirstOrThrow();
expect(slot.status).toBe("satisfied");
const plans = await e.db.selectFrom("acquisition_plan").select("id").where("slot_id", "=", ids.D.slotId).execute();
expect(plans).toHaveLength(1);
expect(e.qb.callsTo("torrents/add")).toHaveLength(4);
expect(e.qb.callsTo("torrents/delete")).toHaveLength(externalDeletes);
}, 60_000);
it("缓存清理:纳管的 C 被清,去标签的 A 不碰(qB 调用日志)", async () => {
await e.settings({ cache: { hardThresholdPercent: 30, resumePercent: 25, cleanupTargetPercent: 20 } });
const r = await e.api("POST", "/download-cache/cleanup", {});
expect(r.status).toBe(202);
await waitFor(
async () => {
const t = await e.db
.selectFrom("task")
.select("status")
.where("id", "=", r.body.data.taskId)
.executeTakeFirst();
return t?.status === "succeeded" ? t : null;
},
60_000,
"cleanup 任务完成",
);
expect((await dt("C")).status).toBe("removed");
const deleted = hashesOf("torrents/delete");
expect(deleted).toContain(ids.C.hash);
expect(deleted).not.toContain(ids.A.hash);
expect(deleted).not.toContain(ids.B.hash);
// 系统发出的带数据删除只有 C;D 那一条是外部 deleteFiles=false
const withData = e.qb.callsTo("torrents/delete").filter((c) => c.form.deleteFiles === "true");
expect(withData.flatMap((c) => c.form.hashes!.split("|"))).toEqual([ids.C.hash]);
expect(e.qb.get(ids.A.hash)).toBeDefined();
expect(await dt("A")).toMatchObject({ managed: false, status: "imported", removed_at: null });
expect(e.qb.callsTo("torrents/add")).toHaveLength(4);
}, 90_000);
});