Open History to see the last commit
1import { describe, expect, it } from "@effect/vitest";
2import {
3 CommandId,
4 RuntimeRequestId,
5 EnvironmentId,
6 MessageId,
7 RunId,
8 ORCHESTRATION_V2_WS_METHODS,
9 ProjectId,
10 ProviderInstanceId,
11 ThreadId,
12 type OrchestrationV2Command,
13 type OrchestrationV2ShellSnapshot,
14} from "@t3tools/contracts";
15import * as DateTime from "effect/DateTime";
16import * as SecureRandom from "effect/crypto/SecureRandom";
17import * as Deferred from "effect/Deferred";
18import * as Effect from "effect/Effect";
19import * as Exit from "effect/Exit";
20import * as Fiber from "effect/Fiber";
21import * as Layer from "effect/Layer";
22import * as Option from "effect/Option";
23import * as Queue from "effect/Queue";
24import * as Scope from "effect/Scope";
25import * as Stream from "effect/Stream";
26import * as SubscriptionRef from "effect/SubscriptionRef";
27import { Atom, AtomRegistry } from "effect/reactivity";
28
29import * as EnvironmentRegistry from "../connection/registry.ts";
30import {
31 AVAILABLE_CONNECTION_STATE,
32 BearerConnectionTarget,
33 type SupervisorConnectionState,
34} from "../connection/model.ts";
35import * as EnvironmentSupervisor from "../connection/supervisor.ts";
36import type { RpcSession } from "../rpc/session.ts";
37import { createThreadEnvironmentAtoms } from "./threadCommands.ts";
38import * as ThreadHistoryController from "./threadHistoryController.ts";
39
40const ENVIRONMENT_ID = EnvironmentId.make("remote");
41const THREAD_ID = ThreadId.make("thread");
42const NOW = DateTime.makeUnsafe("2026-09-12T10:00:00.000Z");
43const FUTURE = DateTime.makeUnsafe("2099-01-01T00:00:00.000Z");
44const APPROVAL = {
45 id: RuntimeRequestId.make("approval"),
46 kind: "command" as const,
47 createdAt: NOW,
48};
49const USER_INPUT = {
50 id: RuntimeRequestId.make("input"),
51 kind: "user_input" as const,
52 createdAt: NOW,
53};
54const SNAPSHOT: OrchestrationV2ShellSnapshot = {
55 snapshotSequence: 1,
56 schemaVersion: 2,
57 archivedThreads: [],
58 projects: [],
59 threads: [
60 {
61 id: THREAD_ID,
62 projectId: ProjectId.make("project"),
63 title: "Remote thread",
64 modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" },
65 runtimeMode: "full-access",
66 interactionMode: "default",
67 branch: null,
68 worktreePath: null,
69 createdBy: "user",
70 creationSource: "web",
71 providerInstanceId: ProviderInstanceId.make("codex"),
72 lineage: { rootThreadId: THREAD_ID, parentThreadId: null, relationshipToParent: null },
73 forkedFrom: null,
74 activeProviderThreadId: null,
75 latestRunId: null,
76 activeRunId: null,
77 status: "idle",
78 pendingRuntimeRequest: null,
79 latestVisibleMessage: null,
80 itemCount: 0,
81 visibleItemCount: 0,
82 deletedAt: null,
83 createdAt: NOW,
84 updatedAt: NOW,
85 archivedAt: null,
86 settledOverride: null,
87 settledAt: null,
88 pullRequests: [],
89 latestUserMessageAt: null,
90 hasActionableProposedPlan: false,
91 },
92 ],
93};
94
95const makeHarness = Effect.fn("TestThreadCommands.makeHarness")(function* (options?: {
96 readonly loadEarlier?: () => Effect.Effect<ThreadHistoryController.ThreadHistoryLoadEarlierResult>;
97 readonly reconcilePendingWork?: (threadId: ThreadId) => Effect.Effect<void>;
98}) {
99 const requests = yield* Queue.unbounded<{
100 command: OrchestrationV2Command;
101 reply: Deferred.Deferred<{ sequence: number }, Error>;
102 }>();
103 const reconciled = yield* Queue.unbounded<ThreadId>();
104 const connectionFollowed = yield* Deferred.make<void>();
105 const connection = yield* SubscriptionRef.make<SupervisorConnectionState>({
106 ...AVAILABLE_CONNECTION_STATE,
107 desired: true,
108 phase: "connected",
109 });
110 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
111 target: new BearerConnectionTarget({
112 environmentId: ENVIRONMENT_ID,
113 label: "Test",
114 connectionId: "test",
115 }),
116 state: connection,
117 session: yield* SubscriptionRef.make(
118 Option.some({
119 client: {
120 [ORCHESTRATION_V2_WS_METHODS.dispatchCommand]: (command: OrchestrationV2Command) =>
121 Effect.gen(function* () {
122 const reply = yield* Deferred.make<{ sequence: number }, Error>();
123 yield* Queue.offer(requests, { command, reply });
124 return yield* Deferred.await(reply);
125 }),
126 },
127 } as unknown as RpcSession),
128 ),
129 } as EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
130 const runtime = Atom.runtime(
131 Layer.mergeAll(
132 Layer.succeed(EnvironmentRegistry.EnvironmentRegistry, {
133 run: (_environmentId, effect) =>
134 Effect.provideService(effect, EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
135 followStream: (_environmentId, stream) =>
136 Stream.provideService(
137 stream,
138 EnvironmentSupervisor.EnvironmentSupervisor,
139 supervisor,
140 ).pipe(Stream.onStart(Deferred.succeed(connectionFollowed, undefined))),
141 } as EnvironmentRegistry.EnvironmentRegistry["Service"]),
142 Layer.succeed(
143 SecureRandom.SecureRandom,
144 SecureRandom.make((size) => new Uint8Array(size)),
145 ),
146 Layer.succeed(
147 ThreadHistoryController.ThreadHistoryController,
148 ThreadHistoryController.ThreadHistoryController.of({
149 register: () => Effect.die("unused"),
150 unregister: () => Effect.void,
151 loadEarlier: () => options?.loadEarlier?.() ?? Effect.die("unused"),
152 reconcilePendingWork: (_environmentId, threadId) =>
153 options?.reconcilePendingWork?.(threadId) ?? Queue.offer(reconciled, threadId),
154 }),
155 ),
156 ),
157 );
158 const snapshotAtom = Atom.family((_environmentId: EnvironmentId) =>
159 Atom.make<OrchestrationV2ShellSnapshot | null>(SNAPSHOT),
160 );
161 const commands = createThreadEnvironmentAtoms(runtime, snapshotAtom);
162 const registry = AtomRegistry.make();
163 yield* Effect.addFinalizer(() => Effect.sync(() => registry.dispose()));
164 const visibleAtom = commands.snapshotAtom(ENVIRONMENT_ID);
165 registry.mount(visibleAtom);
166 return {
167 registry,
168 commands,
169 snapshotAtom,
170 visibleAtom,
171 requests,
172 reconciled,
173 connection,
174 connectionFollowed,
175 };
176});
177
178describe("history loading", () => {
179 it.effect("sends a message on the thread while earlier history is loading", () =>
180 Effect.gen(function* () {
181 const historyStarted = yield* Deferred.make<void>();
182 const historyReply =
183 yield* Deferred.make<ThreadHistoryController.ThreadHistoryLoadEarlierResult>();
184 const h = yield* makeHarness({
185 loadEarlier: () =>
186 Deferred.succeed(historyStarted, undefined).pipe(
187 Effect.andThen(Deferred.await(historyReply)),
188 ),
189 });
190 yield* Effect.addFinalizer(() => Deferred.succeed(historyReply, { _tag: "loaded" }));
191 const history = h.commands.loadEarlierHistory.run(h.registry, {
192 environmentId: ENVIRONMENT_ID,
193 input: { threadId: THREAD_ID },
194 });
195 yield* Deferred.await(historyStarted);
196 const send = h.commands.startTurn.run(h.registry, {
197 environmentId: ENVIRONMENT_ID,
198 input: {
199 threadId: THREAD_ID,
200 message: {
201 messageId: MessageId.make("message"),
202 role: "user",
203 text: "Send while history loads",
204 attachments: [],
205 },
206 runtimeMode: "full-access",
207 interactionMode: "default",
208 dispatchMode: "start",
209 },
210 });
211 // Dispatched after Send on another thread, so it arrives first only if Send is queued.
212 const controlThreadId = ThreadId.make("control-thread");
213 const control = h.commands.interruptTurn.run(h.registry, {
214 environmentId: ENVIRONMENT_ID,
215 input: { threadId: controlThreadId, runId: RunId.make("control-run") },
216 });
217 const first = yield* Queue.take(h.requests);
218 yield* Deferred.succeed(first.reply, { sequence: 1 });
219 yield* Deferred.succeed(historyReply, { _tag: "loaded" });
220 const second = yield* Queue.take(h.requests);
221 yield* Deferred.succeed(second.reply, { sequence: 2 });
222 const results = yield* Effect.promise(() => Promise.all([history, send, control]));
223
224 expect(first.command).toMatchObject({ type: "message.dispatch", threadId: THREAD_ID });
225 expect(second.command).toMatchObject({ type: "run.interrupt", threadId: controlThreadId });
226 expect(results.map((value) => value._tag)).toEqual(["Success", "Success", "Success"]);
227 }),
228 );
229});
230
231describe("interrupt turn", () => {
232 // A stale cached item has nothing left to interrupt, so the server can reject Stop.
233 it.effect("reconciles pending work when the server rejects the interrupt", () =>
234 Effect.gen(function* () {
235 const h = yield* makeHarness();
236 const result = h.commands.interruptTurn.run(h.registry, {
237 environmentId: ENVIRONMENT_ID,
238 input: { threadId: THREAD_ID, runId: RunId.make("finished-run") },
239 });
240 const request = yield* Queue.take(h.requests);
241 expect(request.command.type).toBe("run.interrupt");
242 yield* Deferred.fail(request.reply, new Error("Run is not interruptible"));
243 expect((yield* Effect.promise(() => result))._tag).toBe("Failure");
244 expect(yield* Queue.take(h.reconciled)).toBe(THREAD_ID);
245 }),
246 );
247
248 // Mirrors the thread state's handler: the lookup runs in the thread's scope and is joined.
249 const stopWithBlockedLookup = Effect.fn("stopWithBlockedLookup")(function* () {
250 const threadScope = yield* Scope.make();
251 const asked = yield* Deferred.make<void>();
252 const h = yield* makeHarness({
253 reconcilePendingWork: () =>
254 Deferred.succeed(asked, undefined).pipe(
255 Effect.andThen(Effect.never),
256 Effect.forkIn(threadScope),
257 Effect.flatMap(Fiber.join),
258 ),
259 });
260 const result = h.commands.interruptTurn.run(h.registry, {
261 environmentId: ENVIRONMENT_ID,
262 input: { threadId: THREAD_ID, runId: RunId.make("running-run") },
263 });
264 const request = yield* Queue.take(h.requests);
265 yield* Deferred.succeed(request.reply, { sequence: 1 });
266 yield* Deferred.await(asked);
267 return { threadScope, result };
268 });
269
270 it.effect("resolves Stop without waiting for the pending work lookup", () =>
271 Effect.gen(function* () {
272 const { threadScope, result } = yield* stopWithBlockedLookup();
273 expect((yield* Effect.promise(() => result))._tag).toBe("Success");
274 yield* Scope.close(threadScope, Exit.void);
275 }),
276 );
277
278 it.effect("keeps an accepted Stop when the thread closes during the lookup", () =>
279 Effect.gen(function* () {
280 const { threadScope, result } = yield* stopWithBlockedLookup();
281 yield* Scope.close(threadScope, Exit.void);
282 expect((yield* Effect.promise(() => result))._tag).toBe("Success");
283 }),
284 );
285});
286
287describe("title regeneration", () => {
288 it.effect("resolves with the outcome recorded for its own request", () =>
289 Effect.gen(function* () {
290 const h = yield* makeHarness();
291 const source = h.snapshotAtom(ENVIRONMENT_ID);
292 const thread = SNAPSHOT.threads[0]!;
293 const target = { environmentId: ENVIRONMENT_ID, input: { threadId: THREAD_ID } };
294 const earlier = {
295 requestId: CommandId.make("earlier"),
296 status: "renamed" as const,
297 completedAt: NOW,
298 };
299
300 h.registry.set(source, null);
301 const result = h.commands.regenerateTitle.run(h.registry, target);
302 const request = yield* Queue.take(h.requests);
303 const requestId = request.command.commandId;
304 yield* Deferred.succeed(request.reply, { sequence: 2 });
305 yield* Deferred.await(h.connectionFollowed);
306 h.registry.set(source, {
307 ...SNAPSHOT,
308 snapshotSequence: 2,
309 threads: [
310 {
311 ...thread,
312 titleRegeneration: { requestId, startedAt: NOW },
313 lastTitleRegeneration: earlier,
314 },
315 ],
316 });
317 h.registry.set(source, {
318 ...SNAPSHOT,
319 snapshotSequence: 3,
320 threads: [],
321 archivedThreads: [
322 {
323 ...thread,
324 archivedAt: NOW,
325 titleRegeneration: null,
326 lastTitleRegeneration: {
327 requestId,
328 status: "failed",
329 message: "model unavailable",
330 completedAt: NOW,
331 },
332 },
333 ],
334 });
335 const outcome = yield* Effect.promise(() => result);
336 expect(outcome).toMatchObject({
337 _tag: "Success",
338 value: { status: "failed", message: "model unavailable" },
339 });
340
341 const superseded = h.commands.regenerateTitle.run(h.registry, target);
342 const second = yield* Queue.take(h.requests);
343 h.registry.set(source, {
344 ...SNAPSHOT,
345 snapshotSequence: 4,
346 threads: [{ ...thread, titleRegeneration: null, lastTitleRegeneration: earlier }],
347 });
348 yield* Deferred.succeed(second.reply, { sequence: 4 });
349 expect(yield* Effect.promise(() => superseded)).toMatchObject({
350 _tag: "Success",
351 value: null,
352 });
353 }),
354 );
355
356 it.effect("settles without an outcome when the environment disconnects while waiting", () =>
357 Effect.gen(function* () {
358 const h = yield* makeHarness();
359 const result = h.commands.regenerateTitle.run(h.registry, {
360 environmentId: ENVIRONMENT_ID,
361 input: { threadId: THREAD_ID },
362 });
363 const request = yield* Queue.take(h.requests);
364 yield* Deferred.succeed(request.reply, { sequence: 2 });
365 yield* SubscriptionRef.update(h.connection, (state) => ({
366 ...state,
367 phase: "backoff" as const,
368 }));
369 expect(yield* Effect.promise(() => result)).toMatchObject({ _tag: "Success", value: null });
370 }),
371 );
372});
373
374describe("remote thread lifecycle commands", () => {
375 const actions = [
376 ["settle", {}, { settledOverride: "settled", pinnedAt: null, snoozedUntil: null }],
377 ["unsettle", { reason: "user" }, { settledOverride: "active", settledAt: null }],
378 ["snooze", { snoozedUntil: "2099-01-01T00:00:00.000Z" }, { snoozedUntil: FUTURE }],
379 ["unsnooze", { reason: "user" }, { snoozedUntil: null, snoozedAt: null }],
380 ["pin", { orderKey: "a" }, { pinnedAt: expect.any(Object), pinOrderKey: "a" }],
381 ["unpin", {}, { pinnedAt: null, pinOrderKey: null }],
382 ["setAutoSettle", { enabled: false }, { autoSettleDisabledAt: expect.any(Object) }],
383 ["reorderPin", { orderKey: "b" }, { pinOrderKey: "b" }],
384 ["reorderActive", { orderKey: "b" }, { activeOrderKey: "b" }],
385 ] as const;
386
387 it.effect.each(actions)(
388 "shows %s before a delayed remote reply and rolls back a rejection",
389 ([action, input, expected]) =>
390 Effect.gen(function* () {
391 const h = yield* makeHarness();
392 const source = h.snapshotAtom(ENVIRONMENT_ID);
393 const initial = {
394 ...SNAPSHOT,
395 threads: [
396 {
397 ...SNAPSHOT.threads[0]!,
398 ...(action === "unsettle" || action === "pin"
399 ? { settledOverride: "settled" as const, settledAt: NOW }
400 : {}),
401 ...(action === "unsnooze" || action === "settle" || action === "pin"
402 ? { snoozedUntil: FUTURE, snoozedAt: NOW }
403 : {}),
404 ...(action === "unpin" || action === "settle"
405 ? { pinnedAt: NOW, pinOrderKey: "a" }
406 : {}),
407 },
408 ],
409 };
410 h.registry.set(source, initial);
411 const result = h.commands[action].run(h.registry, {
412 environmentId: ENVIRONMENT_ID,
413 input: {
414 threadId: THREAD_ID,
415 commandId: CommandId.make(action),
416 reason: "user",
417 enabled: false,
418 orderKey: "a",
419 snoozedUntil: "2099-01-01T00:00:00.000Z",
420 ...input,
421 },
422 });
423 expect(h.registry.get(h.visibleAtom)?.threads[0]).toMatchObject(expected);
424 const request = yield* Queue.take(h.requests);
425 expect(h.registry.get(source)).toBe(initial);
426 yield* Deferred.fail(request.reply, new Error("Remote rejected the action"));
427 expect((yield* Effect.promise(() => result))._tag).toBe("Failure");
428 expect(h.registry.get(h.visibleAtom)).toBe(initial);
429 }),
430 );
431
432 it.effect("restamps snoozedAt when re-snoozing to the same wake time", () =>
433 Effect.gen(function* () {
434 const h = yield* makeHarness();
435 const earlierSnoozedAt = DateTime.makeUnsafe("2000-01-01T00:00:00.000Z");
436 h.registry.set(h.snapshotAtom(ENVIRONMENT_ID), {
437 ...SNAPSHOT,
438 threads: [{ ...SNAPSHOT.threads[0]!, snoozedUntil: FUTURE, snoozedAt: earlierSnoozedAt }],
439 });
440 const result = h.commands.snooze.run(h.registry, {
441 environmentId: ENVIRONMENT_ID,
442 input: { threadId: THREAD_ID, snoozedUntil: DateTime.formatIso(FUTURE) },
443 });
444 const snoozedAt = h.registry.get(h.visibleAtom)?.threads[0]?.snoozedAt;
445 expect(DateTime.toEpochMillis(snoozedAt!)).toBeGreaterThan(
446 DateTime.toEpochMillis(earlierSnoozedAt),
447 );
448 const request = yield* Queue.take(h.requests);
449 yield* Deferred.succeed(request.reply, { sequence: 2 });
450 expect((yield* Effect.promise(() => result))._tag).toBe("Success");
451 }),
452 );
453
454 it.effect("keeps the preview after acknowledgement until the matching shell update arrives", () =>
455 Effect.gen(function* () {
456 const h = yield* makeHarness();
457 const result = h.commands.settle.run(h.registry, {
458 environmentId: ENVIRONMENT_ID,
459 input: { threadId: THREAD_ID },
460 });
461 const request = yield* Queue.take(h.requests);
462 yield* Deferred.succeed(request.reply, { sequence: 3 });
463 expect((yield* Effect.promise(() => result))._tag).toBe("Success");
464 const changed = {
465 ...SNAPSHOT,
466 snapshotSequence: 2,
467 threads: [{ ...SNAPSHOT.threads[0]!, title: "Renamed remotely" }],
468 };
469 h.registry.set(h.snapshotAtom(ENVIRONMENT_ID), changed);
470 expect(h.registry.get(h.visibleAtom)?.threads[0]).toMatchObject({
471 title: "Renamed remotely",
472 settledOverride: "settled",
473 });
474 const confirmed = {
475 ...changed,
476 snapshotSequence: 3,
477 threads: [
478 {
479 ...changed.threads[0]!,
480 settledOverride: "settled" as const,
481 settledAt: DateTime.makeUnsafe("2026-09-12T12:00:00.000Z"),
482 },
483 ],
484 };
485 h.registry.set(h.snapshotAtom(ENVIRONMENT_ID), confirmed);
486 expect(h.registry.get(h.visibleAtom)).toBe(confirmed);
487 h.registry.set(h.snapshotAtom(ENVIRONMENT_ID), { ...SNAPSHOT, snapshotSequence: 4 });
488 expect(h.registry.get(h.visibleAtom)?.threads[0]?.settledOverride).toBeNull();
489 }),
490 );
491
492 it.effect(
493 "shows a queued reverse action immediately and preserves it if the earlier action fails",
494 () =>
495 Effect.gen(function* () {
496 const h = yield* makeHarness();
497 const settle = h.commands.settle.run(h.registry, {
498 environmentId: ENVIRONMENT_ID,
499 input: { threadId: THREAD_ID },
500 });
501 const first = yield* Queue.take(h.requests);
502 const unsettle = h.commands.unsettle.run(h.registry, {
503 environmentId: ENVIRONMENT_ID,
504 input: { threadId: THREAD_ID, reason: "user" },
505 });
506 expect(h.registry.get(h.visibleAtom)?.threads[0]?.settledOverride).toBe("active");
507 yield* Deferred.fail(first.reply, new Error("Settle rejected"));
508 yield* Effect.promise(() => settle);
509 expect(h.registry.get(h.visibleAtom)?.threads[0]?.settledOverride).toBe("active");
510 const second = yield* Queue.take(h.requests);
511 expect(second.command.type).toBe("thread.unsettle");
512 const confirmed = {
513 ...SNAPSHOT,
514 snapshotSequence: 2,
515 threads: [{ ...SNAPSHOT.threads[0]!, settledOverride: "active" as const }],
516 };
517 h.registry.set(h.snapshotAtom(ENVIRONMENT_ID), confirmed);
518 yield* Deferred.succeed(second.reply, { sequence: 2 });
519 yield* Effect.promise(() => unsettle);
520 expect(h.registry.get(h.visibleAtom)).toBe(confirmed);
521 }),
522 );
523
524 it.effect("isolates environments and does not restore a remotely removed thread", () =>
525 Effect.gen(function* () {
526 const h = yield* makeHarness();
527 const otherEnvironment = EnvironmentId.make("other-remote");
528 const result = h.commands.settle.run(h.registry, {
529 environmentId: ENVIRONMENT_ID,
530 input: { threadId: THREAD_ID },
531 });
532 const request = yield* Queue.take(h.requests);
533 expect(h.registry.get(h.commands.snapshotAtom(otherEnvironment))).toBe(SNAPSHOT);
534 expect(h.registry.get(h.commands.rowActionsAtom(ENVIRONMENT_ID))).toBe(1);
535 expect(h.registry.get(h.commands.rowActionsAtom(otherEnvironment))).toBe(0);
536 const removed = { ...SNAPSHOT, snapshotSequence: 2, threads: [] };
537 h.registry.set(h.snapshotAtom(ENVIRONMENT_ID), removed);
538 expect(h.registry.get(h.visibleAtom)?.threads).toEqual([]);
539 yield* Deferred.fail(request.reply, new Error("Thread removed"));
540 yield* Effect.promise(() => result);
541 expect(h.registry.get(h.visibleAtom)).toBe(removed);
542 }),
543 );
544
545 const undoableActions = ["settle", "snooze"] as const;
546
547 it.effect.each(undoableActions)("restores a confirmed %s when a queued undo fails", (action) =>
548 Effect.gen(function* () {
549 const h = yield* makeHarness();
550 const parked =
551 action === "settle" ? { settledOverride: "settled" as const } : { snoozedUntil: FUTURE };
552 const awake = action === "settle" ? { settledOverride: "active" } : { snoozedUntil: null };
553 const result = h.commands[action].run(h.registry, {
554 environmentId: ENVIRONMENT_ID,
555 input: { threadId: THREAD_ID, snoozedUntil: "2099-01-01T00:00:00.000Z" },
556 });
557 const first = yield* Queue.take(h.requests);
558 const undo = h.commands[action === "settle" ? "unsettle" : "unsnooze"].run(h.registry, {
559 environmentId: ENVIRONMENT_ID,
560 input: { threadId: THREAD_ID, reason: "user" },
561 });
562 expect(h.registry.get(h.visibleAtom)?.threads[0]).toMatchObject(awake);
563 yield* Deferred.succeed(first.reply, { sequence: 2 });
564 expect((yield* Effect.promise(() => result))._tag).toBe("Success");
565 expect(h.registry.get(h.visibleAtom)?.threads[0]).toMatchObject(awake);
566 const confirmed = {
567 ...SNAPSHOT,
568 snapshotSequence: 2,
569 threads: [{ ...SNAPSHOT.threads[0]!, ...parked }],
570 };
571 h.registry.set(h.snapshotAtom(ENVIRONMENT_ID), confirmed);
572 expect(h.registry.get(h.visibleAtom)?.threads[0]).toMatchObject(awake);
573 const second = yield* Queue.take(h.requests);
574 expect(second.command.type).toBe(action === "settle" ? "thread.unsettle" : "thread.unsnooze");
575 yield* Deferred.fail(second.reply, new Error("Undo rejected"));
576 expect((yield* Effect.promise(() => undo))._tag).toBe("Failure");
577 expect(h.registry.get(h.visibleAtom)).toBe(confirmed);
578 }),
579 );
580
581 it.effect.each(undoableActions)(
582 "preserves a newer approval when the %s reply arrives after the shell",
583 (action) =>
584 Effect.gen(function* () {
585 const h = yield* makeHarness();
586 const result = h.commands[action].run(h.registry, {
587 environmentId: ENVIRONMENT_ID,
588 input: { threadId: THREAD_ID, snoozedUntil: "2099-01-01T00:00:00.000Z" },
589 });
590 const request = yield* Queue.take(h.requests);
591 const newer = {
592 ...SNAPSHOT,
593 snapshotSequence: 3,
594 threads: [{ ...SNAPSHOT.threads[0]!, pendingRuntimeRequest: APPROVAL }],
595 };
596 h.registry.set(h.snapshotAtom(ENVIRONMENT_ID), newer);
597 expect(h.registry.get(h.visibleAtom)?.threads[0]).toBe(newer.threads[0]);
598 yield* Deferred.succeed(request.reply, { sequence: 2 });
599 expect((yield* Effect.promise(() => result))._tag).toBe("Success");
600 expect(h.registry.get(h.visibleAtom)).toBe(newer);
601 }),
602 );
603
604 it.effect.each(undoableActions)(
605 "shows an accepted %s while the shell still has an old input request",
606 (action) =>
607 Effect.gen(function* () {
608 const h = yield* makeHarness();
609 const stale = {
610 ...SNAPSHOT,
611 threads: [{ ...SNAPSHOT.threads[0]!, pendingRuntimeRequest: USER_INPUT }],
612 };
613 h.registry.set(h.snapshotAtom(ENVIRONMENT_ID), stale);
614 const result = h.commands[action].run(h.registry, {
615 environmentId: ENVIRONMENT_ID,
616 input: { threadId: THREAD_ID, snoozedUntil: "2099-01-01T00:00:00.000Z" },
617 });
618 const request = yield* Queue.take(h.requests);
619 if (action === "settle") {
620 expect(h.registry.get(h.visibleAtom)?.threads[0]).toBe(stale.threads[0]);
621 } else {
622 expect(h.registry.get(h.visibleAtom)?.threads[0]).toMatchObject({
623 snoozedUntil: FUTURE,
624 pendingRuntimeRequest: USER_INPUT,
625 });
626 }
627 yield* Deferred.succeed(request.reply, { sequence: 2 });
628 expect((yield* Effect.promise(() => result))._tag).toBe("Success");
629 expect(h.registry.get(h.visibleAtom)?.threads[0]).toMatchObject(
630 action === "settle" ? { settledOverride: "settled" } : { snoozedUntil: FUTURE },
631 );
632 expect(h.registry.get(h.visibleAtom)?.threads[0]?.pendingRuntimeRequest).toEqual(
633 action === "settle" ? null : USER_INPUT,
634 );
635 expect(h.registry.get(h.snapshotAtom(ENVIRONMENT_ID))).toBe(stale);
636 }),
637 );
638 it.effect("acknowledges a new question when snoozing again to the same wake time", () =>
639 Effect.gen(function* () {
640 const h = yield* makeHarness();
641 const initial = {
642 ...SNAPSHOT,
643 threads: [
644 {
645 ...SNAPSHOT.threads[0]!,
646 pendingRuntimeRequest: USER_INPUT,
647 snoozedAt: DateTime.makeUnsafe("2026-09-12T09:00:00.000Z"),
648 snoozedUntil: FUTURE,
649 },
650 ],
651 };
652 h.registry.set(h.snapshotAtom(ENVIRONMENT_ID), initial);
653 const result = h.commands.snooze.run(h.registry, {
654 environmentId: ENVIRONMENT_ID,
655 input: { threadId: THREAD_ID, snoozedUntil: "2099-01-01T00:00:00.000Z" },
656 });
657 const request = yield* Queue.take(h.requests);
658 const snoozed = h.registry.get(h.visibleAtom)?.threads[0];
659 expect(DateTime.toEpochMillis(snoozed!.snoozedAt!)).toBeGreaterThanOrEqual(
660 DateTime.toEpochMillis(USER_INPUT.createdAt),
661 );
662 expect(snoozed?.pendingRuntimeRequest).toEqual(USER_INPUT);
663 yield* Deferred.succeed(request.reply, { sequence: 2 });
664 expect((yield* Effect.promise(() => result))._tag).toBe("Success");
665 }),
666 );
667});