From 1dbc457ad97977fdfb593fa8ad0c204276fe4ad3 Mon Sep 17 00:00:00 2001
From: Marc <marc@kernel-oops.com>
Date: Tue, 11 Aug 2026 17:29:45 +0100
Subject: [PATCH] fix: harden plugin shutdown and cold passes

---
 package-lock.json              |   3 +
 package.json                   |   3 +
 src/cards.ts                   |   9 +-
 src/distill.ts                 | 245 +++++++++++++++++----
 src/opencode-session-recall.ts | 128 +++++++++--
 src/summarize.ts               | 139 +++++++++---
 test/cards.test.ts             |  37 ++++
 test/distill.test.ts           | 386 ++++++++++++++++++++++++++++++++-
 test/plugin.test.ts            | 333 +++++++++++++++++++++++++++-
 test/summarize.test.ts         |  71 +++++-
 test/tools.test.ts             |   9 +-
 11 files changed, 1254 insertions(+), 109 deletions(-)

diff --git a/package-lock.json b/package-lock.json
index 57da9b1..8967f4b 100644
--- a/package-lock.json
+++ b/package-lock.json
@@ -26,6 +26,9 @@
         "typescript-eslint": "^8.58.1",
         "vitest": "^4.1.5"
       },
+      "engines": {
+        "opencode": ">=1.15.11"
+      },
       "peerDependencies": {
         "@opencode-ai/plugin": ">=1.2.0"
       }
diff --git a/package.json b/package.json
index 2ab7f70..3e773e7 100644
--- a/package.json
+++ b/package.json
@@ -62,6 +62,9 @@
     "url": "https://github.com/rmk40/opencode-session-recall/issues"
   },
   "license": "MIT",
+  "engines": {
+    "opencode": ">=1.15.11"
+  },
   "peerDependencies": {
     "@opencode-ai/plugin": ">=1.2.0"
   },
