From 05e4f888a0d8591c790d036140080092cf024d30 Mon Sep 17 00:00:00 2001 From: Yash Singh Date: Tue, 13 Oct 2026 12:12:47 -0500 Subject: [PATCH] fix(server): snoozed threads allow queued automatic wakes (#19099) --- .../orchestration-v2/NotificationMailbox.ts | 13 ++ .../src/orchestration-v2/Orchestrator.ts | 62 +++++++- .../ProjectionRecovery.test.ts | 96 ++++++++++++ .../ProviderRuntimeRecoveryService.test.ts | 1 + .../ProviderRuntimeRecoveryService.ts | 34 ++++- .../src/orchestration-v2/ThreadStop.test.ts | 56 +++++++ .../src/orchestration-v2/runtimeLayer.test.ts | 143 ++++++++++++++++++ 7 files changed, 395 insertions(+), 10 deletions(-) diff --git a/apps/server/src/orchestration-v2/NotificationMailbox.ts b/apps/server/src/orchestration-v2/NotificationMailbox.ts index e7045e28d9e145a4022ee52d1bed17a679962cae..cbf76bf2842671c74a665d49ab51ccb052dc8b92 100644 --- a/apps/server/src/orchestration-v2/NotificationMailbox.ts +++ b/apps/server/src/orchestration-v2/NotificationMailbox.ts @@ -1,9 +1,22 @@ import { latestProviderTurnForAttempt, type MessageId, + type OrchestrationV2ConversationMessage, type OrchestrationV2ThreadProjection, } from "@t3tools/contracts"; +/** A wake attaches to output held in provider memory, which Stop or process loss discards. */ +export function isBufferedProviderNotification( + message: OrchestrationV2ConversationMessage, +): boolean { + return ( + message.createdBy === "agent" && + message.creationSource === "provider" && + message.notification !== undefined && + message.delegatedCompletion === undefined + ); +} + /** * A persisted steer without an acceptance receipt can be retried as a continuation. * Delivery is at least once: provider acceptance and our receipt cannot commit diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index e1497473f1de94a8b993c543d69e99e8be570370..12d4df5001e30c9af54950ab1fd0d0b2105f12d0 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -89,7 +89,10 @@ import { CommandReceiptStoreV2 } from "./CommandReceiptStore.ts"; import { ContextHandoffServiceV2 } from "./ContextHandoffService.ts"; import { notificationTurnItem } from "@t3tools/provider-core/server/notification"; import { isRestartNoteSource } from "./RestartBackgroundNote.ts"; -import { isUndeliveredMailboxSteer } from "./NotificationMailbox.ts"; +import { + isBufferedProviderNotification, + isUndeliveredMailboxSteer, +} from "./NotificationMailbox.ts"; import { EventSinkV2 } from "./EventSink.ts"; import * as EffectOutbox from "./EffectOutbox.ts"; import type { OrchestrationEffectRequestV2, PendingOrchestrationEffectV2 } from "./EffectOutbox.ts"; @@ -2798,7 +2801,24 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio cause: `Thread ${command.threadId} has a pending approval or user-input request and cannot be snoozed.`, }); } - if (projection.runs.some((run) => run.status === "queued")) { + const queuedRuns = projection.runs.filter((run) => run.status === "queued"); + const queuedMessages = + queuedRuns.length === 0 + ? [] + : (yield* loadProjectionForCommand(command, ["messages"], { + messageIds: queuedRuns.map((run) => run.userMessageId), + })).messages; + // Automatic wakes are hidden from the user queue and must not block snooze. + // Snooze only changes visibility, so keep those deliveries intact. + const automaticMessageIds = new Set( + queuedMessages + .filter( + (message) => + message.notification !== undefined || message.delegatedCompletion !== undefined, + ) + .map((message) => message.id), + ); + if (queuedRuns.some((run) => !automaticMessageIds.has(run.userMessageId))) { return yield* new OrchestratorDispatchError({ commandId: command.commandId, commandType: command.type, @@ -8797,9 +8817,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }); /** - * What Stop holds besides the run it interrupts: queued runs wait for the user, the - * thread's pull request watches end, and every wake its delegated tasks still owe is - * dropped, since Stop stops those tasks too. Nothing automatic starts the thread again. + * What Stop holds besides the run it interrupts: durable queued messages wait for the + * user, buffered provider wakes are cancelled, and the thread's pull request watches end. + * Every wake its delegated tasks still owe is dropped, since Stop stops those tasks too. * `cohortRunIds` are dropped even when they owe nothing, like a plain interrupt's. */ const holdStoppedThread = (input: { @@ -8815,6 +8835,22 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio Effect.gen(function* () { const emitEvent = emit(input.events, input.command); const { thread, runs, subagents } = input.projection; + const queuedRuns = runs.filter((run) => run.status === "queued"); + const queuedMessages = + queuedRuns.length === 0 + ? [] + : (yield* projectionStore + .getThreadRecords(thread.id, ["messages"], { + messageIds: queuedRuns.map((run) => run.userMessageId), + }) + .pipe( + Effect.mapError( + (cause) => new OrchestratorProjectionError({ threadId: thread.id, cause }), + ), + )).messages; + const bufferedMessageIds = new Set( + queuedMessages.filter(isBufferedProviderNotification).map((message) => message.id), + ); for (const run of runs) { // A resumed queue can have a lower ordinal than a pending continuation source. // Stop must reach those sources even when interrupting the resumed run succeeds. @@ -8824,7 +8860,21 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ) { yield* markStoppedRun({ ...input, thread, run }); } - if (run.status !== "queued" || run.queueHeld === true) continue; + if (run.status !== "queued") continue; + // Stop discards the provider's wake buffer, so its queued placeholders cannot resume. + if (bufferedMessageIds.has(run.userMessageId)) { + yield* dispatchQueuedRunCancel( + { + type: "queued-run.cancel", + commandId: input.command.commandId, + threadId: thread.id, + runId: run.id, + }, + input.events, + ); + continue; + } + if (run.queueHeld === true) continue; yield* emitEvent({ type: "run.updated", threadId: thread.id, diff --git a/apps/server/src/orchestration-v2/ProjectionRecovery.test.ts b/apps/server/src/orchestration-v2/ProjectionRecovery.test.ts index 95efab067f807e5c6557ffa75473ad25efcbbebe..e616b1d4f3b5ffcbb6eafbdc9ebb35b60750ee24 100644 --- a/apps/server/src/orchestration-v2/ProjectionRecovery.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionRecovery.test.ts @@ -23,14 +23,34 @@ import * as Layer from "effect/Layer"; import * as SqlClient from "effect/sql/SqlClient"; import * as SqlitePersistence from "../persistence/Sqlite.ts"; +import * as ServerSettings from "../serverSettings.ts"; import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts"; import * as EffectOutbox from "./EffectOutbox.ts"; import * as ProjectionStore from "./ProjectionStore.ts"; +import * as EventSink from "./EventSink.ts"; +import * as EventStore from "./EventStore.ts"; +import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; +import * as ProviderRuntimeRecovery from "./ProviderRuntimeRecoveryService.ts"; import { restartContinuationRun } from "./RestartContinuation.ts"; const layerTest = Layer.mergeAll(ProjectionStore.layer, EffectOutbox.layer).pipe( Layer.provideMerge(SqlitePersistence.layerMemory), ); +const layerRecoveryTest = ProviderRuntimeRecovery.layer.pipe( + Layer.provideMerge( + EventSink.layer.pipe( + Layer.provideMerge( + Layer.mergeAll( + EventStore.layer, + ProjectionStore.layer, + EffectOutbox.layer, + IdAllocator.layer, + ).pipe(Layer.provideMerge(SqlitePersistence.layerMemory)), + ), + ), + ), + Layer.provide(ServerSettings.layerTest()), +); const providerInstanceId = ProviderInstanceId.make("codex"); const modelSelection = { instanceId: providerInstanceId, model: "gpt-5.4" }; const driver = ProviderDriverKind.make("codex"); @@ -113,6 +133,82 @@ const createRun = Effect.fn(function* ( return runId; }); +it.effect.each(["startup", "shutdown"] as const)( + "%s cancels queued buffered wakes and holds durable messages", + (trigger) => + Effect.gen(function* () { + const projections = yield* ProjectionStore.ProjectionStoreV2; + const recovery = yield* ProviderRuntimeRecovery.ProviderRuntimeRecoveryService; + const now = yield* DateTime.now; + const bufferedThreads = [] as ThreadId[]; + const durableThreads = [] as ThreadId[]; + for (const kind of ["buffered", "user", "notification", "delegated"] as const) { + const threadId = yield* createThread(`wake:${trigger}:${kind}`); + const buffered = kind === "buffered"; + (buffered ? bufferedThreads : durableThreads).push(threadId); + const messageId = MessageId.make(`message:${threadId}`); + const runId = yield* createRun(threadId, "queued", { + userMessageId: messageId, + startedAt: null, + queuePosition: 1, + }); + yield* createRun(threadId, "completed", { ordinal: 2 }); + yield* projections.apply({ + id: EventId.make(`message:${threadId}`), + type: "message.updated", + threadId, + runId, + occurredAt: now, + payload: { + id: messageId, + threadId, + runId, + nodeId: null, + role: "user", + text: "Queued work", + attachments: [], + streaming: false, + createdAt: now, + updatedAt: now, + createdBy: kind === "user" ? "user" : "agent", + creationSource: buffered ? "provider" : "server", + ...(kind === "user" || kind === "delegated" + ? {} + : { + notification: { + source: { kind: "background_task" as const }, + outcome: "updated" as const, + summary: "Background activity updated", + }, + }), + ...(kind === "delegated" + ? { delegatedCompletion: { parentRunId: runId, generation: 1, taskIds: [] } } + : {}), + }, + }); + } + assert.deepEqual( + new Set(yield* projections.getRecoveryThreadIds("runtime")), + new Set([...bufferedThreads, ...durableThreads]), + ); + const summary = yield* recovery.reconcile(trigger); + assert.equal(summary.terminalizedRuns, 1); + for (const threadId of bufferedThreads) { + const projection = yield* projections.getThreadProjection(threadId); + assert.equal(projection.runs[0]?.status, "cancelled"); + assert.isNotNull(projection.runs[0]?.completedAt); + assert.isNull(projection.runs[0]?.queuePosition); + assert.equal(projection.runs[1]?.status, "completed"); + } + for (const threadId of durableThreads) { + const projection = yield* projections.getThreadProjection(threadId); + assert.equal(projection.runs[0]?.status, "queued"); + assert.isTrue(projection.runs[0]?.queueHeld); + } + assert.deepEqual(yield* projections.getRecoveryThreadIds("runtime"), []); + }).pipe(Effect.provide(layerRecoveryTest)), +); + it.effect("selects unfinished recovery work without reading settled thread histories", () => Effect.gen(function* () { const projections = yield* ProjectionStore.ProjectionStoreV2; diff --git a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts index 05c2b41c011648af006acedd12596ff0bf3d9f3c..1e2fcc44248de0692f5f8ce8e7fca2ba55504da6 100644 --- a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts @@ -690,6 +690,7 @@ it.effect("holds accepted queued work without cancelling its execution state aft Layer.mock(ProjectionStore.ProjectionStoreV2)({ getRecoveryThreadIds: () => Effect.succeed([threadId]), getRuntimeRecoveryProjection: () => Effect.succeed(projection), + getThreadRecords: () => Effect.succeed(projection), }), Layer.mock(EventSink.EventSinkV2)({ commitCommand: (input) => { diff --git a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts index ad57247e03847eb27b2cde47ea30fe3680eedf07..27cacf50da9416e1bbbc37426eb019850d05c551 100644 --- a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts +++ b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts @@ -22,6 +22,7 @@ import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; import * as ProjectionStore from "./ProjectionStore.ts"; import * as ServerSettings from "../serverSettings.ts"; import { restartContinuationRun } from "./RestartContinuation.ts"; +import { isBufferedProviderNotification } from "./NotificationMailbox.ts"; import { endOrphanedNativeSubagent, endRunlessRootTurns, @@ -284,10 +285,35 @@ export const make = Effect.gen(function* () { cancelledBackgroundNativeIds.add(item.nativeItemRef.nativeId); } }; - // Queued runs have not started provider work. Preserve their execution - // identities and order, but require explicit consent before draining them. - for (const run of projection.runs) { - if (run.status !== "queued" || run.queueHeld === true) continue; + const queuedRuns = projection.runs.filter((run) => run.status === "queued"); + const queuedMessages = + queuedRuns.length === 0 + ? [] + : (yield* projections + .getThreadRecords(projection.thread.id, ["messages"], { + messageIds: queuedRuns.map((run) => run.userMessageId), + }) + .pipe( + Effect.mapError( + (cause) => + new ProviderRuntimeRecoveryError({ + operation: "read-projections", + threadId: projection.thread.id, + cause, + }), + ), + )).messages; + const bufferedMessageIds = new Set( + queuedMessages.filter(isBufferedProviderNotification).map((message) => message.id), + ); + // Buffered wakes lost their output with the process. Durable queued messages + // keep their execution identities and order until the user resumes them. + for (const run of queuedRuns) { + if (bufferedMessageIds.has(run.userMessageId)) { + runs.push(run); + continue; + } + if (run.queueHeld === true) continue; events.push({ id: yield* allocateEventId(), type: "run.updated", diff --git a/apps/server/src/orchestration-v2/ThreadStop.test.ts b/apps/server/src/orchestration-v2/ThreadStop.test.ts index 73fc29a05569fda320df7790ab7f1e0dc4cda408..2357c149090129720fb7035cc325380ae2773071 100644 --- a/apps/server/src/orchestration-v2/ThreadStop.test.ts +++ b/apps/server/src/orchestration-v2/ThreadStop.test.ts @@ -225,6 +225,62 @@ it.effect("Stop ends watches, holds queues, and stops the delegated tasks under }).pipe(Effect.provide(layerTest)), ); +it.effect.each(["thread.stop", "run.interrupt"] as const)( + "%s cancels buffered wakes and preserves durable queued messages", + (type) => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const threadId = ThreadId.make(`thread:stop-buffered:${type}`); + yield* createWatchingThread(threadId, 5); + yield* send(threadId, "active", "start_immediately"); + yield* send(threadId, "user follow-up", "queue_after_active"); + for (const creationSource of ["provider", "server"] as const) { + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make(`notification:${type}:${creationSource}`), + threadId, + messageId: MessageId.make(`notification:${type}:${creationSource}`), + text: "Background activity updated", + attachments: [], + createdBy: "agent", + creationSource, + notification: { + source: { kind: "background_task" }, + outcome: "updated", + summary: "Background activity updated", + }, + dispatchMode: { type: "queue_after_active" }, + }); + } + const before = yield* orchestrator.getThreadProjection(threadId); + const buffered = before.runs.find((run) => + before.messages.some( + (message) => message.id === run.userMessageId && message.creationSource === "provider", + ), + )!; + const stop = { commandId: CommandId.make(`stop:${type}`), threadId }; + yield* orchestrator.dispatch( + type === "run.interrupt" + ? { ...stop, type, runId: before.runs[0]!.id, holdQueue: true } + : { ...stop, type }, + ); + const stopped = yield* orchestrator.getThreadProjection(threadId); + assert.equal(stopped.runs.find((run) => run.id === buffered.id)?.status, "cancelled"); + assert.equal( + stopped.attempts.find((attempt) => attempt.runId === buffered.id)?.status, + "cancelled", + ); + assert.equal( + stopped.nodes.find((node) => node.id === buffered.rootNodeId)?.status, + "cancelled", + ); + assert.deepEqual( + stopped.runs.filter((run) => run.status === "queued").map((run) => run.queueHeld), + [true, true], + ); + }).pipe(Effect.provide(layerTest)), +); + it.effect( "thread.stop ends an idle thread's watches and accepts a thread with nothing to stop", () => diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index d122289ecc08ee3997a733a7873da85374ddfc10..c4f5b8295fd9c8144920b57b5089482792ab08de 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -4516,6 +4516,149 @@ it.layer(layerSharedApplicationDataPlaneTest)("pending provider interruption", ( ); }); +it.layer(layerTest)("snooze queued runs", (it) => { + it.effect.each(["notification", "delegated-completion", "user"] as const)( + "only user messages block snooze with a held %s run", + (kind) => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const eventSink = yield* EventSink.EventSinkV2; + const threadId = ThreadId.make(`snooze-queued:${kind}`); + const snoozedUntil = "2099-07-25T09:00:00.000Z"; + yield* orchestrator.dispatch({ + type: "thread.create", + commandId: CommandId.make(`create:${threadId}`), + threadId, + projectId: ProjectId.make("snooze-queued-project"), + title: "Snooze with held queued work", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: "/tmp/snooze-queued", + createdBy: "user", + creationSource: "web", + }); + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make(`active:${threadId}`), + threadId, + messageId: MessageId.make(`active:${threadId}`), + text: "Active run", + attachments: [], + modelSelection, + dispatchMode: { type: "start_immediately" }, + createdBy: "user", + creationSource: "web", + }); + const activeRun = (yield* orchestrator.getThreadProjection(threadId)).runs[0]!; + const now = yield* DateTime.now; + if (kind === "delegated-completion") { + yield* eventSink.write({ + events: [ + { + id: EventId.make(`completion:${threadId}`), + type: "run.updated", + threadId, + runId: activeRun.id, + occurredAt: now, + payload: { + ...activeRun, + delegatedCompletion: { + disposition: "open", + nextGeneration: 2, + delivery: { + messageId: MessageId.make(`queued:${threadId}`), + generation: 1, + taskIds: [NodeId.make("snooze-queued-task")], + }, + }, + }, + }, + ], + }); + } + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make(`queued:${threadId}`), + threadId, + messageId: MessageId.make(`queued:${threadId}`), + text: "Queued work", + attachments: [], + modelSelection, + dispatchMode: { type: "queue_after_active" }, + createdBy: kind === "user" ? "user" : "agent", + creationSource: kind === "user" ? "web" : "server", + ...(kind === "notification" + ? { + notification: { + source: { kind: "background_task" as const }, + outcome: "updated" as const, + summary: "Background activity updated", + }, + } + : kind === "delegated-completion" + ? { + delegatedCompletion: { + parentRunId: activeRun.id, + generation: 1, + taskIds: [NodeId.make("snooze-queued-task")], + }, + } + : {}), + }); + // Recovery holds the queue while the thread's active run ends. + const beforeRecovery = yield* orchestrator.getThreadProjection(threadId); + yield* eventSink.commitCommand({ + commandId: CommandId.make(`recovery:${threadId}`), + threadId, + commandType: "provider-runtime.reconcile", + acceptedAt: now, + events: beforeRecovery.runs.map((run) => ({ + id: EventId.make(`recovery:${run.id}`), + type: "run.updated" as const, + threadId, + runId: run.id, + occurredAt: now, + payload: + run.status === "queued" + ? { ...run, queueHeld: true } + : { ...run, status: "completed" as const, completedAt: now }, + })), + effects: [], + }); + const beforeSnooze = yield* orchestrator.getThreadProjection(threadId); + assert.isTrue(beforeSnooze.runs.some((run) => run.status === "queued" && run.queueHeld)); + const snooze = orchestrator.dispatch({ + type: "thread.snooze", + commandId: CommandId.make(`snooze:${threadId}`), + threadId, + snoozedUntil, + }); + if (kind === "user") { + const error = yield* snooze.pipe(Effect.flip); + assert.equal(error._tag, "OrchestratorDispatchError"); + assert.equal(error.cause, `Thread ${threadId} has a queued run and cannot be snoozed.`); + assert.isNull((yield* orchestrator.getThreadProjection(threadId)).thread.snoozedUntil); + } else { + yield* snooze; + const snoozed = yield* orchestrator.getThreadProjection(threadId); + assert.equal(DateTime.formatIso(snoozed.thread.snoozedUntil!), snoozedUntil); + assert.deepEqual(snoozed.runs, beforeSnooze.runs); + yield* orchestrator.dispatch({ + type: "thread.unsnooze", + commandId: CommandId.make(`unsnooze:${threadId}`), + threadId, + reason: "user", + }); + const awake = yield* orchestrator.getThreadProjection(threadId); + assert.isNull(awake.thread.snoozedUntil); + assert.deepEqual(awake.runs, beforeSnooze.runs); + } + }), + ); +}); + it.layer(layerSharedApplicationDataPlaneTest)("snooze projection", (it) => { it.effect("carries snooze state through the V2 shell projection", () => Effect.gen(function* () { -- 2.54.0