fix(dsh-plugin): stabilize skill loading and browser sessions

Register bundled browser instructions at agent scope, prioritize foreground commands over thumbnail capture, and unify cancellation and session cleanup.

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());
     };
   });
 }

packages/dsh-plugin-browserskill/src/observation.ts

@@ -18,7 +18,12 @@ 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 +95,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 +184,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 +197,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 +255,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 +284,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 +348,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 +376,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 +413,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 +429,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 +449,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);
     }
@@ -415,13 +491,20 @@ export class ObservationService {
     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. */

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<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)`);

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<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();
+  };
+}

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<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) => ({