UHD_Media_Tool/apps/server/test/browser/browser-keepalive.test.ts

import { afterAll, beforeAll, describe, expect, it } from "vitest";
import { browserServiceFor } from "../../src/modules/browser/index.ts";
import { BrowserKeepalive } from "../../src/modules/browser/keepalive.ts";
import { type BrowserHarness, createBrowserHarness, type Loose, ORIGIN, pgAvailable } from "./harness.ts";

// AG-4 浏览器会话保活(对 testkit mock camofox):桌面连接 / 人工接管 / 智能体会话正在用的会话定期做一次
// GET /tabs/{id}/downloads(不带 includeData / consume);用完不再保活;保活时标签页已被回收 → BR-3 自愈重连。

describe("BrowserKeepalive(登记与复查)", () => {
  const fake = () => {
    const calls: string[] = [];
    const alive = new Set<string>(["s1", "s2"]);
    return {
      calls,
      alive,
      target: {
        keepAlive: async (id: string) => {
          calls.push(id);
          return alive.has(id);
        },
      },
    };
  };

  it("引用计数:同一会话多个桌面连接,全部断开才停;撤销可重复调用", async () => {
    const f = fake();
    const k = new BrowserKeepalive(f.target, { intervalMs: 60_000 });
    const a = k.hold("s1", "desktop", "desktop");
    const b = k.hold("s1", "desktop", "desktop");
    await k.tick();
    expect(f.calls).toEqual(["s1"]);
    a();
    a();
    expect(k.sessions()).toEqual([{ sessionId: "s1", reasons: ["desktop"] }]);
    b();
    expect(k.sessions()).toEqual([]);
    await k.tick();
    expect(f.calls).toEqual(["s1"]);
  });

  it("复查不通过的登记撤销;会话已关闭(keepAlive 返回 false)整条撤销;ensure 不加引用计数", async () => {
    const f = fake();
    const k = new BrowserKeepalive(f.target, { intervalMs: 60_000 });
    let running = true;
    k.ensure("s1", "agent:r1", "agent", async () => running);
    k.ensure("s1", "agent:r1", "agent", async () => running);
    k.hold("s2", "desktop", "desktop");
    await k.tick();
    expect(f.calls.sort()).toEqual(["s1", "s2"]);
    running = false;
    f.alive.delete("s2");
    await k.tick();
    expect(f.calls).toHaveLength(3); // s1 复查不通过不再保活;s2 保活一次后发现已关闭
    expect(k.sessions()).toEqual([]);
  });

  it("保活失败只记日志,下一轮继续;intervalMs = 0 时不登记", async () => {
    const logs: string[] = [];
    let fail = true;
    const k = new BrowserKeepalive(
      {
        keepAlive: async () => {
          if (fail) throw new Error("浏览器服务不可用");
          return true;
        },
      },
      { intervalMs: 60_000, log: (_l, msg, extra) => logs.push(`${msg} ${String(extra?.err)}`) },
    );
    k.hold("s1", "takeover", "takeover");
    await k.tick();
    expect(logs).toEqual(["浏览器会话保活失败 浏览器服务不可用"]);
    fail = false;
    await k.tick();
    expect(k.sessions()).toHaveLength(1);
    k.stop();

    const off = new BrowserKeepalive(fake().target, { intervalMs: 0 });
    off.hold("s1", "desktop", "desktop");
    expect(off.sessions()).toEqual([]);
  });
});

const up = await pgAvailable();

