diff --git a/packages/dsh-plugin-browserskill/src/index.ts b/packages/dsh-plugin-browserskill/src/index.ts index 59ffe17..e5660a0 100644 --- a/packages/dsh-plugin-browserskill/src/index.ts +++ b/packages/dsh-plugin-browserskill/src/index.ts @@ -14,6 +14,7 @@ import type { Context } from "@deepseek-ai/cordis"; import Schema from "@deepseek-ai/schemastery"; import { ObservationService } from "./observation"; import { registerObservationRoutes } from "./observation-http"; +import { KeyedExecutor } from "./queue"; import { type BskRunner, createBskRunner } from "./runner"; import { SessionRegistry } from "./sessions"; import { type PluginConfig, registerTools } from "./tools"; @@ -65,10 +66,12 @@ export function apply( }; const runner = options.runnerFactory?.(resolved.bskPath) ?? createBskRunner(resolved.bskPath); const registry = new SessionRegistry(resolved.maxSessions); + const queue = new KeyedExecutor(); const observation = new ObservationService({ ctx, runner, registry, + queue, options: { enabled: resolved.observationEnabled, thumbnailIntervalMs: resolved.thumbnailIntervalMs, @@ -76,7 +79,7 @@ export function apply( }, }); - registerTools({ ctx, runner, registry, config: resolved, observation }); + registerTools({ ctx, runner, registry, config: resolved, observation, queue }); // Route registration rides ctx.inject: the webServer service may be provided // AFTER this plugin loads, and in headless compositions it never appears (the // callback simply never runs, leaving the rest of the plugin unaffected). @@ -121,6 +124,7 @@ export function apply( export type { ObservationEvent, ObservationOptions, SessionObservation } from "./observation"; export { ObservationService } from "./observation"; export { registerObservationRoutes } from "./observation-http"; +export { KeyedExecutor } from "./queue"; export type { BskRunner, BskRunResult, SpawnImpl } from "./runner"; export { BskError, createBskRunner } from "./runner"; export { SessionRegistry } from "./sessions"; diff --git a/packages/dsh-plugin-browserskill/src/observation.ts b/packages/dsh-plugin-browserskill/src/observation.ts index 1176b1f..1522913 100644 --- a/packages/dsh-plugin-browserskill/src/observation.ts +++ b/packages/dsh-plugin-browserskill/src/observation.ts @@ -15,6 +15,7 @@ import { readFile } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; import type { Context } from "@deepseek-ai/cordis"; +import type { KeyedExecutor } from "./queue"; import type { BskRunner } from "./runner"; import type { SessionRegistry } from "./sessions"; @@ -98,6 +99,7 @@ export class ObservationService { ctx: Context; runner: BskRunner; registry: SessionRegistry; + queue: KeyedExecutor; options: ObservationOptions; scheduler?: ObservationScheduler; }, @@ -287,9 +289,11 @@ export class ObservationService { this.captureInFlight.add(sessionId); try { const outPath = join(tmpdir(), `bsk-obs-${sessionId}-${this.scheduler.now()}.png`); - const result = await this.deps.runner.run( - ["screenshot", "--session", sessionId, "--out", outPath], - { timeoutMs: 15_000, tag: `observation:${sessionId}` }, + const result = await this.deps.queue.run(sessionId, () => + this.deps.runner.run(["screenshot", "--session", sessionId, "--out", outPath], { + timeoutMs: 15_000, + tag: `observation:${sessionId}`, + }), ); if (result.code !== 0) { let code: string | undefined; diff --git a/packages/dsh-plugin-browserskill/src/queue.ts b/packages/dsh-plugin-browserskill/src/queue.ts new file mode 100644 index 0000000..2e53204 --- /dev/null +++ b/packages/dsh-plugin-browserskill/src/queue.ts @@ -0,0 +1,62 @@ +/** + * Per-key FIFO executor. The bsk daemon serializes commands per session (a + * second command while one is unfinished is rejected), so every plugin + * command — model-facing tool calls AND observation captures — funnels + * through one queue per session. A queued task rejects early if its abort + * signal fires before it starts; once running, cancellation is owned by the + * task's own signal handling (the runner kills the child). + */ + +export class KeyedExecutor { + private readonly tails = new Map>(); + + /** Run `fn` after every previously queued task for `key` settled. */ + run(key: string, fn: () => Promise, signal?: AbortSignal): Promise { + const previous = this.tails.get(key) ?? Promise.resolve(); + const task = new Promise((resolve, reject) => { + let settled = false; + const onAbort = () => { + if (!settled) { + settled = true; + reject(abortError()); + } + }; + signal?.addEventListener("abort", onAbort, { once: true }); + void previous.then(() => { + if (settled) return; + if (signal?.aborted) { + settled = true; + reject(abortError()); + return; + } + // Running now: cancellation flows through the task's own signal. + signal?.removeEventListener("abort", onAbort); + fn().then( + (value) => { + settled = true; + resolve(value); + }, + (error: unknown) => { + settled = true; + reject(error); + }, + ); + }); + }); + // The stored tail never rejects, so one failure never strands the queue. + this.tails.set( + key, + task.then( + () => {}, + () => {}, + ), + ); + return task; + } +} + +function abortError(): Error { + const error = new Error("tool call aborted"); + error.name = "AbortError"; + return error; +} diff --git a/packages/dsh-plugin-browserskill/src/tools.ts b/packages/dsh-plugin-browserskill/src/tools.ts index d78e8c3..c4b8c46 100644 --- a/packages/dsh-plugin-browserskill/src/tools.ts +++ b/packages/dsh-plugin-browserskill/src/tools.ts @@ -14,6 +14,7 @@ import type { ContentBlock } from "@deepseek-ai/dsh-llm"; import { defineTool, type ToolResult, type ToolRunContext } from "@deepseek-ai/dsh-tools"; import { trySaveScreenshot } from "./image"; import { actionForLabel, type ObservationService } from "./observation"; +import type { KeyedExecutor } from "./queue"; import { type BskRunner, bskInstallMessage, isCommandNotFound, parseBskJson } from "./runner"; import type { SessionRegistry } from "./sessions"; @@ -33,6 +34,8 @@ export interface ToolDeps { registry: SessionRegistry; config: PluginConfig; observation: ObservationService; + /** Per-session FIFO: the daemon rejects a second command while one is unfinished. */ + queue: KeyedExecutor; } /** Device presets supported by `bsk emulate --device`. */ @@ -74,11 +77,16 @@ async function runBsk( try { let result; try { - result = await deps.runner.run(args, { - signal: exec.signal, - timeoutMs: deps.config.defaultTimeoutMs, - ...(observeSession !== undefined ? { tag: observeSession } : {}), - }); + const runOnce = () => + deps.runner.run(args, { + signal: exec.signal, + timeoutMs: deps.config.defaultTimeoutMs, + ...(observeSession !== undefined ? { tag: observeSession } : {}), + }); + result = + observeSession !== undefined + ? await deps.queue.run(observeSession, runOnce, exec.signal) + : await runOnce(); } catch (error) { if (isCommandNotFound(error)) { throw new Error(bskInstallMessage(deps.config.bskPath)); diff --git a/packages/dsh-plugin-browserskill/tests/observation.test.ts b/packages/dsh-plugin-browserskill/tests/observation.test.ts index b2d52eb..bb19e68 100644 --- a/packages/dsh-plugin-browserskill/tests/observation.test.ts +++ b/packages/dsh-plugin-browserskill/tests/observation.test.ts @@ -14,6 +14,7 @@ import { ObservationService, } from "../src/observation"; import { registerObservationRoutes } from "../src/observation-http"; +import { KeyedExecutor } from "../src/queue"; import type { BskRunner, BskRunOptions, BskRunResult } from "../src/runner"; import { SessionRegistry } from "../src/sessions"; @@ -128,6 +129,7 @@ function setup(opts: { ctx: fakeCtx(attachments), runner, registry, + queue: new KeyedExecutor(), options: OPTIONS, scheduler, }); diff --git a/packages/dsh-plugin-browserskill/tests/tools.test.ts b/packages/dsh-plugin-browserskill/tests/tools.test.ts index 12fcf3a..688bf4a 100644 --- a/packages/dsh-plugin-browserskill/tests/tools.test.ts +++ b/packages/dsh-plugin-browserskill/tests/tools.test.ts @@ -4,6 +4,7 @@ import { join } from "node:path"; import type { ToolDefinition, ToolRunContext } from "@deepseek-ai/dsh-tools"; import { describe, expect, it } from "vitest"; import { ObservationService } from "../src/observation"; +import { KeyedExecutor } from "../src/queue"; import type { BskRunner, BskRunOptions, BskRunResult } from "../src/runner"; import { SessionRegistry } from "../src/sessions"; import { type PluginConfig, registerTools } from "../src/tools"; @@ -23,6 +24,7 @@ function disabledObservation(deps: { ctx: unknown; runner: BskRunner; registry: ctx: deps.ctx as never, runner: deps.runner, registry: deps.registry, + queue: new KeyedExecutor(), options: { enabled: false, thumbnailIntervalMs: 1500, idleIntervalMs: 8000 }, }); } @@ -121,6 +123,7 @@ function setup( registry, config, observation: disabledObservation({ ctx, runner: runner as BskRunner, registry }), + queue: new KeyedExecutor(), }); return { tools, calls, registry }; } @@ -576,6 +579,7 @@ describe("error and cancellation semantics", () => { registry, config: CONFIG, observation: disabledObservation({ ctx, runner, registry }), + queue: new KeyedExecutor(), }); const start = tools.get("browser_session_start"); await expect(start?.execute({}, makeExec())).rejects.toThrow(/BrowserSkill must be installed/);