diff --git a/apps/spark-daemon/src/channels/delivery-outbox.test.ts b/apps/spark-daemon/src/channels/delivery-outbox.test.ts index 14c6cb2a..e5c6f88b 100644 --- a/apps/spark-daemon/src/channels/delivery-outbox.test.ts +++ b/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", diff --git a/apps/spark-daemon/src/channels/delivery-outbox.ts b/apps/spark-daemon/src/channels/delivery-outbox.ts index bedb0b22..64973de4 100644 --- a/apps/spark-daemon/src/channels/delivery-outbox.ts +++ b/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) { diff --git a/apps/spark-daemon/src/daemon-start.ts b/apps/spark-daemon/src/daemon-start.ts index 8a2fcf05..43aaeab5 100644 --- a/apps/spark-daemon/src/daemon-start.ts +++ b/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; channelReply?: Promise; notification?: Promise; + sessionCompletion?: Promise; taskClaims?: Promise; } @@ -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 { scheduler.processBatch(); await scheduler.wait(); } + await reconcileSessionRequestCompletionBatch(runtime); if (channelIngress && !runtimeSignal.aborted) { await reconcileDaemonChannelDeliveries( { @@ -615,6 +633,29 @@ async function runDaemonOnce(runtime: PreparedDaemonRuntime): Promise { await runSparkDaemonServerConnectionsOnce(daemonServerConnectionOptions(runtime)); } +async function runSessionCompletionReconcileLoop(runtime: PreparedDaemonRuntime): Promise { + while (!runtime.runtimeSignal.aborted) { + await reconcileSessionRequestCompletionBatch(runtime); + await delayUnlessAborted(500, runtime.runtimeSignal); + } +} + +async function reconcileSessionRequestCompletionBatch( + runtime: PreparedDaemonRuntime, +): Promise { + 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); }); diff --git a/apps/spark-daemon/src/daemon.test.ts b/apps/spark-daemon/src/daemon.test.ts index 32bc10fb..995c0c21 100644 --- a/apps/spark-daemon/src/daemon.test.ts +++ b/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(); diff --git a/apps/spark-daemon/src/session-request-completion-notify.test.ts b/apps/spark-daemon/src/session-request-completion-notify.test.ts index b80ec1c6..cefcb976 100644 --- a/apps/spark-daemon/src/session-request-completion-notify.test.ts +++ b/apps/spark-daemon/src/session-request-completion-notify.test.ts @@ -3,14 +3,17 @@ import { tmpdir } from "node:os"; import { join } from "node:path"; import { DatabaseSync } from "node:sqlite"; +import { SparkSessionMailStore } from "@zendev-lab/spark-session"; import type { SparkSessionRegistryRecord } from "@zendev-lab/spark-protocol"; import { describe, expect, it, vi } from "vitest"; import { notifySessionRequestCompletion, + reconcileSessionRequestCompletions, renderSessionRequestCompletionPrompt, SESSION_REQUEST_COMPLETION_SOURCE_KIND, } from "./session-request-completion-notify.ts"; +import { SessionRequestCompletionDeliveryStore } from "./store/session-request-completion-deliveries.ts"; import { SparkInvocationStore } from "./store/invocations.ts"; import { migrateSparkDaemonDatabase } from "./store/schema.ts"; @@ -208,6 +211,230 @@ describe("session request completion notify", () => { } }); + it("persists one completion mail and retries sender wake after restart or unavailability", async () => { + const harness = createHarness(); + const sender = localSession("sess_sender_retry", harness.cwd); + const target = localSession("sess_target_retry", harness.cwd); + const mailStore = new SparkSessionMailStore({ sparkHome: harness.cwd }); + const deliveryStore = new SessionRequestCompletionDeliveryStore(harness.db); + const source = harness.store.submit({ + sessionId: target.sessionId, + prompt: "delegated work", + task: requestMailTask(sender.sessionId, target.sessionId, true), + }); + harness.store.claimNext("completion-worker"); + harness.store.complete(source.invocationId, { + status: "succeeded", + result: { assistantText: "durable result" }, + }); + deliveryStore.enqueue(source.invocationId); + deliveryStore.enqueue(source.invocationId); + let senderAvailable = false; + const recordTurnQueued = vi.fn(async () => sender); + const deps = { + invocationStore: harness.store, + deliveryStore, + mailStore, + sessionRegistry: { + get: async (sessionId: string) => + sessionId === sender.sessionId ? (senderAvailable ? sender : undefined) : target, + recordTurnQueued, + }, + }; + + try { + await expect(reconcileSessionRequestCompletions(deps)).resolves.toEqual({ + attempted: 1, + delivered: 0, + failed: 1, + }); + expect(await mailStore.list(sender.sessionId, { includeAcked: true })).toHaveLength(1); + expect(harness.store.listPendingForSession(sender.sessionId)).toHaveLength(0); + + senderAvailable = true; + const restartedStore = new SessionRequestCompletionDeliveryStore(harness.db); + await expect( + reconcileSessionRequestCompletions({ ...deps, deliveryStore: restartedStore }), + ).resolves.toEqual({ attempted: 1, delivered: 1, failed: 0 }); + expect(await mailStore.list(sender.sessionId, { includeAcked: true })).toHaveLength(1); + expect(harness.store.listPendingForSession(sender.sessionId)).toHaveLength(1); + expect(recordTurnQueued).toHaveBeenCalledOnce(); + + await expect( + reconcileSessionRequestCompletions({ ...deps, deliveryStore: restartedStore }), + ).resolves.toEqual({ attempted: 0, delivered: 0, failed: 0 }); + expect(harness.store.listPendingForSession(sender.sessionId)).toHaveLength(1); + expect(recordTurnQueued).toHaveBeenCalledOnce(); + } finally { + harness.close(); + } + }); + + it("retries a persisted wake when turn queue projection fails without resubmitting", async () => { + const harness = createHarness(); + const sender = localSession("sess_sender_wake_retry", harness.cwd); + const target = localSession("sess_target_wake_retry", harness.cwd); + const mailStore = new SparkSessionMailStore({ sparkHome: harness.cwd }); + const deliveryStore = new SessionRequestCompletionDeliveryStore(harness.db); + const source = harness.store.submit({ + sessionId: target.sessionId, + prompt: "delegated work", + task: requestMailTask(sender.sessionId, target.sessionId, true), + }); + harness.store.claimNext("completion-worker"); + harness.store.complete(source.invocationId, { + status: "succeeded", + result: { assistantText: "done" }, + }); + deliveryStore.enqueue(source.invocationId); + let failWake = true; + const recordTurnQueued = vi.fn(async () => { + if (failWake) throw new Error("sender unavailable"); + return sender; + }); + const deps = { + invocationStore: harness.store, + deliveryStore, + mailStore, + sessionRegistry: { + get: async () => sender, + recordTurnQueued, + }, + }; + + try { + await expect(reconcileSessionRequestCompletions(deps)).resolves.toEqual({ + attempted: 1, + delivered: 0, + failed: 1, + }); + expect(harness.store.listPendingForSession(sender.sessionId)).toHaveLength(1); + failWake = false; + await expect(reconcileSessionRequestCompletions(deps)).resolves.toEqual({ + attempted: 1, + delivered: 1, + failed: 0, + }); + expect(harness.store.listPendingForSession(sender.sessionId)).toHaveLength(1); + expect(await mailStore.list(sender.sessionId, { includeAcked: true })).toHaveLength(1); + expect(recordTurnQueued).toHaveBeenCalledTimes(2); + } finally { + harness.close(); + } + }); + + it("atomically claims one delivery across two reconcilers and delivers side effects once", async () => { + const harness = createHarness(); + const sender = localSession("sess_sender_concurrent", harness.cwd); + const target = localSession("sess_target_concurrent", harness.cwd); + const mailStore = new SparkSessionMailStore({ sparkHome: harness.cwd }); + const firstStore = new SessionRequestCompletionDeliveryStore(harness.db); + const secondStore = new SessionRequestCompletionDeliveryStore(harness.db); + const source = completedRequest(harness, sender.sessionId, target.sessionId, "concurrent"); + firstStore.enqueue(source.invocationId); + let senderWakeCount = 0; + const deps = { + invocationStore: harness.store, + mailStore, + sessionRegistry: { + get: async () => sender, + recordTurnQueued: async () => { + senderWakeCount += 1; + return sender; + }, + }, + }; + + try { + const [first, second] = await Promise.all([ + reconcileSessionRequestCompletions({ ...deps, deliveryStore: firstStore }), + reconcileSessionRequestCompletions({ ...deps, deliveryStore: secondStore }), + ]); + expect(first.attempted + second.attempted).toBe(1); + expect(first.delivered + second.delivered).toBe(1); + expect(await mailStore.list(sender.sessionId, { includeAcked: true })).toHaveLength(1); + expect(harness.store.listPendingForSession(sender.sessionId)).toHaveLength(1); + expect(senderWakeCount).toBe(1); + expect(firstStore.require(source.invocationId).status).toBe("delivered"); + } finally { + harness.close(); + } + }); + + it("recovers an expired claim after restart without duplicating completion effects", async () => { + const harness = createHarness(); + const sender = localSession("sess_sender_expired", harness.cwd); + const target = localSession("sess_target_expired", harness.cwd); + const mailStore = new SparkSessionMailStore({ sparkHome: harness.cwd }); + let now = new Date("2026-07-29T00:00:00.000Z"); + const firstStore = new SessionRequestCompletionDeliveryStore(harness.db, () => now); + const source = completedRequest(harness, sender.sessionId, target.sessionId, "expired"); + firstStore.enqueue(source.invocationId); + expect(firstStore.claim(source.invocationId, 1_000)?.status).toBe("processing"); + now = new Date("2026-07-29T00:00:02.000Z"); + const restartedStore = new SessionRequestCompletionDeliveryStore(harness.db, () => now); + let senderWakeCount = 0; + + try { + await expect( + reconcileSessionRequestCompletions({ + invocationStore: harness.store, + deliveryStore: restartedStore, + mailStore, + sessionRegistry: { + get: async () => sender, + recordTurnQueued: async () => { + senderWakeCount += 1; + return sender; + }, + }, + }), + ).resolves.toEqual({ attempted: 1, delivered: 1, failed: 0 }); + expect(await mailStore.list(sender.sessionId, { includeAcked: true })).toHaveLength(1); + expect(harness.store.listPendingForSession(sender.sessionId)).toHaveLength(1); + expect(senderWakeCount).toBe(1); + } finally { + harness.close(); + } + }); + + it("lets immediate notify and reconciliation race without duplicate target or wake execution", async () => { + const harness = createHarness(); + const sender = localSession("sess_sender_immediate", harness.cwd); + const target = localSession("sess_target_immediate", harness.cwd); + const mailStore = new SparkSessionMailStore({ sparkHome: harness.cwd }); + const deliveryStore = new SessionRequestCompletionDeliveryStore(harness.db); + const source = completedRequest(harness, sender.sessionId, target.sessionId, "immediate"); + deliveryStore.enqueue(source.invocationId); + let senderWakeCount = 0; + const deps = { + invocationStore: harness.store, + deliveryStore, + mailStore, + sessionRegistry: { + get: async () => sender, + recordTurnQueued: async () => { + senderWakeCount += 1; + return sender; + }, + }, + }; + + try { + const [immediate, loop] = await Promise.all([ + reconcileSessionRequestCompletions(deps, 1, source.invocationId), + reconcileSessionRequestCompletions(deps), + ]); + expect(immediate.attempted + loop.attempted).toBe(1); + expect(await mailStore.list(sender.sessionId, { includeAcked: true })).toHaveLength(1); + expect(harness.store.listPendingForSession(sender.sessionId)).toHaveLength(1); + expect(senderWakeCount).toBe(1); + expect(deliveryStore.require(source.invocationId).status).toBe("delivered"); + } finally { + harness.close(); + } + }); + it("renders a synthesis prompt with failure details", () => { const prompt = renderSessionRequestCompletionPrompt({ mail: { @@ -231,6 +458,25 @@ describe("session request completion notify", () => { }); }); +function completedRequest( + harness: ReturnType, + senderSessionId: string, + targetSessionId: string, + suffix: string, +) { + const source = harness.store.submit({ + sessionId: targetSessionId, + prompt: `delegated work ${suffix}`, + task: requestMailTask(senderSessionId, targetSessionId, true), + }); + harness.store.claimNext(`target-worker-${suffix}`); + harness.store.complete(source.invocationId, { + status: "succeeded", + result: { assistantText: `done ${suffix}` }, + }); + return harness.store.require(source.invocationId); +} + function createHarness() { const cwd = mkdtempSync(join(tmpdir(), "spark-session-request-completion-")); const db = new DatabaseSync(":memory:"); diff --git a/apps/spark-daemon/src/session-request-completion-notify.ts b/apps/spark-daemon/src/session-request-completion-notify.ts index d249fb2d..332553eb 100644 --- a/apps/spark-daemon/src/session-request-completion-notify.ts +++ b/apps/spark-daemon/src/session-request-completion-notify.ts @@ -1,8 +1,10 @@ +import { SparkSessionMailStore } from "@zendev-lab/spark-session"; import type { SparkSessionRegistryRecord } from "@zendev-lab/spark-protocol"; import type { SparkDaemonModelControl } from "./model-control.ts"; import type { DaemonSessionRegistry } from "./session-registry.ts"; import type { SparkDaemonSessionRunTask, SparkDaemonTask } from "./core/types.ts"; +import { SessionRequestCompletionDeliveryStore } from "./store/session-request-completion-deliveries.ts"; import { type CompleteSparkInvocationInput, type SparkInvocationRecord, @@ -12,7 +14,12 @@ import { export const SESSION_REQUEST_COMPLETION_SOURCE_KIND = "session.request.completion"; export interface SessionRequestCompletionNotifyDependencies { - invocationStore: Pick; + invocationStore: Pick; + deliveryStore?: Pick< + SessionRequestCompletionDeliveryStore, + "claim" | "claimPending" | "recordFailure" | "markDelivered" + >; + mailStore?: Pick; sessionRegistry: Pick; modelControl?: Pick< SparkDaemonModelControl, @@ -49,6 +56,15 @@ interface CompletionNotificationSender { * Idempotent on `session.request.completion:${sourceInvocationId}`. Does not * fire for `wait=completed` callers (`notifyOnCompletion: false`). */ +export function sessionRequestCompletionRequested(task: SparkDaemonTask): boolean { + const mail = sessionMailFromTask(task); + return Boolean( + mail?.notifyOnCompletion === true && + mail.kind === "request" && + trimmedString(mail.fromSessionId), + ); +} + export async function notifySessionRequestCompletion( deps: SessionRequestCompletionNotifyDependencies, input: { @@ -57,8 +73,10 @@ export async function notifySessionRequestCompletion( completion: CompleteSparkInvocationInput; }, ): Promise { - const candidate = completionNotificationCandidate(deps, input); + const candidate = completionNotificationCandidate(input); if ("submitted" in candidate) return candidate; + await persistCompletionMail(deps, input, candidate); + if (!canAdmit(deps)) return skipped("admission_closed"); const sender = await completionNotificationSender(deps, candidate.fromSessionId); if ("submitted" in sender) return sender; const prompt = renderSessionRequestCompletionPrompt({ @@ -69,26 +87,143 @@ export async function notifySessionRequestCompletion( }); const task = completionNotificationTask({ input, candidate, sender, prompt }); if (!canAdmit(deps)) return skipped("admission_closed"); - const submitted = deps.invocationStore.submit({ - sessionId: candidate.fromSessionId, - idempotencyKey: candidate.idempotencyKey, - prompt, - task, - sourceKind: SESSION_REQUEST_COMPLETION_SOURCE_KIND, - sourceRef: input.invocation.invocationId, - }); + const existing = deps.invocationStore.findByIdempotencyKey(candidate.idempotencyKey); + const submitted = + existing ?? + deps.invocationStore.submit({ + sessionId: candidate.fromSessionId, + idempotencyKey: candidate.idempotencyKey, + prompt, + task, + sourceKind: SESSION_REQUEST_COMPLETION_SOURCE_KIND, + sourceRef: input.invocation.invocationId, + }); await deps.sessionRegistry.recordTurnQueued(candidate.fromSessionId); - return { submitted: true, invocationId: submitted.invocationId }; + return existing + ? { submitted: false, skippedReason: "already_notified", invocationId: submitted.invocationId } + : { submitted: true, invocationId: submitted.invocationId }; +} + +export async function reconcileSessionRequestCompletions( + deps: SessionRequestCompletionNotifyDependencies, + limit = 50, + sourceInvocationId?: string, +): Promise<{ attempted: number; delivered: number; failed: number }> { + const deliveryStore = deps.deliveryStore; + if (!deliveryStore) return { attempted: 0, delivered: 0, failed: 0 }; + const claimed = sourceInvocationId + ? [deliveryStore.claim(sourceInvocationId)].filter((value) => value !== undefined) + : deliveryStore.claimPending(limit); + let delivered = 0; + let failed = 0; + for (const delivery of claimed) { + const claimToken = delivery.claimToken; + if (!claimToken) continue; + const invocation = deps.invocationStore.get(delivery.sourceInvocationId); + if (!invocation || !invocation.task || !isTerminalStatus(invocation.status)) { + deliveryStore.recordFailure( + delivery.sourceInvocationId, + claimToken, + "source invocation is unavailable", + ); + failed += 1; + continue; + } + try { + const result = await notifySessionRequestCompletion(deps, { + invocation, + task: invocation.task as SparkDaemonTask, + completion: completionFromInvocation(invocation), + }); + if (isCompletedDeliveryResult(result)) { + deliveryStore.markDelivered(delivery.sourceInvocationId, claimToken); + delivered += 1; + } else { + deliveryStore.recordFailure( + delivery.sourceInvocationId, + claimToken, + result.skippedReason ?? "completion wake was not submitted", + ); + failed += 1; + } + } catch (error) { + deliveryStore.recordFailure( + delivery.sourceInvocationId, + claimToken, + error instanceof Error ? error.message : String(error), + ); + failed += 1; + } + } + return { attempted: claimed.length, delivered, failed }; +} + +function isCompletedDeliveryResult(result: SessionRequestCompletionNotifyResult): boolean { + return ( + Boolean(result.invocationId) || + result.skippedReason === "notify_disabled" || + result.skippedReason === "not_request" || + result.skippedReason === "no_session_mail" || + result.skippedReason === "missing_from_session" || + result.skippedReason === "same_session" + ); } -function completionNotificationCandidate( +async function persistCompletionMail( deps: SessionRequestCompletionNotifyDependencies, input: { invocation: SparkInvocationRecord; - task: SparkDaemonTask; + completion: CompleteSparkInvocationInput; }, -): CompletionNotificationCandidate | SessionRequestCompletionNotifyResult { - if (!canAdmit(deps)) return skipped("admission_closed"); + candidate: CompletionNotificationCandidate, +): Promise { + if (!deps.mailStore) return; + await deps.mailStore.send({ + toSessionId: candidate.fromSessionId, + fromSessionId: input.invocation.sessionId ?? candidate.mail.toSessionId, + kind: "notification", + visibility: "internal", + delivery: "mailbox", + intent: SESSION_REQUEST_COMPLETION_SOURCE_KIND, + correlationId: candidate.mail.correlationId, + idempotencyKey: `session.request.completion.mail:${input.invocation.invocationId}`, + payload: { + sourceInvocationId: input.invocation.invocationId, + sourceSessionId: input.invocation.sessionId ?? candidate.mail.toSessionId, + messageId: candidate.mail.messageId, + status: input.completion.status, + summary: completionSummaryText(input.completion), + }, + body: renderSessionRequestCompletionPrompt({ + mail: candidate.mail, + targetSessionId: input.invocation.sessionId ?? candidate.mail.toSessionId, + sourceInvocationId: input.invocation.invocationId, + completion: input.completion, + }), + source: "tool", + }); +} + +function completionFromInvocation(invocation: SparkInvocationRecord): CompleteSparkInvocationInput { + if (invocation.status === "succeeded") return { status: "succeeded", result: invocation.result }; + if (invocation.status === "cancelled") { + return { status: "cancelled", cancelReason: invocation.cancelReason }; + } + return { + status: "failed", + errorCode: invocation.errorCode, + errorMessage: invocation.errorMessage, + }; +} + +function isTerminalStatus(status: SparkInvocationRecord["status"]): boolean { + return status === "succeeded" || status === "failed" || status === "cancelled"; +} + +function completionNotificationCandidate(input: { + invocation: SparkInvocationRecord; + task: SparkDaemonTask; +}): CompletionNotificationCandidate | SessionRequestCompletionNotifyResult { const mail = sessionMailFromTask(input.task); if (!mail) return skipped("no_session_mail"); if (mail.notifyOnCompletion !== true) return skipped("notify_disabled"); @@ -96,10 +231,7 @@ function completionNotificationCandidate( const fromSessionId = trimmedString(mail.fromSessionId); if (!fromSessionId) return skipped("missing_from_session"); if (fromSessionId === input.invocation.sessionId) return skipped("same_session"); - const idempotencyKey = `session.request.completion:${input.invocation.invocationId}`; - if (deps.invocationStore.findByIdempotencyKey(idempotencyKey)) { - return skipped("already_notified", input.invocation.invocationId); - } + const idempotencyKey = `session.request.completion.wake:${input.invocation.invocationId}`; return { mail, fromSessionId, idempotencyKey }; } diff --git a/apps/spark-daemon/src/store/schema.test.ts b/apps/spark-daemon/src/store/schema.test.ts index 7f7ee65b..b8f97236 100644 --- a/apps/spark-daemon/src/store/schema.test.ts +++ b/apps/spark-daemon/src/store/schema.test.ts @@ -28,6 +28,41 @@ describe("migrateSparkDaemonDatabase", () => { } }); + it("upgrades pending completion deliveries with claim lease columns", () => { + const db = new DatabaseSync(":memory:"); + try { + db.exec(` + CREATE TABLE session_request_completion_deliveries ( + source_invocation_id TEXT PRIMARY KEY, + status TEXT NOT NULL CHECK (status IN ('pending', 'delivered')), + attempt_count INTEGER NOT NULL DEFAULT 0, + last_error TEXT, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + delivered_at TEXT + ); + INSERT INTO session_request_completion_deliveries + (source_invocation_id, status, attempt_count, created_at, updated_at) + VALUES ('inv_pending', 'pending', 2, '2026-07-29T00:00:00.000Z', '2026-07-29T00:00:01.000Z'); + `); + + migrateSparkDaemonDatabase(db); + + const columns = db.prepare("PRAGMA table_info(session_request_completion_deliveries)").all(); + expect(columns).toEqual( + expect.arrayContaining([ + expect.objectContaining({ name: "claim_token" }), + expect.objectContaining({ name: "claim_expires_at" }), + ]), + ); + expect( + db.prepare("SELECT status, attempt_count FROM session_request_completion_deliveries").get(), + ).toEqual({ status: "pending", attempt_count: 2 }); + } finally { + db.close(); + } + }); + it("expands legacy workspace client rows with nullable session lease columns", () => { const db = new DatabaseSync(":memory:"); try { diff --git a/apps/spark-daemon/src/store/schema.ts b/apps/spark-daemon/src/store/schema.ts index 1e8a9b92..892ab93f 100644 --- a/apps/spark-daemon/src/store/schema.ts +++ b/apps/spark-daemon/src/store/schema.ts @@ -109,6 +109,18 @@ export function migrateSparkDaemonDatabase(db: DatabaseSync): void { updated_at TEXT NOT NULL ); + CREATE TABLE IF NOT EXISTS session_request_completion_deliveries ( + source_invocation_id TEXT PRIMARY KEY, + status TEXT NOT NULL CHECK (status IN ('pending', 'processing', 'delivered')), + attempt_count INTEGER NOT NULL DEFAULT 0, + last_error TEXT, + claim_token TEXT, + claim_expires_at TEXT, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + delivered_at TEXT + ); + CREATE TABLE IF NOT EXISTS channel_deliveries ( id TEXT PRIMARY KEY, kind TEXT NOT NULL CHECK (kind IN ('reply', 'ask', 'interaction_ack', 'inbound', 'notification')), @@ -212,6 +224,7 @@ export function migrateSparkDaemonDatabase(db: DatabaseSync): void { WHERE terminal_json IS NOT NULL; CREATE INDEX IF NOT EXISTS daemon_human_waits_status_idx ON daemon_human_waits(status, created_at); `); + migrateSessionRequestCompletionDeliverySchema(db); migrateChannelDeliverySchema(db); addMissingRuntimeCommandReceiptColumns(db); addMissingInvocationColumns(db); @@ -234,6 +247,36 @@ export function migrateSparkDaemonDatabase(db: DatabaseSync): void { backfillSparkDaemonRegistrationTables(db); } +function migrateSessionRequestCompletionDeliverySchema(db: DatabaseSync): void { + const columns = workspaceColumns(db, "session_request_completion_deliveries"); + if (!columns.has("claim_token") || !columns.has("claim_expires_at")) { + db.exec(` + ALTER TABLE session_request_completion_deliveries + RENAME TO session_request_completion_deliveries_legacy; + CREATE TABLE session_request_completion_deliveries ( + source_invocation_id TEXT PRIMARY KEY, + status TEXT NOT NULL CHECK (status IN ('pending', 'processing', 'delivered')), + attempt_count INTEGER NOT NULL DEFAULT 0, + last_error TEXT, + claim_token TEXT, + claim_expires_at TEXT, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + delivered_at TEXT + ); + INSERT INTO session_request_completion_deliveries + (source_invocation_id, status, attempt_count, last_error, created_at, updated_at, delivered_at) + SELECT source_invocation_id, status, attempt_count, last_error, created_at, updated_at, delivered_at + FROM session_request_completion_deliveries_legacy; + DROP TABLE session_request_completion_deliveries_legacy; + `); + } + db.exec(` + CREATE INDEX IF NOT EXISTS idx_session_request_completion_due + ON session_request_completion_deliveries(status, claim_expires_at, updated_at) + `); +} + function addMissingDriverColumns(db: DatabaseSync): void { const columns = workspaceColumns(db, "driver_wakeups"); if (!columns.has("wake_prompt")) { diff --git a/apps/spark-daemon/src/store/session-request-completion-deliveries.ts b/apps/spark-daemon/src/store/session-request-completion-deliveries.ts new file mode 100644 index 00000000..635c670f --- /dev/null +++ b/apps/spark-daemon/src/store/session-request-completion-deliveries.ts @@ -0,0 +1,160 @@ +import { randomUUID } from "node:crypto"; +import type { DatabaseSync } from "node:sqlite"; + +export interface SessionRequestCompletionDeliveryRecord { + sourceInvocationId: string; + status: "pending" | "processing" | "delivered"; + attemptCount: number; + lastError?: string; + claimToken?: string; + claimExpiresAt?: string; + createdAt: string; + updatedAt: string; + deliveredAt?: string; +} + +export class SessionRequestCompletionDeliveryStore { + private readonly db: DatabaseSync; + private readonly now: () => Date; + + constructor(db: DatabaseSync, now: () => Date = () => new Date()) { + this.db = db; + this.now = now; + } + + enqueue(sourceInvocationId: string): SessionRequestCompletionDeliveryRecord { + const now = this.now().toISOString(); + this.db + .prepare( + `INSERT INTO session_request_completion_deliveries + (source_invocation_id, status, attempt_count, created_at, updated_at) + VALUES (?, 'pending', 0, ?, ?) + ON CONFLICT(source_invocation_id) DO NOTHING`, + ) + .run(sourceInvocationId, now, now); + return this.require(sourceInvocationId); + } + + claimPending(limit = 50, leaseMs = 300_000): SessionRequestCompletionDeliveryRecord[] { + const claimed: SessionRequestCompletionDeliveryRecord[] = []; + const boundedLimit = Math.max(1, Math.min(500, Math.floor(limit))); + for (let index = 0; index < boundedLimit; index += 1) { + const record = this.claimNext(leaseMs); + if (!record) break; + claimed.push(record); + } + return claimed; + } + + claim( + sourceInvocationId: string, + leaseMs = 300_000, + ): SessionRequestCompletionDeliveryRecord | undefined { + return this.claimNext(leaseMs, sourceInvocationId); + } + + recordFailure( + sourceInvocationId: string, + claimToken: string, + error: string, + ): SessionRequestCompletionDeliveryRecord { + const result = this.db + .prepare( + `UPDATE session_request_completion_deliveries + SET status = 'pending', attempt_count = attempt_count + 1, + last_error = ?, claim_token = NULL, claim_expires_at = NULL, updated_at = ? + WHERE source_invocation_id = ? AND status = 'processing' AND claim_token = ?`, + ) + .run(error, this.now().toISOString(), sourceInvocationId, claimToken); + if (result.changes !== 1) return this.require(sourceInvocationId); + return this.require(sourceInvocationId); + } + + markDelivered( + sourceInvocationId: string, + claimToken: string, + ): SessionRequestCompletionDeliveryRecord { + const now = this.now().toISOString(); + this.db + .prepare( + `UPDATE session_request_completion_deliveries + SET status = 'delivered', attempt_count = attempt_count + 1, + last_error = NULL, claim_token = NULL, claim_expires_at = NULL, + updated_at = ?, delivered_at = ? + WHERE source_invocation_id = ? AND status = 'processing' AND claim_token = ?`, + ) + .run(now, now, sourceInvocationId, claimToken); + return this.require(sourceInvocationId); + } + + require(sourceInvocationId: string): SessionRequestCompletionDeliveryRecord { + const row = this.db + .prepare( + `SELECT source_invocation_id AS sourceInvocationId, status, + attempt_count AS attemptCount, last_error AS lastError, + claim_token AS claimToken, claim_expires_at AS claimExpiresAt, + created_at AS createdAt, updated_at AS updatedAt, + delivered_at AS deliveredAt + FROM session_request_completion_deliveries + WHERE source_invocation_id = ?`, + ) + .get(sourceInvocationId); + if (!row) throw new Error(`unknown session request completion delivery: ${sourceInvocationId}`); + return parseRecord(row); + } + + private claimNext( + leaseMs: number, + sourceInvocationId?: string, + ): SessionRequestCompletionDeliveryRecord | undefined { + const now = this.now(); + const nowIso = now.toISOString(); + const claimToken = randomUUID(); + const claimExpiresAt = new Date(now.getTime() + Math.max(1, leaseMs)).toISOString(); + const sourceFilter = sourceInvocationId ? "AND source_invocation_id = ?" : ""; + const parameters = sourceInvocationId + ? [claimToken, claimExpiresAt, nowIso, nowIso, sourceInvocationId] + : [claimToken, claimExpiresAt, nowIso, nowIso]; + const row = this.db + .prepare( + `UPDATE session_request_completion_deliveries + SET status = 'processing', claim_token = ?, claim_expires_at = ?, updated_at = ? + WHERE source_invocation_id = ( + SELECT source_invocation_id + FROM session_request_completion_deliveries + WHERE (status = 'pending' OR (status = 'processing' AND claim_expires_at <= ?)) + ${sourceFilter} + ORDER BY created_at ASC, source_invocation_id ASC + LIMIT 1 + ) + RETURNING source_invocation_id AS sourceInvocationId, status, + attempt_count AS attemptCount, last_error AS lastError, + claim_token AS claimToken, claim_expires_at AS claimExpiresAt, + created_at AS createdAt, updated_at AS updatedAt, + delivered_at AS deliveredAt`, + ) + .get(...parameters); + return row ? parseRecord(row) : undefined; + } +} + +function parseRecord(value: unknown): SessionRequestCompletionDeliveryRecord { + const row = value as Record; + const status = + row.status === "delivered" + ? "delivered" + : row.status === "processing" + ? "processing" + : "pending"; + return { + sourceInvocationId: String(row.sourceInvocationId), + status, + attemptCount: Number(row.attemptCount), + ...(typeof row.lastError === "string" ? { lastError: row.lastError } : {}), + ...(typeof row.claimToken === "string" ? { claimToken: row.claimToken } : {}), + ...(typeof row.claimExpiresAt === "string" ? { claimExpiresAt: row.claimExpiresAt } : {}), + createdAt: String(row.createdAt), + updatedAt: String(row.updatedAt), + ...(typeof row.deliveredAt === "string" ? { deliveredAt: row.deliveredAt } : {}), + }; +} diff --git a/packages/spark-session/src/action-tool.test.ts b/packages/spark-session/src/action-tool.test.ts index 9af32a4c..db2f254d 100644 --- a/packages/spark-session/src/action-tool.test.ts +++ b/packages/spark-session/src/action-tool.test.ts @@ -580,7 +580,7 @@ describe("blocking session requests", () => { }); }); - it("continues a timed-out request by invocation id without mail or resubmission", async () => { + it("continues wait=completed by invocation id without resubmission", async () => { const continuationInvocationId = "inv_continueonly"; let terminalStatus = false; let terminalReads = 0;