diff --git a/src/cards.ts b/src/cards.ts
index 90a90de..f924283 100644
--- a/src/cards.ts
+++ b/src/cards.ts
@@ -200,6 +200,8 @@ export type CardsRuntime = {
   semanticStatus(): SemanticStatus | undefined;
   /** Force the next rank() to rebuild from the source (tests). */
   invalidate(): void;
+  /** Prevent deferred semantic warm-up from touching its source after shutdown. */
+  dispose(): void;
 };
 
 function basename(path: string): string {
@@ -366,6 +368,7 @@ export function createCardsRuntime(deps: CardsRuntimeDeps): CardsRuntime {
   let lastRevision: string | undefined;
   let lastVectorsRevision: string | undefined;
   let loaded = false;
+  let disposed = false;
 
   /** Recompute/reuse card vectors and persist newly computed ones. Returns the
    *  store's `{ revision, committed }` result for the caller's own-write accounting,
@@ -543,7 +546,7 @@ export function createCardsRuntime(deps: CardsRuntimeDeps): CardsRuntime {
   // no snapshot is loaded yet (the next query then rebuilds and embeds).
   if (embedder && semanticWeight > 0 && deps.semanticReady) {
     void deps.semanticReady.then(() => {
-      if (!embedder.ready || !loaded) return;
+      if (disposed || !embedder.ready || !loaded) return;
       if (cardVectors && cardVectors.size > 0) return;
       // Deliberately leave lastVectorsRevision untouched: this pass writes vectors
       // (bumping vectors_rev past the cached value), so the next refresh reloads
@@ -554,6 +557,10 @@ export function createCardsRuntime(deps: CardsRuntimeDeps): CardsRuntime {
   }
 
   return {
+    dispose(): void {
+      disposed = true;
+    },
+
     rank(query, filters): CardHit[] {
       refreshIfStale();
       const excluded = filters.excludeFamilyOf
diff --git a/src/distill.ts b/src/distill.ts
index f000c34..d54718c 100644
--- a/src/distill.ts
+++ b/src/distill.ts
@@ -44,11 +44,21 @@ const DEFAULT_LEASE_RETRY_MS = 60_000;
 const DEFAULT_COLD_PASS_RETRY_MS = 60_000;
 const LEASE_TTL_MS = 30_000;
 const HEARTBEAT_MS = 10_000;
+/** Bound process-lifetime state retained for malformed legacy sessions. */
+const MAX_QUARANTINED_SESSIONS = 1_000;
 
 // ── Shared shapes ────────────────────────────────────────────────────────────
 
 type MsgWithParts = { info: Message; parts: Part[] };
 
+class MalformedSessionError extends Error {
+  override name = "MalformedSessionError";
+}
+
+class SessionMetadataTransportError extends Error {
+  override name = "SessionMetadataTransportError";
+}
+
 /** Human-layer field extracted from one part; the FTS `norm` column and the
  *  per-row ids are added when this becomes a {@link PartTextRow}. */
 export type DistillField = {
@@ -100,7 +110,12 @@ export type DistillStatus = {
 
 export type Distiller = {
   start(): void;
-  stop(): void;
+  /** Stop accepting or scheduling work immediately, but keep renewing an
+   *  already-held lease until {@link stop} finalizes the handoff. */
+  quiesce(): void;
+  /** Whether this instance still owns a live lease, verified against the store. */
+  ownsLease(): boolean;
+  stop(): Promise<void>;
   onEvent(event: Event): void;
   status(): DistillStatus;
 };
@@ -562,7 +577,10 @@ export async function fetchMessagePage(
       : { sessionID: opts.sessionID, limit: opts.limit };
   const resp = await client.session.messages(params);
   if (resp.error) throw new Error(errmsg(resp.error));
-  const items = Array.isArray(resp.data) ? (resp.data as MsgWithParts[]) : [];
+  if (!Array.isArray(resp.data)) {
+    throw new MalformedSessionError("successful message response was not an array");
+  }
+  const items = resp.data as MsgWithParts[];
   return { items, nextCursor: readNextCursor(resp.response) };
 }
 
@@ -603,7 +621,14 @@ export function createDistiller(options: DistillerOptions): Distiller {
 
   // Store unavailable (degraded mode): every method is a clean no-op.
   if (!options.store) {
-    return { start() {}, stop() {}, onEvent() {}, status: noopStatus };
+    return {
+      start() {},
+      quiesce() {},
+      ownsLease: () => false,
+      async stop() {},
+      onEvent() {},
+      status: noopStatus,
+    };
   }
   const store: Store = options.store;
 
@@ -628,6 +653,7 @@ export function createDistiller(options: DistillerOptions): Distiller {
   type Timer = ReturnType<typeof setTimeout>;
 
   let stopped = false;
+  let finalized = false;
   let leaseHeld = false;
   let coldPassState: DistillStatus["coldPass"] = "idle";
   let lastError: string | undefined;
@@ -638,8 +664,11 @@ export function createDistiller(options: DistillerOptions): Distiller {
   let leaseRetryTimer: Timer | undefined;
   let coldPassRetryTimer: Timer | undefined;
   const debounceTimers = new Map<string, Timer>();
-  const inFlight = new Set<string>();
+  let coldPassPromise: Promise<void> | undefined;
+  const inFlight = new Map<string, Promise<void>>();
   const pendingRerun = new Set<string>();
+  const quarantined = new Map<string, number>();
+  let stopPromise: Promise<void> | undefined;
   /** Sessions that saw a removal-shaped event since their last full distill, so
    *  the next re-distill takes the full path rather than appending. */
   const removalSince = new Set<string>();
@@ -665,15 +694,30 @@ export function createDistiller(options: DistillerOptions): Distiller {
   // ── Fetch primitives (all through the gate at background priority) ──
 
   async function discoverSessions(): Promise<DistillSessionMeta[]> {
-    const sessions = await gate.runBackground(() => discover());
+    const sessions = await gate.runBackground(() => {
+      if (stopped || !leaseHeld) return Promise.resolve([]);
+      return discover();
+    });
     // Never distill the summarizer's worker session — its prompts embed card
     // digests, which recall must not surface (see isSummarizerTitle).
     return sessions.map(toMeta).filter((meta) => !isSummarizerTitle(meta.title));
   }
 
   async function fetchSessionMeta(sessionID: string): Promise<DistillSessionMeta | null> {
-    const resp = await gate.runBackground(() => client.session.get({ sessionID }));
-    if (resp.error || !resp.data || typeof resp.data !== "object") return null;
+    const resp = await gate.runBackground(() => {
+      if (stopped || !leaseHeld) return Promise.resolve(null);
+      return client.session.get({ sessionID });
+    });
+    if (!resp) return null;
+    if (resp.error) {
+      throw new SessionMetadataTransportError(
+        `session ${sessionID} metadata fetch failed: ${errmsg(resp.error)}`,
+      );
+    }
+    if (resp.data == null) return null;
+    if (typeof resp.data !== "object") {
+      throw new MalformedSessionError(`session ${sessionID} metadata response was not an object`);
+    }
     return toMeta(resp.data as Session);
   }
 
@@ -686,15 +730,24 @@ export function createDistiller(options: DistillerOptions): Distiller {
     let rowCount = 0;
     let first = true;
     do {
-      if (!first && limits.distillDelayMs > 0) await sleep(limits.distillDelayMs);
+      if (stopped || !leaseHeld) return all;
+      if (!first && limits.distillDelayMs > 0) {
+        await sleep(limits.distillDelayMs);
+        if (stopped || !leaseHeld) return all;
+      }
       first = false;
       const before = cursor;
-      const page = await gate.runBackground(() =>
-        fetchMessagePage(client, { sessionID, limit: pageMessages, before }),
-      );
-      for (const msg of page.items) {
-        all.push(msg);
-        for (const part of msg.parts) rowCount += distillFields(part).length;
+      const page = await gate.runBackground(() => {
+        if (stopped || !leaseHeld) return Promise.resolve({ items: [], nextCursor: null });
+        return fetchMessagePage(client, { sessionID, limit: pageMessages, before });
+      });
+      try {
+        for (const msg of page.items) {
+          all.push(msg);
+          for (const part of msg.parts) rowCount += distillFields(part).length;
+        }
+      } catch (error) {
+        throw new MalformedSessionError(errmsg(error));
       }
       cursor = page.nextCursor ?? undefined;
     } while (cursor && rowCount < maxRows);
@@ -715,12 +768,17 @@ export function createDistiller(options: DistillerOptions): Distiller {
     let first = true;
     let reached = false;
     do {
-      if (!first && limits.distillDelayMs > 0) await sleep(limits.distillDelayMs);
+      if (stopped || !leaseHeld) return { messages: collected, reached: false };
+      if (!first && limits.distillDelayMs > 0) {
+        await sleep(limits.distillDelayMs);
+        if (stopped || !leaseHeld) return { messages: collected, reached: false };
+      }
       first = false;
       const before = cursor;
-      const page = await gate.runBackground(() =>
-        fetchMessagePage(client, { sessionID, limit: pageMessages, before }),
-      );
+      const page = await gate.runBackground(() => {
+        if (stopped || !leaseHeld) return Promise.resolve({ items: [], nextCursor: null });
+        return fetchMessagePage(client, { sessionID, limit: pageMessages, before });
+      });
       for (const msg of page.items) {
         if (msg.info.id === checkpoint) {
           reached = true;
@@ -805,6 +863,7 @@ export function createDistiller(options: DistillerOptions): Distiller {
       parentById: parentChainOf(session),
       caps,
     });
+    if (stopped || !leaseHeld) return;
     store.replaceSessionParts(session.id, rows, card);
     bumpCardsRev();
   }
@@ -816,6 +875,7 @@ export function createDistiller(options: DistillerOptions): Distiller {
       return;
     }
     const { messages: newMessages, reached } = await fetchNewMessages(session.id, checkpoint);
+    if (stopped || !leaseHeld) return;
     // If the checkpoint message was never found, the newest pages are NOT a clean
     // append tail (they'd re-insert already-stored parts and hit a UNIQUE
     // violation, wedging the session). Fall back to a full re-distill.
@@ -907,6 +967,12 @@ export function createDistiller(options: DistillerOptions): Distiller {
     try {
       const discovered = await discoverSessions();
       knownCount = discovered.length;
+      const liveUpdates = new Map(discovered.map((session) => [session.id, session.timeUpdated]));
+      for (const [sessionId, timeUpdated] of quarantined) {
+        // Deletions and updates both invalidate quarantine. An updated session is
+        // retried below; a deleted one no longer consumes retained state.
+        if (liveUpdates.get(sessionId) !== timeUpdated) quarantined.delete(sessionId);
+      }
       const parentById = new Map(discovered.map((s) => [s.id, s.parentId] as const));
       const sorted = [...discovered].sort(
         (a, b) => b.timeUpdated - a.timeUpdated || a.id.localeCompare(b.id),
@@ -924,12 +990,33 @@ export function createDistiller(options: DistillerOptions): Distiller {
         const upToDate =
           existing?.distillState === "full" && existing.timeUpdated === session.timeUpdated;
         if (!upToDate) {
-          const { card, rows } = deriveCard({
-            session,
-            messages: await fetchSessionMessages(session.id, caps.ftsRowsPerSession),
-            parentById,
-            caps,
-          });
+          if (quarantined.get(session.id) === session.timeUpdated) {
+            examined++;
+            recordProgress(session.timeUpdated);
+            return;
+          }
+
+          let messages: MsgWithParts[];
+          try {
+            messages = await fetchSessionMessages(session.id, caps.ftsRowsPerSession);
+          } catch (error) {
+            if (!(error instanceof MalformedSessionError)) throw error;
+            quarantine(session, error);
+            examined++;
+            recordProgress(session.timeUpdated);
+            return;
+          }
+
+          let card: Card;
+          let rows: PartTextRow[];
+          try {
+            ({ card, rows } = deriveCard({ session, messages, parentById, caps }));
+          } catch (error) {
+            quarantine(session, error);
+            examined++;
+            recordProgress(session.timeUpdated);
+            return;
+          }
           // The lease can drop (heartbeat takeover) during the fetch above; a
           // non-holder must never write. Re-check right before the write and
           // skip it, counting the session as not distilled.
@@ -966,11 +1053,37 @@ export function createDistiller(options: DistillerOptions): Distiller {
     }
   }
 
+  function quarantine(session: DistillSessionMeta, error: unknown): void {
+    // Reinsertion keeps FIFO eviction aligned with the latest failure.
+    quarantined.delete(session.id);
+    quarantined.set(session.id, session.timeUpdated);
+    while (quarantined.size > MAX_QUARANTINED_SESSIONS) {
+      const oldest = quarantined.keys().next().value;
+      if (oldest === undefined) break;
+      quarantined.delete(oldest);
+    }
+    logMsg(
+      `session ${session.id} quarantined at timeUpdated ${session.timeUpdated}: ${errmsg(error)}`,
+    );
+  }
+
+  function startColdPass(): Promise<void> {
+    if (stopped || !leaseHeld) return Promise.resolve();
+    if (coldPassPromise) return coldPassPromise;
+    const running = runColdPass().finally(() => {
+      if (coldPassPromise === running) coldPassPromise = undefined;
+    });
+    coldPassPromise = running;
+    return running;
+  }
+
   function scheduleColdPassRetry(): void {
+    if (stopped || !leaseHeld) return;
     clearTimer(coldPassRetryTimer);
     coldPassRetryTimer = setTimeout(() => {
+      coldPassRetryTimer = undefined;
       if (stopped || !leaseHeld) return;
-      if (coldPassState === "idle" && lastError !== undefined) void runColdPass();
+      if (coldPassState === "idle" && lastError !== undefined) void startColdPass();
     }, coldPassRetryMs);
   }
 
@@ -979,6 +1092,7 @@ export function createDistiller(options: DistillerOptions): Distiller {
   // (Stage 3). Resume correctness comes from the per-card skip above.
   let progressFloor: number | undefined;
   function recordProgress(timeUpdated: number): void {
+    if (stopped || !leaseHeld) return;
     if (progressFloor == null || timeUpdated < progressFloor) {
       progressFloor = timeUpdated;
       store.setMeta("coldpass_cursor", String(timeUpdated));
@@ -994,18 +1108,25 @@ export function createDistiller(options: DistillerOptions): Distiller {
       sessionID,
       setTimeout(() => {
         debounceTimers.delete(sessionID);
-        void runReDistill(sessionID);
+        startReDistill(sessionID);
       }, idleDebounceMs),
     );
   }
 
-  async function runReDistill(sessionID: string): Promise<void> {
+  function startReDistill(sessionID: string): void {
     if (stopped || !leaseHeld) return; // non-holders and stopped instances write nothing
     if (inFlight.has(sessionID)) {
       pendingRerun.add(sessionID); // coalesce: run once more after the in-flight pass
       return;
     }
-    inFlight.add(sessionID);
+    const running = runReDistill(sessionID).finally(() => {
+      inFlight.delete(sessionID);
+      if (!stopped && pendingRerun.delete(sessionID)) scheduleReDistill(sessionID);
+    });
+    inFlight.set(sessionID, running);
+  }
+
+  async function runReDistill(sessionID: string): Promise<void> {
     // Snapshot-and-clear the removal flag BEFORE any await: a removal event that
     // arrives mid-distill re-adds it independently, so the coalesced rerun still
     // forces a full re-distill instead of appending onto a compacted transcript.
@@ -1025,6 +1146,7 @@ export function createDistiller(options: DistillerOptions): Distiller {
           !hadRemoval;
         if (canAppend && existing) await distillAppend(session, existing);
         else await distillFull(session);
+        if (stopped || !leaseHeld) return;
         // Keep the root's family highlights current with its live children.
         const stored = store.getCard(sessionID);
         if (stored && stored.rootId !== sessionID) recomputeRootRollup(stored.rootId);
@@ -1033,14 +1155,10 @@ export function createDistiller(options: DistillerOptions): Distiller {
       }
     } catch (error) {
       lastError = errmsg(error);
+      logMsg(`session ${sessionID} re-distill failed: ${lastError}`);
     } finally {
       // If the distill did not land, preserve the removal signal for the retry.
       if (!succeeded && hadRemoval) removalSince.add(sessionID);
-      inFlight.delete(sessionID);
-      if (pendingRerun.has(sessionID)) {
-        pendingRerun.delete(sessionID);
-        scheduleReDistill(sessionID);
-      }
     }
   }
 
@@ -1055,6 +1173,7 @@ export function createDistiller(options: DistillerOptions): Distiller {
       debounceTimers.delete(sessionID);
     }
     removalSince.delete(sessionID);
+    quarantined.delete(sessionID);
     if (rootId && rootId !== sessionID) recomputeRootRollup(rootId);
   }
 
@@ -1067,7 +1186,8 @@ export function createDistiller(options: DistillerOptions): Distiller {
   function scheduleHeartbeat(): void {
     clearTimer(heartbeatTimer);
     heartbeatTimer = setTimeout(() => {
-      if (stopped || !leaseHeld) return;
+      heartbeatTimer = undefined;
+      if (finalized || !leaseHeld) return;
       if (store.heartbeatLease(instanceId)) scheduleHeartbeat();
       else {
         // Lost the lease (taken over): stop acting as holder and try to regain.
@@ -1080,14 +1200,18 @@ export function createDistiller(options: DistillerOptions): Distiller {
             ? `lease lost to ${taker.holder} (build ${taker.build || "?"}, gen ${taker.gen})`
             : "lease lost",
         );
-        scheduleLeaseRetry();
+        if (!stopped) scheduleLeaseRetry();
       }
     }, HEARTBEAT_MS);
   }
 
   function scheduleLeaseRetry(): void {
+    if (stopped || finalized) return;
     clearTimer(leaseRetryTimer);
-    leaseRetryTimer = setTimeout(acquire, leaseRetryMs);
+    leaseRetryTimer = setTimeout(() => {
+      leaseRetryTimer = undefined;
+      acquire();
+    }, leaseRetryMs);
   }
 
   function acquire(): void {
@@ -1096,29 +1220,61 @@ export function createDistiller(options: DistillerOptions): Distiller {
       leaseHeld = true;
       logMsg(`lease acquired (build ${build}, gen ${gen})`);
       scheduleHeartbeat();
-      if (limits.coldPass && coldPassState === "idle") void runColdPass();
+      if (limits.coldPass && coldPassState === "idle") void startColdPass();
     } else {
       leaseHeld = false;
       scheduleLeaseRetry();
     }
   }
 
+  function ownsLease(): boolean {
+    if (!leaseHeld || finalized) return false;
+    const current = store.leaseStatus();
+    const owned = current?.holder === instanceId && current.heartbeat + current.ttl > now();
+    if (!owned) {
+      leaseHeld = false;
+      clearTimer(heartbeatTimer);
+      heartbeatTimer = undefined;
+      if (!stopped) scheduleLeaseRetry();
+    }
+    return owned;
+  }
+
+  function quiesce(): void {
+    if (stopped) return;
+    stopped = true;
+    clearTimer(leaseRetryTimer);
+    clearTimer(coldPassRetryTimer);
+    leaseRetryTimer = undefined;
+    coldPassRetryTimer = undefined;
+    for (const timer of debounceTimers.values()) clearTimeout(timer);
+    debounceTimers.clear();
+    pendingRerun.clear();
+    // Deliberately retain heartbeatTimer: summary worker cleanup is lease-owned
+    // and plugin disposal finalizes this distiller only after that cleanup has
+    // settled or reached its shutdown bound.
+  }
+
   return {
     start(): void {
       if (stopped) return;
       acquire();
     },
 
-    stop(): void {
-      stopped = true;
+    quiesce,
+
+    ownsLease,
+
+    stop(): Promise<void> {
+      if (stopPromise) return stopPromise;
+      quiesce();
+      finalized = true;
       clearTimer(heartbeatTimer);
       clearTimer(leaseRetryTimer);
       clearTimer(coldPassRetryTimer);
       heartbeatTimer = undefined;
       leaseRetryTimer = undefined;
       coldPassRetryTimer = undefined;
-      for (const timer of debounceTimers.values()) clearTimeout(timer);
-      debounceTimers.clear();
       if (leaseHeld) {
         try {
           store.releaseLease(instanceId);
@@ -1127,6 +1283,11 @@ export function createDistiller(options: DistillerOptions): Distiller {
         }
       }
       leaseHeld = false;
+      const running = [coldPassPromise, ...inFlight.values()].filter(
+        (promise): promise is Promise<void> => promise !== undefined,
+      );
+      stopPromise = Promise.allSettled(running).then(() => {});
+      return stopPromise;
     },
 
     onEvent(event: Event): void {
@@ -1158,7 +1319,7 @@ export function createDistiller(options: DistillerOptions): Distiller {
     status(): DistillStatus {
       const lease = store.leaseStatus();
       return {
-        leaseHeld,
+        leaseHeld: ownsLease(),
         coldPass: coldPassState,
         distilledCount,
         knownCount,
diff --git a/src/opencode-session-recall.ts b/src/opencode-session-recall.ts
index 46f7b13..8f796cf 100644
--- a/src/opencode-session-recall.ts
+++ b/src/opencode-session-recall.ts
@@ -1,4 +1,4 @@
-import type { Plugin } from "@opencode-ai/plugin";
+import type { Plugin, ToolDefinition } from "@opencode-ai/plugin";
 import { createOpencodeClient, type Session } from "@opencode-ai/sdk/v2";
 import { sessions, type SessionEnrichment } from "./sessions.js";
 import { search, DISCOVERY_LIMIT, type SearchDeps, type SemanticSearchConfig } from "./search.js";
@@ -9,7 +9,7 @@ import { systemNudge } from "./hooks/system-nudge.js";
 import { autoRecall } from "./hooks/auto-recall.js";
 import { compactionRecall } from "./hooks/compaction-recall.js";
 import { createFetchGate } from "./fetch-gate.js";
-import { openSqlite } from "./sqlite.js";
+import { openSqlite, type SqliteDb } from "./sqlite.js";
 import {
   openStore,
   defaultStorePath,
@@ -26,6 +26,15 @@ import { createDistiller } from "./distill.js";
 import { createSummarizer, parseModelId, type Summarizer } from "./summarize.js";
 import { TOOLS, DEFAULTS, optionalString, errmsg, type Limits } from "./types.js";
 
+// `dispose` was added to the host/plugin contract in @opencode-ai/plugin 1.15.11.
+// Keep the source compatible with this repository's older tool-result typings
+// while declaring the exact minimum-host hook that the package engine requires.
+declare module "@opencode-ai/plugin" {
+  interface Hooks {
+    dispose?: () => Promise<void>;
+  }
+}
+
 /** Guarded, Node-free logger: `console` is a std global, but `src/` declares no
  *  types, so reach it defensively. */
 function pluginLog(message: string): void {
@@ -95,6 +104,18 @@ const server: Plugin = async (ctx, options) => {
   const nudge = opts.nudge !== false;
   const autoRecallEnabled = opts.autoRecall === true;
   const compactionRecallEnabled = opts.compactionRecall === true;
+  let disposed = false;
+  let disposePromise: Promise<void> | undefined;
+  const operations = new Set<Promise<unknown>>();
+
+  const track = <T>(operation: Promise<T>): Promise<T> => {
+    operations.add(operation);
+    void operation.then(
+      () => operations.delete(operation),
+      () => operations.delete(operation),
+    );
+    return operation;
+  };
 
   const clamp = (val: number | undefined, fallback: number, min = 1) =>
     Math.max(min, Math.floor(val ?? fallback));
@@ -196,8 +217,9 @@ const server: Plugin = async (ctx, options) => {
   // cards-lite built from the session list.
   const storePath = optionalString(opts.storePath) ?? (await defaultStorePath());
   let store: Store | null = null;
+  let db: SqliteDb | null = null;
   if (storePath) {
-    const db = await openSqlite(storePath);
+    db = await openSqlite(storePath);
     if (db) store = openStore(db);
   }
 
@@ -214,13 +236,16 @@ const server: Plugin = async (ctx, options) => {
       }
     : { getCards: () => liteCards, revision: () => undefined, degraded: true };
   if (!store) {
-    void discover()
-      .then((list) => {
-        liteCards = cardsLiteFromSessions(list as Parameters<typeof cardsLiteFromSessions>[0]);
-      })
-      .catch(() => {
-        // Best-effort; a failed list just leaves cards-lite empty until retried.
-      });
+    void track(
+      discover()
+        .then((list) => {
+          if (!disposed)
+            liteCards = cardsLiteFromSessions(list as Parameters<typeof cardsLiteFromSessions>[0]);
+        })
+        .catch(() => {
+          // Best-effort; a failed list just leaves cards-lite empty until retried.
+        }),
+    );
   }
 
   // One shared fetch gate gates every SDK call in the query/distill paths so the
@@ -255,9 +280,18 @@ const server: Plugin = async (ctx, options) => {
     gen: EMBED_REPRESENTATION,
     discover,
     onColdPassDone: () => {
-      void summarizer?.runColdPass();
+      if (!disposed && summarizer) {
+        // Summarizer.stop() owns the bounded shutdown of this drain. Do not add
+        // it to the plugin's general operation set, which is intentionally
+        // unbounded for DB-capable hooks and distiller work.
+        void summarizer
+          .runColdPass()
+          .catch((error) => pluginLog(`summarizer cold pass failed: ${errmsg(error)}`));
+      }
+    },
+    onSessionDistilled: (sessionId) => {
+      if (!disposed) summarizer?.queue(sessionId);
     },
-    onSessionDistilled: (sessionId) => summarizer?.queue(sessionId),
   });
   if (store && opts.summaries?.enabled === true) {
     const model = optionalString(opts.summaries.model);
@@ -279,7 +313,8 @@ const server: Plugin = async (ctx, options) => {
           ...(agent != null && { agent }),
           ...(maxPromptsPerPass != null && { maxPromptsPerPass }),
         },
-        leaseHeld: () => distiller.status().leaseHeld,
+        ownerToken: instanceId,
+        leaseHeld: () => distiller.ownsLease(),
         log: pluginLog,
       });
     } else {
@@ -298,32 +333,57 @@ const server: Plugin = async (ctx, options) => {
     ? { cards: () => store.allCards() }
     : undefined;
 
+  const guardTool = (definition: ToolDefinition): ToolDefinition => ({
+    ...definition,
+    execute: (args, context) => {
+      if (disposed) {
+        return Promise.reject(new Error("opencode-session-recall: plugin has been disposed"));
+      }
+      return track(definition.execute(args, context));
+    },
+  });
+
+  const guardHook =
+    <TArgs extends unknown[]>(
+      hook: (...args: TArgs) => Promise<void>,
+    ): ((...args: TArgs) => Promise<void>) =>
+    async (...args) => {
+      if (disposed) return;
+      await track(hook(...args));
+    };
+
+  const nudgeHook = nudge ? systemNudge() : undefined;
+  const autoRecallHook = autoRecallEnabled ? autoRecall(deps) : undefined;
+  const compactionHook = compactionRecallEnabled ? compactionRecall(deps) : undefined;
+
   return {
     tool: {
-      recall_sessions: sessions(client, unscoped, global, limits, enrichment),
-      recall: search(client, unscoped, global, limits, deps),
-      recall_get: get(client, gate),
-      recall_context: context(client, gate, limits),
-      recall_messages: messages(client, gate, limits),
+      recall_sessions: guardTool(sessions(client, unscoped, global, limits, enrichment)),
+      recall: guardTool(search(client, unscoped, global, limits, deps)),
+      recall_get: guardTool(get(client, gate)),
+      recall_context: guardTool(context(client, gate, limits)),
+      recall_messages: guardTool(messages(client, gate, limits)),
     },
     event: async ({ event }) => {
+      if (disposed) return;
       // The plugin `event` hook is typed against the default SDK vintage; the
       // distiller compiles against the v2 event union the live bus actually
       // delivers. Bridge the vintage gap at this one boundary.
       distiller.onEvent(event as unknown as Parameters<typeof distiller.onEvent>[0]);
     },
-    ...(nudge && {
-      "experimental.chat.system.transform": systemNudge(),
+    ...(nudgeHook && {
+      "experimental.chat.system.transform": guardHook(nudgeHook),
     }),
-    ...(autoRecallEnabled && {
-      "chat.message": autoRecall(deps),
+    ...(autoRecallHook && {
+      "chat.message": guardHook(autoRecallHook),
     }),
-    ...(compactionRecallEnabled && {
-      "experimental.session.compacting": compactionRecall(deps),
+    ...(compactionHook && {
+      "experimental.session.compacting": guardHook(compactionHook),
     }),
     ...(primary && {
       // eslint-disable-next-line @typescript-eslint/no-explicit-any -- opencode config type not exported
       config: async (c: any) => {
+        if (disposed) return;
         c.experimental ??= {};
         const existing: string[] = c.experimental.primary_tools ?? [];
         const deduped = new Set(existing);
@@ -331,6 +391,26 @@ const server: Plugin = async (ctx, options) => {
         c.experimental.primary_tools = [...deduped];
       },
     }),
+    dispose: () => {
+      if (disposePromise) return disposePromise;
+      disposed = true;
+      // Phase 1 is synchronous: no distill/retry/incremental work can start
+      // after dispose() returns its promise. The heartbeat deliberately remains
+      // active so lease-owned summarizer cleanup can finish safely.
+      distiller.quiesce();
+      cards.dispose();
+      disposePromise = (async () => {
+        // Summarizer.stop() is bounded. Its late SDK promises are detached and
+        // ownership/lease guarded, so phase 2 may safely release the lease once
+        // this settles even when an SDK request itself never does.
+        if (summarizer) await Promise.allSettled([summarizer.stop()]);
+        await distiller.stop();
+        await Promise.allSettled([...operations]);
+        db?.close();
+        db = null;
+      })();
+      return disposePromise;
+    },
   };
 };
 
diff --git a/src/summarize.ts b/src/summarize.ts
index be22a0e..bcf97e0 100644
--- a/src/summarize.ts
+++ b/src/summarize.ts
@@ -36,6 +36,10 @@ const DEFAULT_BATCH_SIZE = 15;
 const DEFAULT_MAX_PROMPTS_PER_PASS = 200;
 const DEFAULT_POLITENESS_MS = 250;
 const DEFAULT_PROMPT_TIMEOUT_MS = 60_000;
+/** Disposal may detach a stuck SDK request after this bound. The request itself
+ *  cannot be cancelled by the SDK; lease/title guards make its late settlement
+ *  incapable of touching SQLite or another instance's worker. */
+const DEFAULT_SHUTDOWN_TIMEOUT_MS = 15_000;
 const DEFAULT_IDLE_DEBOUNCE_MS = 3_000;
 /** Abort a drain after this many prompts fail in a row: a misconfigured model
  *  must not burn the whole per-pass budget. Latches until the next cold pass. */
@@ -92,6 +96,9 @@ export type SummarizerDeps = {
   store: Store;
   gate: FetchGate;
   config: SummariesConfig;
+  /** Unique to this plugin instance; embedded in worker titles and used to keep
+   *  normal cleanup scoped to workers this instance created. */
+  ownerToken: string;
   /** Only the distill-lease holder writes; checked before every persisted write. */
   leaseHeld: () => boolean;
   log?: (message: string) => void;
@@ -101,6 +108,7 @@ export type SummarizerDeps = {
   batchSize?: number;
   politenessMs?: number;
   promptTimeoutMs?: number;
+  shutdownTimeoutMs?: number;
   idleDebounceMs?: number;
 };
 
@@ -111,12 +119,34 @@ export type Summarizer = {
   /** Debounced incremental re-summarize of one session (the idle-debounce path);
    *  the drain skips it when the content hash is unchanged. */
   queue(sessionId: string): void;
-  stop(): void;
+  /** Stop accepting work and settle the active serialized drain. */
+  stop(): Promise<void>;
   status(): { summarized: number; lastError?: string };
 };
 
 type Timer = ReturnType<typeof setTimeout>;
 
+type TimedResult<T> = { timedOut: false; value: T } | { timedOut: true };
+
+async function settleWithin<T>(promise: Promise<T>, timeoutMs: number): Promise<TimedResult<T>> {
+  let timer: Timer | undefined;
+  const timeout = new Promise<TimedResult<T>>((resolve) => {
+    timer = setTimeout(() => resolve({ timedOut: true }), timeoutMs);
+  });
+  try {
+    return await Promise.race([
+      promise.then((value): TimedResult<T> => ({ timedOut: false, value })),
+      timeout,
+    ]);
+  } finally {
+    if (timer) clearTimeout(timer);
+  }
+}
+
+export function summarizerWorkerTitle(ownerToken: string): string {
+  return `${SUMMARIZER_SENTINEL} owner=${ownerToken}`;
+}
+
 // ── Text helpers ─────────────────────────────────────────────────────────────
 
 function cap(text: string, limit: number): string {
@@ -235,13 +265,14 @@ function replyText(parts: Part[]): string {
 // ── Summarizer ───────────────────────────────────────────────────────────────
 
 export function createSummarizer(deps: SummarizerDeps): Summarizer {
-  const { client, store, gate, config, leaseHeld } = deps;
+  const { client, store, gate, config, leaseHeld, ownerToken } = deps;
   const now = deps.now ?? Date.now;
   const log = deps.log;
   const rev = deps.rev ?? SUMMARY_REV;
   const batchSize = Math.max(1, deps.batchSize ?? DEFAULT_BATCH_SIZE);
   const politenessMs = Math.max(0, deps.politenessMs ?? DEFAULT_POLITENESS_MS);
   const promptTimeoutMs = Math.max(1, deps.promptTimeoutMs ?? DEFAULT_PROMPT_TIMEOUT_MS);
+  const shutdownTimeoutMs = Math.max(1, deps.shutdownTimeoutMs ?? DEFAULT_SHUTDOWN_TIMEOUT_MS);
   const idleDebounceMs = Math.max(0, deps.idleDebounceMs ?? DEFAULT_IDLE_DEBOUNCE_MS);
   const maxPromptsPerPass = Math.max(1, config.maxPromptsPerPass ?? DEFAULT_MAX_PROMPTS_PER_PASS);
 
@@ -263,24 +294,50 @@ export function createSummarizer(deps: SummarizerDeps): Summarizer {
   const inFlight = new Set<string>();
   const attempts = new Map<string, number>();
   let drainPromise: Promise<void> | null = null;
+  let stopPromise: Promise<void> | undefined;
   const debounceTimers = new Map<string, Timer>();
+  const ownedWorkers = new Set<string>();
+  const workerTitle = summarizerWorkerTitle(ownerToken);
 
   // ── Worker session lifecycle ──
   // A fresh worker per batch (create, prompt once, delete): create/delete are
   // unbilled and this keeps every batch's context clean with zero accumulation.
 
+  async function leaseSdk<T>(
+    label: string,
+    allowWhileStopped: boolean,
+    operation: () => Promise<T>,
+  ): Promise<T | undefined> {
+    if ((!allowWhileStopped && stopped) || !leaseHeld()) return undefined;
+    const result = await settleWithin(
+      gate.runBackground(() => {
+        // A gate permit may arrive after shutdown or an involuntary lease loss.
+        if ((!allowWhileStopped && stopped) || !leaseHeld()) return Promise.resolve(undefined);
+        return operation();
+      }),
+      shutdownTimeoutMs,
+    );
+    if (result.timedOut) {
+      logMsg(`${label} timed out after ${shutdownTimeoutMs}ms; late SDK settlement detached`);
+      return undefined;
+    }
+    return result.value;
+  }
+
   async function createWorker(): Promise<string | null> {
     try {
       // Probe the deny-all permission ruleset once; if the server rejects the
       // shape, remember that and create plainly thereafter (never let a rejected
       // ruleset silently disable summaries).
       if (permissionMode !== "without") {
-        const resp = await gate.runBackground(() =>
-          client.session.create({ title: SUMMARIZER_SENTINEL, permission: DENY_ALL_PERMISSION }),
+        const resp = await leaseSdk("worker create", false, () =>
+          client.session.create({ title: workerTitle, permission: DENY_ALL_PERMISSION }),
         );
+        if (!resp) return null;
         const created = resp.data as Session | undefined;
         if (!resp.error && created?.id) {
           permissionMode = "with";
+          ownedWorkers.add(created.id);
           return created.id;
         }
         if (permissionMode === undefined) {
@@ -288,10 +345,12 @@ export function createSummarizer(deps: SummarizerDeps): Summarizer {
           logMsg("worker permission ruleset rejected; relying on tool-disable + exclusion");
         }
       }
-      const resp = await gate.runBackground(() =>
-        client.session.create({ title: SUMMARIZER_SENTINEL }),
+      const resp = await leaseSdk("worker create", false, () =>
+        client.session.create({ title: workerTitle }),
       );
+      if (!resp) return null;
       const created = resp.data as Session | undefined;
+      if (created?.id) ownedWorkers.add(created.id);
       return created?.id ?? null;
     } catch (error) {
       lastError = errmsg(error);
@@ -299,17 +358,19 @@ export function createSummarizer(deps: SummarizerDeps): Summarizer {
     }
   }
 
-  async function deleteWorker(sessionID: string): Promise<void> {
+  async function deleteOwnedWorker(sessionID: string): Promise<void> {
+    if (!ownedWorkers.has(sessionID)) return;
     try {
-      await gate.runBackground(() => client.session.delete({ sessionID }));
+      await leaseSdk("worker delete", true, () => client.session.delete({ sessionID }));
     } catch {
       // Best-effort; a lingering sentinel session is excluded everywhere.
     }
   }
 
-  async function abortWorker(sessionID: string): Promise<void> {
+  async function abortOwnedWorker(sessionID: string): Promise<void> {
+    if (!ownedWorkers.has(sessionID)) return;
     try {
-      await gate.runBackground(() => client.session.abort({ sessionID }));
+      await leaseSdk("worker abort", true, () => client.session.abort({ sessionID }));
     } catch {
       // Best-effort.
     }
@@ -319,13 +380,20 @@ export function createSummarizer(deps: SummarizerDeps): Summarizer {
    *  than adopting one whose accumulated context is unknown. Runs once. */
   async function deleteOrphans(): Promise<void> {
     try {
-      const resp = await gate.runBackground(() =>
+      const resp = await leaseSdk("worker orphan list", false, () =>
         client.session.list({ search: SUMMARIZER_SENTINEL, limit: 100 }),
       );
+      if (!resp || stopped || !leaseHeld()) return;
       const rows = Array.isArray(resp.data) ? (resp.data as Session[]) : [];
       for (const row of rows) {
+        // Ownership can change while list/delete is in flight. Re-check after
+        // every await and immediately before each destructive request.
+        if (stopped || !leaseHeld()) return;
         if (isSummarizerTitle(row.title) && typeof row.id === "string" && row.id) {
-          await deleteWorker(row.id);
+          await leaseSdk("orphan worker delete", false, () =>
+            client.session.delete({ sessionID: row.id }),
+          );
+          if (stopped || !leaseHeld()) return;
         }
       }
     } catch {
@@ -342,6 +410,7 @@ export function createSummarizer(deps: SummarizerDeps): Summarizer {
     const workerId = await createWorker();
     if (!workerId) return new Map();
     try {
+      if (stopped || !leaseHeld()) return new Map();
       const { text, keyToSession } = renderBatch(cards);
       const reply = await promptWorker(workerId, text);
       if (reply == null) return new Map();
@@ -353,17 +422,15 @@ export function createSummarizer(deps: SummarizerDeps): Summarizer {
       }
       return out;
     } finally {
-      await deleteWorker(workerId);
+      // Only ids created by this owner token enter ownedWorkers; normal cleanup
+      // can therefore never target another instance's worker.
+      await deleteOwnedWorker(workerId);
     }
   }
 
   async function promptWorker(workerId: string, text: string): Promise<string | null> {
-    let timer: Timer | undefined;
-    const timeout = new Promise<"timeout">((resolve) => {
-      timer = setTimeout(() => resolve("timeout"), promptTimeoutMs);
-    });
     try {
-      const outcome = await Promise.race([
+      const outcome = await settleWithin(
         client.session.prompt({
           sessionID: workerId,
           model: { providerID: config.providerID, modelID: config.modelID },
@@ -374,25 +441,24 @@ export function createSummarizer(deps: SummarizerDeps): Summarizer {
           tools: DISABLE_ALL_TOOLS,
           parts: [{ type: "text", text }],
         }),
-        timeout,
-      ]);
-      if (outcome === "timeout") {
+        promptTimeoutMs,
+      );
+      if (outcome.timedOut) {
         // Stop the generation so the timeout caps SPEND, not just our waiting.
-        await abortWorker(workerId);
+        await abortOwnedWorker(workerId);
         lastError = "summary prompt timed out";
         return null;
       }
-      if (outcome.error || !outcome.data) {
-        lastError = outcome.error ? errmsg(outcome.error) : "empty prompt response";
+      const response = outcome.value;
+      if (response.error || !response.data) {
+        lastError = response.error ? errmsg(response.error) : "empty prompt response";
         return null;
       }
-      const data = outcome.data as { parts?: Part[] };
+      const data = response.data as { parts?: Part[] };
       return replyText(Array.isArray(data.parts) ? data.parts : []);
     } catch (error) {
       lastError = errmsg(error);
       return null;
-    } finally {
-      if (timer) clearTimeout(timer);
     }
   }
 
@@ -443,6 +509,7 @@ export function createSummarizer(deps: SummarizerDeps): Summarizer {
     if (!cleanedOrphans) {
       cleanedOrphans = true;
       await deleteOrphans();
+      if (stopped || !leaseHeld()) return;
     }
     let prompts = 0;
     while (
@@ -478,7 +545,7 @@ export function createSummarizer(deps: SummarizerDeps): Summarizer {
 
       prompts++;
       const summaries = await promptBatchFor(cards);
-      if (!leaseHeld() || stopped) {
+      if (stopped || !leaseHeld()) {
         for (const id of batchIds) inFlight.delete(id);
         break;
       }
@@ -547,10 +614,24 @@ export function createSummarizer(deps: SummarizerDeps): Summarizer {
       );
     },
 
-    stop(): void {
+    stop(): Promise<void> {
+      if (stopPromise) return stopPromise;
       stopped = true;
       for (const timer of debounceTimers.values()) clearTimeout(timer);
       debounceTimers.clear();
+      pending.length = 0;
+      pendingSet.clear();
+      const active = drainPromise;
+      stopPromise = active
+        ? settleWithin(active, shutdownTimeoutMs).then((result) => {
+            if (result.timedOut) {
+              logMsg(
+                `shutdown timed out after ${shutdownTimeoutMs}ms; late SDK work is lease-guarded and detached`,
+              );
+            }
+          })
+        : Promise.resolve();
+      return stopPromise;
     },
 
     status() {
diff --git a/test/cards.test.ts b/test/cards.test.ts
index 206da85..db9de3f 100644
--- a/test/cards.test.ts
+++ b/test/cards.test.ts
@@ -428,6 +428,43 @@ describe("cards semantic persistence", () => {
     db.close();
   });
 
+  it("cancels deferred semantic warm-up when disposed", async () => {
+    let ready = false;
+    let resolveReady!: () => void;
+    const readyPromise = new Promise<void>((resolve) => {
+      resolveReady = () => {
+        ready = true;
+        resolve();
+      };
+    });
+    const embed = vi.fn(() => Float32Array.from([1, 0]));
+    const embedder = {
+      get ready() {
+        return ready;
+      },
+      embed,
+    };
+    const { db, store } = await freshStore();
+    store.upsertCard(makeCard("s1", { inventory: `alpha topic ${SUBST}` }));
+    store.setMeta("cards_rev", "1");
+    const runtime = createCardsRuntime({
+      source: persistentSource(store),
+      embedder,
+      semanticWeight: 0.5,
+      semanticModel: "model-x",
+      semanticReady: readyPromise,
+    });
+    runtime.rank(parseQuery("alpha"), {}); // load the warm store before model readiness
+
+    runtime.dispose();
+    db.close();
+    resolveReady();
+    await readyPromise;
+    await Promise.resolve();
+
+    expect(embed).not.toHaveBeenCalled();
+  });
+
   it("reloads on a cross-process vectors_rev bump, but not on its own vector write", async () => {
     // Two handles on one store file (mixed-version skew): the runtime reads
     // through storeB; storeA stands in for another process. Vector writes do NOT
diff --git a/test/distill.test.ts b/test/distill.test.ts
index 02b5e6e..1ef1f84 100644
--- a/test/distill.test.ts
+++ b/test/distill.test.ts
@@ -1,4 +1,4 @@
-import { afterAll, describe, expect, it } from "vitest";
+import { afterAll, describe, expect, it, vi } from "vitest";
 import { mkdtempSync, rmSync } from "node:fs";
 import { tmpdir } from "node:os";
 import { join } from "node:path";
@@ -66,7 +66,11 @@ type SdkCalls = {
 
 function makeDistillFake(
   graph: Graph,
-  opts: { throwOnce?: Set<string> } = {},
+  opts: {
+    throwOnce?: Set<string>;
+    nonArray?: Set<string>;
+    metadataErrorOnce?: Set<string>;
+  } = {},
 ): { client: OpencodeClient; sdk: SdkCalls } {
   const sdk: SdkCalls = { list: 0, get: 0, messages: [] };
   const threw = new Set<string>();
@@ -78,6 +82,10 @@ function makeDistillFake(
       },
       get: async ({ sessionID }: { sessionID: string }) => {
         sdk.get++;
+        if (opts.metadataErrorOnce?.has(sessionID) && !threw.has(`meta:${sessionID}`)) {
+          threw.add(`meta:${sessionID}`);
+          return { error: apiFailure(`metadata transport failed: ${sessionID}`) };
+        }
         const found = graph.sessions.find((s) => s.id === sessionID);
         return found ? { data: found } : { error: apiFailure(`not found: ${sessionID}`) };
       },
@@ -91,6 +99,9 @@ function makeDistillFake(
           threw.add(params.sessionID);
           throw new Error(`throw once: ${params.sessionID}`);
         }
+        if (opts.nonArray?.has(params.sessionID)) {
+          return { data: { messages: "not-an-array" } };
+        }
         const data = graph.messagesBySession[params.sessionID] ?? [];
         const { items, nextCursor } = paginateBundles(
           data,
@@ -543,6 +554,322 @@ describe("fetchMessagePage", () => {
 // ── Cold pass ────────────────────────────────────────────────────────────────
 
 describe("cold pass", () => {
+  function malformedBundle(sessionId: string): MessageBundle {
+    return {
+      info: userMessage(`m-${sessionId}`, sessionId, 100),
+      parts: undefined,
+    } as unknown as MessageBundle;
+  }
+
+  it("quarantines a malformed session and continues processing the pass", async () => {
+    const bad = session("bad", "Bad", PROJECT_DIR, 3000);
+    const good = session("good", "Good", PROJECT_DIR, 2000);
+    const graph: Graph = {
+      sessions: [bad, good],
+      messagesBySession: {
+        bad: [malformedBundle("bad")],
+        good: [
+          bundle(userMessage("m-good", "good", 100), [
+            textPart("p-good", "good", "m-good", "valid session content"),
+          ]),
+        ],
+      },
+    };
+    const { client } = makeDistillFake(graph);
+    const { db, store } = await freshStore();
+    const { gate } = makeSpyGate();
+    const logs: string[] = [];
+    const distiller = createDistiller({
+      client,
+      store,
+      gate,
+      limits: { ...TEST_LIMITS, distillConcurrency: 1 },
+      instanceId: "malformed-continue",
+      log: (message) => logs.push(message),
+    });
+
+    distiller.start();
+    await waitFor(() => distiller.status().coldPass === "done");
+
+    expect(store.getCard("bad")).toBeUndefined();
+    expect(store.getCard("good")?.distillState).toBe("full");
+    expect(logs.some((line) => line.includes("session bad quarantined at timeUpdated 3000:"))).toBe(
+      true,
+    );
+    await distiller.stop();
+    db.close();
+  });
+
+  it("skips an unchanged quarantined session on a later cold-pass retry", async () => {
+    const graph: Graph = {
+      sessions: [
+        session("bad", "Bad", PROJECT_DIR, 3000),
+        session("flaky", "Flaky", PROJECT_DIR, 2000),
+        session("good", "Good", PROJECT_DIR, 1000),
+      ],
+      messagesBySession: {
+        bad: [malformedBundle("bad")],
+        flaky: [
+          bundle(userMessage("m-flaky", "flaky", 100), [
+            textPart("p-flaky", "flaky", "m-flaky", "flaky content"),
+          ]),
+        ],
+        good: [
+          bundle(userMessage("m-good", "good", 100), [
+            textPart("p-good", "good", "m-good", "good content"),
+          ]),
+        ],
+      },
+    };
+    const { client, sdk } = makeDistillFake(graph, { throwOnce: new Set(["flaky"]) });
+    const { db, store } = await freshStore();
+    const { gate } = makeSpyGate();
+    const distiller = createDistiller({
+      client,
+      store,
+      gate,
+      limits: { ...TEST_LIMITS, distillConcurrency: 1 },
+      instanceId: "malformed-skip",
+      coldPassRetryMs: 15,
+    });
+
+    distiller.start();
+    await waitFor(() => distiller.status().coldPass === "done");
+
+    expect(sdk.list).toBe(2);
+    expect(sdk.messages.filter((call) => call.sessionID === "bad")).toHaveLength(1);
+    expect(store.getCard("good")?.distillState).toBe("full");
+    await distiller.stop();
+    db.close();
+  });
+
+  it("preserves an existing card and FTS rows for a non-array message payload", async () => {
+    const current = session("bad", "Bad", PROJECT_DIR, 2000);
+    const old = session("bad", "Bad", PROJECT_DIR, 1000);
+    const oldMessages = [
+      bundle(userMessage("m-old", "bad", 100), [
+        textPart("p-old", "bad", "m-old", "preserved_fts_marker"),
+      ]),
+    ];
+    const graph: Graph = { sessions: [current], messagesBySession: { bad: oldMessages } };
+    const { client } = makeDistillFake(graph, { nonArray: new Set(["bad"]) });
+    const { db, store } = await freshStore();
+    const seeded = deriveCard({
+      session: metaFromSession(old),
+      messages: oldMessages,
+      parentById: new Map([["bad", null]]),
+    });
+    store.replaceSessionParts("bad", seeded.rows, seeded.card);
+    const { gate } = makeSpyGate();
+    const distiller = createDistiller({
+      client,
+      store,
+      gate,
+      limits: { ...TEST_LIMITS, distillConcurrency: 1 },
+      instanceId: "non-array-preserve",
+    });
+
+    distiller.start();
+    await waitFor(() => distiller.status().coldPass === "done");
+
+    expect(store.getCard("bad")).toEqual(seeded.card);
+    expect(ftsSessions(store.ftsSearch({ strong: ["preserved_fts_marker"], weak: [] }))).toEqual([
+      "bad",
+    ]);
+    await distiller.stop();
+    db.close();
+  });
+
+  it("does not arm a cold-pass retry when discovery fails after stop", async () => {
+    vi.useFakeTimers();
+    const { db, store } = await freshStore();
+    try {
+      let rejectDiscovery!: (error: Error) => void;
+      const discovery = new Promise<Session[]>((_, reject) => {
+        rejectDiscovery = reject;
+      });
+      const discover = vi.fn(() => discovery);
+      const { client } = makeDistillFake({ sessions: [], messagesBySession: {} });
+      const { gate } = makeSpyGate();
+      const distiller = createDistiller({
+        client,
+        store,
+        gate,
+        limits: TEST_LIMITS,
+        instanceId: "late-discovery-failure",
+        coldPassRetryMs: 100,
+        discover,
+      });
+
+      distiller.start();
+      await Promise.resolve();
+      await Promise.resolve();
+      expect(discover).toHaveBeenCalledOnce();
+      const stopping = distiller.stop();
+      rejectDiscovery(new Error("late discovery failure"));
+      await stopping;
+
+      expect(vi.getTimerCount()).toBe(0);
+      expect(discover).toHaveBeenCalledOnce();
+    } finally {
+      db.close();
+      vi.useRealTimers();
+    }
+  });
+
+  it("does not fetch another message page or write progress after stop during a delay", async () => {
+    vi.useFakeTimers();
+    const { db, store } = await freshStore();
+    try {
+      const meta = session("s1", "Paged", PROJECT_DIR, 2000);
+      let messageCalls = 0;
+      const first = bundle(userMessage("m1", "s1", 100), [
+        textPart("p1", "s1", "m1", "first page"),
+      ]);
+      const client = {
+        session: {
+          messages: async () => {
+            messageCalls++;
+            return messagesResponse(
+              messageCalls === 1 ? [first] : [],
+              messageCalls === 1 ? "next" : null,
+            );
+          },
+        },
+      } as unknown as OpencodeClient;
+      const { gate } = makeSpyGate();
+      const distiller = createDistiller({
+        client,
+        store,
+        gate,
+        limits: { ...TEST_LIMITS, distillDelayMs: 100 },
+        instanceId: "stop-pagination",
+        discover: async () => [meta],
+      });
+
+      distiller.start();
+      for (let i = 0; i < 8 && messageCalls === 0; i++) await Promise.resolve();
+      expect(messageCalls).toBe(1);
+      const stopping = distiller.stop();
+      await vi.advanceTimersByTimeAsync(100);
+      await stopping;
+
+      expect(messageCalls).toBe(1);
+      expect(store.getCard("s1")).toBeUndefined();
+      expect(store.getMeta("coldpass_cursor")).toBeUndefined();
+    } finally {
+      db.close();
+      vi.useRealTimers();
+    }
+  });
+
+  it("does not start a message request that was queued in the gate when stopped", async () => {
+    const { db, store } = await freshStore();
+    try {
+      const meta = session("s1", "Queued", PROJECT_DIR, 2000);
+      let messageCalls = 0;
+      let releaseQuery!: () => void;
+      const queryBlocked = new Promise<void>((resolve) => {
+        releaseQuery = resolve;
+      });
+      let queryStarted!: () => void;
+      const queryIsRunning = new Promise<void>((resolve) => {
+        queryStarted = resolve;
+      });
+      let messageQueued!: () => void;
+      const messageIsQueued = new Promise<void>((resolve) => {
+        messageQueued = resolve;
+      });
+      const realGate = createFetchGate({ concurrency: 1 });
+      let backgroundCalls = 0;
+      const gate: FetchGate = {
+        runQuery: (fn) => realGate.runQuery(fn),
+        runBackground: (fn) => {
+          backgroundCalls++;
+          if (backgroundCalls === 2) messageQueued();
+          return realGate.runBackground(fn);
+        },
+        activeQueries: () => realGate.activeQueries(),
+      };
+      const client = {
+        session: {
+          messages: async () => {
+            messageCalls++;
+            return messagesResponse([], null);
+          },
+        },
+      } as unknown as OpencodeClient;
+      const distiller = createDistiller({
+        client,
+        store,
+        gate,
+        limits: TEST_LIMITS,
+        instanceId: "stop-gate-queue",
+        discover: async () => {
+          void gate.runQuery(async () => {
+            queryStarted();
+            await queryBlocked;
+          });
+          return [meta];
+        },
+      });
+
+      distiller.start();
+      await queryIsRunning;
+      await messageIsQueued;
+      const stopping = distiller.stop();
+      releaseQuery();
+      await stopping;
+
+      expect(messageCalls).toBe(0);
+      expect(store.getCard("s1")).toBeUndefined();
+      expect(store.getMeta("coldpass_cursor")).toBeUndefined();
+    } finally {
+      db.close();
+    }
+  });
+
+  it("retries a quarantined session after its update timestamp changes", async () => {
+    const bad = session("bad", "Bad", PROJECT_DIR, 3000);
+    const graph: Graph = {
+      sessions: [bad, session("flaky", "Flaky", PROJECT_DIR, 2000)],
+      messagesBySession: {
+        bad: [malformedBundle("bad")],
+        flaky: [
+          bundle(userMessage("m-flaky", "flaky", 100), [
+            textPart("p-flaky", "flaky", "m-flaky", "flaky content"),
+          ]),
+        ],
+      },
+    };
+    const { client, sdk } = makeDistillFake(graph, { throwOnce: new Set(["flaky"]) });
+    const { db, store } = await freshStore();
+    const { gate } = makeSpyGate();
+    const distiller = createDistiller({
+      client,
+      store,
+      gate,
+      limits: { ...TEST_LIMITS, distillConcurrency: 1 },
+      instanceId: "malformed-update",
+      coldPassRetryMs: 100,
+    });
+
+    distiller.start();
+    await waitFor(() => distiller.status().lastError != null);
+    bad.time.updated = 4000;
+    graph.messagesBySession.bad = [
+      bundle(userMessage("m-bad-fixed", "bad", 100), [
+        textPart("p-bad-fixed", "bad", "m-bad-fixed", "repaired content"),
+      ]),
+    ];
+    await waitFor(() => distiller.status().coldPass === "done");
+
+    expect(sdk.messages.filter((call) => call.sessionID === "bad")).toHaveLength(2);
+    expect(store.getCard("bad")?.timeUpdated).toBe(4000);
+    await distiller.stop();
+    db.close();
+  });
+
   it("distills newest-updated-first, skips up-to-date cards, routes every fetch through the gate", async () => {
     const s1 = session("s1", "Alpha", PROJECT_DIR, 3000);
     const s2 = session("s2", "Bravo", PROJECT_DIR, 2000);
@@ -922,6 +1249,61 @@ describe("lease", () => {
 // ── Incremental ──────────────────────────────────────────────────────────────
 
 describe("incremental", () => {
+  it("logs metadata transport errors without changing a valid card and retries on a later event", async () => {
+    const original = session("s1", "Alpha", PROJECT_DIR, 3000);
+    const graph: Graph = {
+      sessions: [original],
+      messagesBySession: {
+        s1: [
+          bundle(userMessage("m1", "s1", 100), [
+            textPart("p1", "s1", "m1", "preserved metadata transport marker"),
+          ]),
+        ],
+      },
+    };
+    const { client, sdk } = makeDistillFake(graph, {
+      metadataErrorOnce: new Set(["s1"]),
+    });
+    const { db, store } = await freshStore();
+    const { gate } = makeSpyGate();
+    const logs: string[] = [];
+    const distiller = createDistiller({
+      client,
+      store,
+      gate,
+      limits: TEST_LIMITS,
+      instanceId: "metadata-transport",
+      idleDebounceMs: 5,
+      log: (message) => logs.push(message),
+    });
+    distiller.start();
+    await waitFor(() => distiller.status().coldPass === "done");
+    const preserved = store.getCard("s1");
+
+    graph.sessions[0] = session("s1", "Alpha updated", PROJECT_DIR, 4000);
+    graph.messagesBySession.s1 = [
+      bundle(userMessage("m2", "s1", 200), [
+        textPart("p2", "s1", "m2", "fresh metadata transport marker"),
+      ]),
+    ];
+    distiller.onEvent(idleEvent("s1"));
+    await waitFor(() => sdk.get === 1);
+    await new Promise((resolve) => setTimeout(resolve, 10));
+
+    expect(store.getCard("s1")).toEqual(preserved);
+    expect(logs.some((line) => line.includes("session s1 re-distill failed:"))).toBe(true);
+    expect(logs.some((line) => line.includes("metadata transport failed: s1"))).toBe(true);
+
+    // A later idle event retries normally; the transport failure was neither
+    // interpreted as deletion nor quarantined as malformed session data.
+    distiller.onEvent(idleEvent("s1"));
+    await waitFor(() => store.getCard("s1")?.timeUpdated === 4000);
+    expect(sdk.get).toBe(2);
+    expect(store.getCard("s1")?.summaryHead).toContain("fresh metadata transport marker");
+    await distiller.stop();
+    db.close();
+  });
+
   it("coalesces an idle burst into a single re-distill", async () => {
     const graph: Graph = {
       sessions: [session("s1", "Alpha", PROJECT_DIR, 3000)],
diff --git a/test/plugin.test.ts b/test/plugin.test.ts
index 9b5660b..e5fdf18 100644
--- a/test/plugin.test.ts
+++ b/test/plugin.test.ts
@@ -1,12 +1,47 @@
-import { beforeEach, describe, expect, it, vi } from "vitest";
-import { tool, type PluginInput, type ToolDefinition } from "@opencode-ai/plugin";
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+import { tool, type Hooks, type PluginInput, type ToolDefinition } from "@opencode-ai/plugin";
+import { mkdtempSync, rmSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
 import { TOOLS } from "../src/types.js";
-import { PROJECT_DIR } from "./helpers.js";
+import { openSqlite } from "../src/sqlite.js";
+import { openStore } from "../src/store.js";
+import { bundle, PROJECT_DIR, textPart, userMessage } from "./helpers.js";
 
 const createOpencodeClient = vi.hoisted(() => vi.fn((options: unknown) => options));
+const sqliteLifecycle = vi.hoisted(() => ({ closes: 0, postCloseCalls: 0 }));
 
 vi.mock("@opencode-ai/sdk/v2", () => ({ createOpencodeClient }));
 
+vi.mock("../src/sqlite.js", async (importOriginal) => {
+  const actual = await importOriginal<typeof import("../src/sqlite.js")>();
+  return {
+    ...actual,
+    openSqlite: async (...args: Parameters<typeof actual.openSqlite>) => {
+      const db = await actual.openSqlite(...args);
+      if (!db) return db;
+      let closed = false;
+      return new Proxy(db, {
+        get(target, property, receiver) {
+          const value = Reflect.get(target, property, receiver);
+          if (typeof value !== "function") return value;
+          if (property === "close") {
+            return () => {
+              sqliteLifecycle.closes++;
+              closed = true;
+              return value.call(target);
+            };
+          }
+          return (...methodArgs: unknown[]) => {
+            if (closed) sqliteLifecycle.postCloseCalls++;
+            return value.apply(target, methodArgs);
+          };
+        },
+      });
+    },
+  };
+});
+
 // Mock the semantic embedder so the wiring test never downloads a model or
 // touches the network: init resolves immediately and the model stays unready.
 vi.mock("../src/semantic/embedder.js", () => ({
@@ -22,12 +57,19 @@ vi.mock("../src/semantic/embedder.js", () => ({
 }));
 
 const plugin = await import("../src/opencode-session-recall.js");
+const activeHooks: Hooks[] = [];
 
 // Every entry call opens a card store and (with coldPass) starts a background
 // distiller. Default tests use an in-memory store with the cold pass off so they
 // exercise only wiring, with no filesystem side effect or leaked timers.
-function server(input: PluginInput, opts: Record<string, unknown> = {}) {
-  return plugin.default.server(input, { storePath: ":memory:", coldPass: false, ...opts });
+async function server(input: PluginInput, opts: Record<string, unknown> = {}) {
+  const hooks = await plugin.default.server(input, {
+    storePath: ":memory:",
+    coldPass: false,
+    ...opts,
+  });
+  activeHooks.push(hooks);
+  return hooks;
 }
 
 function mustTool(definition: ToolDefinition | undefined): ToolDefinition {
@@ -66,6 +108,12 @@ function ctx(config: {
 describe("plugin entry", () => {
   beforeEach(() => {
     createOpencodeClient.mockClear();
+    sqliteLifecycle.closes = 0;
+    sqliteLifecycle.postCloseCalls = 0;
+  });
+
+  afterEach(async () => {
+    await Promise.all(activeHooks.splice(0).map((hooks) => hooks.dispose?.()));
   });
 
   it("registers all tools and strips project scoping only from the unscoped client", async () => {
@@ -214,6 +262,281 @@ describe("plugin entry", () => {
     expect(() => sessionsArgs.parse({ limit: 5 })).toThrow();
   });
 
+  it("waits for an in-flight cold-pass fetch before closing SQLite", async () => {
+    let resolveMessages: ((value: { data: [] }) => void) | undefined;
+    const messages = vi.fn(
+      () =>
+        new Promise<{ data: [] }>((resolve) => {
+          resolveMessages = resolve;
+        }),
+    );
+    createOpencodeClient
+      .mockImplementationOnce((options: unknown) => ({
+        ...(options as object),
+        session: { messages, list: vi.fn(async () => ({ data: [] })) },
+      }))
+      .mockImplementationOnce((options: unknown) => ({
+        ...(options as object),
+        experimental: {
+          session: {
+            list: vi.fn(async () => ({
+              data: [
+                {
+                  id: "s1",
+                  title: "T",
+                  directory: PROJECT_DIR,
+                  time: { created: 1, updated: 2 },
+                },
+              ],
+            })),
+          },
+        },
+      }));
+    const hooks = await server(ctx({ fetch: vi.fn() }), { coldPass: true });
+    await vi.waitFor(() => expect(messages).toHaveBeenCalled());
+
+    let disposed = false;
+    const stopping = hooks.dispose?.().then(() => {
+      disposed = true;
+    });
+    await Promise.resolve();
+    expect(disposed).toBe(false);
+    expect(sqliteLifecycle.closes).toBe(0);
+
+    resolveMessages?.({ data: [] });
+    await stopping;
+    expect(sqliteLifecycle.closes).toBe(1);
+    expect(sqliteLifecycle.postCloseCalls).toBe(0);
+  });
+
+  it("disposes idempotently and rejects tools without touching SQLite afterwards", async () => {
+    const hooks = await server(ctx({ fetch: vi.fn() }), {});
+
+    const first = hooks.dispose?.();
+    const second = hooks.dispose?.();
+    expect(second).toBe(first);
+    await first;
+    expect(sqliteLifecycle.closes).toBe(1);
+
+    await hooks.event?.({
+      event: { type: "session.idle", properties: { sessionID: "s" } },
+    } as never);
+    await expect(mustTool(hooks.tool?.recall_sessions).execute({}, {} as never)).rejects.toThrow(
+      "opencode-session-recall: plugin has been disposed",
+    );
+    expect(sqliteLifecycle.closes).toBe(1);
+    expect(sqliteLifecycle.postCloseCalls).toBe(0);
+  });
+
+  it("cancels scheduler timers before closing SQLite", async () => {
+    vi.useFakeTimers();
+    try {
+      createOpencodeClient
+        .mockImplementationOnce((options: unknown) => ({
+          ...(options as object),
+          session: { messages: vi.fn(async () => ({ data: [] })) },
+        }))
+        .mockImplementationOnce((options: unknown) => ({
+          ...(options as object),
+          experimental: { session: { list: vi.fn(async () => ({ data: [] })) } },
+        }));
+      const hooks = await server(ctx({ fetch: vi.fn() }), { coldPass: true });
+
+      await hooks.dispose?.();
+      expect(sqliteLifecycle.closes).toBe(1);
+      await vi.advanceTimersByTimeAsync(20_000);
+
+      expect(sqliteLifecycle.closes).toBe(1);
+      expect(sqliteLifecycle.postCloseCalls).toBe(0);
+    } finally {
+      vi.useRealTimers();
+    }
+  });
+
+  it("holds the distill lease until active summarizer cleanup finishes", async () => {
+    const dir = mkdtempSync(join(tmpdir(), "recall-plugin-handoff-"));
+    const storePath = join(dir, "recall.sqlite");
+    let resolvePrompt: ((value: unknown) => void) | undefined;
+    const prompt = vi.fn(
+      () =>
+        new Promise<unknown>((resolve) => {
+          resolvePrompt = resolve;
+        }),
+    );
+    const deleteWorker = vi.fn(async () => ({ data: true }));
+    const messages = vi.fn(async () => ({
+      data: [
+        bundle(userMessage("m1", "s1", 1), [
+          textPart("p1", "s1", "m1", "Summarize this lease handoff session."),
+        ]),
+      ],
+    }));
+    createOpencodeClient
+      .mockImplementationOnce((options: unknown) => ({
+        ...(options as object),
+        session: {
+          messages,
+          list: vi.fn(async () => ({ data: [] })),
+          create: vi.fn(async () => ({ data: { id: "worker-1" } })),
+          prompt,
+          delete: deleteWorker,
+          abort: vi.fn(async () => ({ data: true })),
+        },
+      }))
+      .mockImplementationOnce((options: unknown) => ({
+        ...(options as object),
+        experimental: {
+          session: {
+            list: vi.fn(async () => ({
+              data: [
+                {
+                  id: "s1",
+                  title: "Lease handoff",
+                  directory: PROJECT_DIR,
+                  time: { created: 1, updated: 2 },
+                },
+              ],
+            })),
+          },
+        },
+      }));
+
+    let rivalDb: Awaited<ReturnType<typeof openSqlite>> = null;
+    try {
+      const hooks = await server(ctx({ fetch: vi.fn() }), {
+        storePath,
+        coldPass: true,
+        summaries: { enabled: true, model: "test/cheap" },
+      });
+      await vi.waitFor(() => expect(prompt).toHaveBeenCalledOnce());
+
+      const stopping = hooks.dispose?.();
+      await Promise.resolve();
+      rivalDb = await openSqlite(storePath);
+      if (!rivalDb) throw new Error("failed to open rival store");
+      const rival = openStore(rivalDb);
+      if (!rival) throw new Error("failed to initialize rival store");
+      expect(rival.acquireLease("rival", 60_000, "test", 1)).toBe(false);
+
+      resolvePrompt?.({ data: { info: {}, parts: [{ type: "text", text: "[]" }] } });
+      await stopping;
+      expect(deleteWorker).toHaveBeenCalledWith({ sessionID: "worker-1" });
+      expect(rival.acquireLease("rival", 60_000, "test", 1)).toBe(true);
+    } finally {
+      rivalDb?.close();
+      rmSync(dir, { recursive: true, force: true });
+    }
+  });
+
+  it("quiesces immediately but renews the lease only until blocked cleanup times out", async () => {
+    vi.useFakeTimers({ now: 1_000_000 });
+    const dir = mkdtempSync(join(tmpdir(), "recall-plugin-bounded-cleanup-"));
+    const storePath = join(dir, "recall.db");
+    let resolveDelete: ((value: { data: true }) => void) | undefined;
+    try {
+      const coldSession = {
+        id: "s1",
+        title: "Cold session",
+        slug: "cold-session",
+        directory: PROJECT_DIR,
+        projectID: "p",
+        time: { created: 1000, updated: 2000 },
+      };
+      const getSession = vi.fn(async () => ({ data: coldSession }));
+      const messages = vi.fn(async () => ({
+        data: [
+          bundle(userMessage("m1", "s1", 1100), [
+            textPart("p1", "s1", "m1", "bounded cleanup source"),
+          ]),
+        ],
+      }));
+      const deleteWorker = vi.fn(
+        () =>
+          new Promise<{ data: true }>((resolve) => {
+            resolveDelete = resolve;
+          }),
+      );
+      const scopedClient = {
+        session: {
+          get: getSession,
+          messages,
+          list: vi.fn(async () => ({ data: [] })),
+          create: vi.fn(async () => ({ data: { id: "blocked-worker" } })),
+          prompt: vi.fn(async () => ({
+            data: {
+              info: { role: "assistant" },
+              parts: [{ type: "text", text: "[]" }],
+            },
+          })),
+          delete: deleteWorker,
+          abort: vi.fn(async () => ({ data: true })),
+        },
+      };
+      const discover = vi.fn(async () => ({ data: [coldSession] }));
+      const unscopedClient = { experimental: { session: { list: discover } } };
+      createOpencodeClient.mockReturnValueOnce(scopedClient).mockReturnValueOnce(unscopedClient);
+
+      const hooks = await server(ctx({ fetch: vi.fn() }), {
+        storePath,
+        coldPass: true,
+        summaries: { enabled: true, model: "test/cheap" },
+      });
+      for (let i = 0; i < 100 && deleteWorker.mock.calls.length === 0; i++) {
+        await Promise.resolve();
+      }
+      expect(deleteWorker).toHaveBeenCalledWith({ sessionID: "blocked-worker" });
+
+      // Queue incremental work immediately before disposal. Quiescence must
+      // cancel it synchronously even though summarizer cleanup is still blocked.
+      await hooks.event?.({
+        event: { type: "session.idle", properties: { sessionID: "s1" } },
+      } as Parameters<NonNullable<Hooks["event"]>>[0]);
+      let disposed = false;
+      const stopping = hooks.dispose?.().then(() => {
+        disposed = true;
+      });
+      expect(stopping).toBeDefined();
+
+      const rivalDb = await openSqlite(storePath);
+      if (!rivalDb) throw new Error("openSqlite returned null for rival");
+      const rival = openStore(rivalDb);
+      if (!rival) throw new Error("openStore returned null for rival");
+      const initialLease = rival.leaseStatus();
+      expect(initialLease).toBeDefined();
+      expect(rival.acquireLease("rival", 60_000, "test", 1)).toBe(false);
+
+      await vi.advanceTimersByTimeAsync(10_000);
+      const renewedLease = rival.leaseStatus();
+      expect(renewedLease?.heartbeat).toBeGreaterThan(initialLease?.heartbeat ?? 0);
+      expect(disposed).toBe(false);
+      expect(rival.acquireLease("rival", 60_000, "test", 1)).toBe(false);
+      expect(discover).toHaveBeenCalledTimes(1);
+      expect(messages).toHaveBeenCalledTimes(1);
+      expect(getSession).not.toHaveBeenCalled();
+
+      // Summarizer shutdown is bounded. Once its timeout expires, distiller
+      // finalization releases the lease and disposal closes the old DB.
+      await vi.advanceTimersByTimeAsync(5_001);
+      await stopping;
+      expect(disposed).toBe(true);
+      expect(rival.acquireLease("rival", 60_000, "test", 1)).toBe(true);
+      expect(discover).toHaveBeenCalledTimes(1);
+      expect(messages).toHaveBeenCalledTimes(1);
+      expect(getSession).not.toHaveBeenCalled();
+      expect(sqliteLifecycle.postCloseCalls).toBe(0);
+
+      // The SDK promise itself cannot be cancelled. Its late settlement only
+      // releases the fetch-gate permit and cannot touch SQLite or another worker.
+      resolveDelete?.({ data: true });
+      await Promise.resolve();
+      expect(sqliteLifecycle.postCloseCalls).toBe(0);
+      rivalDb.close();
+    } finally {
+      vi.useRealTimers();
+      rmSync(dir, { recursive: true, force: true });
+    }
+  });
+
   it("fails clearly if SDK internals needed for transport extraction change", async () => {
     await expect(server({ client: {} } as unknown as PluginInput, {})).rejects.toThrow(
       "SDK internals changed",
diff --git a/test/summarize.test.ts b/test/summarize.test.ts
index 9f9009a..e6d594e 100644
--- a/test/summarize.test.ts
+++ b/test/summarize.test.ts
@@ -1,4 +1,4 @@
-import { afterAll, describe, expect, it } from "vitest";
+import { afterAll, describe, expect, it, vi } from "vitest";
 import { mkdtempSync, rmSync } from "node:fs";
 import { tmpdir } from "node:os";
 import { join } from "node:path";
@@ -16,6 +16,7 @@ import {
   createSummarizer,
   parseModelId,
   parseSummaryReply,
+  summarizerWorkerTitle,
   type Summarizer,
 } from "../src/summarize.js";
 import {
@@ -119,6 +120,7 @@ function makeSummarizer(
     store,
     gate,
     config: { providerID: "test", modelID: "cheap" },
+    ownerToken: "test-owner",
     leaseHeld: () => true,
     politenessMs: 0,
     idleDebounceMs: 0,
@@ -306,6 +308,11 @@ describe("summarizer worker lifecycle", () => {
 
     expect(client.calls.prompts.length).toBe(3);
     expect(client.calls.creates.length).toBe(3); // one worker per batch
+    expect(client.calls.creates.map((call) => call.title)).toEqual([
+      summarizerWorkerTitle("test-owner"),
+      summarizerWorkerTitle("test-owner"),
+      summarizerWorkerTitle("test-owner"),
+    ]);
     expect(client.calls.deletes.length).toBe(3); // each disposed
     expect(client.liveWorkers()).toHaveLength(0);
   });
@@ -325,6 +332,46 @@ describe("summarizer worker lifecycle", () => {
     expect(client.liveWorkers()).toHaveLength(0); // orphans + this batch's worker all gone
   });
 
+  it("does not delete a new holder's worker when an orphan list returns after lease loss", async () => {
+    const store = await freshStore();
+    store.upsertCard(fullCard("c1"));
+    const gate = createFetchGate({ concurrency: 2 });
+    let leaseHeld = true;
+    let resolveList!: (value: { data: unknown[] }) => void;
+    const listResult = new Promise<{ data: unknown[] }>((resolve) => {
+      resolveList = resolve;
+    });
+    const deleteWorker = vi.fn(async () => ({ data: true }));
+    const createWorker = vi.fn(async () => ({ data: { id: "old-worker" } }));
+    const listWorkers = vi.fn(async () => listResult);
+    const client = {
+      session: {
+        list: listWorkers,
+        delete: deleteWorker,
+        create: createWorker,
+      },
+    } as unknown as OpencodeClient;
+    const summarizer = makeSummarizer(store, client, gate, {
+      ownerToken: "old-owner",
+      leaseHeld: () => leaseHeld,
+      shutdownTimeoutMs: 100,
+    });
+
+    const run = summarizer.runColdPass();
+    while (listWorkers.mock.calls.length === 0) {
+      await Promise.resolve();
+    }
+    leaseHeld = false;
+    resolveList({
+      data: [session("new-worker", summarizerWorkerTitle("new-owner"), PROJECT_DIR, 4000)],
+    });
+    await run;
+
+    expect(deleteWorker).not.toHaveBeenCalled();
+    expect(createWorker).not.toHaveBeenCalled();
+    await summarizer.stop();
+  });
+
   it("disables tools and applies a deny-all permission on the worker prompt", async () => {
     const store = await freshStore();
     store.upsertCard(fullCard("c1"));
@@ -365,6 +412,28 @@ describe("summarizer worker lifecycle", () => {
 });
 
 describe("summarizer drain (budget, latch, gating)", () => {
+  it("waits for an active prompt to settle when stopped", async () => {
+    const store = await freshStore();
+    store.upsertCard(fullCard("c1"));
+    const gate = createFetchGate({ concurrency: 2 });
+    const client = makeSummarizerClient(() => ({ text: "[]", delayMs: 30 }));
+    const summarizer = makeSummarizer(store, client.client, gate);
+
+    const run = summarizer.runColdPass();
+    while (client.calls.prompts.length === 0)
+      await new Promise((resolve) => setTimeout(resolve, 1));
+    let stopped = false;
+    const stopping = summarizer.stop().then(() => {
+      stopped = true;
+    });
+    await Promise.resolve();
+
+    expect(stopped).toBe(false);
+    await Promise.all([run, stopping]);
+    expect(stopped).toBe(true);
+    expect(store.getCard("c1")?.nlSummary).toBe("");
+  });
+
   it("summarizes every needing card, then skips them on the content-hash gate", async () => {
     const store = await freshStore();
     store.upsertCard(fullCard("c1"));
diff --git a/test/tools.test.ts b/test/tools.test.ts
index c993d9d..d7ef17f 100644
--- a/test/tools.test.ts
+++ b/test/tools.test.ts
@@ -163,14 +163,13 @@ describe("recall_messages", () => {
     );
     expect(errorOut.error).toContain("Unauthorized");
 
-    // A session with no data now returns an empty page rather than an error.
+    // A successful response without an array body is malformed, not an empty page.
     const noData = makeFakeHarness({ noMessageData: new Set(["s-current"]) });
-    const noDataOut = await runTool<MessagesOutput>(
+    const noDataOut = await runTool<ErrorOutput>(
       messagesTool(noData.client, gate, TEST_LIMITS),
       {},
     );
-    expect(noDataOut.ok).toBe(true);
-    expect(noDataOut.pagination.returned).toBe(0);
+    expect(noDataOut.error).toBe("successful message response was not an array");
   });
 
   it("survives raw MCP-bypass args (undefined role/limit must not filter everything)", async () => {
@@ -383,7 +382,7 @@ describe("recall_context", () => {
       sessionID: "s-current",
       messageID: "m-current-1",
     });
-    expect(noDataOut.error).toBe("No messages returned");
+    expect(noDataOut.error).toBe("successful message response was not an array");
   });
 
   it("survives raw MCP-bypass args (undefined window must not break slice bounds)", async () => {