describe.skipIf(!up)("AG-4 浏览器会话保活(mock camofox)", () => {
  let h: BrowserHarness;
  const svc = () => browserServiceFor(h.ctx)!;
  const open = async (path: string) =>
    (await h.req("/browser/sessions", { body: { siteId: h.siteId, accountId: h.accountId, url: `${ORIGIN}${path}` } }))
      .body.data as Loose;
  const pings = (tabId: string) =>
    h.mock.callsTo(`/tabs/${encodeURIComponent(tabId)}/downloads`).filter((c) => c.method === "GET");
  const view = async (id: string) => (await h.req(`/browser/sessions/${id}`)).body.data as Loose;

  beforeAll(async () => {
    h = await createBrowserHarness("ag4ka");
    await h.grant("browser.navigate");
  }, 120_000);
  afterAll(async () => {
    svc()?.keepalive.stop();
    await h?.close();
  });

  it("人工接管期间保活:只读调用、不记自动化活动;交还后停止", async () => {
    const s = await open("/torrents.php");
    const before = (await view(s.id)).lastAutomationAt;
    const acq = await h.req(`/browser/sessions/${s.id}/control/acquire`, { body: { reason: "核对" } });
    expect(acq.status).toBe(200);
    expect(svc().keepalive.sessions()).toEqual([{ sessionId: s.id, reasons: ["takeover"] }]);

    await svc().keepalive.tick();
    const p = pings(s.currentTabId);
    expect(p).toHaveLength(1);
    // 不带 includeData / consume:不读下载内容、不清下载记录
    expect(p[0]!.query).toEqual({ userId: s.profileKey });
    const v = await view(s.id);
    expect(v.controller).toBe("human");
    expect(v.lastAutomationAt).toBe(before);

    await h.req(`/browser/sessions/${s.id}/control/release`, { body: { to: "automation", reason: "done" } });
    expect(svc().keepalive.sessions()).toEqual([]);
    await svc().keepalive.tick();
    expect(pings(s.currentTabId)).toHaveLength(1);
    await h.req(`/browser/sessions/${s.id}`, { method: "DELETE" });
  });

  it("桌面连接期间标签页被空闲回收 → 保活时按 BR-3 自愈重连回到原地址;断开后停止", async () => {
    const s = await open("/details.php?id=42");
    const release = svc().keepalive.hold(s.id, "desktop", "desktop");
    // 上游空闲回收了标签页
    h.mock.tabs.delete(s.currentTabId);
    await svc().keepalive.tick();
    const v = await view(s.id);
    expect(v.currentTabId).not.toBe(s.currentTabId);
    // 回到会话记录的地址(测试账号没有登录态,mock 站点跳登录页并带 returnto)
    expect(decodeURIComponent(v.currentUrl)).toContain("/details.php?id=42");
    expect(v.reconnect).toMatchObject({ status: "ok", trigger: "tab_gone" });
    // 新标签页上重试了保活
    expect(pings(v.currentTabId)).toHaveLength(1);

    release();
    expect(svc().keepalive.sessions()).toEqual([]);
    await h.req(`/browser/sessions/${s.id}`, { method: "DELETE" });
  });

  it("智能体会话在用:运行中保活;会话结束后停止;浏览器会话关闭也停止", async () => {
    const run = await h.db
      .insertInto("agent_run")
      .values({
        trigger: "user",
        goal: "看看种子列表",
        status: "running",
        mode: "execute",
        scope: JSON.stringify({}),
        budget: JSON.stringify({}),
        created_by: "user:admin",
      })
      .returning("id")
      .executeTakeFirstOrThrow();
    const nav = (await h.registry.invoke(
      "browser.navigate",
      { siteId: h.siteId, url: `${ORIGIN}/torrents.php` },
      { ...h.agentCtx(), agentRunId: run.id },
    )) as Loose;
    const sid = nav.sessionId as string;
    expect(svc().keepalive.sessions()).toEqual([{ sessionId: sid, reasons: ["agent"] }]);
    const tab = (await view(sid)).currentTabId as string;
    await svc().keepalive.tick();
    expect(pings(tab)).toHaveLength(1);

    await h.db.updateTable("agent_run").set({ status: "succeeded" }).where("id", "=", run.id).execute();
    await svc().keepalive.tick();
    expect(pings(tab)).toHaveLength(1);
    expect(svc().keepalive.sessions()).toEqual([]);

    // 再用一次(新会话在跑)→ 登记;关闭浏览器会话 → 撤销
    await h.db.updateTable("agent_run").set({ status: "running" }).where("id", "=", run.id).execute();
    await h.registry.invoke("browser.snapshot", { sessionId: sid }, { ...h.agentCtx(), agentRunId: run.id });
    expect(svc().keepalive.sessions()).toHaveLength(1);
    await h.req(`/browser/sessions/${sid}`, { method: "DELETE" });
    expect(svc().keepalive.sessions()).toEqual([]);
  });
});