🐛 fix(session): deliver completion notifications reliably

apps/spark-daemon/src/channels/delivery-outbox.test.ts

@@ -938,6 +938,7 @@ describe("daemon channel delivery outbox", () => {
         messageId: "source-message-1",
       },
     };
+    let completionIntentCount = 0;
     try {
       const invocation = invocations.submit({
         sessionId: task.sessionId,
@@ -946,10 +947,24 @@ describe("daemon channel delivery outbox", () => {
       });
       invocations.claimNext("worker");
 
-      completeInvocationWithChannelDelivery({ db, invocations, deliveries }, invocation, task, {
-        status: "succeeded",
-        result: { assistantText: "done" },
-      });
+      completeInvocationWithChannelDelivery(
+        { db, invocations, deliveries, afterComplete: () => void (completionIntentCount += 1) },
+        invocation,
+        task,
+        {
+          status: "succeeded",
+          result: { assistantText: "done" },
+        },
+      );
+      expect(() =>
+        completeInvocationWithChannelDelivery(
+          { db, invocations, deliveries, afterComplete: () => void (completionIntentCount += 1) },
+          invocation,
+          task,
+          { status: "succeeded", result: { assistantText: "done again" } },
+        ),
+      ).toThrow("Invalid Spark invocation transition");
+      expect(completionIntentCount).toBe(1);
 
       expect(invocations.require(invocation.invocationId)).toMatchObject({
         status: "succeeded",

apps/spark-daemon/src/channels/delivery-outbox.ts

@@ -190,6 +190,7 @@ export function completeInvocationWithChannelDelivery(
     db: DatabaseSync;
     invocations: SparkInvocationStore;
     deliveries: SparkChannelDeliveryStore;
+    afterComplete?: (completed: SparkInvocationRecord) => void;
   },
   invocation: SparkInvocationRecord,
   task: SparkDaemonTask,
@@ -204,15 +205,18 @@ export function completeInvocationWithChannelDelivery(
           completion.status === "succeeded" ? "final" : "failure",
           completion.result,
         );
-  if (!delivery) {
+  if (!delivery && !deps.afterComplete) {
     return deps.invocations.complete(invocation.invocationId, completion);
   }
 
   deps.db.exec("BEGIN IMMEDIATE");
   try {
     const completed = deps.invocations.complete(invocation.invocationId, completion);
-    const { idempotencyKey, ...payload } = delivery;
-    deps.deliveries.enqueue({ kind: "reply", idempotencyKey, payload });
+    if (delivery) {
+      const { idempotencyKey, ...payload } = delivery;
+      deps.deliveries.enqueue({ kind: "reply", idempotencyKey, payload });
+    }
+    deps.afterComplete?.(completed);
     deps.db.exec("COMMIT");
     return completed;
   } catch (error) {

apps/spark-daemon/src/daemon-start.ts

@@ -55,6 +55,7 @@ import { SparkDaemonHumanWaitRegistry } from "./core/human-waits.ts";
 import type { DaemonSessionRegistry } from "./session-registry.ts";
 import { SparkInvocationScheduler } from "./core/invocation-scheduler.ts";
 import { recoverInterruptedRuntimeCommandReceipts } from "./runtime-command-receipts.ts";
+import { SessionRequestCompletionDeliveryStore } from "./store/session-request-completion-deliveries.ts";
 import { migrateLegacyQueueHistory } from "./store/legacy-queue-migration.ts";
 import { SparkChannelDeliveryStore } from "./store/channel-deliveries.ts";
 import {
@@ -80,7 +81,10 @@ import {
 import { runSparkCommandBridge, cancelSparkBridgeInvocation } from "./spark/bridge.js";
 import { createChannelAwareTaskExecutor, sessionSourceForTask } from "./spark/session-run.js";
 import { reconcileSessionNotificationDeliveries } from "./session-notification-delivery.ts";
-import { notifySessionRequestCompletion } from "./session-request-completion-notify.ts";
+import {
+  reconcileSessionRequestCompletions,
+  sessionRequestCompletionRequested,
+} from "./session-request-completion-notify.ts";
 import {
   nextSparkDaemonTokenRefreshDelayMs,
   refreshSparkDaemonCredentials,
@@ -145,6 +149,7 @@ interface DaemonServingLoops {
   channelDelivery?: Promise<void>;
   channelReply?: Promise<void>;
   notification?: Promise<void>;
+  sessionCompletion?: Promise<void>;
   taskClaims?: Promise<void>;
 }
 
@@ -176,6 +181,7 @@ interface PreparedDaemonRuntime {
   channelReplyDeliveryStore: ChannelReplyDeliveryStore;
   scheduler: SparkInvocationScheduler | null;
   mailStore: SparkSessionMailStore;
+  sessionCompletionDeliveryStore: SessionRequestCompletionDeliveryStore;
   servingGate: ServingLoopGate;
   loops: DaemonServingLoops;
   restartDrain: RestartDrainController;
@@ -249,6 +255,19 @@ async function createPreparedDaemonRuntime(
   const corruptMailboxReporter = createRepeatedErrorReporter(
     "[spark-daemon] corrupt session mailbox skipped",
   );
+  const mailStore =
+    options.mailStore ??
+    new SparkSessionMailStore({
+      sparkHome: userPaths.dataRoot,
+      onCorruptMailbox: ({ path, error }) => {
+        corruptMailboxReporter.report(
+          new Error(`Unable to read mailbox ${path}`, {
+            cause: error,
+          }),
+        );
+      },
+    });
+  const sessionCompletionDeliveryStore = new SessionRequestCompletionDeliveryStore(options.db);
   const scheduler = createDaemonScheduler({
     options,
     runtimeSignal,
@@ -262,6 +281,8 @@ async function createPreparedDaemonRuntime(
     eventHub,
     controlSparkHome: userPaths.configRoot,
     channelsSparkHome: userPaths.dataRoot,
+    mailStore,
+    sessionCompletionDeliveryStore,
   });
   const closeRestartAdmission = () => {
     admission.open = false;
@@ -310,18 +331,8 @@ async function createPreparedDaemonRuntime(
     nextDriverGcAtMs: Date.now() + 60_000,
     channelReplyDeliveryStore,
     scheduler,
-    mailStore:
-      options.mailStore ??
-      new SparkSessionMailStore({
-        sparkHome: userPaths.dataRoot,
-        onCorruptMailbox: ({ path, error }) => {
-          corruptMailboxReporter.report(
-            new Error(`Unable to read mailbox ${path}`, {
-              cause: error,
-            }),
-          );
-        },
-      }),
+    mailStore,
+    sessionCompletionDeliveryStore,
     servingGate,
     loops: {},
     restartDrain,
@@ -548,6 +559,12 @@ function startDaemonServingLoops(runtime: PreparedDaemonRuntime): void {
       );
     });
   }
+  if (options.sessionRegistry && !options.once) {
+    loops.sessionCompletion = servingGate.promise.then(async (committed) => {
+      if (!committed || runtimeSignal.aborted) return;
+      await runSessionCompletionReconcileLoop(runtime);
+    });
+  }
   if (channelIngress && options.sessionRegistry && !options.once) {
     loops.notification = servingGate.promise.then(async (committed) => {
       if (!committed || runtimeSignal.aborted) return;
@@ -602,6 +619,7 @@ async function runDaemonOnce(runtime: PreparedDaemonRuntime): Promise<void> {
     scheduler.processBatch();
     await scheduler.wait();
   }
+  await reconcileSessionRequestCompletionBatch(runtime);
   if (channelIngress && !runtimeSignal.aborted) {
     await reconcileDaemonChannelDeliveries(
       {
@@ -615,6 +633,29 @@ async function runDaemonOnce(runtime: PreparedDaemonRuntime): Promise<void> {
   await runSparkDaemonServerConnectionsOnce(daemonServerConnectionOptions(runtime));
 }
 
+async function runSessionCompletionReconcileLoop(runtime: PreparedDaemonRuntime): Promise<void> {
+  while (!runtime.runtimeSignal.aborted) {
+    await reconcileSessionRequestCompletionBatch(runtime);
+    await delayUnlessAborted(500, runtime.runtimeSignal);
+  }
+}
+
+async function reconcileSessionRequestCompletionBatch(
+  runtime: PreparedDaemonRuntime,
+): Promise<void> {
+  if (!runtime.options.sessionRegistry || !runtime.admission.open) return;
+  await reconcileSessionRequestCompletions({
+    invocationStore: runtime.invocationStore,
+    deliveryStore: runtime.sessionCompletionDeliveryStore,
+    mailStore: runtime.mailStore,
+    sessionRegistry: runtime.options.sessionRegistry,
+    ...(runtime.options.modelControl ? { modelControl: runtime.options.modelControl } : {}),
+    resolveWorkspaceCwd: (workspaceId) =>
+      resolveWorkspaceLocalPath(runtime.options.db, workspaceId),
+    canAdmit: () => runtime.admission.open && !runtime.runtimeSignal.aborted,
+  });
+}
+
 async function reconcileDriverHiddenSessionGc(
   runtime: PreparedDaemonRuntime,
   force = false,
@@ -664,6 +705,7 @@ async function cleanupPreparedDaemonRuntime(runtime: PreparedDaemonRuntime): Pro
   await runtime.loops.scheduler;
   await runtime.loops.channelDelivery;
   await runtime.loops.notification;
+  await runtime.loops.sessionCompletion;
   await runtime.loops.channelReply;
   await runtime.loops.taskClaims;
   if (options.managePidFile !== false && existsSync(options.paths.pidFile)) {
@@ -684,6 +726,8 @@ function createDaemonScheduler(input: {
   eventHub: InvocationEventHub;
   controlSparkHome: string;
   channelsSparkHome: string;
+  mailStore: SparkSessionMailStore;
+  sessionCompletionDeliveryStore: SessionRequestCompletionDeliveryStore;
 }): SparkInvocationScheduler | null {
   if (input.options.runScheduler === false) return null;
   const { options } = input;
@@ -795,22 +839,30 @@ function completeScheduledInvocation(
       db: input.options.db,
       invocations: input.invocationStore,
       deliveries: input.channelDeliveryStore,
+      afterComplete: () => {
+        if (sessionRequestCompletionRequested(task)) {
+          input.sessionCompletionDeliveryStore.enqueue(invocation.invocationId);
+        }
+      },
     },
     invocation,
     task,
     completion,
   );
-  if (input.options.sessionRegistry) {
-    void notifySessionRequestCompletion(
+  if (input.options.sessionRegistry && sessionRequestCompletionRequested(task)) {
+    void reconcileSessionRequestCompletions(
       {
         invocationStore: input.invocationStore,
+        deliveryStore: input.sessionCompletionDeliveryStore,
+        mailStore: input.mailStore,
         sessionRegistry: input.options.sessionRegistry,
         ...(input.options.modelControl ? { modelControl: input.options.modelControl } : {}),
         resolveWorkspaceCwd: (workspaceId) =>
           resolveWorkspaceLocalPath(input.options.db, workspaceId),
         canAdmit: () => input.admission.open && !input.runtimeSignal.aborted,
       },
-      { invocation, task, completion },
+      1,
+      invocation.invocationId,
     ).catch((error) => {
       console.error("[spark-daemon] session request completion notify failed", error);
     });

apps/spark-daemon/src/daemon.test.ts

@@ -7,6 +7,7 @@ import { join } from "node:path";
 import { WebSocketServer } from "ws";
 import { describe, expect, it, vi } from "vitest";
 import {
+  type SparkSessionRegistryRecord,
   SPARK_PROTOCOL_VERSION,
   createId,
   runtimeProtocolVersion,
@@ -27,6 +28,8 @@ import type { DaemonChannelIngressRuntime } from "./channels/ingress.ts";
 import type { SparkDaemonModelControl } from "./model-control.ts";
 import type { CancelSparkInvocationFn, RunSparkCommandFn } from "./spark/bridge.js";
 import { SparkDriverStore } from "./store/drivers.ts";
+import { createDaemonSessionRegistry } from "./session-registry.ts";
+import { SessionRequestCompletionDeliveryStore } from "./store/session-request-completion-deliveries.ts";
 import { SparkInvocationStore } from "./store/invocations.ts";
 import { openSparkDaemonDatabase } from "./store/schema.js";
 import {
@@ -684,6 +687,103 @@ describe("Spark daemon handleCommand task.start.request", () => {
     }
   });
 
+  it("reconciles a restarted daemon-local completion without channel ingress", async () => {
+    const harness = makeHarness();
+    const store = new SparkInvocationStore(harness.db);
+    const deliveries = new SessionRequestCompletionDeliveryStore(harness.db);
+    const source = store.submit({
+      sessionId: "target-session-restart",
+      prompt: "delegated request",
+      task: {
+        type: "session.run",
+        sessionId: "target-session-restart",
+        prompt: "delegated request",
+        cwd: harness.workspace.localPath,
+        messageMetadata: {
+          sessionMail: {
+            messageId: "mail:daemon-restart",
+            kind: "request",
+            fromSessionId: "sender-session-restart",
+            toSessionId: "target-session-restart",
+            notifyOnCompletion: true,
+          },
+        },
+      },
+    });
+    store.claimNext("target-executor");
+    store.complete(source.invocationId, {
+      status: "succeeded",
+      result: { assistantText: "recovered result" },
+    });
+    deliveries.enqueue(source.invocationId);
+    const sender: SparkSessionRegistryRecord = {
+      sessionId: "sender-session-restart",
+      scope: { kind: "workspace", workspaceId: harness.workspace.id },
+      workspaceId: harness.workspace.id,
+      cwd: harness.workspace.localPath,
+      status: "ready",
+      bindings: [],
+      createdAt: "2026-07-29T00:00:00.000Z",
+      updatedAt: "2026-07-29T00:00:00.000Z",
+    };
+    const sessionRegistry = createDaemonSessionRegistry(harness.sparkHome, {
+      daemonId: "install-test",
+      daemonCwd: harness.workspace.localPath,
+    });
+    const firstShutdown = new AbortController();
+    const first = startSparkDaemon({
+      paths: harness.paths,
+      sparkHome: harness.sparkHome,
+      db: harness.db,
+      config: { installationId: "install-test", displayName: "Test daemon" },
+      signal: firstShutdown.signal,
+      runScheduler: false,
+      sessionRegistry,
+      notificationReconcileIntervalMs: 5,
+    });
+
+    try {
+      await vi.waitFor(() =>
+        expect(deliveries.require(source.invocationId).attemptCount).toBeGreaterThan(0),
+      );
+      firstShutdown.abort();
+      await first;
+      expect(store.listPendingForSession(sender.sessionId)).toHaveLength(0);
+
+      await sessionRegistry.create({
+        sessionId: sender.sessionId,
+        scope: sender.scope,
+        workspaceId: sender.workspaceId,
+        cwd: sender.cwd,
+      });
+      const recordTurnQueued = vi.spyOn(sessionRegistry, "recordTurnQueued");
+      const successorShutdown = new AbortController();
+      const successor = startSparkDaemon({
+        paths: harness.paths,
+        sparkHome: harness.sparkHome,
+        db: harness.db,
+        config: { installationId: "install-test", displayName: "Test daemon" },
+        signal: successorShutdown.signal,
+        runScheduler: false,
+        sessionRegistry,
+        notificationReconcileIntervalMs: 5,
+      });
+      await vi.waitFor(() =>
+        expect(deliveries.require(source.invocationId).status).toBe("delivered"),
+      );
+      successorShutdown.abort();
+      await successor;
+
+      expect(store.listPendingForSession(sender.sessionId)).toHaveLength(1);
+      expect(recordTurnQueued).toHaveBeenCalledOnce();
+      expect(store.require(source.invocationId).status).toBe("succeeded");
+    } finally {
+      firstShutdown.abort();
+      await first.catch(() => undefined);
+      harness.cleanup();
+    }
+  });
+
   it("DRV-DRAIN-001 drains active scheduler work and leaves queued work for the restart successor", async () => {
     const harness = makeHarness();
     const shutdown = new AbortController();