authorgravatar for saiansh2525@gmail.comYash Singh <saiansh2525@gmail.com> 2026-10-13 12:12:47-05:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-10-13 17:12:47+00:00
log05e4f888a0d8591c790d036140080092cf024d30
treea1e44c647d6b8c7ecbd1e0af040efedb75f5bc96
parent8d7523248964ec7b37cc08f91e15122def171821
signature Signed by PGP key B5690EEEBB952194

fix(server): snoozed threads allow queued automatic wakes (#19099)

7 files changed, 395 insertions(+), 10 deletions(-)

apps/server/src/orchestration-v2/NotificationMailbox.ts+13
......@@ -1,9 +1,22 @@
11import {
22 latestProviderTurnForAttempt,
33 type MessageId,
4 type OrchestrationV2ConversationMessage,
45 type OrchestrationV2ThreadProjection,
56} from "@t3tools/contracts";
67
8/** A wake attaches to output held in provider memory, which Stop or process loss discards. */
9export function isBufferedProviderNotification(
10 message: OrchestrationV2ConversationMessage,
11): boolean {
12 return (
13 message.createdBy === "agent" &&
14 message.creationSource === "provider" &&
15 message.notification !== undefined &&
16 message.delegatedCompletion === undefined
17 );
18}
19
720/**
821 * A persisted steer without an acceptance receipt can be retried as a continuation.
922 * Delivery is at least once: provider acceptance and our receipt cannot commit
apps/server/src/orchestration-v2/Orchestrator.ts+56-6
......@@ -89,7 +89,10 @@ import { CommandReceiptStoreV2 } from "./CommandReceiptStore.ts";
8989import { ContextHandoffServiceV2 } from "./ContextHandoffService.ts";
9090import { notificationTurnItem } from "@t3tools/provider-core/server/notification";
9191import { isRestartNoteSource } from "./RestartBackgroundNote.ts";
92import { isUndeliveredMailboxSteer } from "./NotificationMailbox.ts";
92import {
93 isBufferedProviderNotification,
94 isUndeliveredMailboxSteer,
95} from "./NotificationMailbox.ts";
9396import { EventSinkV2 } from "./EventSink.ts";
9497import * as EffectOutbox from "./EffectOutbox.ts";
9598import type { OrchestrationEffectRequestV2, PendingOrchestrationEffectV2 } from "./EffectOutbox.ts";
......@@ -2798,7 +2801,24 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
27982801 cause: `Thread ${command.threadId} has a pending approval or user-input request and cannot be snoozed.`,
27992802 });
28002803 }
2801 if (projection.runs.some((run) => run.status === "queued")) {
2804 const queuedRuns = projection.runs.filter((run) => run.status === "queued");
2805 const queuedMessages =
2806 queuedRuns.length === 0
2807 ? []
2808 : (yield* loadProjectionForCommand(command, ["messages"], {
2809 messageIds: queuedRuns.map((run) => run.userMessageId),
2810 })).messages;
2811 // Automatic wakes are hidden from the user queue and must not block snooze.
2812 // Snooze only changes visibility, so keep those deliveries intact.
2813 const automaticMessageIds = new Set(
2814 queuedMessages
2815 .filter(
2816 (message) =>
2817 message.notification !== undefined || message.delegatedCompletion !== undefined,
2818 )
2819 .map((message) => message.id),
2820 );
2821 if (queuedRuns.some((run) => !automaticMessageIds.has(run.userMessageId))) {
28022822 return yield* new OrchestratorDispatchError({
28032823 commandId: command.commandId,
28042824 commandType: command.type,
......@@ -8797,9 +8817,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
87978817 });
87988818
87998819 /**
8800 * What Stop holds besides the run it interrupts: queued runs wait for the user, the
8801 * thread's pull request watches end, and every wake its delegated tasks still owe is
8802 * dropped, since Stop stops those tasks too. Nothing automatic starts the thread again.
8820 * What Stop holds besides the run it interrupts: durable queued messages wait for the
8821 * user, buffered provider wakes are cancelled, and the thread's pull request watches end.
8822 * Every wake its delegated tasks still owe is dropped, since Stop stops those tasks too.
88038823 * `cohortRunIds` are dropped even when they owe nothing, like a plain interrupt's.
88048824 */
88058825 const holdStoppedThread = (input: {
......@@ -8815,6 +8835,22 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
88158835 Effect.gen(function* () {
88168836 const emitEvent = emit(input.events, input.command);
88178837 const { thread, runs, subagents } = input.projection;
8838 const queuedRuns = runs.filter((run) => run.status === "queued");
8839 const queuedMessages =
8840 queuedRuns.length === 0
8841 ? []
8842 : (yield* projectionStore
8843 .getThreadRecords(thread.id, ["messages"], {
8844 messageIds: queuedRuns.map((run) => run.userMessageId),
8845 })
8846 .pipe(
8847 Effect.mapError(
8848 (cause) => new OrchestratorProjectionError({ threadId: thread.id, cause }),
8849 ),
8850 )).messages;
8851 const bufferedMessageIds = new Set(
8852 queuedMessages.filter(isBufferedProviderNotification).map((message) => message.id),
8853 );
88188854 for (const run of runs) {
88198855 // A resumed queue can have a lower ordinal than a pending continuation source.
88208856 // Stop must reach those sources even when interrupting the resumed run succeeds.
......@@ -8824,7 +8860,21 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
88248860 ) {
88258861 yield* markStoppedRun({ ...input, thread, run });
88268862 }
8827 if (run.status !== "queued" || run.queueHeld === true) continue;
8863 if (run.status !== "queued") continue;
8864 // Stop discards the provider's wake buffer, so its queued placeholders cannot resume.
8865 if (bufferedMessageIds.has(run.userMessageId)) {
8866 yield* dispatchQueuedRunCancel(
8867 {
8868 type: "queued-run.cancel",
8869 commandId: input.command.commandId,
8870 threadId: thread.id,
8871 runId: run.id,
8872 },
8873 input.events,
8874 );
8875 continue;
8876 }
8877 if (run.queueHeld === true) continue;
88288878 yield* emitEvent({
88298879 type: "run.updated",
88308880 threadId: thread.id,
apps/server/src/orchestration-v2/ProjectionRecovery.test.ts+96
......@@ -23,14 +23,34 @@ import * as Layer from "effect/Layer";
2323import * as SqlClient from "effect/sql/SqlClient";
2424
2525import * as SqlitePersistence from "../persistence/Sqlite.ts";
26import * as ServerSettings from "../serverSettings.ts";
2627import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts";
2728import * as EffectOutbox from "./EffectOutbox.ts";
2829import * as ProjectionStore from "./ProjectionStore.ts";
30import * as EventSink from "./EventSink.ts";
31import * as EventStore from "./EventStore.ts";
32import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
33import * as ProviderRuntimeRecovery from "./ProviderRuntimeRecoveryService.ts";
2934import { restartContinuationRun } from "./RestartContinuation.ts";
3035
3136const layerTest = Layer.mergeAll(ProjectionStore.layer, EffectOutbox.layer).pipe(
3237 Layer.provideMerge(SqlitePersistence.layerMemory),
3338);
39const layerRecoveryTest = ProviderRuntimeRecovery.layer.pipe(
40 Layer.provideMerge(
41 EventSink.layer.pipe(
42 Layer.provideMerge(
43 Layer.mergeAll(
44 EventStore.layer,
45 ProjectionStore.layer,
46 EffectOutbox.layer,
47 IdAllocator.layer,
48 ).pipe(Layer.provideMerge(SqlitePersistence.layerMemory)),
49 ),
50 ),
51 ),
52 Layer.provide(ServerSettings.layerTest()),
53);
3454const providerInstanceId = ProviderInstanceId.make("codex");
3555const modelSelection = { instanceId: providerInstanceId, model: "gpt-5.4" };
3656const driver = ProviderDriverKind.make("codex");
......@@ -113,6 +133,82 @@ const createRun = Effect.fn(function* (
113133 return runId;
114134});
115135
136it.effect.each(["startup", "shutdown"] as const)(
137 "%s cancels queued buffered wakes and holds durable messages",
138 (trigger) =>
139 Effect.gen(function* () {
140 const projections = yield* ProjectionStore.ProjectionStoreV2;
141 const recovery = yield* ProviderRuntimeRecovery.ProviderRuntimeRecoveryService;
142 const now = yield* DateTime.now;
143 const bufferedThreads = [] as ThreadId[];
144 const durableThreads = [] as ThreadId[];
145 for (const kind of ["buffered", "user", "notification", "delegated"] as const) {
146 const threadId = yield* createThread(`wake:${trigger}:${kind}`);
147 const buffered = kind === "buffered";
148 (buffered ? bufferedThreads : durableThreads).push(threadId);
149 const messageId = MessageId.make(`message:${threadId}`);
150 const runId = yield* createRun(threadId, "queued", {
151 userMessageId: messageId,
152 startedAt: null,
153 queuePosition: 1,
154 });
155 yield* createRun(threadId, "completed", { ordinal: 2 });
156 yield* projections.apply({
157 id: EventId.make(`message:${threadId}`),
158 type: "message.updated",
159 threadId,
160 runId,
161 occurredAt: now,
162 payload: {
163 id: messageId,
164 threadId,
165 runId,
166 nodeId: null,
167 role: "user",
168 text: "Queued work",
169 attachments: [],
170 streaming: false,
171 createdAt: now,
172 updatedAt: now,
173 createdBy: kind === "user" ? "user" : "agent",
174 creationSource: buffered ? "provider" : "server",
175 ...(kind === "user" || kind === "delegated"
176 ? {}
177 : {
178 notification: {
179 source: { kind: "background_task" as const },
180 outcome: "updated" as const,
181 summary: "Background activity updated",
182 },
183 }),
184 ...(kind === "delegated"
185 ? { delegatedCompletion: { parentRunId: runId, generation: 1, taskIds: [] } }
186 : {}),
187 },
188 });
189 }
190 assert.deepEqual(
191 new Set(yield* projections.getRecoveryThreadIds("runtime")),
192 new Set([...bufferedThreads, ...durableThreads]),
193 );
194 const summary = yield* recovery.reconcile(trigger);
195 assert.equal(summary.terminalizedRuns, 1);
196 for (const threadId of bufferedThreads) {
197 const projection = yield* projections.getThreadProjection(threadId);
198 assert.equal(projection.runs[0]?.status, "cancelled");
199 assert.isNotNull(projection.runs[0]?.completedAt);
200 assert.isNull(projection.runs[0]?.queuePosition);
201 assert.equal(projection.runs[1]?.status, "completed");
202 }
203 for (const threadId of durableThreads) {
204 const projection = yield* projections.getThreadProjection(threadId);
205 assert.equal(projection.runs[0]?.status, "queued");
206 assert.isTrue(projection.runs[0]?.queueHeld);
207 }
208 assert.deepEqual(yield* projections.getRecoveryThreadIds("runtime"), []);
209 }).pipe(Effect.provide(layerRecoveryTest)),
210);
211
116212it.effect("selects unfinished recovery work without reading settled thread histories", () =>
117213 Effect.gen(function* () {
118214 const projections = yield* ProjectionStore.ProjectionStoreV2;
apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts+1
......@@ -690,6 +690,7 @@ it.effect("holds accepted queued work without cancelling its execution state aft
690690 Layer.mock(ProjectionStore.ProjectionStoreV2)({
691691 getRecoveryThreadIds: () => Effect.succeed([threadId]),
692692 getRuntimeRecoveryProjection: () => Effect.succeed(projection),
693 getThreadRecords: () => Effect.succeed(projection),
693694 }),
694695 Layer.mock(EventSink.EventSinkV2)({
695696 commitCommand: (input) => {
apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts+30-4
......@@ -22,6 +22,7 @@ import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
2222import * as ProjectionStore from "./ProjectionStore.ts";
2323import * as ServerSettings from "../serverSettings.ts";
2424import { restartContinuationRun } from "./RestartContinuation.ts";
25import { isBufferedProviderNotification } from "./NotificationMailbox.ts";
2526import {
2627 endOrphanedNativeSubagent,
2728 endRunlessRootTurns,
......@@ -284,10 +285,35 @@ export const make = Effect.gen(function* () {
284285 cancelledBackgroundNativeIds.add(item.nativeItemRef.nativeId);
285286 }
286287 };
287 // Queued runs have not started provider work. Preserve their execution
288 // identities and order, but require explicit consent before draining them.
289 for (const run of projection.runs) {
290 if (run.status !== "queued" || run.queueHeld === true) continue;
288 const queuedRuns = projection.runs.filter((run) => run.status === "queued");
289 const queuedMessages =
290 queuedRuns.length === 0
291 ? []
292 : (yield* projections
293 .getThreadRecords(projection.thread.id, ["messages"], {
294 messageIds: queuedRuns.map((run) => run.userMessageId),
295 })
296 .pipe(
297 Effect.mapError(
298 (cause) =>
299 new ProviderRuntimeRecoveryError({
300 operation: "read-projections",
301 threadId: projection.thread.id,
302 cause,
303 }),
304 ),
305 )).messages;
306 const bufferedMessageIds = new Set(
307 queuedMessages.filter(isBufferedProviderNotification).map((message) => message.id),
308 );
309 // Buffered wakes lost their output with the process. Durable queued messages
310 // keep their execution identities and order until the user resumes them.
311 for (const run of queuedRuns) {
312 if (bufferedMessageIds.has(run.userMessageId)) {
313 runs.push(run);
314 continue;
315 }
316 if (run.queueHeld === true) continue;
291317 events.push({
292318 id: yield* allocateEventId(),
293319 type: "run.updated",
apps/server/src/orchestration-v2/ThreadStop.test.ts+56
......@@ -225,6 +225,62 @@ it.effect("Stop ends watches, holds queues, and stops the delegated tasks under
225225 }).pipe(Effect.provide(layerTest)),
226226);
227227
228it.effect.each(["thread.stop", "run.interrupt"] as const)(
229 "%s cancels buffered wakes and preserves durable queued messages",
230 (type) =>
231 Effect.gen(function* () {
232 const orchestrator = yield* Orchestrator.OrchestratorV2;
233 const threadId = ThreadId.make(`thread:stop-buffered:${type}`);
234 yield* createWatchingThread(threadId, 5);
235 yield* send(threadId, "active", "start_immediately");
236 yield* send(threadId, "user follow-up", "queue_after_active");
237 for (const creationSource of ["provider", "server"] as const) {
238 yield* orchestrator.dispatch({
239 type: "message.dispatch",
240 commandId: CommandId.make(`notification:${type}:${creationSource}`),
241 threadId,
242 messageId: MessageId.make(`notification:${type}:${creationSource}`),
243 text: "Background activity updated",
244 attachments: [],
245 createdBy: "agent",
246 creationSource,
247 notification: {
248 source: { kind: "background_task" },
249 outcome: "updated",
250 summary: "Background activity updated",
251 },
252 dispatchMode: { type: "queue_after_active" },
253 });
254 }
255 const before = yield* orchestrator.getThreadProjection(threadId);
256 const buffered = before.runs.find((run) =>
257 before.messages.some(
258 (message) => message.id === run.userMessageId && message.creationSource === "provider",
259 ),
260 )!;
261 const stop = { commandId: CommandId.make(`stop:${type}`), threadId };
262 yield* orchestrator.dispatch(
263 type === "run.interrupt"
264 ? { ...stop, type, runId: before.runs[0]!.id, holdQueue: true }
265 : { ...stop, type },
266 );
267 const stopped = yield* orchestrator.getThreadProjection(threadId);
268 assert.equal(stopped.runs.find((run) => run.id === buffered.id)?.status, "cancelled");
269 assert.equal(
270 stopped.attempts.find((attempt) => attempt.runId === buffered.id)?.status,
271 "cancelled",
272 );
273 assert.equal(
274 stopped.nodes.find((node) => node.id === buffered.rootNodeId)?.status,
275 "cancelled",
276 );
277 assert.deepEqual(
278 stopped.runs.filter((run) => run.status === "queued").map((run) => run.queueHeld),
279 [true, true],
280 );
281 }).pipe(Effect.provide(layerTest)),
282);
283
228284it.effect(
229285 "thread.stop ends an idle thread's watches and accepts a thread with nothing to stop",
230286 () =>
apps/server/src/orchestration-v2/runtimeLayer.test.ts+143
......@@ -4516,6 +4516,149 @@ it.layer(layerSharedApplicationDataPlaneTest)("pending provider interruption", (
45164516 );
45174517});
45184518
4519it.layer(layerTest)("snooze queued runs", (it) => {
4520 it.effect.each(["notification", "delegated-completion", "user"] as const)(
4521 "only user messages block snooze with a held %s run",
4522 (kind) =>
4523 Effect.gen(function* () {
4524 const orchestrator = yield* Orchestrator.OrchestratorV2;
4525 const eventSink = yield* EventSink.EventSinkV2;
4526 const threadId = ThreadId.make(`snooze-queued:${kind}`);
4527 const snoozedUntil = "2099-07-25T09:00:00.000Z";
4528 yield* orchestrator.dispatch({
4529 type: "thread.create",
4530 commandId: CommandId.make(`create:${threadId}`),
4531 threadId,
4532 projectId: ProjectId.make("snooze-queued-project"),
4533 title: "Snooze with held queued work",
4534 modelSelection,
4535 runtimeMode: "full-access",
4536 interactionMode: "default",
4537 branch: null,
4538 worktreePath: "/tmp/snooze-queued",
4539 createdBy: "user",
4540 creationSource: "web",
4541 });
4542 yield* orchestrator.dispatch({
4543 type: "message.dispatch",
4544 commandId: CommandId.make(`active:${threadId}`),
4545 threadId,
4546 messageId: MessageId.make(`active:${threadId}`),
4547 text: "Active run",
4548 attachments: [],
4549 modelSelection,
4550 dispatchMode: { type: "start_immediately" },
4551 createdBy: "user",
4552 creationSource: "web",
4553 });
4554 const activeRun = (yield* orchestrator.getThreadProjection(threadId)).runs[0]!;
4555 const now = yield* DateTime.now;
4556 if (kind === "delegated-completion") {
4557 yield* eventSink.write({
4558 events: [
4559 {
4560 id: EventId.make(`completion:${threadId}`),
4561 type: "run.updated",
4562 threadId,
4563 runId: activeRun.id,
4564 occurredAt: now,
4565 payload: {
4566 ...activeRun,
4567 delegatedCompletion: {
4568 disposition: "open",
4569 nextGeneration: 2,
4570 delivery: {
4571 messageId: MessageId.make(`queued:${threadId}`),
4572 generation: 1,
4573 taskIds: [NodeId.make("snooze-queued-task")],
4574 },
4575 },
4576 },
4577 },
4578 ],
4579 });
4580 }
4581 yield* orchestrator.dispatch({
4582 type: "message.dispatch",
4583 commandId: CommandId.make(`queued:${threadId}`),
4584 threadId,
4585 messageId: MessageId.make(`queued:${threadId}`),
4586 text: "Queued work",
4587 attachments: [],
4588 modelSelection,
4589 dispatchMode: { type: "queue_after_active" },
4590 createdBy: kind === "user" ? "user" : "agent",
4591 creationSource: kind === "user" ? "web" : "server",
4592 ...(kind === "notification"
4593 ? {
4594 notification: {
4595 source: { kind: "background_task" as const },
4596 outcome: "updated" as const,
4597 summary: "Background activity updated",
4598 },
4599 }
4600 : kind === "delegated-completion"
4601 ? {
4602 delegatedCompletion: {
4603 parentRunId: activeRun.id,
4604 generation: 1,
4605 taskIds: [NodeId.make("snooze-queued-task")],
4606 },
4607 }
4608 : {}),
4609 });
4610 // Recovery holds the queue while the thread's active run ends.
4611 const beforeRecovery = yield* orchestrator.getThreadProjection(threadId);
4612 yield* eventSink.commitCommand({
4613 commandId: CommandId.make(`recovery:${threadId}`),
4614 threadId,
4615 commandType: "provider-runtime.reconcile",
4616 acceptedAt: now,
4617 events: beforeRecovery.runs.map((run) => ({
4618 id: EventId.make(`recovery:${run.id}`),
4619 type: "run.updated" as const,
4620 threadId,
4621 runId: run.id,
4622 occurredAt: now,
4623 payload:
4624 run.status === "queued"
4625 ? { ...run, queueHeld: true }
4626 : { ...run, status: "completed" as const, completedAt: now },
4627 })),
4628 effects: [],
4629 });
4630 const beforeSnooze = yield* orchestrator.getThreadProjection(threadId);
4631 assert.isTrue(beforeSnooze.runs.some((run) => run.status === "queued" && run.queueHeld));
4632 const snooze = orchestrator.dispatch({
4633 type: "thread.snooze",
4634 commandId: CommandId.make(`snooze:${threadId}`),
4635 threadId,
4636 snoozedUntil,
4637 });
4638 if (kind === "user") {
4639 const error = yield* snooze.pipe(Effect.flip);
4640 assert.equal(error._tag, "OrchestratorDispatchError");
4641 assert.equal(error.cause, `Thread ${threadId} has a queued run and cannot be snoozed.`);
4642 assert.isNull((yield* orchestrator.getThreadProjection(threadId)).thread.snoozedUntil);
4643 } else {
4644 yield* snooze;
4645 const snoozed = yield* orchestrator.getThreadProjection(threadId);
4646 assert.equal(DateTime.formatIso(snoozed.thread.snoozedUntil!), snoozedUntil);
4647 assert.deepEqual(snoozed.runs, beforeSnooze.runs);
4648 yield* orchestrator.dispatch({
4649 type: "thread.unsnooze",
4650 commandId: CommandId.make(`unsnooze:${threadId}`),
4651 threadId,
4652 reason: "user",
4653 });
4654 const awake = yield* orchestrator.getThreadProjection(threadId);
4655 assert.isNull(awake.thread.snoozedUntil);
4656 assert.deepEqual(awake.runs, beforeSnooze.runs);
4657 }
4658 }),
4659 );
4660});
4661
45194662it.layer(layerSharedApplicationDataPlaneTest)("snooze projection", (it) => {
45204663 it.effect("carries snooze state through the V2 shell projection", () =>
45214664 Effect.gen(function* () {