From adffe368464db98358816df50a331d548c6dc91d Mon Sep 17 00:00:00 2001 From: root Date: Mon, 28 Sep 2026 20:17:46 +0800 Subject: [PATCH] feat(server): browser module with DB-backed sessions, takeover lease and browser.* capabilities /browser/providers, /browser/sessions (open/close/list), navigate/snapshot/screenshot/ click/type/scroll/select/extract/download and control acquire/release. Human input needs an active lease; expiry or captcha moves the session to waiting with a decision; release forces a fresh snapshot. Downloads are staged under DATA_DIR/tmp, verified by infohash and submitted through POST /acquisition-plans. Capability count 21 -> 25. Co-Authored-By: Claude Opus 5.5 --- .../browser/__tests__/browser-api.test.ts | 275 +++++ .../browser/__tests__/browser-e2e.test.ts | 114 ++ .../src/modules/browser/__tests__/harness.ts | 209 ++++ .../src/modules/browser/capabilities.ts | 188 +++ apps/server/src/modules/browser/index.ts | 41 + apps/server/src/modules/browser/routes.ts | 292 +++++ apps/server/src/modules/browser/service.ts | 1003 +++++++++++++++++ apps/server/src/modules/index.ts | 2 + .../server/test/e2e/e2-agent-delegate.test.ts | 4 +- apps/server/test/e2e/fixtures/modules.ts | 3 + apps/server/test/modules.test.ts | 3 +- .../site-capabilities.test.ts | 4 +- 12 files changed, 2133 insertions(+), 5 deletions(-) create mode 100644 apps/server/src/modules/browser/__tests__/browser-api.test.ts create mode 100644 apps/server/src/modules/browser/__tests__/browser-e2e.test.ts create mode 100644 apps/server/src/modules/browser/__tests__/harness.ts create mode 100644 apps/server/src/modules/browser/capabilities.ts create mode 100644 apps/server/src/modules/browser/index.ts create mode 100644 apps/server/src/modules/browser/routes.ts create mode 100644 apps/server/src/modules/browser/service.ts diff --git a/apps/server/src/modules/browser/__tests__/browser-api.test.ts b/apps/server/src/modules/browser/__tests__/browser-api.test.ts new file mode 100644 index 00000000..ab3c6d4d --- /dev/null +++ b/apps/server/src/modules/browser/__tests__/browser-api.test.ts @@ -0,0 +1,275 @@ +import { createHash } from "node:crypto"; +import { readdir, readFile } from "node:fs/promises"; +import { join } from "node:path"; +import { MOCK_ROW_H, MOCK_ROW_Y0, MOCK_VIEWPORT } from "@umt/testkit"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { registerBrowserCapabilities } from "../capabilities.ts"; +import { createBrowserModule } from "../index.ts"; +import { + ACCESS_KEY, + type BrowserHarness, + createBrowserHarness, + type Loose, + ORIGIN, + OTHER_KEY, + PASSKEY, + PASSWORD, + pgAvailable, + READ_KEY, +} from "./harness.ts"; + +// C1-E 浏览器 HTTP 端点(对 testkit 交互式 mock camofox 验证,未连真实实例): +// 浏览器服务列表与探测、会话开关(数据库为准)、接管租约、坐标点击与凭证注入、截图、下载暂存核验与提交、异常路径。 + +const up = await pgAvailable(); +const row = (i: number) => ({ x: 200, y: MOCK_ROW_Y0 + MOCK_ROW_H * i + MOCK_ROW_H / 2 }); + +describe.skipIf(!up)("浏览器 HTTP 端点", () => { + let h: BrowserHarness; + let sid: string; + + beforeAll(async () => { + h = await createBrowserHarness("api"); + }, 120_000); + afterAll(async () => { + await h?.close(); + }); + + it("GET /browser/providers:凭证只给引用;probe 写回能力与健康,verified 只来自人工核验配置", async () => { + const r = await h.req("/browser/providers"); + expect(r.status).toBe(200); + expect(r.body.data.items).toHaveLength(1); + expect(r.body.data.items[0]).toMatchObject({ kind: "camofox", enabled: true, capabilities: null, health: null }); + expect(JSON.stringify(r.body)).not.toContain(ACCESS_KEY); + const p = await h.req("/browser/providers?probe=true"); + const item = p.body.data.items[0]; + expect(item.health.status).toBe("ok"); + expect(item.healthStatus).toBe("healthy"); + expect(item.capabilities.upstreamVersion).toBe("1.17.0"); + expect(item.capabilities.snapshot).toMatchObject({ supported: true, verified: true, source: "probed" }); + expect(item.capabilities.interact).toMatchObject({ supported: true, verified: false }); + expect(item.capabilities.takeover).toMatchObject({ supported: true, verified: false }); + // 探测带 Bearer 访问凭据 + expect(h.mock.callsTo("/openapi.json")[0]?.headers.authorization).toBe(`Bearer ${ACCESS_KEY}`); + expect((await h.req("/browser/providers", { key: READ_KEY })).status).toBe(403); + }); + + it("打开会话:profile 作 camofox userId,会话/标签页落库;同站点账号再开返回同一会话", async () => { + const r = await h.req("/browser/sessions", { body: { siteId: h.siteId, url: `${ORIGIN}/torrents.php` } }); + expect(r.status).toBe(200); + const s = r.body.data; + sid = s.id; + expect(s).toMatchObject({ siteId: h.siteId, accountId: h.accountId, controller: "automation", lease: null }); + expect(s.profileKey).toBe(`umt-site-${h.siteId}-acct-${h.accountId}`); + // 登录失效:落到登录页 + expect(s.currentUrl).toContain("/login.php"); + expect(s.tabs).toHaveLength(1); + expect([...h.mock.tabs.values()][0]?.userId).toBe(s.profileKey); + const dbRow = await h.db.selectFrom("browser_session").selectAll().where("id", "=", sid).executeTakeFirstOrThrow(); + expect(dbRow.controller).toBe("automation"); + const again = await h.req("/browser/sessions", { body: { siteId: h.siteId } }); + expect(again.body.data.id).toBe(sid); + const list = await h.req(`/browser/sessions?siteId=${h.siteId}`); + expect(list.body.data.items.map((x: Loose) => x.id)).toEqual([sid]); + }); + + it("输入类动作要先接管;他人接管中 409;读(快照/截图)随时可用", async () => { + const nav = await h.req(`/browser/sessions/${sid}/navigate`, { body: { url: `${ORIGIN}/` } }); + expect(nav.status).toBe(409); + expect(nav.body.error.message).toContain("先接管"); + const shot = await h.req(`/browser/sessions/${sid}/screenshot`); + expect(shot.status).toBe(200); + expect(shot.body.data).toMatchObject({ + mimeType: "image/png", + width: MOCK_VIEWPORT.width, + height: MOCK_VIEWPORT.height, + modelSafe: false, + controller: "automation", + }); + const png = Buffer.from(shot.body.data.dataBase64, "base64"); + expect(png.subarray(1, 4).toString()).toBe("PNG"); + + const a = await h.req(`/browser/sessions/${sid}/control/acquire`, { body: { reason: "登录" } }); + expect(a.status).toBe(200); + expect(a.body.data.lease).toMatchObject({ holderKind: "human", status: "active" }); + expect(a.body.data.session.controller).toBe("human"); + const other = await h.req(`/browser/sessions/${sid}/control/acquire`, { body: {}, key: OTHER_KEY }); + expect(other.status).toBe(409); + const otherClick = await h.req(`/browser/sessions/${sid}/click`, { body: { target: row(1) }, key: OTHER_KEY }); + expect(otherClick.status).toBe(409); + }); + + it("人工:导航限站点域名;坐标点击 + 凭证注入登录;快照遮蔽密码与 passkey", async () => { + const evil = await h.req(`/browser/sessions/${sid}/navigate`, { body: { url: "https://evil.example/" } }); + expect(evil.status).toBe(400); + expect(evil.body.error.message).toContain("白名单"); + expect((await h.req(`/browser/sessions/${sid}/navigate`, { body: {} })).status).toBe(400); + const nav = await h.req(`/browser/sessions/${sid}/navigate`, { body: { url: `${ORIGIN}/torrents.php` } }); + expect(nav.body.data).toMatchObject({ loginRequired: true, ok: true }); + + expect((await h.req(`/browser/sessions/${sid}/click`, { body: { target: row(1) } })).status).toBe(200); + await h.req(`/browser/sessions/${sid}/type`, { body: { credential: "username" } }); + await h.req(`/browser/sessions/${sid}/click`, { body: { target: row(2) } }); + const typed = await h.req(`/browser/sessions/${sid}/type`, { body: { credential: "password" } }); + expect(typed.status).toBe(200); + expect(JSON.stringify(typed.body)).not.toContain(PASSWORD); + // 登录前快照:密码框有值,返回前已遮蔽 + const snap = await h.req(`/browser/sessions/${sid}/snapshot`); + expect(snap.body.data.text).toContain("登录"); + expect(snap.body.data.text).not.toContain(PASSWORD); + // 凭证确实下发到了浏览器(mock 记录),点击按钮完成登录 + expect(h.mock.actions.some((x) => x.action === "type" && x.body.text === PASSWORD)).toBe(true); + await h.req(`/browser/sessions/${sid}/type`, { body: { key: "Enter" } }); + const after = await h.req(`/browser/sessions/${sid}/snapshot`); + expect(after.body.data.text).toContain("种子列表"); + expect(after.body.data.text).not.toContain(PASSKEY); + expect(after.body.data.url).toContain("/torrents.php"); + // 后退 / 滚动 + expect((await h.req(`/browser/sessions/${sid}/scroll`, { body: { direction: "down" } })).status).toBe(200); + const back = await h.req(`/browser/sessions/${sid}/navigate`, { body: { action: "reload" } }); + expect(back.status).toBe(200); + }); + + it("下载:点击捕获 → 暂存 0600 文件、sha256/infohash 核验、按 infohash 匹配资源,经 /acquisition-plans 提交;同一文件不重复提交", async () => { + await h.req(`/browser/sessions/${sid}/snapshot`); + const r = await h.req(`/browser/sessions/${sid}/download`, { + body: { target: { ref: "e11" }, submit: { note: "browser" } }, + }); + expect(r.status, JSON.stringify(r.body)).toBe(200); + const d = r.body.data; + expect(d.artifact.sha256).toBe(createHash("sha256").update(h.torrent.bytes).digest("hex")); + expect(d.artifact.sourceUrl).not.toContain(PASSKEY); + expect(d.verification).toMatchObject({ valid: true, infohashV1: h.torrent.infoHash, private: true }); + expect(d.sourceReleaseId).toBe(h.releaseId); + // 资源未识别:与 POST /acquisition-plans 相同的拒绝,产物照常返回 + expect(d.submission).toBeNull(); + expect(d.note).toContain("提交未通过"); + const files = await readdir(h.stagingDir); + expect(files).toHaveLength(1); + expect(new Uint8Array(await readFile(join(h.stagingDir, files[0]!)))).toEqual(h.torrent.bytes); + + // 识别到作品后再下载:走同一流程(无版本目标 → no_slot);同一文件第二次捕获标 duplicate、不提交 + const entity = await h.db + .insertInto("media_entity") + .values({ kind: "movie", title: "奥本海默", year: 2023, status: "active" }) + .returning("id") + .executeTakeFirstOrThrow(); + await h.db + .updateTable("source_release") + .set({ media_entity_id: entity.id }) + .where("id", "=", h.releaseId) + .execute(); + await h.req(`/browser/sessions/${sid}/snapshot`); + const dup = await h.req(`/browser/sessions/${sid}/download`, { body: { target: { ref: "e11" }, submit: {} } }); + expect(dup.body.data).toMatchObject({ duplicate: true, submission: null }); + expect(dup.body.data.note).toContain("未重复提交"); + // 未接管时不能点击触发下载 + await h.req(`/browser/sessions/${sid}/control/release`, { body: {} }); + const denied = await h.req(`/browser/sessions/${sid}/download`, { body: { target: { ref: "e11" } } }); + expect(denied.status).toBe(409); + }); + + it("交还后需重新读取页面;租约到期 → 等待处理 + 待处理事项,自动化被拒", async () => { + const v = await h.req(`/browser/sessions/${sid}`); + expect(v.body.data).toMatchObject({ controller: "automation", needsSnapshot: true, lease: null }); + await h.grant("browser.interact"); + await expect( + h.registry.invoke("browser.interact", { sessionId: sid, action: "click", target: { ref: "e11" } }, h.agentCtx()), + ).rejects.toMatchObject({ code: "conflict" }); + + await h.req(`/browser/sessions/${sid}/control/acquire`, { body: {} }); + await h.db + .updateTable("browser_control_lease") + .set({ acquired_at: new Date(Date.now() - 120_000), expires_at: new Date(Date.now() - 1_000) }) + .where("session_id", "=", sid) + .where("status", "=", "active") + .execute(); + const w = await h.req(`/browser/sessions/${sid}`); + expect(w.body.data).toMatchObject({ controller: "waiting", waitingReason: "lease_expired", lease: null }); + const dec = await h.db.selectFrom("decision").selectAll().where("subject_id", "=", sid).execute(); + expect(dec).toHaveLength(1); + expect(dec[0]).toMatchObject({ status: "awaiting_user", subject_type: "browser_session" }); + await expect( + h.registry.invoke("browser.interact", { sessionId: sid, action: "scroll" }, h.agentCtx()), + ).rejects.toMatchObject({ code: "conflict" }); + // 人工重新接管并交还 → 自动化恢复,待处理事项关闭 + await h.req(`/browser/sessions/${sid}/control/acquire`, { body: {} }); + await h.req(`/browser/sessions/${sid}/control/release`, { body: { reason: "处理完" } }); + const d2 = await h.db + .selectFrom("decision") + .select("status") + .where("subject_id", "=", sid) + .executeTakeFirstOrThrow(); + expect(d2.status).toBe("resolved"); + }); + + it("服务重启:按数据库恢复会话,旧引用失效;上游回收的标签页标 lost", async () => { + // 新模块实例 = 进程重启(适配器内存为空) + const fresh = createBrowserModule({ stagingDir: h.stagingDir }); + const reg = (await import("../../agent/capability-registry.ts")).createCapabilityRegistry({ + logger: h.ctx.logger, + db: h.db, + settings: h.ctx.settings, + }); + registerBrowserCapabilities(reg, () => fresh.service(h.ctx)); + await expect( + reg.invoke("browser.interact", { sessionId: sid, action: "click", target: { ref: "e11" } }, h.agentCtx()), + ).rejects.toMatchObject({ code: "conflict" }); + const snap = (await reg.invoke("browser.snapshot", { siteId: h.siteId }, h.agentCtx())) as Loose; + expect(snap.snapshot.text).toContain("种子列表"); + expect(snap).toMatchObject({ controller: "automation", loginRequired: false }); + expect((await fresh.service(h.ctx).view(sid)).needsSnapshot).toBe(false); + + const tab = ( + await h.db.selectFrom("browser_tab").selectAll().where("session_id", "=", sid).executeTakeFirstOrThrow() + ).upstream_tab_id; + h.mock.dropTab(tab); + await expect(reg.invoke("browser.snapshot", { sessionId: sid }, h.agentCtx())).rejects.toMatchObject({ + code: "conflict", + }); + const t = await h.db + .selectFrom("browser_tab") + .select("status") + .where("upstream_tab_id", "=", tab) + .executeTakeFirstOrThrow(); + expect(t.status).toBe("lost"); + }); + + it("站点停用 → 会话关闭、自动访问停止;关闭会话撤销租约并清上游标签页", async () => { + const s2 = await h.req("/browser/sessions", { body: { siteId: h.siteId, accountId: null } }); + expect(s2.status).toBe(200); + const id2 = s2.body.data.id; + expect(id2).not.toBe(sid); + await h.req(`/browser/sessions/${id2}/control/acquire`, { body: {} }); + const closed = await h.req(`/browser/sessions/${id2}`, { method: "DELETE" }); + expect(closed.body.data).toMatchObject({ controller: "closed", closeReason: "closed_by_user", lease: null }); + const lease = await h.db.selectFrom("browser_control_lease").selectAll().where("session_id", "=", id2).execute(); + expect(lease.every((l) => l.status === "revoked")).toBe(true); + + await h.db.updateTable("site").set({ enabled: false }).where("id", "=", h.siteId).execute(); + const r = await h.req(`/browser/sessions/${sid}/snapshot`); + expect(r.status).toBe(412); + const s = await h.db.selectFrom("browser_session").selectAll().where("id", "=", sid).executeTakeFirstOrThrow(); + expect(s).toMatchObject({ controller: "closed", close_reason: "site_disabled" }); + expect((await h.req("/browser/sessions", { body: { siteId: h.siteId } })).status).toBe(412); + await h.db.updateTable("site").set({ enabled: true }).where("id", "=", h.siteId).execute(); + }); + + it("浏览器服务认证失败(401)与超时:报上游不可用,不无限重试", async () => { + const r = await h.req("/browser/sessions", { body: { siteId: h.siteId } }); + const id = r.body.data.id; + await h.req(`/browser/sessions/${id}/control/acquire`, { body: {} }); + h.mock.fault(/\/navigate$/, { status: 401, times: 5 }); + const a = await h.req(`/browser/sessions/${id}/navigate`, { body: { url: `${ORIGIN}/` } }); + expect(a.status).toBe(412); + expect(JSON.stringify(a.body)).not.toContain(ACCESS_KEY); + h.mock.clearFaults(); + h.mock.fault(/\/navigate$/, { delayMs: 2_500, times: 5 }); + const before = h.mock.callsTo(/\/navigate$/).length; + const t = await h.req(`/browser/sessions/${id}/navigate`, { body: { url: `${ORIGIN}/` } }); + expect(t.status).toBe(502); + // maxRetries = 1:共 2 次 + expect(h.mock.callsTo(/\/navigate$/).length - before).toBe(2); + h.mock.clearFaults(); + }, 30_000); +}); diff --git a/apps/server/src/modules/browser/__tests__/browser-e2e.test.ts b/apps/server/src/modules/browser/__tests__/browser-e2e.test.ts new file mode 100644 index 00000000..0cd4bce2 --- /dev/null +++ b/apps/server/src/modules/browser/__tests__/browser-e2e.test.ts @@ -0,0 +1,114 @@ +import { MOCK_ROW_H, MOCK_ROW_Y0 } from "@umt/testkit"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { + type BrowserHarness, + createBrowserHarness, + type Loose, + ORIGIN, + PASSKEY, + PASSWORD, + pgAvailable, +} from "./harness.ts"; + +// C1-E e2e(mock 站点,未连真实 camofox): +// Agent 打开站点 → 登录失效 + 验证码 → 会话转等待处理(待处理事项)→ 自动化被拒 → +// 人在原站视图接管(截图坐标点击 + 输入,密码由服务端注入)完成登录 → 交还 → +// Agent 的旧引用被拒,重新读取页面后继续:捕获种子下载、核验并匹配到本地资源。 + +const up = await pgAvailable(); +const at = (i: number) => ({ x: 320, y: MOCK_ROW_Y0 + MOCK_ROW_H * i + 10 }); + +describe.skipIf(!up)("e2e:登录失效 → 人工接管 → 交还 → Agent 重新读取并继续", () => { + let h: BrowserHarness; + beforeAll(async () => { + h = await createBrowserHarness("e2e", { captcha: true }); + await h.grant("browser.navigate", "browser.interact", "browser.download"); + }, 120_000); + afterAll(async () => { + await h?.close(); + }); + + it("全流程", async () => { + const agent = h.agentCtx(); + const invoke = (id: string, input: unknown) => h.registry.invoke(id, input, agent) as Promise; + + // 1. Agent 打开种子列表:登录已失效,落到登录页 + const nav = await invoke("browser.navigate", { siteId: h.siteId, url: `${ORIGIN}/torrents.php` }); + expect(nav).toMatchObject({ loginRequired: true, controller: "automation" }); + const sid = nav.sessionId as string; + + // 2. Agent 读页面:有验证码 → 会话转等待处理,生成待处理事项 + const snap = await invoke("browser.snapshot", { sessionId: sid }); + expect(snap).toMatchObject({ controller: "waiting", waitingReason: "captcha", loginRequired: true }); + expect(snap.snapshot.text).toContain("验证码"); + const decision = await h.db + .selectFrom("decision") + .selectAll() + .where("subject_id", "=", sid) + .executeTakeFirstOrThrow(); + expect(decision).toMatchObject({ status: "awaiting_user", kind: "other" }); + expect(decision.title).toContain("观众"); + + // 3. 等待处理期间 Agent 不能输入(不无限重试) + await expect( + invoke("browser.interact", { sessionId: sid, action: "type", target: { ref: "e3" }, credential: "password" }), + ).rejects.toMatchObject({ code: "conflict" }); + + // 4. 人在原站视图接管:截图 → 坐标点击 → 输入(密码走凭证注入)→ 验证码 → 登录 + const view = await h.req(`/browser/sessions/${sid}`); + expect(view.body.data).toMatchObject({ controller: "waiting", waitingReason: "captcha" }); + const acq = await h.req(`/browser/sessions/${sid}/control/acquire`, { body: { reason: "验证码" } }); + expect(acq.body.data.session.controller).toBe("human"); + const shot = await h.req(`/browser/sessions/${sid}/screenshot`); + expect(shot.body.data).toMatchObject({ controller: "human", modelSafe: false }); + const human = async (path: string, body: unknown) => { + const r = await h.req(`/browser/sessions/${sid}/${path}`, { body }); + expect(r.status, JSON.stringify(r.body)).toBe(200); + }; + await human("click", { target: at(1) }); + await human("type", { text: "fixture-user" }); + await human("click", { target: at(2) }); + await human("type", { credential: "password" }); + await human("click", { target: at(4) }); + await human("type", { text: "4321" }); + await human("click", { target: at(5) }); + const loggedIn = await h.req(`/browser/sessions/${sid}/snapshot`); + expect(loggedIn.body.data.text).toContain("种子列表"); + expect(JSON.stringify(loggedIn.body)).not.toContain(PASSWORD); + expect(JSON.stringify(loggedIn.body)).not.toContain(PASSKEY); + + // 5. 交还:Agent 恢复前必须重新读页面;待处理事项关闭 + const rel = await h.req(`/browser/sessions/${sid}/control/release`, { body: { reason: "已登录" } }); + expect(rel.body.data.session).toMatchObject({ controller: "automation", needsSnapshot: true, waitingReason: null }); + const d2 = await h.db + .selectFrom("decision") + .select("status") + .where("id", "=", decision.id) + .executeTakeFirstOrThrow(); + expect(d2.status).toBe("resolved"); + + // 6. 旧引用(交接前拿到的 eN)被拒,不重放 + await expect( + invoke("browser.interact", { sessionId: sid, action: "click", target: { ref: "e6" } }), + ).rejects.toMatchObject({ code: "conflict", message: expect.stringContaining("重新读取页面") }); + + // 7. 重新读取:人工登录后的状态;继续捕获下载 → 暂存核验 → 匹配本地资源(提交仍走 download.plan) + const fresh = await invoke("browser.snapshot", { siteId: h.siteId }); + expect(fresh).toMatchObject({ controller: "automation", loginRequired: false, waitingReason: null }); + expect(fresh.snapshot.text).toContain("种子列表"); + expect((await h.req(`/browser/sessions/${sid}`)).body.data.needsSnapshot).toBe(false); + const dl = await invoke("browser.download", { sessionId: sid, target: { ref: "e11" }, timeoutMs: 2_000 }); + expect(dl).toMatchObject({ + sourceReleaseId: h.releaseId, + duplicate: false, + verification: { valid: true, infohashV1: h.torrent.infoHash }, + }); + expect(dl.note).toContain("download.plan"); + expect(JSON.stringify(dl)).not.toContain(PASSKEY); + + // 审计:能力调用留痕,输入里没有凭证 + const audits = await h.db.selectFrom("audit_event").select(["action", "target_id", "payload"]).execute(); + expect(audits.some((a) => a.target_id === "browser.download" && a.action === "capability.invoke")).toBe(true); + expect(JSON.stringify(audits)).not.toContain(PASSWORD); + }, 60_000); +}); diff --git a/apps/server/src/modules/browser/__tests__/harness.ts b/apps/server/src/modules/browser/__tests__/harness.ts new file mode 100644 index 00000000..1bbab667 --- /dev/null +++ b/apps/server/src/modules/browser/__tests__/harness.ts @@ -0,0 +1,209 @@ +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import type { CapabilityRegistry, InvokeContext } from "@umt/agent"; +import { type DB, type Kysely, migrateToLatest, sql } from "@umt/db"; +import { buildFakeTorrent, createMockCamofox, type FakeTorrent, type MockCamofox } from "@umt/testkit"; +import { Kysely as K, PostgresDialect } from "kysely"; +import pg from "pg"; +import { withSchemaLock } from "../../../../test/auth-settings/harness.ts"; +import { DEV_KEY, memoryLogger, memoryStore } from "../../../../test/helpers.ts"; +import { fakeEngine, fakeSecrets, PG_URL, plainSettings } from "../../../../test/sources-identity/harness.ts"; +import { createOutboxPort } from "../../../context.ts"; +import { createApp } from "../../../http/app.ts"; +import { EventBus } from "../../../http/sse.ts"; +import { createAcquisitionOrganizeModule } from "../../acquisition-organize/index.ts"; +import { createCapabilityRegistry } from "../../agent/capability-registry.ts"; +import { createAgentModule } from "../../agent/index.ts"; +import { createSourcesIdentityModule } from "../../sources-identity/index.ts"; +import type { ServerContext } from "../../types.ts"; +import { registerBrowserCapabilities } from "../capabilities.ts"; +import { createBrowserModule } from "../index.ts"; + +// C1-E 浏览器模块测试装配:独立 PG schema + 真实迁移 + 交互式 mock camofox(观众风格站点)+ 真实能力注册中心。 + +export { pgAvailable } from "../../../../test/sources-identity/harness.ts"; + +export const ORIGIN = "https://audiences.example"; +export const PASSWORD = "brw-s3cret-login-pw"; +export const PASSKEY = "0f0e0d0c0b0a09080706050403020100"; +export const ACCESS_KEY = "camofox-access-key-c1e"; +/** 第二位管理员(接管冲突) */ +export const OTHER_KEY = "brw-other-admin-key"; +export const READ_KEY = "brw-read-key"; + +// biome-ignore lint/suspicious/noExplicitAny: 测试断言直接读 JSON +export type Loose = any; + +export interface BrowserHarness { + db: Kysely; + ctx: ServerContext; + mock: MockCamofox; + torrent: FakeTorrent; + registry: CapabilityRegistry; + siteId: string; + accountId: string; + providerId: string; + releaseId: string; + stagingDir: string; + secrets: ReturnType; + req(path: string, init?: { method?: string; body?: unknown; key?: string }): Promise<{ status: number; body: Loose }>; + agentCtx(): InvokeContext; + grant(...actions: string[]): Promise; + close(): Promise; +} + +export async function createBrowserHarness(name: string, opts: { captcha?: boolean } = {}): Promise { + const torrent = buildFakeTorrent({ name: "Oppenheimer.2023.2160p.UHD.BluRay.x265-GRP" }); + const mock = await createMockCamofox({ + accessKey: ACCESS_KEY, + site: { + origin: ORIGIN, + username: "fixture-user", + password: PASSWORD, + passkey: PASSKEY, + ...(opts.captcha ? { captcha: true, captchaAnswer: "4321" } : {}), + releases: [{ id: "42", title: "Oppenheimer 2023 2160p UHD BluRay" }], + torrent: () => torrent.bytes, + }, + }); + const stagingDir = await mkdtemp(join(tmpdir(), "umt-browser-")); + const schema = `c1e_${name}_${process.pid.toString(36)}${Math.random().toString(36).slice(2, 7)}`; + const admin = new K({ + dialect: new PostgresDialect({ pool: new pg.Pool({ connectionString: PG_URL, max: 1 }) }), + }); + await sql.raw(`create schema ${schema}`).execute(admin); + await import("@umt/db"); + const db = new K({ + dialect: new PostgresDialect({ + pool: new pg.Pool({ connectionString: PG_URL, max: 8, options: `-c search_path=${schema},public` }), + }), + }) as unknown as Kysely; + await withSchemaLock(() => migrateToLatest(db, { migrationTableSchema: schema })); + + const { logger } = memoryLogger(); + const secrets = fakeSecrets(() => db); + const settings = plainSettings(() => db); + const registry = createCapabilityRegistry({ logger, db, settings }); + const ctx = { + role: "all", + db, + config: { NODE_ENV: "development", DEV_ADMIN_KEY: DEV_KEY } as never, + logger, + events: new EventBus(5), + engine: fakeEngine(), + outbox: createOutboxPort(), + secrets, + settings, + capabilities: registry, + } as ServerContext; + const sources = createSourcesIdentityModule({ tmdb: () => null }); + const browser = createBrowserModule({ runtime: (c) => sources.runtime(c), stagingDir, downloadPollMs: 20 }); + registerBrowserCapabilities(registry, () => browser.service(ctx)); + const app = createApp({ + config: { NODE_ENV: "development", DEV_ADMIN_KEY: DEV_KEY }, + logger, + bus: new EventBus(5), + role: "all", + version: "0.0.0-test", + // POST /acquisition-plans 在 agent 模块 + modules: [sources, createAcquisitionOrganizeModule(), createAgentModule(), browser], + ctx, + apiKeyStore: memoryStore({ + [OTHER_KEY]: { id: "other-admin", scopes: ["admin"] }, + [READ_KEY]: { id: "reader", scopes: ["read"] }, + }), + }); + + const siteId = ( + await db + .insertInto("site") + .values({ + key: "audiences", + name: "观众", + adapter: "nexusphp", + base_url: `${ORIGIN}/`, + rate_limit: JSON.stringify({ minIntervalMs: 0, concurrency: 5 }), + }) + .returning("id") + .executeTakeFirstOrThrow() + ).id; + const pwRef = await secrets.create({ purpose: "site_password", plaintext: PASSWORD }); + const accountId = ( + await db + .insertInto("site_account") + .values({ site_id: siteId, label: "main", auth_kind: "password", secret_ref: pwRef, login_name: "fixture-user" }) + .returning("id") + .executeTakeFirstOrThrow() + ).id; + const keyRef = await secrets.create({ purpose: "other", plaintext: ACCESS_KEY }); + const providerId = ( + await db + .insertInto("browser_provider") + .values({ + name: "camofox", + kind: "camofox", + base_url: mock.url, + secret_ref: keyRef, + enabled: true, + config: JSON.stringify({ verified: { snapshot: true }, navigationTimeoutMs: 2_000, maxRetries: 1 }), + }) + .returning("id") + .executeTakeFirstOrThrow() + ).id; + const releaseId = ( + await db + .insertInto("source_release") + .values({ + site_id: siteId, + site_release_id: "42", + title: "Oppenheimer 2023 2160p UHD BluRay", + infohash_v1: torrent.infoHash, + }) + .returning("id") + .executeTakeFirstOrThrow() + ).id; + + return { + db, + ctx, + mock, + torrent, + registry, + siteId, + accountId, + providerId, + releaseId, + stagingDir, + secrets, + async req(path, init = {}) { + const res = await app.request(`/api/v1${path}`, { + method: init.method ?? (init.body !== undefined ? "POST" : "GET"), + headers: { + authorization: `Bearer ${init.key ?? DEV_KEY}`, + ...(init.body !== undefined ? { "content-type": "application/json" } : {}), + }, + ...(init.body !== undefined ? { body: JSON.stringify(init.body) } : {}), + }); + return { status: res.status, body: (await res.json()) as Loose }; + }, + agentCtx: () => ({ + actorId: "agent:c1e", + taskId: crypto.randomUUID(), + scopeIds: [], + signal: new AbortController().signal, + source: "agent", + }), + async grant(...actions) { + for (const action of actions) + await db.insertInto("automation_grant").values({ action, granted_by: "test" }).execute(); + }, + async close() { + await db.destroy(); + await withSchemaLock(() => sql.raw(`drop schema if exists ${schema} cascade`).execute(admin)); + await admin.destroy(); + await mock.close(); + await rm(stagingDir, { recursive: true, force: true }); + }, + }; +} diff --git a/apps/server/src/modules/browser/capabilities.ts b/apps/server/src/modules/browser/capabilities.ts new file mode 100644 index 00000000..4b109fcd --- /dev/null +++ b/apps/server/src/modules/browser/capabilities.ts @@ -0,0 +1,188 @@ +import { isLoginUrl } from "@umt/adapters"; +import { CapabilityError, type CapabilityRegistry } from "@umt/agent"; +import { capabilitySpec } from "@umt/contracts"; +import type { Actor, BrowserService } from "./service.ts"; +import { BrowserServiceError } from "./service.ts"; + +// C1-E Agent 浏览器能力:browser.snapshot(read)/ browser.navigate / browser.interact / browser.download(execute)。 +// 一律以自动化身份执行:会话被人工接管或在等待处理时拒绝;元素引用只来自最近一次 snapshot; +// 凭证用 credential 字段由服务端注入;下载只暂存核验并匹配本地资源,提交仍走 download.plan / download.submit。 + +const CODE = { + not_found: "not_found", + conflict: "conflict", + precondition: "precondition", + bad_request: "precondition", + upstream: "precondition", + rate_limited: "budget", +} as const; + +async function guard(fn: () => Promise): Promise { + try { + return await fn(); + } catch (e) { + if (e instanceof BrowserServiceError) throw new CapabilityError(CODE[e.code], e.message); + throw e; + } +} + +const agent = (actorId: string): Actor => ({ kind: "automation", holder: actorId }); + +export function registerBrowserCapabilities(registry: CapabilityRegistry, svc: () => BrowserService): void { + /** sessionId 优先;否则取站点的未关闭会话 */ + const resolveSession = async (input: { sessionId?: string | undefined; siteId?: string | undefined }) => { + if (input.sessionId) return input.sessionId; + const row = input.siteId ? await svc().liveSessionFor(input.siteId) : undefined; + if (!row) throw new BrowserServiceError("not_found", "该站点没有浏览器会话,先用 browser.navigate 打开页面"); + return row.id; + }; + + registry.register(capabilitySpec("browser.snapshot"), (input, ctx) => + guard(async () => { + const id = await resolveSession(input); + const snap = await svc().run( + id, + agent(ctx.actorId), + (p, ref) => p.snapshot(ref, input.offset !== undefined ? { offset: input.offset } : {}), + { write: false, patch: () => ({ needsSnapshot: false }), ...(input.tabId ? { tabId: input.tabId } : {}) }, + ); + const s = svc(); + const loginPage = isLoginUrl(snap.url || null); + await s.afterAutomationSnapshot(id, snap.text, loginPage); + const view = await s.view(id); + return { + snapshot: snap, + controller: view.controller, + waitingReason: view.waitingReason, + loginRequired: loginPage, + }; + }), + ); + + registry.register(capabilitySpec("browser.navigate"), (input, ctx) => + guard(async () => { + let id = input.sessionId; + if (!id) { + const live = await svc().liveSessionFor(input.siteId!, input.accountId); + id = + live?.id ?? + (await svc().open({ + siteId: input.siteId!, + ...(input.accountId ? { accountId: input.accountId } : {}), + url: input.url, + })); + if (!live) { + const v = await svc().view(id); + const info = v.currentUrl ?? ""; + return { + sessionId: id, + url: info, + httpStatus: null, + ok: true, + attempts: 1, + loginRequired: isLoginUrl(info || null), + controller: v.controller, + }; + } + } + const sid = id; + const nav = await svc().run( + sid, + agent(ctx.actorId), + async (p, ref, x) => { + await svc().limited(x.site); + const r = await p.navigate(ref, input.url, x.budget); + return { ...r, loginRequired: p.sessionInfo(sid)?.loginPage ?? false }; + }, + input.tabId ? { tabId: input.tabId } : {}, + ); + const v = await svc().view(sid); + return { sessionId: sid, ...nav, controller: v.controller }; + }), + ); + + registry.register(capabilitySpec("browser.interact"), (input, ctx) => + guard(async () => { + const s = svc(); + if (input.action === "request_human") { + await s.requestHuman(input.sessionId, input.reason ?? "agent_request"); + const v = await s.view(input.sessionId); + return { ok: true as const, controller: v.controller, needsSnapshot: true }; + } + const needTarget = input.action === "click" || input.action === "select"; + if (needTarget && !input.target) throw new BrowserServiceError("bad_request", `${input.action} 需要 target`); + await s.run( + input.sessionId, + agent(ctx.actorId), + async (p, ref, x) => { + switch (input.action) { + case "click": + return p.click(ref, input.target!); + case "select": + return p.select(ref, input.target!, input.value ?? ""); + case "scroll": + return p.scroll(ref, { direction: input.direction ?? "down", amount: input.amount ?? 600 }); + case "press": + return p.press(ref, input.key ?? "Enter"); + case "type": { + const extra = input.submit !== undefined ? { submit: input.submit } : {}; + if (input.credential) + return p.type(ref, input.target ?? null, { + kind: "secret", + siteId: x.row.site_id, + accountId: x.row.account_id, + field: input.credential, + ...extra, + }); + if (input.text === undefined) + throw new BrowserServiceError("bad_request", "type 需要 text 或 credential"); + return p.type(ref, input.target ?? null, { kind: "text", text: input.text, ...extra }); + } + } + }, + input.tabId ? { tabId: input.tabId } : {}, + ); + const info = (await s.view(input.sessionId)).controller; + const changes = input.action === "click" || input.action === "press" || !!input.submit; + return { ok: true as const, controller: info, needsSnapshot: changes }; + }), + ); + + registry.register(capabilitySpec("browser.download"), (input, ctx) => + guard(async () => { + const s = svc(); + const artifact = await s.run( + input.sessionId, + agent(ctx.actorId), + async (p, ref) => { + if (input.target) await p.click(ref, input.target); + return p.waitForDownload(ref, { timeoutMs: input.timeoutMs, maxBytes: 10 * 1024 * 1024 }); + }, + input.tabId ? { tabId: input.tabId } : {}, + ); + const v = await s.verifyDownload(input.sessionId, artifact); + await s.recordDownload(input.sessionId, { + sha256: artifact.sha256, + artifactId: artifact.id, + sourceReleaseId: v.sourceReleaseId, + planId: v.duplicate?.planId ?? null, + }); + return { + artifactId: artifact.id, + filename: artifact.filename, + sizeBytes: artifact.sizeBytes, + sha256: artifact.sha256, + verification: v.verification, + sourceReleaseId: v.sourceReleaseId, + duplicate: !!v.duplicate, + note: v.duplicate + ? "同一会话已捕获过这个文件;先查已有下载计划,不要重复提交" + : !v.verification.valid + ? "不是有效的种子文件" + : v.sourceReleaseId + ? "已暂存核验;用 download.plan 提交这个 sourceReleaseId" + : "资源不在本地索引;先 release.search 该站点后再提交", + }; + }), + ); +} diff --git a/apps/server/src/modules/browser/index.ts b/apps/server/src/modules/browser/index.ts new file mode 100644 index 00000000..ccd71a05 --- /dev/null +++ b/apps/server/src/modules/browser/index.ts @@ -0,0 +1,41 @@ +import { sourcesIdentityModule } from "../sources-identity/index.ts"; +import type { ServerContext, ServerModule } from "../types.ts"; +import { registerBrowserCapabilities } from "./capabilities.ts"; +import { registerBrowserRoutes } from "./routes.ts"; +import { type BrowserModuleOptions, BrowserService } from "./service.ts"; + +/** + * C1-E 浏览器:BrowserProvider(camofox)之上的站点浏览器会话、原站视图与人工接管、Agent 浏览器能力。 + * 会话/标签/租约以数据库(0023 四张表)为准;站点限流与 sources-identity 共享;下载产物只进后端暂存核验, + * 提交走 POST /acquisition-plans 同一流程。运维与真实实例核验见 docs/runbook/browser.md。 + */ +export function createBrowserModule(opts: BrowserModuleOptions = {}): ServerModule & { + service(ctx: ServerContext): BrowserService; +} { + const services = new WeakMap(); + const options: BrowserModuleOptions = { + runtime: (ctx) => sourcesIdentityModule.runtime(ctx), + ...opts, + }; + const service = (ctx: ServerContext) => { + let s = services.get(ctx); + if (!s) { + s = new BrowserService(ctx, options); + services.set(ctx, s); + } + return s; + }; + return { + name: "browser", + service, + registerRoutes(app, ctx) { + registerBrowserRoutes(app, () => service(ctx)); + }, + registerTasks() {}, + registerCapabilities(registry, ctx) { + registerBrowserCapabilities(registry, () => service(ctx)); + }, + }; +} + +export const browserModule = createBrowserModule(); diff --git a/apps/server/src/modules/browser/routes.ts b/apps/server/src/modules/browser/routes.ts new file mode 100644 index 00000000..b5643407 --- /dev/null +++ b/apps/server/src/modules/browser/routes.ts @@ -0,0 +1,292 @@ +import { createRoute, type RouteConfig } from "@hono/zod-openapi"; +import { + Api, + type ApiScope, + apiResponse, + type BrowserDownloadResult, + type BrowserNavigation, + type BrowserScreenshotView, + type EndpointDef, + type ManualAcquisitionResult, +} from "@umt/contracts"; +import type { Context } from "hono"; +import type { z } from "zod"; +import { type AuthEnv, requireScope } from "../../http/auth.ts"; +import { ApiException } from "../../http/errors.ts"; +import { InternalDispatcher } from "../search/dispatch.ts"; +import type { ModuleApp } from "../types.ts"; +import { type Actor, type BrowserService, BrowserServiceError, imageSize } from "./service.ts"; + +// C1-E 浏览器 HTTP 端点:原站视图与外部程序共用。读(会话/快照/截图)任何时候可用; +// 输入类动作(导航/点击/输入/滚动/选择/标签页/触发下载)只接受持有人工租约的调用者(先接管)。 + +type Infer = T extends z.ZodType ? z.infer : Record; +interface Input { + params: Infer; + query: Infer; + body: Infer; + c: Context; +} + +const CODE = { + not_found: "not_found", + conflict: "conflict", + precondition: "precondition_failed", + bad_request: "bad_request", + upstream: "upstream_unavailable", + rate_limited: "rate_limited", +} as const; + +function route(app: ModuleApp, def: D, handler: (input: Input) => Promise) { + const request: RouteConfig["request"] = {}; + if (def.params) request.params = def.params as never; + if (def.query) request.query = def.query as never; + if (def.body) request.body = { content: { "application/json": { schema: def.body as never } }, required: true }; + const r = createRoute({ + method: def.method, + path: def.path, + summary: def.summary, + tags: ["browser"], + middleware: [requireScope(def.scope as ApiScope)] as never, + request, + responses: { + 200: { description: def.summary, content: { "application/json": { schema: apiResponse(def.response) } } }, + }, + }); + // biome-ignore lint/suspicious/noExplicitAny: 路由泛型在封装里擦除,入参已由 zod-openapi 校验 + app.openapi(r, (async (c: any) => { + const input = { + params: def.params ? c.req.valid("param") : {}, + query: def.query ? c.req.valid("query") : {}, + body: def.body ? c.req.valid("json") : {}, + c, + } as Input; + try { + const data = def.response.parse(await handler(input)); + return c.json({ ok: true, data }, 200); + } catch (e) { + if (e instanceof BrowserServiceError) throw new ApiException(CODE[e.code], e.message); + throw e; + } + // biome-ignore lint/suspicious/noExplicitAny: 同上 + }) as any); +} + +/** 人工操作者:控制台会话 / API Key 的身份(租约 holder) */ +function human(c: Context): Actor { + const p = c.get("principal"); + return { kind: "human", holder: `${p?.kind ?? "unknown"}:${p?.name ?? p?.id ?? "?"}` }; +} + +const b64 = (data: Uint8Array) => Buffer.from(data).toString("base64"); + +export function registerBrowserRoutes(app: ModuleApp, svc: () => BrowserService) { + const dispatch = new InternalDispatcher(app); + const B = Api.browser; + + route(app, B.providers, async ({ query }) => ({ items: await svc().listProviders(query.probe === "true") })); + + route(app, B.sessions, async ({ query }) => ({ + items: await svc().list({ + ...(query.siteId ? { siteId: query.siteId } : {}), + includeClosed: query.includeClosed === "true", + }), + })); + + route(app, B.open, async ({ body }) => { + const id = await svc().open({ + siteId: body.siteId, + ...(body.accountId !== undefined ? { accountId: body.accountId } : {}), + ...(body.providerId ? { providerId: body.providerId } : {}), + ...(body.url ? { url: body.url } : {}), + }); + return svc().view(id); + }); + + route(app, B.get, async ({ params }) => svc().view(params.id)); + + route(app, B.close, async ({ params }) => { + await svc().close(params.id, "closed_by_user"); + return svc().view(params.id); + }); + + route(app, B.openTab, async ({ params, body, c }) => { + await svc().run(params.id, human(c), async (p, ref, x) => { + if (body.url) await svc().limited(x.site); + return p.openTab(ref.sessionId, { ...(body.url ? { url: body.url } : {}), budget: x.budget, as: "human" }); + }); + return svc().view(params.id); + }); + + route(app, B.closeTab, async ({ params, c }) => { + await svc().run(params.id, human(c), (p, ref) => p.closeTab(ref), { tabId: params.tabId }); + return svc().view(params.id); + }); + + route(app, B.navigate, async ({ params, body, c }): Promise => { + if (!body.url && !body.action) throw new ApiException("bad_request", "url 与 action 二选一"); + const out = await svc().run( + params.id, + human(c), + async (p, ref, x) => { + await svc().limited(x.site); + const nav = body.url ? await p.navigate(ref, body.url, x.budget) : await p.history(ref, body.action!, x.budget); + return { ...nav, loginRequired: p.sessionInfo(ref.sessionId)?.loginPage ?? false }; + }, + body.tabId ? { tabId: body.tabId } : {}, + ); + return { ...out, session: await svc().view(params.id) }; + }); + + route(app, B.snapshot, async ({ params, query, c }) => + svc().run( + params.id, + human(c), + (p, ref) => p.snapshot(ref, query.offset !== undefined ? { offset: query.offset } : {}), + { write: false, ...(query.tabId ? { tabId: query.tabId } : {}) }, + ), + ); + + route(app, B.screenshot, async ({ params, query, c }): Promise => { + const shot = await svc().run(params.id, human(c), (p, ref) => p.screenshot(ref, { maxBytes: 8 * 1024 * 1024 }), { + write: false, + persist: false, + ...(query.tabId ? { tabId: query.tabId } : {}), + }); + const view = await svc().view(params.id); + return { + mimeType: shot.mimeType, + dataBase64: b64(shot.data), + ...imageSize(shot.data), + url: shot.url, + modelSafe: shot.modelSafe, + capturedAt: shot.capturedAt, + controller: view.controller, + }; + }); + + const ok = async (id: string) => ({ ok: true as const, session: await svc().view(id) }); + + route(app, B.click, async ({ params, body, c }) => { + await svc().run( + params.id, + human(c), + (p, ref) => p.click(ref, body.target), + body.tabId ? { tabId: body.tabId } : {}, + ); + return ok(params.id); + }); + + route(app, B.type, async ({ params, body, c }) => { + await svc().run( + params.id, + human(c), + async (p, ref, x) => { + if (body.key) return p.press(ref, body.key); + if (!body.credential && body.text === undefined) + throw new BrowserServiceError("bad_request", "text / credential / key 至少给一个"); + const extra = { + ...(body.submit !== undefined ? { submit: body.submit } : {}), + ...(body.clear !== undefined ? { clear: body.clear } : {}), + }; + return p.type( + ref, + body.target ?? null, + body.credential + ? { kind: "secret", siteId: x.row.site_id, accountId: x.row.account_id, field: body.credential, ...extra } + : { kind: "text", text: body.text ?? "", ...extra }, + ); + }, + body.tabId ? { tabId: body.tabId } : {}, + ); + return ok(params.id); + }); + + route(app, B.scroll, async ({ params, body, c }) => { + await svc().run( + params.id, + human(c), + (p, ref) => p.scroll(ref, { direction: body.direction, amount: body.amount }), + body.tabId ? { tabId: body.tabId } : {}, + ); + return ok(params.id); + }); + + route(app, B.select, async ({ params, body, c }) => { + await svc().run( + params.id, + human(c), + (p, ref) => p.select(ref, body.target, body.value), + body.tabId ? { tabId: body.tabId } : {}, + ); + return ok(params.id); + }); + + route(app, B.extract, async ({ params, body, c }) => ({ + data: await svc().run(params.id, human(c), (p, ref) => p.extract(ref, body.schema), { + write: false, + ...(body.tabId ? { tabId: body.tabId } : {}), + }), + })); + + // 下载:给 target 视为人工输入(需接管);不给 target 只等待已触发的下载。 + // 产物暂存核验后,给 submit 时转调 POST /acquisition-plans(同一鉴权、同一计划/去重/阈值/授权检查)。 + route(app, B.download, async ({ params, body, c }): Promise => { + const artifact = await svc().run( + params.id, + human(c), + async (p, ref) => { + if (body.target) await p.click(ref, body.target); + return p.waitForDownload(ref, { timeoutMs: body.timeoutMs, maxBytes: 10 * 1024 * 1024 }); + }, + { write: !!body.target, ...(body.tabId ? { tabId: body.tabId } : {}) }, + ); + const v = await svc().verifyDownload(params.id, artifact); + let submission: ManualAcquisitionResult | null = null; + let note: string | null = null; + if (v.duplicate) note = "同一会话已捕获过这个文件,未重复提交"; + else if (!v.verification.valid) note = "不是有效的种子文件,未提交"; + else if (!v.sourceReleaseId) note = "资源不在本地索引(先快照或搜索该站点),未提交"; + else if (body.submit) { + try { + submission = await dispatch.call(c, "POST", "/acquisition-plans", { + ...body.submit, + sourceReleaseId: v.sourceReleaseId, + }); + if (!submission.submitted) note = "提交未生成下载计划,见 rejected"; + } catch (e) { + // 与 POST /acquisition-plans 相同的拒绝(未识别、无授权等):产物已暂存,结果照常返回 + if (!(e instanceof ApiException) || e.code === "internal") throw e; + note = `提交未通过:${e.message}`; + } + } + await svc().recordDownload(params.id, { + sha256: artifact.sha256, + artifactId: artifact.id, + sourceReleaseId: v.sourceReleaseId, + planId: submission?.plan?.id ?? v.duplicate?.planId ?? null, + }); + return { + artifact, + verification: v.verification, + sourceReleaseId: v.sourceReleaseId, + duplicate: !!v.duplicate, + submission, + note, + }; + }); + + route(app, B.acquire, async ({ params, body, c }) => { + const lease = await svc().acquire(params.id, human(c).holder, body.reason, body.ttlMs); + return { lease, session: await svc().view(params.id) }; + }); + + route(app, B.release, async ({ params, body }) => { + const lease = await svc().release(params.id, { + ...(body.leaseId ? { leaseId: body.leaseId } : {}), + reason: body.reason, + to: body.to, + }); + return { lease, session: await svc().view(params.id) }; + }); +} diff --git a/apps/server/src/modules/browser/service.ts b/apps/server/src/modules/browser/service.ts new file mode 100644 index 00000000..16a72cc4 --- /dev/null +++ b/apps/server/src/modules/browser/service.ts @@ -0,0 +1,1003 @@ +import { mkdir, readdir, rm, stat, writeFile } from "node:fs/promises"; +import { join, resolve } from "node:path"; +import { + BrowserError, + type BrowserProvider, + type BrowserSecretResolver, + createCamofoxBrowserProvider, + type DownloadSink, + type ElementTarget, + parseTorrent, + profileKeyFor, + type TabRef, + totpCode, +} from "@umt/adapters"; +import type { + BrowserCapabilities, + BrowserControlLease, + BrowserController, + BrowserDownloadVerification, + BrowserHealth, + BrowserProviderView, + BrowserSession, + BrowserSessionView, + BrowserTab, + DownloadArtifact, + NavigationBudget, +} from "@umt/contracts"; +import type { BrowserControlLeaseRow, BrowserProviderRow, BrowserSessionRow, BrowserTabRow, SiteRow } from "@umt/db"; +import type { SourcesRuntime } from "../sources-identity/runtime.ts"; +import type { ServerContext } from "../types.ts"; + +// C1-E 浏览器服务层:BrowserProvider(camofox)之上的会话/标签/控制权持久化(数据库为准)、 +// 站点域名白名单 + 站点限流、凭证注入、接管状态机、下载产物暂存核验。 +// 适配器只是内存镜像:服务重启后按数据库恢复,旧元素引用一律失效。 + +export type Actor = { kind: "automation" | "human"; holder: string }; + +export interface BrowserModuleOptions { + /** camofox 请求用的 fetch(测试指向 mock) */ + fetch?: typeof globalThis.fetch; + /** 站点限流器所在的运行时(与快照/实时搜索共享) */ + runtime?: (ctx: ServerContext) => SourcesRuntime; + now?: () => Date; + /** 下载产物暂存目录;缺省 DATA_DIR/tmp/browser-downloads */ + stagingDir?: string; + downloadPollMs?: number; + /** 测试替换 provider 实现 */ + providerFactory?: (row: BrowserProviderRow, deps: ProviderDeps) => BrowserProvider; +} + +export interface ProviderDeps { + secrets: BrowserSecretResolver; + sink: DownloadSink; + fetch?: typeof globalThis.fetch; + downloadPollMs?: number; +} + +/** 业务错误(路由层转 ApiException,能力层转 CapabilityError) */ +export class BrowserServiceError extends Error { + constructor( + readonly code: "not_found" | "conflict" | "precondition" | "bad_request" | "upstream" | "rate_limited", + message: string, + ) { + super(message); + this.name = "BrowserServiceError"; + } +} + +/** BrowserError → 业务错误;消息只含方法/路径/状态,不含凭证 */ +export function fromBrowserError(e: unknown): unknown { + if (!(e instanceof BrowserError)) return e; + switch (e.kind) { + case "control_conflict": + case "lock_timeout": + return new BrowserServiceError("conflict", e.message); + case "stale_ref": + return new BrowserServiceError("conflict", e.message); + case "not_found": + return new BrowserServiceError("not_found", e.message); + case "domain_blocked": + case "bad_request": + case "budget_exceeded": + return new BrowserServiceError("bad_request", e.message); + case "auth": + return new BrowserServiceError("precondition", e.message); + case "rate_limited": + return new BrowserServiceError("rate_limited", e.message); + default: + return new BrowserServiceError("upstream", `浏览器服务不可用:${e.message}`); + } +} + +const HEALTH: Record = { + ok: "healthy", + degraded: "degraded", + down: "offline", + disabled: "unknown", +}; + +/** 验证码 / 二次认证 / Cloudflare 挑战:自动化不能完成,转等待处理 */ +const HUMAN_CHALLENGE = + /验证码|captcha|二次验证|两步验证|二步验证|2fa|two-?step|one-time code|cf-challenge|just a moment|checking your browser/i; + +const iso = (d: Date | string | null | undefined) => (d ? new Date(d).toISOString() : null); +type Json = Record; +const obj = (v: unknown): Json => (v && typeof v === "object" && !Array.isArray(v) ? (v as Json) : {}); + +export interface RecoveryState { + currentTabId?: string | null; + needsSnapshot?: boolean; + waitingReason?: string | null; + waitingDecisionId?: string | null; + releasedAt?: string; + downloads?: { sha256: string; artifactId: string; sourceReleaseId: string | null; planId: string | null }[]; +} + +export class BrowserService { + private readonly providers = new Map(); + readonly now: () => Date; + + constructor( + readonly ctx: ServerContext, + readonly opts: BrowserModuleOptions = {}, + ) { + this.now = opts.now ?? (() => new Date()); + } + + // ---------- 浏览器服务 ---------- + + private stagingDir(): string { + if (this.opts.stagingDir) return this.opts.stagingDir; + const dataDir = (this.ctx.config as { DATA_DIR?: string }).DATA_DIR ?? "./data"; + return join(resolve(dataDir), "tmp", "browser-downloads"); + } + + /** 下载产物落 DATA_DIR/tmp/browser-downloads(0600);超过 24 小时的顺手清掉 */ + private sink(): DownloadSink { + return async (bytes, meta) => { + const dir = this.stagingDir(); + await mkdir(dir, { recursive: true, mode: 0o700 }); + await this.cleanupStaging(dir); + const name = `${meta.id}.bin`; + await writeFile(join(dir, name), bytes, { mode: 0o600 }); + return { kind: "blob", ref: `browser-downloads/${name}` }; + }; + } + + private async cleanupStaging(dir: string): Promise { + const cutoff = Date.now() - 24 * 3_600_000; + for (const f of await readdir(dir).catch(() => [] as string[])) { + const p = join(dir, f); + const st = await stat(p).catch(() => null); + if (st && st.mtimeMs < cutoff) await rm(p, { force: true }).catch(() => {}); + } + } + + stagedPath(ref: string): string { + return join(this.stagingDir(), ref.replace(/^browser-downloads\//, "").replace(/[/\\]/g, "_")); + } + + /** 凭证解析:浏览器服务访问凭据与站点账号登录字段;明文不缓存、不入日志 */ + private secretsFor(row: BrowserProviderRow): BrowserSecretResolver { + const { db, secrets } = this.ctx; + return { + resolve: async (req) => { + if (req.kind === "provider_access_key") return row.secret_ref ? secrets.resolve(row.secret_ref) : null; + let q = db + .selectFrom("site_account") + .selectAll() + .where("site_id", "=", req.siteId) + .where("enabled", "=", true) + .where("auth_kind", "=", "password"); + if (req.accountId) q = q.where("id", "=", req.accountId); + const acc = await q.orderBy("created_at").executeTakeFirst(); + if (!acc) return null; + if (req.field === "username") + return acc.login_name ?? (obj(acc.user_info).username as string | undefined) ?? null; + if (req.field === "password") return acc.secret_ref ? secrets.resolve(acc.secret_ref) : null; + return acc.totp_secret_ref ? totpCode(await secrets.resolve(acc.totp_secret_ref), this.now()) : null; + }, + }; + } + + providerFor(row: BrowserProviderRow): BrowserProvider { + const sig = JSON.stringify([row.base_url, row.secret_ref, row.config, row.kind]); + const hit = this.providers.get(row.id); + if (hit && hit.sig === sig) return hit.provider; + const deps: ProviderDeps = { + secrets: this.secretsFor(row), + sink: this.sink(), + ...(this.opts.fetch ? { fetch: this.opts.fetch } : {}), + ...(this.opts.downloadPollMs ? { downloadPollMs: this.opts.downloadPollMs } : {}), + }; + const cfg = obj(row.config); + const provider = + this.opts.providerFactory?.(row, deps) ?? + createCamofoxBrowserProvider({ + providerId: row.id, + baseUrl: row.base_url ?? "", + secrets: deps.secrets, + downloadSink: deps.sink, + verified: obj(cfg.verified) as Record, + ...(typeof cfg.requestTimeoutMs === "number" ? { timeoutMs: cfg.requestTimeoutMs } : {}), + ...(deps.fetch ? { fetch: deps.fetch } : {}), + ...(deps.downloadPollMs ? { downloadPollMs: deps.downloadPollMs } : {}), + now: this.now, + logger: (e) => this.ctx.logger[e.level === "warn" ? "warn" : "debug"]({ component: "browser", ...e }, e.msg), + }); + this.providers.set(row.id, { sig, provider }); + return provider; + } + + async providerRow(id?: string): Promise { + let q = this.ctx.db + .selectFrom("browser_provider") + .selectAll() + .where("enabled", "=", true) + .where("kind", "<>", "none"); + if (id) q = q.where("id", "=", id); + const row = await q.orderBy("created_at").executeTakeFirst(); + if (!row) throw new BrowserServiceError("precondition", "没有启用的浏览器服务(camofox),见 docs/runbook/browser.md"); + return row; + } + + async listProviders(probe: boolean): Promise { + const rows = await this.ctx.db.selectFrom("browser_provider").selectAll().orderBy("created_at").execute(); + const out: BrowserProviderView[] = []; + for (let row of rows) { + let health: BrowserHealth | null = null; + if (probe && row.enabled && row.kind !== "none") { + const p = this.providerFor(row); + health = await p.health(); + let caps: BrowserCapabilities | null = null; + try { + caps = await p.capabilities({ probe: health.status !== "down" }); + } catch { + caps = await p.capabilities(); + } + row = await this.ctx.db + .updateTable("browser_provider") + .set({ + capabilities: JSON.stringify(caps), + health_status: HEALTH[health.status], + last_health_at: new Date(health.checkedAt), + }) + .where("id", "=", row.id) + .returningAll() + .executeTakeFirstOrThrow(); + } + const caps = obj(row.capabilities); + out.push({ + id: row.id, + name: row.name, + kind: row.kind as "camofox" | "none", + baseUrl: row.base_url, + secretRef: row.secret_ref, + enabled: row.enabled, + capabilities: Object.keys(caps).length ? (caps as unknown as BrowserCapabilities) : null, + verifiedAt: iso(row.verified_at), + healthStatus: row.health_status as BrowserProviderView["healthStatus"], + lastHealthAt: iso(row.last_health_at), + health, + }); + } + return out; + } + + // ---------- 站点与预算 ---------- + + async liveSite(siteId: string): Promise { + const s = await this.ctx.db + .selectFrom("site") + .selectAll() + .where("id", "=", siteId) + .where("deleted_at", "is", null) + .executeTakeFirst(); + if (!s) throw new BrowserServiceError("not_found", "站点不存在"); + return s; + } + + /** 导航预算:站点域名(+ config.browserDomains)、超时、重试、字节上限(浏览器服务 config 可覆盖) */ + budgetFor(site: SiteRow, provider: BrowserProviderRow): NavigationBudget { + const cfg = obj(provider.config); + const extra = obj(site.config).browserDomains; + const hosts = [new URL(site.base_url).hostname, ...(Array.isArray(extra) ? extra.map(String) : [])]; + return { + domainAllowlist: hosts, + timeoutMs: typeof cfg.navigationTimeoutMs === "number" ? cfg.navigationTimeoutMs : 30_000, + maxRetries: typeof cfg.maxRetries === "number" ? cfg.maxRetries : 2, + maxBytes: typeof cfg.snapshotMaxBytes === "number" ? cfg.snapshotMaxBytes : 512 * 1024, + }; + } + + /** 站点限流对快照、实时搜索和浏览器统一生效(REQ §4.6):导航前取同一个限流令牌 */ + private async rateLimit(site: SiteRow): Promise { + const rt = this.opts.runtime?.(this.ctx); + if (!rt) return; + try { + rt.siteAdapter(site); + } catch { + return; // 没有随产品发布的适配器定义:无站点限流器 + } + await rt.limiterOf(site.id)?.acquire(); + } + + // ---------- 会话 ---------- + + private async sessionRow(id: string): Promise { + const row = await this.ctx.db.selectFrom("browser_session").selectAll().where("id", "=", id).executeTakeFirst(); + if (!row) throw new BrowserServiceError("not_found", "浏览器会话不存在"); + return row; + } + + private async activeLease(sessionId: string): Promise { + return this.ctx.db + .selectFrom("browser_control_lease") + .selectAll() + .where("session_id", "=", sessionId) + .where("status", "=", "active") + .executeTakeFirst(); + } + + /** 会话加载进适配器(服务重启后按数据库恢复);返回 provider 与最新行 */ + private async loaded(row: BrowserSessionRow): Promise<{ provider: BrowserProvider; row: BrowserSessionRow }> { + const prow = await this.ctx.db + .selectFrom("browser_provider") + .selectAll() + .where("id", "=", row.provider_id) + .executeTakeFirst(); + if (!prow?.enabled) { + await this.closeRow(row, "provider_disabled"); + throw new BrowserServiceError("precondition", "浏览器服务已停用,会话已关闭"); + } + const site = await this.liveSite(row.site_id).catch(() => null); + if (!site?.enabled) { + // 站点关闭即停止其网络访问(REQ §4.6、§4.7.3) + await this.closeRow(row, "site_disabled", prow); + throw new BrowserServiceError("precondition", "站点已停用,浏览器会话已关闭"); + } + const provider = this.providerFor(prow); + if (!provider.sessionInfo(row.id)) { + const tabs = await this.tabRows(row.id); + provider.restoreSession({ + session: toSession(row), + tabs: tabs.map(toTab), + currentTabId: (obj(row.recovery_state) as RecoveryState).currentTabId ?? null, + budget: this.budgetFor(site, prow), + }); + // 重启后旧引用失效:Agent 必须重新读取页面 + row = await this.patchRecovery(row, { needsSnapshot: true }); + } + return { provider, row }; + } + + private tabRows(sessionId: string): Promise { + return this.ctx.db + .selectFrom("browser_tab") + .selectAll() + .where("session_id", "=", sessionId) + .orderBy("created_at") + .execute(); + } + + private async patchRecovery(row: BrowserSessionRow, patch: RecoveryState): Promise { + return this.ctx.db + .updateTable("browser_session") + .set({ recovery_state: JSON.stringify({ ...obj(row.recovery_state), ...patch }), updated_at: this.now() }) + .where("id", "=", row.id) + .returningAll() + .executeTakeFirstOrThrow(); + } + + /** 适配器内存状态 → 数据库(会话地址/活动时间/控制者、标签页) */ + private async persist(sessionId: string, provider: BrowserProvider, patch: RecoveryState = {}): Promise { + const info = provider.sessionInfo(sessionId); + if (!info) return; + const row = await this.sessionRow(sessionId); + await this.ctx.db.transaction().execute(async (trx) => { + await trx + .updateTable("browser_session") + .set({ + current_url: info.session.currentUrl, + last_activity_at: new Date(info.session.lastActivityAt), + recovery_state: JSON.stringify({ + ...obj(row.recovery_state), + currentTabId: info.currentTabId, + ...patch, + }), + updated_at: this.now(), + }) + .where("id", "=", sessionId) + .execute(); + for (const t of info.tabs) + await trx + .insertInto("browser_tab") + .values({ + id: t.id, + session_id: sessionId, + upstream_tab_id: t.upstreamTabId, + url: t.url, + title: t.title, + status: t.status, + last_snapshot_at: t.lastSnapshotAt ? new Date(t.lastSnapshotAt) : null, + closed_at: t.closedAt ? new Date(t.closedAt) : null, + }) + .onConflict((oc) => + oc.columns(["session_id", "upstream_tab_id"]).doUpdateSet({ + url: t.url, + status: t.status, + last_snapshot_at: t.lastSnapshotAt ? new Date(t.lastSnapshotAt) : null, + closed_at: t.closedAt ? new Date(t.closedAt) : null, + updated_at: this.now(), + }), + ) + .execute(); + }); + } + + async view(id: string): Promise { + let row = await this.sessionRow(id); + row = await this.expireLeases(row); + const [tabs, lease] = await Promise.all([this.tabRows(id), this.activeLease(id)]); + const rs = obj(row.recovery_state) as RecoveryState; + return { + ...toSession(row), + tabs: tabs.map(toTab), + currentTabId: rs.currentTabId ?? tabs.find((t) => t.status === "open")?.upstream_tab_id ?? null, + lease: lease ? toLease(lease) : null, + needsSnapshot: rs.needsSnapshot === true, + waitingReason: row.controller === "waiting" || row.controller === "human" ? (rs.waitingReason ?? null) : null, + }; + } + + async list(q: { siteId?: string; includeClosed?: boolean }): Promise { + let sel = this.ctx.db.selectFrom("browser_session").select("id").orderBy("last_activity_at", "desc").limit(100); + if (q.siteId) sel = sel.where("site_id", "=", q.siteId); + if (!q.includeClosed) sel = sel.where("controller", "<>", "closed"); + const rows = await sel.execute(); + return Promise.all(rows.map((r) => this.view(r.id))); + } + + /** 站点(账号)对应的未关闭会话 */ + async liveSessionFor(siteId: string, accountId?: string | null): Promise { + let q = this.ctx.db + .selectFrom("browser_session") + .selectAll() + .where("site_id", "=", siteId) + .where("controller", "<>", "closed"); + if (accountId !== undefined) + q = accountId === null ? q.where("account_id", "is", null) : q.where("account_id", "=", accountId); + return q.orderBy("last_activity_at", "desc").executeTakeFirst(); + } + + /** 站点缺省账号:第一个启用的非 passkey 账号 */ + private async defaultAccount(siteId: string): Promise { + const a = await this.ctx.db + .selectFrom("site_account") + .select("id") + .where("site_id", "=", siteId) + .where("enabled", "=", true) + .where("auth_kind", "<>", "passkey") + .orderBy("created_at") + .executeTakeFirst(); + return a?.id ?? null; + } + + async open(req: { siteId: string; accountId?: string | null; providerId?: string; url?: string }): Promise { + const site = await this.liveSite(req.siteId); + if (!site.enabled) throw new BrowserServiceError("precondition", "站点已停用"); + const prow = await this.providerRow(req.providerId); + const accountId = req.accountId === undefined ? await this.defaultAccount(site.id) : req.accountId; + const profileKey = profileKeyFor(site.id, accountId); + const existing = await this.ctx.db + .selectFrom("browser_session") + .selectAll() + .where("provider_id", "=", prow.id) + .where("profile_key", "=", profileKey) + .where("controller", "<>", "closed") + .executeTakeFirst(); + if (existing) { + await this.loaded(existing); + return existing.id; + } + const budget = this.budgetFor(site, prow); + const provider = this.providerFor(prow); + await this.rateLimit(site); + let res: Awaited>; + try { + res = await provider.openSession({ + providerId: prow.id, + siteId: site.id, + accountId, + profileKey, + initialUrl: req.url ?? site.base_url, + budget, + }); + } catch (e) { + throw fromBrowserError(e); + } + const s = res.session; + await this.ctx.db.transaction().execute(async (trx) => { + await trx + .insertInto("browser_session") + .values({ + id: s.id, + provider_id: prow.id, + site_id: site.id, + account_id: accountId, + profile_key: profileKey, + controller: "automation", + current_url: s.currentUrl, + last_activity_at: new Date(s.lastActivityAt), + upstream_ref: JSON.stringify(s.upstreamRef), + recovery_state: JSON.stringify({ currentTabId: res.tab.upstreamTabId, needsSnapshot: true }), + }) + .execute(); + await trx + .insertInto("browser_tab") + .values({ + id: res.tab.id, + session_id: s.id, + upstream_tab_id: res.tab.upstreamTabId, + url: res.tab.url, + status: "open", + }) + .execute(); + }); + return s.id; + } + + private async closeRow(row: BrowserSessionRow, reason: string, prow?: BrowserProviderRow): Promise { + const provider = prow ? this.providerFor(prow) : this.providers.get(row.provider_id)?.provider; + if (provider?.sessionInfo(row.id)) await provider.closeSession(row.id, reason).catch(() => {}); + const at = this.now(); + await this.ctx.db.transaction().execute(async (trx) => { + await trx + .updateTable("browser_session") + .set({ controller: "closed", closed_at: at, close_reason: reason, updated_at: at }) + .where("id", "=", row.id) + .where("controller", "<>", "closed") + .execute(); + await trx + .updateTable("browser_tab") + .set({ status: "closed", closed_at: at, updated_at: at }) + .where("session_id", "=", row.id) + .where("status", "=", "open") + .execute(); + await trx + .updateTable("browser_control_lease") + .set({ status: "revoked", released_at: at, release_reason: "session_closed", updated_at: at }) + .where("session_id", "=", row.id) + .where("status", "=", "active") + .execute(); + }); + await this.resolveWaiting(row, "session_closed"); + } + + async close(id: string, reason = "closed"): Promise { + const row = await this.sessionRow(id); + if (row.controller === "closed") return; + const prow = await this.ctx.db + .selectFrom("browser_provider") + .selectAll() + .where("id", "=", row.provider_id) + .executeTakeFirst(); + if (prow?.enabled) { + const provider = this.providerFor(prow); + if (!provider.sessionInfo(row.id)) { + const site = await this.liveSite(row.site_id).catch(() => null); + if (site) + provider.restoreSession({ + session: toSession(row), + tabs: (await this.tabRows(row.id)).map(toTab), + currentTabId: null, + budget: this.budgetFor(site, prow), + }); + } + } + await this.closeRow(row, reason, prow); + } + + // ---------- 控制权 ---------- + + /** 过期的人工租约 → expired,会话转等待处理(不自动从未知页面继续) */ + async expireLeases(row: BrowserSessionRow): Promise { + if (row.controller === "closed") return row; + const lease = await this.activeLease(row.id); + if (!lease || new Date(lease.expires_at).getTime() > this.now().getTime()) return row; + const at = this.now(); + await this.ctx.db + .updateTable("browser_control_lease") + .set({ status: "expired", released_at: at, release_reason: "ttl", updated_at: at }) + .where("id", "=", lease.id) + .where("status", "=", "active") + .execute(); + if (lease.holder_kind !== "human") return row; + return this.enterWaiting(row, "lease_expired"); + } + + /** 转等待处理:自动化动作被拒,生成待处理事项,等人在原站视图接管 */ + async enterWaiting(row: BrowserSessionRow, reason: string): Promise { + const provider = this.providers.get(row.provider_id)?.provider; + if (provider?.sessionInfo(row.id)) provider.setController(row.id, "waiting"); + const site = await this.liveSite(row.site_id).catch(() => null); + const rs = obj(row.recovery_state) as RecoveryState; + let decisionId = rs.waitingDecisionId ?? null; + if (!decisionId) { + const d = await this.ctx.db + .insertInto("decision") + .values({ + kind: "other", + status: "awaiting_user", + severity: "warning", + title: `${site?.name ?? "站点"}:原站需要人工处理`, + summary: `${REASON_TEXT[reason] ?? reason}。在站点详情的原站视图接管会话,完成后交还。`, + subject_type: "browser_session", + subject_id: row.id, + dedupe_key: `browser.waiting:${row.id}`, + evidence: JSON.stringify({ siteId: row.site_id, reason, url: row.current_url }), + }) + .onConflict((oc) => + oc.column("dedupe_key").where("status", "in", ["open", "investigating", "awaiting_user"]).doNothing(), + ) + .returning("id") + .executeTakeFirst(); + decisionId = d?.id ?? null; + } + return this.ctx.db + .updateTable("browser_session") + .set({ + controller: "waiting", + recovery_state: JSON.stringify({ ...rs, waitingReason: reason, waitingDecisionId: decisionId }), + updated_at: this.now(), + }) + .where("id", "=", row.id) + .where("controller", "<>", "closed") + .returningAll() + .executeTakeFirstOrThrow(); + } + + private async resolveWaiting(row: BrowserSessionRow, by: string): Promise { + const id = (obj(row.recovery_state) as RecoveryState).waitingDecisionId; + if (!id) return; + await this.ctx.db + .updateTable("decision") + .set({ + status: "resolved", + resolved_at: this.now(), + resolved_by: by, + resolution: JSON.stringify({ via: "browser" }), + }) + .where("id", "=", id) + .where("status", "in", ["open", "investigating", "awaiting_user"]) + .execute(); + } + + /** + * 人工接管(REQ §4.7.3):先把适配器控制者切到 human(新的自动化输入立即被拒), + * 等该 profile 上进行中的操作收尾,再写租约;其他会话不受影响。 + */ + async acquire(id: string, holder: string, reason: string, ttlMs: number): Promise { + let row = await this.expireLeases(await this.sessionRow(id)); + if (row.controller === "closed") throw new BrowserServiceError("conflict", "会话已关闭"); + const current = await this.activeLease(id); + if (current?.holder_kind === "human") { + if (current.holder === holder) return toLease(current); + throw new BrowserServiceError("conflict", `会话已由 ${holderName(current.holder)} 接管`); + } + const { provider } = await this.loaded(row); + provider.setController(id, "human"); + try { + await provider.drain(id); + } catch (e) { + throw fromBrowserError(e); + } + const at = this.now(); + row = await this.sessionRow(id); + const lease = await this.ctx.db.transaction().execute(async (trx) => { + if (current) + await trx + .updateTable("browser_control_lease") + .set({ status: "revoked", released_at: at, release_reason: "human_takeover", updated_at: at }) + .where("id", "=", current.id) + .execute(); + const l = await trx + .insertInto("browser_control_lease") + .values({ + session_id: id, + holder_kind: "human", + holder, + reason, + status: "active", + acquired_at: at, + expires_at: new Date(at.getTime() + ttlMs), + }) + .returningAll() + .executeTakeFirstOrThrow(); + await trx + .updateTable("browser_session") + .set({ + controller: "human", + recovery_state: JSON.stringify({ + ...obj(row.recovery_state), + takeoverAt: at.toISOString(), + previousController: row.controller, + }), + updated_at: at, + }) + .where("id", "=", id) + .execute(); + return l; + }); + return toLease(lease); + } + + /** 交还:回到自动化(或留在等待处理);Agent 恢复前必须重新读取页面,不复用旧引用、不重放旧点击 */ + async release( + id: string, + opts: { leaseId?: string; reason: string; to: "automation" | "waiting" }, + ): Promise { + const row = await this.sessionRow(id); + const lease = await this.activeLease(id); + if (lease?.holder_kind !== "human" || (opts.leaseId && lease.id !== opts.leaseId)) + throw new BrowserServiceError("conflict", "没有可交还的人工接管"); + const at = this.now(); + const { provider } = await this.loaded(row); + provider.setController(id, opts.to); + const rs = obj(row.recovery_state) as RecoveryState; + const released = await this.ctx.db.transaction().execute(async (trx) => { + const l = await trx + .updateTable("browser_control_lease") + .set({ status: "released", released_at: at, release_reason: opts.reason, updated_at: at }) + .where("id", "=", lease.id) + .returningAll() + .executeTakeFirstOrThrow(); + await trx + .updateTable("browser_session") + .set({ + controller: opts.to, + recovery_state: JSON.stringify({ + ...rs, + needsSnapshot: true, + releasedAt: at.toISOString(), + waitingReason: opts.to === "waiting" ? (rs.waitingReason ?? "human_released_waiting") : null, + ...(opts.to === "automation" ? { waitingDecisionId: null } : {}), + }), + updated_at: at, + }) + .where("id", "=", id) + .execute(); + return l; + }); + if (opts.to === "automation") await this.resolveWaiting(row, lease.holder); + await this.persist(id, provider); + return toLease(released); + } + + // ---------- 动作 ---------- + + /** + * 执行一次会话动作:会话存活 → 租约过期检查 → 发起方与控制者一致(人工须持有自己的租约)→ 适配器 → 回写数据库。 + * 上游标签页已被回收(404)时对账标 lost 并报冲突,不盲目重放。 + */ + async run( + id: string, + actor: Actor, + fn: ( + p: BrowserProvider, + ref: TabRef, + ctx: { row: BrowserSessionRow; site: SiteRow; budget: NavigationBudget }, + ) => Promise, + opts: { tabId?: string; write?: boolean; persist?: boolean; patch?: (out: T) => RecoveryState } = {}, + ): Promise { + let row = await this.expireLeases(await this.sessionRow(id)); + if (row.controller === "closed") throw new BrowserServiceError("conflict", "会话已关闭"); + if (opts.write !== false) { + if (actor.kind === "human") { + const lease = await this.activeLease(id); + if (row.controller !== "human" || !lease || lease.holder !== actor.holder) + throw new BrowserServiceError("conflict", "先接管会话再操作"); + } else if (row.controller !== "automation") { + const rs = obj(row.recovery_state) as RecoveryState; + throw new BrowserServiceError( + "conflict", + row.controller === "human" + ? "会话正由人工接管,自动化动作已停止" + : `会话在等待人工处理(${REASON_TEXT[rs.waitingReason ?? ""] ?? rs.waitingReason ?? "未知原因"}),自动化动作已停止`, + ); + } + } + const loaded = await this.loaded(row); + row = loaded.row; + const prow = await this.providerRow(row.provider_id).catch(() => null); + const site = await this.liveSite(row.site_id); + const budget = prow + ? this.budgetFor(site, prow) + : { domainAllowlist: [], timeoutMs: 30_000, maxRetries: 0, maxBytes: 1 }; + const ref: TabRef = { sessionId: id, as: actor.kind, ...(opts.tabId ? { tabId: opts.tabId } : {}) }; + let out: T; + try { + out = await fn(loaded.provider, ref, { row, site, budget }); + } catch (e) { + if (e instanceof BrowserError && e.kind === "not_found" && e.httpStatus === 404) { + const r = await loaded.provider.reconcileTabs(id).catch(() => null); + await this.persist(id, loaded.provider, { needsSnapshot: true }); + if (r?.lost.length) throw new BrowserServiceError("conflict", "标签页已被浏览器服务回收,重新打开页面"); + } + await this.persist(id, loaded.provider).catch(() => {}); + throw fromBrowserError(e); + } + if (opts.persist !== false) await this.persist(id, loaded.provider, opts.patch?.(out) ?? {}); + return out; + } + + /** 导航前按站点限流取令牌 */ + async limited(site: SiteRow): Promise { + await this.rateLimit(site); + } + + /** 快照后判断是否需要人工:只在自动化控制下、页面出现验证码/二次认证/挑战时转等待处理 */ + async afterAutomationSnapshot(id: string, text: string, loginPage: boolean): Promise { + const row = await this.sessionRow(id); + if (row.controller !== "automation") return; + if (HUMAN_CHALLENGE.test(text)) await this.enterWaiting(row, loginPage ? "captcha" : "challenge"); + } + + async requestHuman(id: string, reason: string): Promise { + const row = await this.sessionRow(id); + if (row.controller === "closed") throw new BrowserServiceError("conflict", "会话已关闭"); + if (row.controller !== "automation") return; + await this.enterWaiting(row, reason || "agent_request"); + } + + // ---------- 下载 ---------- + + /** 暂存产物核验:解析种子、按 infohash / 站点资源 ID 匹配本地资源、会话内去重 */ + async verifyDownload( + id: string, + artifact: DownloadArtifact, + ): Promise<{ + verification: BrowserDownloadVerification; + sourceReleaseId: string | null; + duplicate: { planId: string | null } | null; + }> { + const row = await this.sessionRow(id); + const { readFile } = await import("node:fs/promises"); + let verification: BrowserDownloadVerification = { + valid: false, + infohashV1: null, + infohashV2: null, + name: null, + totalSize: null, + private: null, + error: null, + }; + try { + const bytes = new Uint8Array(await readFile(this.stagedPath(artifact.bytes.ref))); + const meta = parseTorrent(bytes); + verification = { + valid: true, + infohashV1: meta.infohashV1, + infohashV2: meta.infohashV2, + name: meta.name, + totalSize: meta.totalSize, + private: meta.private, + error: null, + }; + } catch (e) { + verification.error = e instanceof Error ? e.message.slice(0, 200) : "无法解析"; + } + let sourceReleaseId: string | null = null; + if (verification.valid) { + const hashes = [verification.infohashV1, verification.infohashV2].filter((h): h is string => !!h); + const hit = await this.ctx.db + .selectFrom("source_release") + .select("id") + .where("site_id", "=", row.site_id) + .where((eb) => eb.or([eb("infohash_v1", "in", hashes), eb("infohash_v2", "in", hashes)])) + .executeTakeFirst(); + sourceReleaseId = hit?.id ?? null; + if (!sourceReleaseId && artifact.sourceUrl) { + const rid = safeUrl(artifact.sourceUrl)?.searchParams.get("id"); + if (rid) { + const byId = await this.ctx.db + .selectFrom("source_release") + .select("id") + .where("site_id", "=", row.site_id) + .where("site_release_id", "=", rid) + .executeTakeFirst(); + sourceReleaseId = byId?.id ?? null; + } + } + } + const rs = obj(row.recovery_state) as RecoveryState; + const prev = (rs.downloads ?? []).find((d) => d.sha256 === artifact.sha256); + return { verification, sourceReleaseId, duplicate: prev ? { planId: prev.planId } : null }; + } + + async recordDownload( + id: string, + entry: { sha256: string; artifactId: string; sourceReleaseId: string | null; planId: string | null }, + ): Promise { + const row = await this.sessionRow(id); + const rs = obj(row.recovery_state) as RecoveryState; + const list = (rs.downloads ?? []).filter((d) => d.sha256 !== entry.sha256); + list.push(entry); + await this.patchRecovery(row, { downloads: list.slice(-50) }); + } +} + +const REASON_TEXT: Record = { + captcha: "登录页需要验证码", + challenge: "页面需要人工验证", + login_required: "登录已失效", + lease_expired: "人工接管超时未交还", + agent_request: "Agent 请求人工处理", + human_released_waiting: "人工交还时选择继续等待", +}; + +function safeUrl(s: string): URL | null { + try { + return new URL(s); + } catch { + return null; + } +} + +function holderName(holder: string): string { + return holder.includes(":") ? holder.slice(holder.indexOf(":") + 1) : holder; +} + +export function toSession(r: BrowserSessionRow): BrowserSession { + return { + id: r.id, + providerId: r.provider_id, + siteId: r.site_id, + accountId: r.account_id, + profileKey: r.profile_key, + controller: r.controller as BrowserController, + currentUrl: r.current_url, + lastActivityAt: iso(r.last_activity_at)!, + upstreamRef: obj(r.upstream_ref), + recoveryState: obj(r.recovery_state), + closedAt: iso(r.closed_at), + closeReason: r.close_reason, + createdAt: iso(r.created_at)!, + updatedAt: iso(r.updated_at)!, + }; +} + +export function toTab(r: BrowserTabRow): BrowserTab { + return { + id: r.id, + sessionId: r.session_id, + upstreamTabId: r.upstream_tab_id, + url: r.url, + title: r.title, + status: r.status as BrowserTab["status"], + lastSnapshotAt: iso(r.last_snapshot_at), + closedAt: iso(r.closed_at), + createdAt: iso(r.created_at)!, + updatedAt: iso(r.updated_at)!, + }; +} + +export function toLease(r: BrowserControlLeaseRow): BrowserControlLease { + return { + id: r.id, + sessionId: r.session_id, + holderKind: r.holder_kind as BrowserControlLease["holderKind"], + holder: r.holder, + reason: r.reason, + status: r.status as BrowserControlLease["status"], + acquiredAt: iso(r.acquired_at)!, + expiresAt: iso(r.expires_at)!, + releasedAt: iso(r.released_at), + releaseReason: r.release_reason, + createdAt: iso(r.created_at)!, + updatedAt: iso(r.updated_at)!, + }; +} + +/** 截图像素尺寸:PNG IHDR / JPEG SOF */ +export function imageSize(data: Uint8Array): { width: number | null; height: number | null } { + if (data.length > 24 && data[0] === 0x89 && data[1] === 0x50) { + const v = new DataView(data.buffer, data.byteOffset, data.byteLength); + return { width: v.getUint32(16), height: v.getUint32(20) }; + } + if (data[0] === 0xff && data[1] === 0xd8) { + let i = 2; + while (i + 9 < data.length) { + if (data[i] !== 0xff) break; + const marker = data[i + 1]!; + const len = (data[i + 2]! << 8) | data[i + 3]!; + if (marker >= 0xc0 && marker <= 0xc3) + return { width: (data[i + 7]! << 8) | data[i + 8]!, height: (data[i + 5]! << 8) | data[i + 6]! }; + i += 2 + len; + } + } + return { width: null, height: null }; +} + +export type { ElementTarget }; diff --git a/apps/server/src/modules/index.ts b/apps/server/src/modules/index.ts index 809edcf1..6c34d1bb 100644 --- a/apps/server/src/modules/index.ts +++ b/apps/server/src/modules/index.ts @@ -1,6 +1,7 @@ import { acquisitionOrganizeModule } from "./acquisition-organize/index.ts"; import { agentModule } from "./agent/index.ts"; import { authSettingsModule } from "./auth-settings/index.ts"; +import { browserModule } from "./browser/index.ts"; import { fileManagerModule } from "./file-manager/index.ts"; import { piModule } from "./pi/index.ts"; import { schedulesModule } from "./schedules/index.ts"; @@ -24,5 +25,6 @@ export const SERVER_MODULES: readonly ServerModule[] = [ piModule, schedulesModule, searchModule, + browserModule, systemModule, ]; diff --git a/apps/server/test/e2e/e2-agent-delegate.test.ts b/apps/server/test/e2e/e2-agent-delegate.test.ts index cb336657..731acdf2 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("启动断言:21 个能力全部注册,并在能力目录里可见", async () => { - expect(e.booted.ctx.capabilities.list()).toHaveLength(21); + it("启动断言:25 个能力全部注册,并在能力目录里可见", async () => { + expect(e.booted.ctx.capabilities.list()).toHaveLength(25); 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/fixtures/modules.ts b/apps/server/test/e2e/fixtures/modules.ts index 20f98dfc..dffe69e5 100644 --- a/apps/server/test/e2e/fixtures/modules.ts +++ b/apps/server/test/e2e/fixtures/modules.ts @@ -5,6 +5,7 @@ import { makeTorrent } from "@umt/testkit"; import { createAcquisitionOrganizeModule } from "../../../src/modules/acquisition-organize/index.ts"; import { createAgentModule } from "../../../src/modules/agent/index.ts"; import { authSettingsModule } from "../../../src/modules/auth-settings/index.ts"; +import { createBrowserModule } from "../../../src/modules/browser/index.ts"; import { createSchedulesModule } from "../../../src/modules/schedules/index.ts"; import { createSourcesIdentityModule } from "../../../src/modules/sources-identity/index.ts"; import type { ServerModule } from "../../../src/modules/types.ts"; @@ -65,6 +66,8 @@ export function buildModules(o: E2EModuleOptions): ServerModule[] { ...(o.fakeScripts ? { fakeScripts: o.fakeScripts } : {}), goalPollSec: 1, }), + // 浏览器能力(browser.*)随首批能力一起注册;e2e 不配置浏览器服务 + createBrowserModule(), // 定时任务调度器(驱动上面模块登记的周期任务;须在末尾) createSchedulesModule({ builtinChecks: false, scheduler: { tickMs: 100, lockRetryMs: 200 } }), ]; diff --git a/apps/server/test/modules.test.ts b/apps/server/test/modules.test.ts index d30c5e5d..820bbaff 100644 --- a/apps/server/test/modules.test.ts +++ b/apps/server/test/modules.test.ts @@ -43,7 +43,7 @@ const pingModule: ServerModule = { }; describe("ServerModule 骨架", () => { - it("清单里是四个 W2 模块 + 文件管理 + Pi 上游 + 定时任务 + 全局搜索 + 进程与健康(末尾),名字唯一", () => { + it("清单里是四个 W2 模块 + 文件管理 + Pi 上游 + 定时任务 + 全局搜索 + 浏览器 + 进程与健康(末尾),名字唯一", () => { expect(SERVER_MODULES.map((m) => m.name)).toEqual([ "auth-settings", "sources-identity", @@ -53,6 +53,7 @@ describe("ServerModule 骨架", () => { "pi", "schedules", "search", + "browser", "system", ]); }); diff --git a/apps/server/test/sources-identity/site-capabilities.test.ts b/apps/server/test/sources-identity/site-capabilities.test.ts index 5a9804e4..cceffd2f 100644 --- a/apps/server/test/sources-identity/site-capabilities.test.ts +++ b/apps/server/test/sources-identity/site-capabilities.test.ts @@ -15,8 +15,8 @@ const TiB = 1024 ** 4; const SITE_IDS = ["site.list", "site.health_check", "site.refresh_login", "site.user_data"] as const; describe("站点能力规格", () => { - it("进入首批能力目录(17 + 4 = 21);类别 / 风险 / 重放按约定", () => { - expect(CAPABILITY_IDS).toHaveLength(21); + it("进入首批能力目录(17 + 4 站点 + 4 浏览器 = 25);类别 / 风险 / 重放按约定", () => { + expect(CAPABILITY_IDS).toHaveLength(25); for (const id of SITE_IDS) expect(CAPABILITY_IDS).toContain(id); expect(capabilitySpec("site.list")).toMatchObject({ kind: "read", riskLevel: "low", replay: "safe" }); expect(capabilitySpec("site.user_data")).toMatchObject({ kind: "read", riskLevel: "low", replay: "safe" });