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