From afdc3987307626ba5baeb3286b5acbfe8b835eec Mon Sep 17 00:00:00 2001 From: BB-fat <1056871944@qq.com> Date: Fri, 14 Aug 2026 03:45:52 +0000 Subject: [PATCH] fix(dsh-plugin-browserskill): strict session ownership on a shared daemon MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review hardening for daemon-sharing deployments (other agents, terminals, or dsh instances on the same bsk daemon): - Ownership split: only sessions created by browser_session_start are 'owned'. Explicitly referenced foreign sessions are still tracked for the current-session pointer, but browser_session_stop refuses them and unload cleanup stops exactly the owned set — no path can stop a session this plugin did not create. - Start race: the capacity check is now a synchronous reserveStart/completeStart/abandonStart protocol, so two concurrent starts can never both pass the cap (check-and-reserve is atomic on the event loop); a rejected or failed start never leaks a session. - Stale handles: dispose already swallowed per-stop errors; the owned set is computed at dispose time so an externally stopped session is a no-op, not a failure. Tests (+7): reservation race (two concurrent starts, cap 1 -> exactly one spawn), foreign-stop refusal (no stop command reaches the daemon), reference adoption never becoming owned, resolveForStop pointer stability, and an apply-level dispose test proving only owned sessions are stopped while stale stops are tolerated. e2e: a manually started foreign session survived the plugin's full lifecycle while both plugin sessions were cleaned up. --- packages/dsh-plugin-browserskill/src/index.ts | 27 +++-- .../dsh-plugin-browserskill/src/sessions.ts | 103 ++++++++++++++---- packages/dsh-plugin-browserskill/src/tools.ts | 23 ++-- .../tests/dispose.test.ts | 89 +++++++++++++++ .../tests/sessions.test.ts | 79 +++++++++++--- .../tests/tools.test.ts | 35 ++++++ 6 files changed, 304 insertions(+), 52 deletions(-) create mode 100644 packages/dsh-plugin-browserskill/tests/dispose.test.ts diff --git a/packages/dsh-plugin-browserskill/src/index.ts b/packages/dsh-plugin-browserskill/src/index.ts index 7e047ad..ab48da1 100644 --- a/packages/dsh-plugin-browserskill/src/index.ts +++ b/packages/dsh-plugin-browserskill/src/index.ts @@ -12,7 +12,7 @@ import type { Context } from "@deepseek-ai/cordis"; import Schema from "@deepseek-ai/schemastery"; -import { createBskRunner } from "./runner"; +import { type BskRunner, createBskRunner } from "./runner"; import { SessionRegistry } from "./sessions"; import { type PluginConfig, registerTools } from "./tools"; @@ -34,13 +34,22 @@ export const Config = Schema.object({ export type Config = PluginConfig; -export function apply(ctx: Context, config: Partial = {}): void { +/** Test seams: swap the process runner (unit tests never spawn a real bsk). */ +export interface ApplyOptions { + runnerFactory?: (bskPath: string) => BskRunner; +} + +export function apply( + ctx: Context, + config: Partial = {}, + options: ApplyOptions = {}, +): void { const resolved: PluginConfig = { bskPath: config.bskPath ?? "bsk", defaultTimeoutMs: config.defaultTimeoutMs ?? 120_000, maxSessions: config.maxSessions ?? 5, }; - const runner = createBskRunner(resolved.bskPath); + const runner = options.runnerFactory?.(resolved.bskPath) ?? createBskRunner(resolved.bskPath); const registry = new SessionRegistry(resolved.maxSessions); registerTools({ ctx, runner, registry, config: resolved }); @@ -58,15 +67,17 @@ export function apply(ctx: Context, config: Partial = {}): void { ); // Unload cleanup: kill in-flight children, then stop every session this - // plugin started (best effort; sessions started outside the plugin are - // left running). + // plugin OWNS (created via browser_session_start). Referenced or unknown + // sessions belonging to other programs on the shared daemon are never + // touched; per-stop failures (already stopped externally, daemon restart) + // are swallowed so one stale handle cannot abort the rest. ctx.effect(() => { return () => { runner.killAll(); const stops = registry - .list() - .map((session) => - runner.run(["session", "stop", session.sessionId], { timeoutMs: 15_000 }).catch(() => {}), + .ownedIds() + .map((sessionId) => + runner.run(["session", "stop", sessionId], { timeoutMs: 15_000 }).catch(() => {}), ); return Promise.all(stops).then(() => {}); }; diff --git a/packages/dsh-plugin-browserskill/src/sessions.ts b/packages/dsh-plugin-browserskill/src/sessions.ts index b470f30..4650d9b 100644 --- a/packages/dsh-plugin-browserskill/src/sessions.ts +++ b/packages/dsh-plugin-browserskill/src/sessions.ts @@ -4,44 +4,72 @@ * The pointer moves to whatever session was most recently started, stopped, * or operated on, so a model can run several browsers side by side without * repeating the id on every call. + * + * Ownership discipline (the bsk daemon may be shared with other agents, + * terminals, or dsh instances): a session is **owned** only when it was + * created by this plugin's browser_session_start; an explicit `session` + * argument naming a foreign session is tracked as a **reference** so the + * current-session pointer keeps working, but referenced sessions are never + * stopped — not by browser_session_stop and not by unload cleanup. */ export interface TrackedSession { sessionId: string; browserInstanceId?: string; startedAtMs: number; + /** True only for sessions this plugin created (and therefore may stop). */ + owned: boolean; } export class SessionRegistry { private readonly sessions = new Map(); private currentId: string | undefined; + /** + * Slots reserved by in-flight starts. reserveStart/completeStart/abandonStart + * run synchronously around the async spawn, so concurrent starts can never + * both pass the capacity check (check-and-reserve is atomic on the event loop). + */ + private pendingStarts = 0; constructor(private readonly maxSessions: number) {} /** - * Register a freshly started session and make it current. - * @throws when the configured concurrency cap is already reached. + * Reserve a start slot synchronously, BEFORE spawning. + * @throws when the configured concurrency cap (tracked + in-flight) is reached. */ - add(session: TrackedSession): void { - if (!this.sessions.has(session.sessionId)) this.assertCapacity(); - this.sessions.set(session.sessionId, session); - this.currentId = session.sessionId; - } - - /** - * Throw when starting another session would exceed the concurrency cap. - * Call BEFORE spawning a new session so a rejected start never leaks one. - */ - assertCapacity(): void { - if (this.sessions.size >= this.maxSessions) { + reserveStart(): void { + if (this.sessions.size + this.pendingStarts >= this.maxSessions) { throw new Error( `session limit reached (${this.maxSessions} concurrent sessions); ` + "stop one with browser_session_stop before starting another", ); } + this.pendingStarts += 1; } - /** Forget a stopped session; falls back to the most recent remaining one. */ + /** Give back a reservation after a start that never produced a session. */ + abandonStart(): void { + this.pendingStarts = Math.max(0, this.pendingStarts - 1); + } + + /** + * Register a freshly started (owned) session, consuming its reservation, + * and make it current. + */ + completeStart(session: Omit): void { + this.pendingStarts = Math.max(0, this.pendingStarts - 1); + // Backstop only: with the reservation protocol above this never fires. + if (!this.sessions.has(session.sessionId) && this.sessions.size >= this.maxSessions) { + throw new Error( + `session limit reached (${this.maxSessions} concurrent sessions); ` + + "stop one with browser_session_stop before starting another", + ); + } + this.sessions.set(session.sessionId, { ...session, owned: true }); + this.currentId = session.sessionId; + } + + /** Forget a session; falls back to the most recent remaining one. */ remove(sessionId: string): void { this.sessions.delete(sessionId); if (this.currentId === sessionId) { @@ -50,7 +78,7 @@ export class SessionRegistry { } } - /** Mark a session as most recently used. Unknown ids are adopted. */ + /** Mark a session as most recently used. Unknown ids are adopted as references (never owned). */ touch(sessionId: string): void { const existing = this.sessions.get(sessionId); if (existing !== undefined) { @@ -58,7 +86,7 @@ export class SessionRegistry { this.sessions.delete(sessionId); this.sessions.set(sessionId, existing); } else { - this.sessions.set(sessionId, { sessionId, startedAtMs: Date.now() }); + this.sessions.set(sessionId, { sessionId, startedAtMs: Date.now(), owned: false }); } this.currentId = sessionId; } @@ -73,13 +101,26 @@ export class SessionRegistry { return [...this.sessions.values()]; } + /** Ids of owned sessions — the exact set unload cleanup is allowed to stop. */ + ownedIds(): string[] { + return this.list() + .filter((session) => session.owned) + .map((session) => session.sessionId); + } + + /** Whether the session was created by this plugin and may be stopped by it. */ + isOwned(sessionId: string): boolean { + return this.sessions.get(sessionId)?.owned === true; + } + size(): number { return this.sessions.size; } /** - * Resolve the session a tool call acts on: an explicit `session` argument - * wins (and becomes current); otherwise fall back to the current session. + * Resolve the session an operation tool call acts on: an explicit `session` + * argument wins (and becomes current); otherwise fall back to the current + * session. Foreign ids are adopted as references. * @throws when neither is available. */ resolve(explicit: string | undefined, toolName: string): string { @@ -97,4 +138,28 @@ export class SessionRegistry { this.touch(current); return current; } + + /** + * Resolve the session a STOP call acts on, without adopting foreign ids: + * only owned sessions may be stopped, and the current pointer is not moved + * by a rejected stop. + * @throws when the target is unknown, foreign, or no session exists. + */ + resolveForStop(explicit: string | undefined): string { + const candidate = + explicit !== undefined && explicit.trim().length > 0 ? explicit : this.current(); + if (candidate === undefined) { + throw new Error( + "browser_session_stop needs a session but none is active — start one with browser_session_start " + + "or pass an explicit `session` id", + ); + } + if (!this.isOwned(candidate)) { + throw new Error( + `browser_session_stop refuses to stop session "${candidate}": it was not created by this plugin. ` + + "Only sessions returned by browser_session_start can be stopped (use `bsk session stop` yourself for others).", + ); + } + return candidate; + } } diff --git a/packages/dsh-plugin-browserskill/src/tools.ts b/packages/dsh-plugin-browserskill/src/tools.ts index 416517b..cc16cbe 100644 --- a/packages/dsh-plugin-browserskill/src/tools.ts +++ b/packages/dsh-plugin-browserskill/src/tools.ts @@ -162,19 +162,23 @@ export function registerTools(deps: ToolDeps): void { if ((args.width === undefined) !== (args.height === undefined)) { throw new Error("width and height must be given together"); } - // Check the cap BEFORE spawning so a rejected start never leaks a session. - registry.assertCapacity(); + // Reserve the slot BEFORE spawning: check-and-reserve is synchronous, + // so concurrent starts can never both pass the cap and leak a session. + registry.reserveStart(); const startArgs = ["session", "start"]; if (args.width !== undefined && args.height !== undefined) { startArgs.push("--width", String(args.width), "--height", String(args.height)); } if (args.noFocus === true) startArgs.push("--no-focus"); if (args.browser !== undefined) startArgs.push("--browser", args.browser); - const reply = (await runBsk(deps, exec, startArgs, "session start")) as { - session_id: string; - browser_instance_id: string; - }; - registry.add({ + let reply: { session_id: string; browser_instance_id: string }; + try { + reply = (await runBsk(deps, exec, startArgs, "session start")) as typeof reply; + } catch (error) { + registry.abandonStart(); + throw error; + } + registry.completeStart({ sessionId: reply.session_id, browserInstanceId: reply.browser_instance_id, startedAtMs: Date.now(), @@ -230,7 +234,8 @@ export function registerTools(deps: ToolDeps): void { name: "browser_session_stop", description: "Stop a browser session and close its Agent Window. Stops the given session, or the " + - "current session when `session` is omitted.", + "current session when `session` is omitted. Only sessions created by browser_session_start " + + "can be stopped — sessions owned by other programs sharing the bsk daemon are refused.", parameters: { session: SESSION_PARAM }, output: { schema: { @@ -243,7 +248,7 @@ export function registerTools(deps: ToolDeps): void { ], }, async execute(args, exec) { - const sessionId = registry.resolve(args.session, "browser_session_stop"); + const sessionId = registry.resolveForStop(args.session); await runBsk(deps, exec, ["session", "stop", sessionId], "session stop"); registry.remove(sessionId); return { stopped: sessionId }; diff --git a/packages/dsh-plugin-browserskill/tests/dispose.test.ts b/packages/dsh-plugin-browserskill/tests/dispose.test.ts new file mode 100644 index 0000000..0962fbb --- /dev/null +++ b/packages/dsh-plugin-browserskill/tests/dispose.test.ts @@ -0,0 +1,89 @@ +// Unload-cleanup ownership discipline: dispose must stop exactly the sessions +// this plugin created — never referenced/foreign sessions on the shared +// daemon — and a stale handle (already stopped elsewhere, daemon restarted) +// must not abort the remaining stops. + +import type { ToolDefinition, ToolRunContext } from "@deepseek-ai/dsh-tools"; +import { describe, expect, it } from "vitest"; +import { apply } from "../src/index"; +import type { BskRunOptions, BskRunResult } from "../src/runner"; + +interface FakeCall { + args: string[]; +} + +function ok(payload: unknown): BskRunResult { + return { code: 0, stdout: JSON.stringify(payload), stderr: "", timedOut: false, aborted: false }; +} + +function fail(message: string): BskRunResult { + return { + code: 4, + stdout: JSON.stringify({ code: "session_not_found", message }), + stderr: "", + timedOut: false, + aborted: false, + }; +} + +function makeExec(): ToolRunContext { + return { + callId: "t", + name: "t", + signal: new AbortController().signal, + } as unknown as ToolRunContext; +} + +describe("dispose cleanup ownership", () => { + it("stops only plugin-owned sessions and tolerates stale handles", async () => { + const calls: FakeCall[] = []; + const startReplies = [ + { session_id: "own1", browser_instance_id: "b1" }, + { session_id: "own2", browser_instance_id: "b1" }, + ]; + const runner = { + async run(args: string[], _options: BskRunOptions = {}): Promise { + calls.push({ args }); + const joined = args.join(" "); + if (joined.startsWith("status")) return ok({}); + if (joined.startsWith("session start")) return ok(startReplies.shift()); + if (joined.startsWith("snapshot")) { + return ok({ text: "x", ref_count: 1, tab_id: 7, truncated: false }); + } + if (joined === "session stop own1") return ok({}); + if (joined === "session stop own2") { + // Stale handle: already stopped by someone else / daemon restarted. + return fail("session not found"); + } + return fail(`unexpected command: ${joined}`); + }, + killAll() {}, + }; + const tools = new Map(); + const disposers: Array<() => Promise> = []; + const ctx = { + tools: { register: (def: ToolDefinition) => tools.set(def.name, def) }, + get: () => undefined, + effect: (fn: () => () => Promise) => { + disposers.push(fn()); + }, + }; + apply(ctx as never, { maxSessions: 5 }, { runnerFactory: () => runner }); + + const start = tools.get("browser_session_start"); + const snapshot = tools.get("browser_snapshot"); + if (start === undefined || snapshot === undefined) throw new Error("tools not registered"); + await start.execute({}, makeExec()); + await start.execute({}, makeExec()); + // A foreign session gets referenced (current-pointer bookkeeping) but never owned. + await snapshot.execute({ session: "ext9" }, makeExec()); + + await Promise.all(disposers.map((dispose) => dispose())); + + const stops = calls + .map((call) => call.args.join(" ")) + .filter((joined) => joined.startsWith("session stop")); + expect(stops.sort()).toEqual(["session stop own1", "session stop own2"]); + expect(stops.some((joined) => joined.includes("ext9"))).toBe(false); + }); +}); diff --git a/packages/dsh-plugin-browserskill/tests/sessions.test.ts b/packages/dsh-plugin-browserskill/tests/sessions.test.ts index c73ca63..63d8a50 100644 --- a/packages/dsh-plugin-browserskill/tests/sessions.test.ts +++ b/packages/dsh-plugin-browserskill/tests/sessions.test.ts @@ -1,13 +1,19 @@ import { describe, expect, it } from "vitest"; import { SessionRegistry } from "../src/sessions"; +/** Drive one full successful start through the reservation protocol. */ +function start(registry: SessionRegistry, sessionId: string): void { + registry.reserveStart(); + registry.completeStart({ sessionId, startedAtMs: Date.now() }); +} + describe("SessionRegistry", () => { - it("tracks the current session across add and remove", () => { + it("tracks the current session across start and remove", () => { const registry = new SessionRegistry(5); expect(registry.current()).toBeUndefined(); - registry.add({ sessionId: "s1", startedAtMs: 1 }); + start(registry, "s1"); expect(registry.current()).toBe("s1"); - registry.add({ sessionId: "s2", startedAtMs: 2 }); + start(registry, "s2"); expect(registry.current()).toBe("s2"); registry.remove("s2"); expect(registry.current()).toBe("s1"); @@ -15,27 +21,42 @@ describe("SessionRegistry", () => { expect(registry.current()).toBeUndefined(); }); - it("enforces the concurrency cap", () => { + it("enforces the concurrency cap including in-flight starts", () => { const registry = new SessionRegistry(2); - registry.add({ sessionId: "s1", startedAtMs: 1 }); - registry.add({ sessionId: "s2", startedAtMs: 2 }); - expect(() => registry.add({ sessionId: "s3", startedAtMs: 3 })).toThrow(/session limit/); + start(registry, "s1"); + start(registry, "s2"); + expect(() => registry.reserveStart()).toThrow(/session limit/); registry.remove("s1"); - registry.add({ sessionId: "s3", startedAtMs: 3 }); + registry.reserveStart(); + registry.completeStart({ sessionId: "s3", startedAtMs: Date.now() }); expect(registry.size()).toBe(2); }); - it("resolve prefers the explicit session and adopts unknown ids", () => { + it("counts reserved (not yet spawned) starts against the cap — the race case", () => { + const registry = new SessionRegistry(2); + registry.reserveStart(); + registry.reserveStart(); + // Both slots are in flight; a third concurrent start must reject even + // though no session has been registered yet. + expect(() => registry.reserveStart()).toThrow(/session limit/); + registry.abandonStart(); + registry.reserveStart(); + registry.completeStart({ sessionId: "s1", startedAtMs: Date.now() }); + expect(registry.ownedIds()).toEqual(["s1"]); + }); + + it("resolve prefers the explicit session and adopts it as a reference", () => { const registry = new SessionRegistry(5); - registry.add({ sessionId: "s1", startedAtMs: 1 }); - expect(registry.resolve("s9", "tool")).toBe("s9"); - expect(registry.current()).toBe("s9"); - expect(registry.list().map((s) => s.sessionId)).toContain("s9"); + start(registry, "s1"); + expect(registry.resolve("foreign", "tool")).toBe("foreign"); + expect(registry.current()).toBe("foreign"); + expect(registry.isOwned("foreign")).toBe(false); + expect(registry.ownedIds()).toEqual(["s1"]); }); it("resolve falls back to the current session", () => { const registry = new SessionRegistry(5); - registry.add({ sessionId: "s1", startedAtMs: 1 }); + start(registry, "s1"); expect(registry.resolve(undefined, "tool")).toBe("s1"); }); @@ -46,10 +67,36 @@ describe("SessionRegistry", () => { it("touch refreshes recency so fallback picks the most recent survivor", () => { const registry = new SessionRegistry(5); - registry.add({ sessionId: "s1", startedAtMs: 1 }); - registry.add({ sessionId: "s2", startedAtMs: 2 }); + start(registry, "s1"); + start(registry, "s2"); registry.touch("s1"); registry.remove("s1"); expect(registry.current()).toBe("s2"); }); + + it("resolveForStop accepts owned sessions and refuses foreign ones", () => { + const registry = new SessionRegistry(5); + start(registry, "s1"); + expect(registry.resolveForStop(undefined)).toBe("s1"); + expect(registry.resolveForStop("s1")).toBe("s1"); + registry.resolve("foreign", "browser_snapshot"); + expect(() => registry.resolveForStop("foreign")).toThrow(/not created by this plugin/); + expect(() => registry.resolveForStop("never-seen")).toThrow(/not created by this plugin/); + }); + + it("a refused stop does not move the current pointer", () => { + const registry = new SessionRegistry(5); + start(registry, "s1"); + expect(() => registry.resolveForStop("foreign")).toThrow(); + expect(registry.current()).toBe("s1"); + }); + + it("ownedIds excludes referenced sessions even after heavy interleaving", () => { + const registry = new SessionRegistry(5); + start(registry, "own1"); + registry.resolve("ext1", "tool"); + start(registry, "own2"); + registry.resolve("ext2", "tool"); + expect(registry.ownedIds().sort()).toEqual(["own1", "own2"]); + }); }); diff --git a/packages/dsh-plugin-browserskill/tests/tools.test.ts b/packages/dsh-plugin-browserskill/tests/tools.test.ts index 9b6aa60..eb2581c 100644 --- a/packages/dsh-plugin-browserskill/tests/tools.test.ts +++ b/packages/dsh-plugin-browserskill/tests/tools.test.ts @@ -238,6 +238,41 @@ describe("multi-session behavior", () => { // A rejected start must not spawn another bsk session (no leaked session). expect(calls.filter((c) => c.args.join(" ").startsWith("session start"))).toHaveLength(1); }); + + it("shares the cap atomically across concurrent starts (reservation race)", async () => { + const { tools, calls } = setup( + { "session start": seq([START_REPLY("s1"), START_REPLY("s2")]) }, + {}, + 1, + ); + const start = tools.get("browser_session_start"); + if (start === undefined) throw new Error("not registered"); + // Two truly concurrent starts against a cap of 1: exactly one may spawn. + const outcomes = await Promise.allSettled([ + start.execute({}, makeExec()), + start.execute({}, makeExec()), + ]); + const fulfilled = outcomes.filter((o) => o.status === "fulfilled"); + const rejected = outcomes.filter((o) => o.status === "rejected"); + expect(fulfilled).toHaveLength(1); + expect(rejected).toHaveLength(1); + expect(String((rejected[0] as PromiseRejectedResult).reason)).toMatch(/session limit/); + expect(calls.filter((c) => c.args.join(" ").startsWith("session start"))).toHaveLength(1); + }); + + it("refuses to stop a session the plugin did not create", async () => { + const { tools, calls } = setup(responses); + await startSession(tools); + // Reference a foreign session (adopted for the current pointer, never owned). + const snapshot = tools.get("browser_snapshot"); + await snapshot?.execute({ session: "ext7" }, makeExec()); + const stop = tools.get("browser_session_stop"); + await expect(stop?.execute({ session: "ext7" }, makeExec())).rejects.toThrow( + /not created by this plugin/, + ); + // No stop command may reach the daemon for a foreign session. + expect(calls.some((c) => c.args.join(" ").startsWith("session stop"))).toBe(false); + }); }); describe("browser_session_stop / list", () => {