Merge pull request #147 from Tencent/fix/dsh-plugin-session-lifecycle

fix(dsh-plugin): stabilize skill loading and browser sessions
This commit is contained in:
Zhang GH
2026-08-28 12:11:08 +08:00
committed by GitHub
10 changed files with 495 additions and 92 deletions
@@ -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",
@@ -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());
};
});
}
@@ -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<string, unknown>();
private readonly captureInFlight = new Set<string>();
/** Captures intentionally cancelled to make way for foreground work. */
private readonly capturePreempted = new Set<string>();
/** Number of queued/running model-facing calls that currently outrank thumbnails. */
private readonly foregroundDepth = new Map<string, number>();
private readonly captureFailures = new Map<string, number>();
private readonly lastActivity = new Map<string, number>();
/** 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<boolean> {
async stopSession(sessionId: string, signal?: AbortSignal): Promise<boolean> {
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. */
+60 -3
View File
@@ -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<ChildProcess, string | undefined>();
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<BskRunResult>,
signal?: AbortSignal,
): Promise<BskRunResult> {
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<void> {
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)`);
@@ -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<Agent, () => 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();
};
}
+36 -24
View File
@@ -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<unknown> {
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) => ({
@@ -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", () => {
@@ -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);
});
});
@@ -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<string, unknown>[] = [];
let disposed = 0;
const agent = {
ctx: fakeCtx({
register(skill: Record<string, unknown>) {
registered.push(skill);
return () => {
disposed += 1;
};
},
}),
} as never;
const listeners = new Map<string, (payload: { agent: never }) => 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);
});
});
+6
View File
@@ -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)