UHD_Media_Tool/apps/server/src/modules/browser/service.ts

import { mkdir, readdir, rm, stat, writeFile } from "node:fs/promises";
import { join, resolve } from "node:path";
import {
  BrowserError,
  type BrowserProvider,
  type BrowserSecretResolver,
  CAMOFOX_VERIFIED_RELEASES,
  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<BrowserHealth["status"], "healthy" | "degraded" | "offline" | "unknown"> = {
  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<string, unknown>;
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 }[];
}

/** 环境变量(BROWSER_PROVIDER_URL)登记的浏览器服务 */
export const ENV_PROVIDER_NAME = "camofox-env";
const ENV_MANAGED = "env";

export class BrowserService {
  private readonly providers = new Map<string, { sig: string; provider: BrowserProvider }>();
  private envSync: Promise<void> | undefined;
  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<void> {
    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") {
          if (row.secret_ref) return secrets.resolve(row.secret_ref);
          // 环境变量登记的服务:凭据只在进程内(BROWSER_ACCESS_KEY),不入库
          return obj(row.config).managedBy === ENV_MANAGED ? (this.ctx.config?.BROWSER_ACCESS_KEY ?? null) : 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<string, boolean>,
        ...(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;
  }

  /**
   * BROWSER_PROVIDER_URL / BROWSER_ACCESS_KEY / BROWSER_ENABLED → 名为 camofox-env 的浏览器服务(每个进程同步一次)。
   * 地址与启用由环境变量决定;config 里的其它键(verified、超时等)保留。没设地址时已有的 camofox-env 置为停用。
   */
  syncEnvProvider(): Promise<void> {
    this.envSync ??= this.doSyncEnvProvider().then(
      (probe) => {
        // 首次登记后在后台探测一次,写回能力与上游版本(不阻塞当前请求;camofox 未就绪时下次"测试连通"再写)
        if (probe) void this.listProviders(true).catch(() => {});
      },
      (e) => {
        this.envSync = undefined;
        this.ctx.logger.warn({ component: "browser", err: String(e) }, "登记环境变量浏览器服务失败");
      },
    );
    return this.envSync;
  }

  /** 返回是否需要后台探测(已启用且还没有探测过能力) */
  private async doSyncEnvProvider(): Promise<boolean> {
    const c = this.ctx.config;
    const url = c?.BROWSER_PROVIDER_URL;
    const enabled = !!url && c.BROWSER_ENABLED !== false;
    const db = this.ctx.db;
    const row = await db
      .selectFrom("browser_provider")
      .selectAll()
      .where("name", "=", ENV_PROVIDER_NAME)
      .executeTakeFirst();
    if (!url) {
      if (row?.enabled)
        await db.updateTable("browser_provider").set({ enabled: false }).where("id", "=", row.id).execute();
      return false;
    }
    const config = JSON.stringify({ ...obj(row?.config), managedBy: ENV_MANAGED });
    if (row)
      await db
        .updateTable("browser_provider")
        .set({ base_url: url, enabled, kind: "camofox", secret_ref: null, config })
        .where("id", "=", row.id)
        .execute();
    else
      await db
        .insertInto("browser_provider")
        .values({ name: ENV_PROVIDER_NAME, kind: "camofox", base_url: url, enabled, config })
        .onConflict((oc) => oc.column("name").doNothing())
        .execute();
    return enabled && !Object.keys(obj(row?.capabilities)).length;
  }

  /** 设置页登记浏览器服务:访问凭据加密存 secret;名称 camofox-env 保留给环境变量 */
  async createProvider(body: {
    name: string;
    baseUrl: string;
    accessKey?: string | undefined;
    enabled?: boolean | undefined;
  }): Promise<BrowserProviderView> {
    if (body.name === ENV_PROVIDER_NAME)
      throw new BrowserServiceError("bad_request", `名称 ${ENV_PROVIDER_NAME} 保留给环境变量`);
    const exists = await this.ctx.db
      .selectFrom("browser_provider")
      .select("id")
      .where("name", "=", body.name)
      .executeTakeFirst();
    if (exists) throw new BrowserServiceError("conflict", `浏览器服务「${body.name}」已存在`);
    const secretRef = body.accessKey
      ? await this.ctx.secrets.create({ purpose: "other", plaintext: body.accessKey, label: `browser:${body.name}` })
      : null;
    const row = await this.ctx.db
      .insertInto("browser_provider")
      .values({
        name: body.name,
        kind: "camofox",
        base_url: body.baseUrl,
        secret_ref: secretRef,
        enabled: body.enabled ?? true,
      })
      .returning("id")
      .executeTakeFirstOrThrow();
    return this.providerView(row.id);
  }

  /** 改地址 / 凭据 / 启用;accessKey = null 清除凭据。camofox-env 的地址与凭据归环境变量管,只能改启用 */
  async updateProvider(
    id: string,
    body: { baseUrl?: string | undefined; accessKey?: string | null | undefined; enabled?: boolean | undefined },
  ): Promise<BrowserProviderView> {
    const row = await this.ctx.db.selectFrom("browser_provider").selectAll().where("id", "=", id).executeTakeFirst();
    if (!row) throw new BrowserServiceError("not_found", "浏览器服务不存在");
    const changesAddr = body.baseUrl !== undefined && body.baseUrl !== row.base_url;
    if (obj(row.config).managedBy === ENV_MANAGED && (changesAddr || body.accessKey !== undefined))
      throw new BrowserServiceError(
        "conflict",
        "该浏览器服务由环境变量 BROWSER_PROVIDER_URL / BROWSER_ACCESS_KEY 管理，改部署配置后重启",
      );
    let secretRef = row.secret_ref;
    if (body.accessKey === null) secretRef = null;
    else if (body.accessKey) {
      if (secretRef) await this.ctx.secrets.replace(secretRef, body.accessKey);
      else
        secretRef = await this.ctx.secrets.create({
          purpose: "other",
          plaintext: body.accessKey,
          label: `browser:${row.name}`,
        });
    }
    await this.ctx.db
      .updateTable("browser_provider")
      .set({
        ...(body.baseUrl !== undefined ? { base_url: body.baseUrl } : {}),
        ...(body.enabled !== undefined ? { enabled: body.enabled } : {}),
        secret_ref: secretRef,
      })
      .where("id", "=", id)
      .execute();
    if (row.secret_ref && secretRef === null) await this.ctx.secrets.remove(row.secret_ref).catch(() => {});
    // 凭据换了但 secret_ref 不变:签名不变,显式丢弃缓存的适配器
    this.providers.delete(id);
    return this.providerView(id);
  }

  /** 删除设置页登记的服务:有未关闭会话时拒绝(先关会话);环境变量登记的只能停用 */
  async deleteProvider(id: string): Promise<{ id: string }> {
    const row = await this.ctx.db.selectFrom("browser_provider").selectAll().where("id", "=", id).executeTakeFirst();
    if (!row) throw new BrowserServiceError("not_found", "浏览器服务不存在");
    if (obj(row.config).managedBy === ENV_MANAGED)
      throw new BrowserServiceError("conflict", "该浏览器服务由环境变量登记，只能停用（BROWSER_ENABLED=false）");
    const open = await this.ctx.db
      .selectFrom("browser_session")
      .select("id")
      .where("provider_id", "=", id)
      .where("controller", "<>", "closed")
      .executeTakeFirst();
    if (open) throw new BrowserServiceError("conflict", "该浏览器服务还有未关闭的会话，先关闭会话");
    await this.ctx.db.deleteFrom("browser_provider").where("id", "=", id).execute();
    if (row.secret_ref) await this.ctx.secrets.remove(row.secret_ref).catch(() => {});
    this.providers.delete(id);
    return { id };
  }

  private async providerView(id: string): Promise<BrowserProviderView> {
    const v = (await this.listProviders(false, id))[0];
    if (!v) throw new BrowserServiceError("not_found", "浏览器服务不存在");
    return v;
  }

  async providerRow(id?: string): Promise<BrowserProviderRow> {
    await this.syncEnvProvider();
    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, onlyId?: string): Promise<BrowserProviderView[]> {
    await this.syncEnvProvider();
    let rows = await this.ctx.db.selectFrom("browser_provider").selectAll().orderBy("created_at").execute();
    if (onlyId) rows = rows.filter((r) => r.id === onlyId);
    const out: BrowserProviderView[] = [];
    for (let row of rows) {
      let health: BrowserHealth | null = null;
      if (probe && (row.enabled || onlyId) && 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 (e) {
          caps = await p.capabilities();
          // /health 免认证,只看它会把"凭据错了"显示成健康:探测被 401/403 拒绝时标为离线并注明
          if (e instanceof BrowserError && e.kind === "auth")
            health = { ...health, status: "down", detail: { ...health.detail, error: "auth" } };
        }
        row = await this.ctx.db
          .updateTable("browser_provider")
          .set({
            capabilities: JSON.stringify(caps),
            // 上游版本有真机核验记录时,verified_at 取核验日期(不是探测时间)
            ...(caps.upstreamVersion && CAMOFOX_VERIFIED_RELEASES[caps.upstreamVersion]
              ? { verified_at: new Date(`${CAMOFOX_VERIFIED_RELEASES[caps.upstreamVersion]!.verifiedAt}T00:00:00Z`) }
              : {}),
            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,
        managedBy: obj(row.config).managedBy === ENV_MANAGED ? "env" : null,
        hasAccessKey:
          !!row.secret_ref || (obj(row.config).managedBy === ENV_MANAGED && !!this.ctx.config?.BROWSER_ACCESS_KEY),
        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<SiteRow> {
    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<void> {
    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<BrowserSessionRow> {
    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<BrowserControlLeaseRow | undefined> {
    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<BrowserTabRow[]> {
    return this.ctx.db
      .selectFrom("browser_tab")
      .selectAll()
      .where("session_id", "=", sessionId)
      .orderBy("created_at")
      .execute();
  }

  private async patchRecovery(row: BrowserSessionRow, patch: RecoveryState): Promise<BrowserSessionRow> {
    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<void> {
    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<BrowserSessionView> {
    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<BrowserSessionView[]> {
    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<BrowserSessionRow | undefined> {
    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<string | null> {
    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<string> {
    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<ReturnType<BrowserProvider["openSession"]>>;
    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<void> {
    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<void> {
    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<BrowserSessionRow> {
    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<BrowserSessionRow> {
    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<void> {
    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<BrowserControlLease> {
    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<BrowserControlLease> {
    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<T>(
    id: string,
    actor: Actor,
    fn: (
      p: BrowserProvider,
      ref: TabRef,
      ctx: { row: BrowserSessionRow; site: SiteRow; budget: NavigationBudget },
    ) => Promise<T>,
    opts: { tabId?: string; write?: boolean; persist?: boolean; patch?: (out: T) => RecoveryState } = {},
  ): Promise<T> {
    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 || e.httpStatus === 410)) {
        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<void> {
    await this.rateLimit(site);
  }

  /** 快照后判断是否需要人工:只在自动化控制下、页面出现验证码/二次认证/挑战时转等待处理 */
  async afterAutomationSnapshot(id: string, text: string, loginPage: boolean): Promise<void> {
    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<void> {
    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<void> {
    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<string, string> = {
  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 };