diff --git a/packages/dsh-plugin-browserskill/package.json b/packages/dsh-plugin-browserskill/package.json index dd18861..be8a194 100644 --- a/packages/dsh-plugin-browserskill/package.json +++ b/packages/dsh-plugin-browserskill/package.json @@ -107,6 +107,8 @@ "@deepseek-ai/dsh-client-ui-primitives": "^0.1.0-rc.6", "@deepseek-ai/dsh-client-ui-tool": "^0.1.0-rc.6", "@deepseek-ai/dsh-llm": "^0.1.0-rc.6", + "@deepseek-ai/dsh-scope": "^0.1.0-rc.6", + "@deepseek-ai/dsh-skill": "^0.1.0-rc.6", "@deepseek-ai/dsh-tools": "^0.1.0-rc.6", "@deepseek-ai/schemastery": "^3.18.1", "@tailwindcss/cli": "^4.3.3", diff --git a/packages/dsh-plugin-browserskill/src/index.ts b/packages/dsh-plugin-browserskill/src/index.ts index 45946d3..721e572 100644 --- a/packages/dsh-plugin-browserskill/src/index.ts +++ b/packages/dsh-plugin-browserskill/src/index.ts @@ -19,7 +19,7 @@ import { registerObservationRoutes } from "./observation-http"; import { KeyedExecutor } from "./queue"; import { type BskRunner, createBskRunner } from "./runner"; import { SessionRegistry } from "./sessions"; -import { registerBskSkill } from "./skill"; +import { armAgentScopedBskSkill, registerBskSkill } from "./skill"; import { type PluginConfig, registerTools } from "./tools"; export const name = "dsh-plugin-browserskill"; @@ -95,6 +95,11 @@ export function apply( // ONLY model-visible advertisement — the tool suite reveals itself on a // successful skill invocation (or a session whose history already has one). const unregisterSkill = registerBskSkill(ctx); + // A same-name CLI skill left in ~/.agents is discovered in the nearer preset + // layer and shadows the global registration above. Re-register through every + // exact agent context at startup so DSH always loads the browser_* protocol + // instructions; the shared CLI skill remains untouched for other agents. + const disarmAgentSkill = armAgentScopedBskSkill(ctx); const removeSuite = resolved.lazyTools ? armLazyTools(ctx, () => registerTools({ ctx, runner, registry, config: resolved, observation, queue }), @@ -135,17 +140,15 @@ export function apply( ctx.effect(() => { return () => { removeSuite(); + disarmAgentSkill(); unregisterSkill(); removeRoutes(); disarmArchiveCleanup(); - observation.dispose(); runner.killAll(); const stops = registry .ownedIds() - .map((sessionId) => - runner.run(["session", "stop", sessionId], { timeoutMs: 15_000 }).catch(() => {}), - ); - return Promise.all(stops).then(() => {}); + .map((sessionId) => observation.stopSession(sessionId).catch(() => false)); + return Promise.all(stops).then(() => observation.dispose()); }; }); } diff --git a/packages/dsh-plugin-browserskill/src/observation.ts b/packages/dsh-plugin-browserskill/src/observation.ts index 975db04..2c0c62f 100644 --- a/packages/dsh-plugin-browserskill/src/observation.ts +++ b/packages/dsh-plugin-browserskill/src/observation.ts @@ -18,7 +18,7 @@ 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 { BskError, type BskRunner, parseBskJson, runWithSessionBusyRetry } from "./runner"; import type { SessionRegistry } from "./sessions"; /** One owned session's live observation record (wire-stable shape). */ @@ -90,6 +90,10 @@ export class ObservationService { private readonly listeners = new Set<(event: ObservationEvent) => void>(); private readonly captureTimers = new Map(); private readonly captureInFlight = new Set(); + /** Captures intentionally cancelled to make way for foreground work. */ + private readonly capturePreempted = new Set(); + /** Number of queued/running model-facing calls that currently outrank thumbnails. */ + private readonly foregroundDepth = new Map(); private readonly captureFailures = new Map(); private readonly lastActivity = new Map(); /** Ephemeral frame id → PNG bytes. Only the live ring members are present. */ @@ -175,6 +179,8 @@ export class ObservationService { /** Drop a session (called from browser_session_stop). */ removeSession(sessionId: string): void { this.cancelCapture(sessionId); + this.foregroundDepth.delete(sessionId); + this.capturePreempted.delete(sessionId); this.captureFailures.delete(sessionId); this.lastActivity.delete(sessionId); this.dropSessionFrames(sessionId); @@ -186,6 +192,37 @@ export class ObservationService { } } + /** + * Give model-facing work priority over the best-effort thumbnail lane. + * The first lease cancels a pending timer and gracefully interrupts an + * active bsk screenshot; the last release schedules one fresh frame. + */ + acquireForeground(sessionId: string): () => void { + if (!this.deps.options.enabled || this.disposed || !this.observations.has(sessionId)) { + return () => {}; + } + const depth = this.foregroundDepth.get(sessionId) ?? 0; + this.foregroundDepth.set(sessionId, depth + 1); + if (depth === 0) { + this.cancelCapture(sessionId); + if (this.deps.runner.killFor(`observation:${sessionId}`) > 0) { + this.capturePreempted.add(sessionId); + } + } + let released = false; + return () => { + if (released) return; + released = true; + const remaining = (this.foregroundDepth.get(sessionId) ?? 1) - 1; + if (remaining > 0) { + this.foregroundDepth.set(sessionId, remaining); + return; + } + this.foregroundDepth.delete(sessionId); + if (!this.disposed && this.observations.has(sessionId)) this.scheduleCapture(sessionId, 0); + }; + } + /** Mark an action starting on a session (tool entry instrumentation). */ beginAction(sessionId: string, action: string): void { if (!this.deps.options.enabled || this.disposed) return; @@ -213,7 +250,7 @@ export class ObservationService { if (error !== undefined) next.lastError = error; else delete next.lastError; this.put(next); - this.scheduleCapture(sessionId, 0); + if (!this.foregroundDepth.has(sessionId)) this.scheduleCapture(sessionId, 0); } /** Record the settled page URL (navigate success / start with url). */ @@ -242,29 +279,48 @@ export class ObservationService { * button — same end state as `browser_session_stop`). Never waits behind a * hung in-flight command: tool children are killed first so the session's * keyed queue drains immediately, and no further captures queue up. A - * session the daemon already forgot (dead strip entries) stops - * idempotently — the goal state (entry gone) is identical. - * @returns whether the session ended up stopped/removed. + * session the daemon already forgot stops idempotently — the goal state + * (entry gone) is identical. + * @returns false only for a foreign session; bsk failures reject so callers + * can preserve the structured error instead of silently leaving a ghost. */ - async stopSession(sessionId: string): Promise { + async stopSession(sessionId: string, signal?: AbortSignal): Promise { if (!this.deps.registry.isOwned(sessionId)) return false; - this.cancelCapture(sessionId); - this.deps.runner.killFor(sessionId); - const result = await this.deps.queue.run(sessionId, () => - this.deps.runner.run(["session", "stop", sessionId], { timeoutMs: 30_000 }), - ); - if (result.code !== 0) { - let code: string | undefined; + const releaseForeground = this.acquireForeground(sessionId); + let actionError: string | undefined; + this.beginAction(sessionId, "stopping"); + try { + this.deps.runner.killFor(sessionId); + const result = await this.deps.queue.run( + sessionId, + () => + runWithSessionBusyRetry( + () => + this.deps.runner.run(["session", "stop", sessionId], { + signal, + timeoutMs: 30_000, + tag: sessionId, + }), + signal, + ), + signal, + ); + if (result.aborted) throw abortError(); try { - code = (JSON.parse(result.stdout) as { code?: string }).code; - } catch { - code = undefined; + parseBskJson(result, "session stop"); + } catch (error) { + if (!isSessionNotFoundError(error)) throw error; } - if (code !== "session_not_found") return false; + this.deps.registry.remove(sessionId); + this.removeSession(sessionId); + return true; + } catch (error) { + actionError = error instanceof Error ? error.message.split("\n")[0] : String(error); + throw error; + } finally { + this.endAction(sessionId, actionError); + releaseForeground(); } - this.deps.registry.remove(sessionId); - this.removeSession(sessionId); - return true; } /** @@ -287,6 +343,8 @@ export class ObservationService { for (const sessionId of [...this.captureTimers.keys()]) this.cancelCapture(sessionId); this.captureTimers.clear(); this.captureInFlight.clear(); + this.capturePreempted.clear(); + this.foregroundDepth.clear(); this.captureFailures.clear(); this.lastActivity.clear(); this.frames.clear(); @@ -313,6 +371,7 @@ export class ObservationService { /** Schedule the next capture for a session; `delayMs` 0 means "as soon as the event loop allows". */ private scheduleCapture(sessionId: string, delayMs?: number): void { if (!this.deps.options.enabled || this.disposed) return; + if (this.foregroundDepth.has(sessionId)) return; const entry = this.observations.get(sessionId); if (entry === undefined || entry.dead === true) return; this.cancelCapture(sessionId); @@ -349,12 +408,15 @@ export class ObservationService { const outPath = join(tmpdir(), `bsk-obs-${this.scratchNamespace}-${sessionKey}.png`); let writtenPath = outPath; try { - const result = await this.deps.queue.run(sessionId, () => - this.deps.runner.run(["screenshot", "--session", sessionId, "--out", outPath], { + const result = await this.deps.queue.run(sessionId, () => { + // A foreground lease may have arrived while this capture was queued. + if (this.foregroundDepth.has(sessionId)) return Promise.resolve(undefined); + return this.deps.runner.run(["screenshot", "--session", sessionId, "--out", outPath], { timeoutMs: 15_000, tag: `observation:${sessionId}`, - }), - ); + }); + }); + if (result === undefined) return; if (result.code !== 0) { let code: string | undefined; try { @@ -362,8 +424,12 @@ export class ObservationService { } catch { code = undefined; } - if (code === "session_not_found") { - this.markDead(sessionId); + if (isSessionNotFoundCode(code)) { + // Observation entries are owned sessions. If the daemon has already + // forgotten one (external close / restart), reconcile both stores so + // the PiP cannot retain a ghost strip or keep retrying screenshots. + this.deps.registry.remove(sessionId); + this.removeSession(sessionId); return; } throw new Error(`screenshot exited ${result.code}`); @@ -378,12 +444,17 @@ export class ObservationService { this.setAvailable(true); this.publishFrame(sessionId, entry, data); } catch { - // Silent by design: keep the previous frame, count toward backoff. - this.captureFailures.set(sessionId, (this.captureFailures.get(sessionId) ?? 0) + 1); - this.globalFailures += 1; - if (this.globalFailures >= FAILURE_BACKOFF_THRESHOLD) this.setAvailable(false); + // Foreground preemption is healthy scheduling, not browser/daemon + // unavailability; do not let normal tool traffic trip capture backoff. + if (!this.capturePreempted.delete(sessionId)) { + // Silent by design: keep the previous frame, count toward backoff. + this.captureFailures.set(sessionId, (this.captureFailures.get(sessionId) ?? 0) + 1); + this.globalFailures += 1; + if (this.globalFailures >= FAILURE_BACKOFF_THRESHOLD) this.setAvailable(false); + } } finally { await unlink(writtenPath).catch(() => {}); + this.capturePreempted.delete(sessionId); this.captureInFlight.delete(sessionId); if (!this.disposed && this.observations.has(sessionId)) this.scheduleCapture(sessionId); } @@ -414,14 +485,20 @@ export class ObservationService { this.hashes.delete(sessionId); this.seqs.delete(sessionId); } +} - /** Mark a session dead: grey it out, stop asking for frames, keep the entry for the strip. */ - private markDead(sessionId: string): void { - const entry = this.observations.get(sessionId); - if (entry === undefined || entry.dead === true) return; - this.cancelCapture(sessionId); - this.put({ ...entry, dead: true, action: "idle" }); - } +function isSessionNotFoundCode(code: string | undefined): boolean { + return code === "not_found" || code === "session_not_found"; +} + +function isSessionNotFoundError(error: unknown): boolean { + return error instanceof BskError && isSessionNotFoundCode(error.code); +} + +function abortError(): Error { + const error = new Error("tool call aborted"); + error.name = "AbortError"; + return error; } /** Map a bsk command label onto its observation action verb. */ diff --git a/packages/dsh-plugin-browserskill/src/runner.ts b/packages/dsh-plugin-browserskill/src/runner.ts index 6505a47..13a5752 100644 --- a/packages/dsh-plugin-browserskill/src/runner.ts +++ b/packages/dsh-plugin-browserskill/src/runner.ts @@ -64,14 +64,17 @@ export interface BskRunner { killFor(tag: string): number; } -const KILL_GRACE_MS = 2000; +// Business RPCs translate SIGINT into the daemon's cancel(rpc_id) protocol. +// Give that bounded reconciliation path time to settle before the hard kill. +const KILL_GRACE_MS = 3000; +const SESSION_BUSY_RETRY_DELAY_MS = 100; export function createBskRunner(bskPath: string, spawnImpl: SpawnImpl = spawn): BskRunner { const live = new Map(); function killChild(child: ChildProcess): void { if (child.exitCode !== null || child.signalCode !== null) return; - child.kill("SIGTERM"); + child.kill("SIGINT"); const force = setTimeout(() => { if (child.exitCode === null && child.signalCode === null) child.kill("SIGKILL"); }, KILL_GRACE_MS); @@ -163,6 +166,60 @@ export function isCommandNotFound(error: unknown): boolean { ); } +/** True only for the daemon's transient per-session reconciliation window. */ +export function isSessionBusyResult(result: BskRunResult): boolean { + if (result.code === 0) return false; + try { + const body = JSON.parse(result.stdout) as BskErrorBody; + const reason = + typeof body.data === "object" && body.data !== null && "reason" in body.data + ? (body.data as { reason?: unknown }).reason + : undefined; + return reason === "session_busy"; + } catch { + return false; + } +} + +/** + * Retry exactly once after the tiny daemon-settlement race that can follow a + * graceful SIGINT cancellation. Other errors and persistent busy states stay + * visible to callers. + */ +export async function runWithSessionBusyRetry( + run: () => Promise, + signal?: AbortSignal, +): Promise { + const first = await run(); + if (!isSessionBusyResult(first)) return first; + await abortableDelay(SESSION_BUSY_RETRY_DELAY_MS, signal); + return run(); +} + +function abortableDelay(ms: number, signal?: AbortSignal): Promise { + if (signal?.aborted) return Promise.reject(abortError()); + return new Promise((resolve, reject) => { + const timer = setTimeout(done, ms); + timer.unref(); + const onAbort = () => { + clearTimeout(timer); + signal?.removeEventListener("abort", onAbort); + reject(abortError()); + }; + function done() { + signal?.removeEventListener("abort", onAbort); + resolve(); + } + signal?.addEventListener("abort", onAbort, { once: true }); + }); +} + +function abortError(): Error { + const error = new Error("tool call aborted"); + error.name = "AbortError"; + return error; +} + /** Install guidance shown when the bsk CLI cannot be spawned. */ export function bskInstallMessage(bskPath: string): string { return ( @@ -182,7 +239,7 @@ export function parseBskJson(result: BskRunResult, commandLabel: string): unknow throw new BskError(`bsk ${commandLabel} timed out`, { timedOut: true }); } const body = result.stdout.trim(); - // Killed by our own interrupt (SIGTERM from killFor), not by the abort path: + // Killed by our own interrupt (SIGINT from killFor), not by the abort path: // say so instead of doubling the generic label into the message. if (result.code === null && !result.aborted && !result.timedOut) { throw new BskError(`bsk ${commandLabel} was interrupted (process killed)`); diff --git a/packages/dsh-plugin-browserskill/src/skill.ts b/packages/dsh-plugin-browserskill/src/skill.ts index 8fb7107..22e479c 100644 --- a/packages/dsh-plugin-browserskill/src/skill.ts +++ b/packages/dsh-plugin-browserskill/src/skill.ts @@ -9,6 +9,7 @@ */ import type { Context } from "@deepseek-ai/cordis"; +import type { Agent } from "@deepseek-ai/dsh-agent"; import { BSK_SKILL_DESCRIPTION, BSK_SKILL_MARKDOWN, @@ -25,6 +26,11 @@ interface SkillsLike { }): () => void; } +/** Structural view of the live-agent registry (absent in bare compositions). */ +interface AgentsLike { + list(): Agent[]; +} + /** * Publish the browser skill as an embedded runtime skill. Returns the unregistration * disposer. A composition without the `skills` service (or with dsh-tool-skill @@ -43,3 +49,54 @@ export function registerBskSkill(ctx: Context): () => void { source: "bundled", }); } + +/** + * Install the DSH-specific browser skill into every exact agent scope. + * + * DSH merges skill layers nearest-first (`agent -> preset -> global`). A legacy + * CLI skill installed at `~/.agents/skills/browser-skill` is discovered by the + * preset filesystem provider and therefore shadows a plugin-level registration. + * Registering the embedded skill through `agent.ctx` makes the DSH protocol + * contract authoritative for that agent without touching the shared CLI skill. + * + * New agents are handled at `agent/session-start`, the first supported startup + * injection point and still before the first prompt assembly. Existing agents + * are registered immediately so plugin reloads take effect without recreating + * the conversation. Returns a disposer for all plugin-owned registrations. + */ +export function armAgentScopedBskSkill(ctx: Context): () => void { + const registrations = new Map void>(); + let active = true; + + const registerForAgent = (agent: Agent): void => { + if (!active || registrations.has(agent)) return; + registrations.set(agent, registerBskSkill(agent.ctx)); + }; + + let stopSessionStart = () => {}; + let stopDisposed = () => {}; + if (typeof ctx.on === "function") { + stopSessionStart = ctx.on("agent/session-start", ({ agent }) => { + registerForAgent(agent); + }); + stopDisposed = ctx.on("agent/disposed", ({ agent }) => { + // Agent-scoped effects have already unwound at this lifecycle edge. Drop + // the retained disposer without invoking it against a disposed context. + registrations.delete(agent); + }); + } + + const agents = ctx.get("agents") as AgentsLike | undefined | null; + if (agents != null && typeof agents.list === "function") { + for (const agent of agents.list()) registerForAgent(agent); + } + + return () => { + if (!active) return; + active = false; + stopDisposed(); + stopSessionStart(); + for (const unregister of [...registrations.values()].reverse()) unregister(); + registrations.clear(); + }; +} diff --git a/packages/dsh-plugin-browserskill/src/tools.ts b/packages/dsh-plugin-browserskill/src/tools.ts index 38ac96e..00ea69f 100644 --- a/packages/dsh-plugin-browserskill/src/tools.ts +++ b/packages/dsh-plugin-browserskill/src/tools.ts @@ -22,7 +22,13 @@ import { trySaveScreenshot } from "./image"; import { actionForLabel, type ObservationService } from "./observation"; import { registerPhaseOneTools } from "./phase-one-tools"; import type { KeyedExecutor } from "./queue"; -import { type BskRunner, bskInstallMessage, isCommandNotFound, parseBskJson } from "./runner"; +import { + type BskRunner, + bskInstallMessage, + isCommandNotFound, + parseBskJson, + runWithSessionBusyRetry, +} from "./runner"; import type { SessionRegistry } from "./sessions"; import { SESSION_PARAM } from "./tool-params"; @@ -79,6 +85,8 @@ async function runBsk( observeSession?: string, runnerTimeoutMs?: number, ): Promise { + const releaseForeground = + observeSession !== undefined ? deps.observation.acquireForeground(observeSession) : undefined; // beginAction must fire only when the task actually starts inside the // per-session queue — marking it while still queued would overwrite the // in-flight action's label (and flash it idle on a queued abort). @@ -92,11 +100,15 @@ async function runBsk( began = true; deps.observation.beginAction(observeSession, actionForLabel(label)); } - return deps.runner.run(args, { - signal: exec.signal, - timeoutMs: runnerTimeoutMs ?? deps.config.defaultTimeoutMs, - ...(observeSession !== undefined ? { tag: observeSession } : {}), - }); + return runWithSessionBusyRetry( + () => + deps.runner.run(args, { + signal: exec.signal, + timeoutMs: runnerTimeoutMs ?? deps.config.defaultTimeoutMs, + ...(observeSession !== undefined ? { tag: observeSession } : {}), + }), + exec.signal, + ); }; result = observeSession !== undefined @@ -121,6 +133,7 @@ async function runBsk( if (observeSession !== undefined && began) { deps.observation.endAction(observeSession, actionError); } + releaseForeground?.(); } } @@ -267,22 +280,15 @@ export function registerTools(deps: ToolDeps): () => void { } } catch (error) { // A half-initialized session must not leak: stop it before surfacing. - // The stop funnels through the session's queue (an in-flight command - // — e.g. an observation capture — would otherwise make the daemon - // refuse it), with one retry for good measure. - for (let attempt = 0; attempt < 2; attempt++) { - try { - const stopped = await deps.queue.run(reply.session_id, () => - deps.runner.run(["session", "stop", reply.session_id], { timeoutMs: 30_000 }), - ); - if (stopped.code === 0) break; - } catch { - // retry once, then give up quietly (registry/observation are - // cleaned below either way; the daemon reaps orphans). - } + // Reuse the same capture-preempting, idempotent path as explicit and + // overlay stops; local ownership is dropped even when daemon cleanup + // itself fails, because this start never becomes usable to the model. + try { + await deps.observation.stopSession(reply.session_id); + } catch { + registry.remove(reply.session_id); + deps.observation.removeSession(reply.session_id); } - registry.remove(reply.session_id); - deps.observation.removeSession(reply.session_id); throw error; } return { @@ -326,9 +332,15 @@ export function registerTools(deps: ToolDeps): () => void { }, async execute(args, exec) { const sessionId = registry.resolveForStop(args.session); - await runBsk(deps, exec, ["session", "stop", sessionId], "session stop", sessionId); - registry.remove(sessionId); - deps.observation.removeSession(sessionId); + try { + const stopped = await deps.observation.stopSession(sessionId, exec.signal); + if (!stopped) throw new Error(`browser session ${sessionId} is not owned by this plugin`); + } catch (error) { + if (isCommandNotFound(error)) { + throw new Error(bskInstallMessage(deps.config.bskPath)); + } + throw error; + } return { stopped: sessionId }; }, presentCall: (args) => ({ diff --git a/packages/dsh-plugin-browserskill/tests/observation.test.ts b/packages/dsh-plugin-browserskill/tests/observation.test.ts index 95cb23f..98182e2 100644 --- a/packages/dsh-plugin-browserskill/tests/observation.test.ts +++ b/packages/dsh-plugin-browserskill/tests/observation.test.ts @@ -78,7 +78,7 @@ function fakeRunner( if (opts.stopNotFound) { return { code: 4, - stdout: JSON.stringify({ code: "session_not_found", message: "no such session" }), + stdout: JSON.stringify({ code: "not_found", message: "no such session" }), stderr: "", timedOut: false, aborted: false, @@ -93,7 +93,7 @@ function fakeRunner( if (opts.screenshotNotFound) { return { code: 4, - stdout: JSON.stringify({ code: "session_not_found", message: "no such session" }), + stdout: JSON.stringify({ code: "not_found", message: "no such session" }), stderr: "", timedOut: false, aborted: false, @@ -224,6 +224,57 @@ describe("state machine", () => { }); describe("thumbnail cadence", () => { + it("gives foreground work priority and resumes with one fresh capture", async () => { + const scheduler = fakeScheduler(); + let finishCapture: ((result: BskRunResult) => void) | undefined; + const killed: string[] = []; + const runner: FakeRunner = { + calls: [], + killed, + run(args) { + if (args[0] !== "screenshot") { + return Promise.resolve({ + code: 0, + stdout: "{}", + stderr: "", + timedOut: false, + aborted: false, + }); + } + return new Promise((resolve) => { + finishCapture = resolve; + }); + }, + killAll() {}, + killFor(tag) { + killed.push(tag); + if (tag === "observation:s1" && finishCapture !== undefined) { + finishCapture({ + code: 5, + stdout: JSON.stringify({ code: "cancelled", message: "cancelled" }), + stderr: "", + timedOut: false, + aborted: false, + }); + finishCapture = undefined; + return 1; + } + return 0; + }, + }; + const { service } = setup({ runner, scheduler }); + service.addSession("s1"); + scheduler.runNext(); + await waitFor(() => finishCapture !== undefined); + + const release = service.acquireForeground("s1"); + expect(killed).toEqual(["observation:s1"]); + await waitFor(() => scheduler.pending().length === 0); + release(); + expect(service.isAvailable()).toBe(true); + expect(scheduler.pending()).toEqual([0]); + }); + it("captures immediately on add and after action end, then fast-cadences while active", async () => { const scheduler = fakeScheduler(); const runner = fakeRunner(); @@ -351,7 +402,7 @@ describe("stopSession", () => { expect(registry.isOwned("s1")).toBe(true); await expect(service.stopSession("s1")).resolves.toBe(true); - expect(runner.killed).toEqual(["s1"]); + expect(runner.killed).toEqual(["observation:s1", "s1"]); const stop = runner.calls.find((c) => c.args[0] === "session" && c.args[1] === "stop"); expect(stop?.args).toEqual(["session", "stop", "s1"]); expect(registry.isOwned("s1")).toBe(false); @@ -380,13 +431,13 @@ describe("stopSession", () => { expect(events[events.length - 1]).toMatchObject({ type: "remove" }); }); - it("returns false and keeps the entry when the stop itself fails", async () => { + it("rejects and keeps the entry when the stop itself fails", async () => { const registry = new SessionRegistry(5); own(registry, "s1"); const runner = fakeRunner({ stopFails: true }); const { service } = setup({ registry, runner }); service.addSession("s1"); - await expect(service.stopSession("s1")).resolves.toBe(false); + await expect(service.stopSession("s1")).rejects.toThrow(/boom/); expect(registry.isOwned("s1")).toBe(true); expect(service.getState().map((s) => s.sessionId)).toEqual(["s1"]); }); @@ -590,7 +641,7 @@ describe("HTTP/SSE interface", () => { }); }); -describe("availability and dead sessions", () => { +describe("availability and missing sessions", () => { it("flips availability off after repeated global failures and back on success", async () => { const scheduler = fakeScheduler(); const runner = fakeRunner({ screenshotFails: true }); @@ -610,23 +661,18 @@ describe("availability and dead sessions", () => { service.dispose(); }); - it("marks a session dead on session_not_found and stops asking for frames", async () => { + it("removes an owned ghost on not_found and stops asking for frames", async () => { const scheduler = fakeScheduler(); const runner = fakeRunner({ screenshotNotFound: true }); - const { service, events } = setup({ runner, scheduler }); + const registry = new SessionRegistry(5); + own(registry, "s1"); + const { service, events } = setup({ runner, scheduler, registry }); service.addSession("s1"); scheduler.runNext(); - await waitFor(() => events.some((e) => e.type === "upsert" && e.session?.dead === true)); - expect(service.getState()[0].dead).toBe(true); - expect(service.getState()[0].action).toBe("idle"); - // No further captures are scheduled for a dead session. - expect(scheduler.pending()).toEqual([]); - // Instrumentation is dropped for dead sessions. - service.beginAction("s1", "clicking"); - expect(service.getState()[0].action).toBe("idle"); - // A dead session leaves cleanly. - service.removeSession("s1"); + await waitFor(() => events.some((e) => e.type === "remove")); expect(service.getState()).toEqual([]); + expect(registry.isOwned("s1")).toBe(false); + expect(scheduler.pending()).toEqual([]); }); it("clears lastError on the next action and keeps it on failure", () => { diff --git a/packages/dsh-plugin-browserskill/tests/runner.test.ts b/packages/dsh-plugin-browserskill/tests/runner.test.ts index 52d3dac..37b2de1 100644 --- a/packages/dsh-plugin-browserskill/tests/runner.test.ts +++ b/packages/dsh-plugin-browserskill/tests/runner.test.ts @@ -1,7 +1,14 @@ import type { ChildProcess } from "node:child_process"; import { EventEmitter } from "node:events"; import { describe, expect, it } from "vitest"; -import { type BskRunResult, createBskRunner, isCommandNotFound, parseBskJson } from "../src/runner"; +import { + type BskRunResult, + createBskRunner, + isCommandNotFound, + isSessionBusyResult, + parseBskJson, + runWithSessionBusyRetry, +} from "../src/runner"; /** Minimal fake ChildProcess driven by the test. */ class FakeChild extends EventEmitter { @@ -63,7 +70,7 @@ describe("createBskRunner", () => { controller.abort(); const result = await promise; expect(result.aborted).toBe(true); - expect(child.killedWith).toContain("SIGTERM"); + expect(child.killedWith).toContain("SIGINT"); }); it("kills the child on timeout", async () => { @@ -80,7 +87,7 @@ describe("createBskRunner", () => { const promise = runner.run(["session", "list"]); runner.killAll(); const result = await promise; - expect(child.killedWith).toContain("SIGTERM"); + expect(child.killedWith).toContain("SIGINT"); expect(result.code).toBeNull(); }); }); @@ -140,3 +147,51 @@ describe("isCommandNotFound", () => { expect(isCommandNotFound(new Error("other"))).toBe(false); }); }); + +describe("session busy reconciliation", () => { + const busy: BskRunResult = { + code: 4, + stdout: JSON.stringify({ + code: "timeout", + message: "session already has an unfinished command", + data: { reason: "session_busy" }, + }), + stderr: "", + timedOut: false, + aborted: false, + }; + const ok: BskRunResult = { + code: 0, + stdout: "{}", + stderr: "", + timedOut: false, + aborted: false, + }; + + it("recognizes only the structured session_busy reason", () => { + expect(isSessionBusyResult(busy)).toBe(true); + expect(isSessionBusyResult({ ...busy, stdout: '{"code":"timeout"}' })).toBe(false); + expect(isSessionBusyResult(ok)).toBe(false); + }); + + it("retries one transient busy result and returns the settled response", async () => { + const replies = [busy, ok]; + let calls = 0; + const result = await runWithSessionBusyRetry(async () => { + calls += 1; + return replies.shift() ?? ok; + }); + expect(result).toBe(ok); + expect(calls).toBe(2); + }); + + it("does not loop on a persistent busy result", async () => { + let calls = 0; + const result = await runWithSessionBusyRetry(async () => { + calls += 1; + return busy; + }); + expect(result).toBe(busy); + expect(calls).toBe(2); + }); +}); diff --git a/packages/dsh-plugin-browserskill/tests/skill.test.ts b/packages/dsh-plugin-browserskill/tests/skill.test.ts index 138c511..d63f1cf 100644 --- a/packages/dsh-plugin-browserskill/tests/skill.test.ts +++ b/packages/dsh-plugin-browserskill/tests/skill.test.ts @@ -1,8 +1,11 @@ // Browser skill injection: registration wiring, catalog content, progressive-load // weight, and silent degradation without the skill seam. +import { Context } from "@deepseek-ai/cordis"; +import { createScope, scopeTarget } from "@deepseek-ai/dsh-scope"; +import { SkillRegistry } from "@deepseek-ai/dsh-skill"; import { describe, expect, it } from "vitest"; -import { registerBskSkill } from "../src/skill"; +import { armAgentScopedBskSkill, registerBskSkill } from "../src/skill"; function fakeCtx(skills?: unknown) { return { get: (key: string) => (key === "skills" ? skills : undefined) } as never; @@ -100,3 +103,88 @@ describe("registerBskSkill", () => { expect(() => disposer()).not.toThrow(); }); }); + +describe("armAgentScopedBskSkill", () => { + it("overrides a nearer legacy CLI skill when the DSH agent starts", async () => { + const root = new Context(); + const skillFiber = root.plugin(SkillRegistry); + await skillFiber; + + // The plugin's original registration is global. + registerBskSkill(skillFiber.ctx); + + // Model the standard preset's filesystem discovery of + // ~/.agents/skills/browser-skill. Its nearer scope wins before the fix. + const presetKey = { preset: "standard" }; + const presetScope = createScope(skillFiber.ctx, presetKey); + presetScope.ctx.skills.register({ + name: "browser-skill", + description: "Legacy BrowserSkill CLI instructions", + content: "Use bsk shell commands.", + source: "user-agents", + }); + + const agentKey = { agent: "test-agent" }; + const agentScope = createScope(skillFiber.ctx, agentKey, { parent: presetKey }); + const agent = { + id: "test-agent", + ctx: agentScope.ctx, + } as never; + const agents = { list: () => [] }; + const pluginCtx = skillFiber.ctx.extend({ agents }); + + const before = await skillFiber.ctx.skills.get("browser-skill", { scope: agentKey }); + expect(before?.content).toBe("Use bsk shell commands."); + expect(before?.source).toBe("user-agents"); + + const disarm = armAgentScopedBskSkill(pluginCtx); + pluginCtx.emit(scopeTarget(agent, agentKey), "agent/session-start", { + agent, + source: "startup", + }); + + const after = await skillFiber.ctx.skills.get("browser-skill", { scope: agentKey }); + expect(after?.content).toContain("All browser operations must use the injected tools directly"); + expect(after?.content).not.toMatch(/\bbsk\b/i); + expect(after?.source).toBe("bundled"); + + disarm(); + await agentScope.dispose(); + await presetScope.dispose(); + await skillFiber.dispose(); + }); + + it("installs on existing agents for plugin reload and disposes registrations", () => { + const registered: Record[] = []; + let disposed = 0; + const agent = { + ctx: fakeCtx({ + register(skill: Record) { + registered.push(skill); + return () => { + disposed += 1; + }; + }, + }), + } as never; + const listeners = new Map void>(); + const ctx = { + get: (key: string) => (key === "agents" ? { list: () => [agent] } : undefined), + on(event: string, listener: (payload: { agent: never }) => void) { + listeners.set(event, listener); + return () => listeners.delete(event); + }, + } as never; + + const disarm = armAgentScopedBskSkill(ctx); + expect(registered).toHaveLength(1); + + // A later lifecycle notification for the same agent must not duplicate it. + listeners.get("agent/session-start")?.({ agent }); + expect(registered).toHaveLength(1); + + disarm(); + expect(disposed).toBe(1); + expect(listeners.size).toBe(0); + }); +}); diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index d3ce398..d60c24f 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -114,6 +114,12 @@ importers: '@deepseek-ai/dsh-llm': specifier: ^0.1.0-rc.6 version: 0.1.0-rc.6(@deepseek-ai/cordis@4.0.1)(@deepseek-ai/dsh-attachment@0.1.0-rc.6(@deepseek-ai/cordis@4.0.1)(@deepseek-ai/dsh-brand@0.1.0-rc.6(@deepseek-ai/cordis@4.0.1)(@deepseek-ai/dsh-invariants@0.1.0-rc.6(@deepseek-ai/cordis@4.0.1)))(@deepseek-ai/dsh-invariants@0.1.0-rc.6(@deepseek-ai/cordis@4.0.1)))(@deepseek-ai/dsh-brand@0.1.0-rc.6(@deepseek-ai/cordis@4.0.1)(@deepseek-ai/dsh-invariants@0.1.0-rc.6(@deepseek-ai/cordis@4.0.1)))(@deepseek-ai/dsh-invariants@0.1.0-rc.6(@deepseek-ai/cordis@4.0.1))(@deepseek-ai/dsh-timeout@0.1.0-rc.6(@deepseek-ai/cordis@4.0.1)(@deepseek-ai/dsh-invariants@0.1.0-rc.6(@deepseek-ai/cordis@4.0.1))) + '@deepseek-ai/dsh-scope': + specifier: ^0.1.0-rc.6 + version: 0.1.0-rc.6(@deepseek-ai/cordis@4.0.1)(@deepseek-ai/dsh-invariants@0.1.0-rc.6(@deepseek-ai/cordis@4.0.1)) + '@deepseek-ai/dsh-skill': + specifier: ^0.1.0-rc.6 + version: 0.1.0-rc.6(@deepseek-ai/cordis@4.0.1)(@deepseek-ai/dsh-invariants@0.1.0-rc.6(@deepseek-ai/cordis@4.0.1))(@deepseek-ai/dsh-llm@0.1.0-rc.6(@deepseek-ai/cordis@4.0.1)(@deepseek-ai/dsh-attachment@0.1.0-rc.6(@deepseek-ai/cordis@4.0.1)(@deepseek-ai/dsh-brand@0.1.0-rc.6(@deepseek-ai/cordis@4.0.1)(@deepseek-ai/dsh-invariants@0.1.0-rc.6(@deepseek-ai/cordis@4.0.1)))(@deepseek-ai/dsh-invariants@0.1.0-rc.6(@deepseek-ai/cordis@4.0.1)))(@deepseek-ai/dsh-brand@0.1.0-rc.6(@deepseek-ai/cordis@4.0.1)(@deepseek-ai/dsh-invariants@0.1.0-rc.6(@deepseek-ai/cordis@4.0.1)))(@deepseek-ai/dsh-invariants@0.1.0-rc.6(@deepseek-ai/cordis@4.0.1))(@deepseek-ai/dsh-timeout@0.1.0-rc.6(@deepseek-ai/cordis@4.0.1)(@deepseek-ai/dsh-invariants@0.1.0-rc.6(@deepseek-ai/cordis@4.0.1))))(@deepseek-ai/dsh-scope@0.1.0-rc.6(@deepseek-ai/cordis@4.0.1)(@deepseek-ai/dsh-invariants@0.1.0-rc.6(@deepseek-ai/cordis@4.0.1))) '@deepseek-ai/dsh-tools': specifier: ^0.1.0-rc.6 version: 0.1.0-rc.6(f8724372086ccc1457fc84e7becee2e0)