1import { describe, expect, it } from "@effect/vitest";
2import {
3 CommandId,
4 RuntimeRequestId,
5 EnvironmentId,
6 ORCHESTRATION_V2_WS_METHODS,
7 ProjectId,
8 ProviderInstanceId,
9 ThreadId,
10 type OrchestrationV2Command,
11 type OrchestrationV2ShellSnapshot,
12} from "@t3tools/contracts";
13import * as DateTime from "effect/DateTime";
14import * as Crypto from "effect/Crypto";
15import * as Deferred from "effect/Deferred";
16import * as Effect from "effect/Effect";
17import * as Layer from "effect/Layer";
18import * as Option from "effect/Option";
19import * as Queue from "effect/Queue";
20import * as SubscriptionRef from "effect/SubscriptionRef";
21import { Atom, AtomRegistry } from "effect/reactivity";
22
23import * as EnvironmentRegistry from "../connection/registry.ts";
24import * as EnvironmentSupervisor from "../connection/supervisor.ts";
25import type { RpcSession } from "../rpc/session.ts";
26import { createThreadEnvironmentAtoms } from "./threadCommands.ts";
27import { applyHubInboxEnvironment } from "./hub.ts";
28
29const ENVIRONMENT_ID = EnvironmentId.make("remote");
30const THREAD_ID = ThreadId.make("thread");
31const NOW = DateTime.makeUnsafe("2026-09-12T10:00:00.000Z");
32const FUTURE = DateTime.makeUnsafe("2099-01-01T00:00:00.000Z");
33const APPROVAL = {
34 id: RuntimeRequestId.make("approval"),
35 kind: "command" as const,
36 createdAt: NOW,
37};
38const USER_INPUT = {
39 id: RuntimeRequestId.make("input"),
40 kind: "user_input" as const,
41 createdAt: NOW,
42};
43const SNAPSHOT: OrchestrationV2ShellSnapshot = {
44 snapshotSequence: 1,
45 schemaVersion: 2,
46 archivedThreads: [],
47 projects: [],
48 threads: [
49 {
50 id: THREAD_ID,
51 projectId: ProjectId.make("project"),
52 title: "Remote thread",
53 modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" },
54 runtimeMode: "full-access",
55 interactionMode: "default",
56 branch: null,
57 worktreePath: null,
58 createdBy: "user",
59 creationSource: "web",
60 providerInstanceId: ProviderInstanceId.make("codex"),
61 lineage: { rootThreadId: THREAD_ID, parentThreadId: null, relationshipToParent: null },
62 forkedFrom: null,
63 activeProviderThreadId: null,
64 latestRunId: null,
65 activeRunId: null,
66 status: "idle",
67 pendingRuntimeRequest: null,
68 latestVisibleMessage: null,
69 itemCount: 0,
70 visibleItemCount: 0,
71 deletedAt: null,
72 createdAt: NOW,
73 updatedAt: NOW,
74 archivedAt: null,
75 settledOverride: null,
76 settledAt: null,
77 pullRequests: [],
78 latestUserMessageAt: null,
79 hasActionableProposedPlan: false,
80 },
81 ],
82};
83
84describe("central inbox shell state", () => {
85 const entry = {
86 environmentId: ENVIRONMENT_ID,
87 pendingThreadIds: [THREAD_ID],
88 snapshot: {
89 ...SNAPSHOT,
90 threads: SNAPSHOT.threads.map((thread) => ({
91 ...thread,
92 settledOverride: "settled" as const,
93 settledAt: NOW,
94 })),
95 },
96 };
97
98 it("shows cached chat metadata on a new client with no source snapshot", () => {
99 const state = applyHubInboxEnvironment(
100 {
101 snapshot: Option.none(),
102 status: "empty",
103 error: Option.none(),
104 },
105 entry,
106 );
107 expect(state.status).toBe("cached");
108 expect(Option.getOrNull(state.snapshot)?.threads[0]?.settledOverride).toBe("settled");
109 });
110
111 it("applies pending organization without hiding live execution", () => {
112 const live = {
113 ...SNAPSHOT,
114 threads: SNAPSHOT.threads.map((thread) => ({ ...thread, status: "running" as const })),
115 };
116 const state = { snapshot: Option.some(live), status: "live" as const, error: Option.none() };
117 const merged = Option.getOrThrow(applyHubInboxEnvironment(state, entry).snapshot);
118 expect(merged.threads[0]?.settledOverride).toBe("settled");
119 expect(merged.threads[0]?.status).toBe("running");
120 expect(merged.snapshotSequence).toBe(live.snapshotSequence);
121 expect(applyHubInboxEnvironment(state, { ...entry, pendingThreadIds: [] })).toBe(state);
122 });
123});
124
125const makeHarness = Effect.fn("TestThreadCommands.makeHarness")(function* () {
126 const requests = yield* Queue.unbounded<{
127 command: OrchestrationV2Command;
128 reply: Deferred.Deferred<{ sequence: number }, Error>;
129 }>();
130 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
131 target: { environmentId: ENVIRONMENT_ID },
132 session: yield* SubscriptionRef.make(
133 Option.some({
134 client: {
135 [ORCHESTRATION_V2_WS_METHODS.dispatchCommand]: (command: OrchestrationV2Command) =>
136 Effect.gen(function* () {
137 const reply = yield* Deferred.make<{ sequence: number }, Error>();
138 yield* Queue.offer(requests, { command, reply });
139 return yield* Deferred.await(reply);
140 }),
141 },
142 } as unknown as RpcSession),
143 ),
144 } as EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
145 const runtime = Atom.runtime(
146 Layer.mergeAll(
147 Layer.succeed(EnvironmentRegistry.EnvironmentRegistry, {
148 run: (_environmentId, effect) =>
149 Effect.provideService(effect, EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
150 } as EnvironmentRegistry.EnvironmentRegistry["Service"]),
151 Layer.succeed(
152 Crypto.Crypto,
153 Crypto.make({
154 randomBytes: (size) => new Uint8Array(size),
155 digest: (_algorithm, data) => Effect.succeed(data),
156 }),
157 ),
158 ),
159 );
160 const snapshotAtom = Atom.family((_environmentId: EnvironmentId) => Atom.make(SNAPSHOT));
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 { registry, commands, snapshotAtom, visibleAtom, requests };
167});
168
169describe("remote thread lifecycle commands", () => {
170 const actions = [
171 ["settle", {}, { settledOverride: "settled", pinnedAt: null, snoozedUntil: null }],
172 ["unsettle", { reason: "user" }, { settledOverride: "active", settledAt: null }],
173 ["snooze", { snoozedUntil: "2099-01-01T00:00:00.000Z" }, { snoozedUntil: FUTURE }],
174 ["unsnooze", { reason: "user" }, { snoozedUntil: null, snoozedAt: null }],
175 ["pin", { orderKey: "a" }, { pinnedAt: expect.any(Object), pinOrderKey: "a" }],
176 ["unpin", {}, { pinnedAt: null, pinOrderKey: null }],
177 ["setAutoSettle", { enabled: false }, { autoSettleDisabledAt: expect.any(Object) }],
178 ["reorderPin", { orderKey: "b" }, { pinOrderKey: "b" }],
179 ["reorderActive", { orderKey: "b" }, { activeOrderKey: "b" }],
180 ] as const;
181
182 it.effect.each(actions)(
183 "shows %s before a delayed remote reply and rolls back a rejection",
184 ([action, input, expected]) =>
185 Effect.gen(function* () {
186 const h = yield* makeHarness();
187 const source = h.snapshotAtom(ENVIRONMENT_ID);
188 const initial = {
189 ...SNAPSHOT,
190 threads: [
191 {
192 ...SNAPSHOT.threads[0]!,
193 ...(action === "unsettle" || action === "pin"
194 ? { settledOverride: "settled" as const, settledAt: NOW }
195 : {}),
196 ...(action === "unsnooze" || action === "settle" || action === "pin"
197 ? { snoozedUntil: FUTURE, snoozedAt: NOW }
198 : {}),
199 ...(action === "unpin" || action === "settle"
200 ? { pinnedAt: NOW, pinOrderKey: "a" }
201 : {}),
202 },
203 ],
204 };
205 h.registry.set(source, initial);
206 const result = h.commands[action].run(h.registry, {
207 environmentId: ENVIRONMENT_ID,
208 input: {
209 threadId: THREAD_ID,
210 commandId: CommandId.make(action),
211 reason: "user",
212 enabled: false,
213 orderKey: "a",
214 snoozedUntil: "2099-01-01T00:00:00.000Z",
215 ...input,
216 },
217 });
218 expect(h.registry.get(h.visibleAtom)?.threads[0]).toMatchObject(expected);
219 const request = yield* Queue.take(h.requests);
220 expect(h.registry.get(source)).toBe(initial);
221 yield* Deferred.fail(request.reply, new Error("Remote rejected the action"));
222 expect((yield* Effect.promise(() => result))._tag).toBe("Failure");
223 expect(h.registry.get(h.visibleAtom)).toBe(initial);
224 }),
225 );
226
227 it.effect("keeps the preview after acknowledgement until the matching shell update arrives", () =>
228 Effect.gen(function* () {
229 const h = yield* makeHarness();
230 const result = h.commands.settle.run(h.registry, {
231 environmentId: ENVIRONMENT_ID,
232 input: { threadId: THREAD_ID },
233 });
234 const request = yield* Queue.take(h.requests);
235 yield* Deferred.succeed(request.reply, { sequence: 3 });
236 expect((yield* Effect.promise(() => result))._tag).toBe("Success");
237 const changed = {
238 ...SNAPSHOT,
239 snapshotSequence: 2,
240 threads: [{ ...SNAPSHOT.threads[0]!, title: "Renamed remotely" }],
241 };
242 h.registry.set(h.snapshotAtom(ENVIRONMENT_ID), changed);
243 expect(h.registry.get(h.visibleAtom)?.threads[0]).toMatchObject({
244 title: "Renamed remotely",
245 settledOverride: "settled",
246 });
247 const confirmed = {
248 ...changed,
249 snapshotSequence: 3,
250 threads: [
251 {
252 ...changed.threads[0]!,
253 settledOverride: "settled" as const,
254 settledAt: DateTime.makeUnsafe("2026-09-12T12:00:00.000Z"),
255 },
256 ],
257 };
258 h.registry.set(h.snapshotAtom(ENVIRONMENT_ID), confirmed);
259 expect(h.registry.get(h.visibleAtom)).toBe(confirmed);
260 h.registry.set(h.snapshotAtom(ENVIRONMENT_ID), { ...SNAPSHOT, snapshotSequence: 4 });
261 expect(h.registry.get(h.visibleAtom)?.threads[0]?.settledOverride).toBeNull();
262 }),
263 );
264
265 it.effect(
266 "shows a queued reverse action immediately and preserves it if the earlier action fails",
267 () =>
268 Effect.gen(function* () {
269 const h = yield* makeHarness();
270 const settle = h.commands.settle.run(h.registry, {
271 environmentId: ENVIRONMENT_ID,
272 input: { threadId: THREAD_ID },
273 });
274 const first = yield* Queue.take(h.requests);
275 const unsettle = h.commands.unsettle.run(h.registry, {
276 environmentId: ENVIRONMENT_ID,
277 input: { threadId: THREAD_ID, reason: "user" },
278 });
279 expect(h.registry.get(h.visibleAtom)?.threads[0]?.settledOverride).toBe("active");
280 yield* Deferred.fail(first.reply, new Error("Settle rejected"));
281 yield* Effect.promise(() => settle);
282 expect(h.registry.get(h.visibleAtom)?.threads[0]?.settledOverride).toBe("active");
283 const second = yield* Queue.take(h.requests);
284 expect(second.command.type).toBe("thread.unsettle");
285 const confirmed = {
286 ...SNAPSHOT,
287 snapshotSequence: 2,
288 threads: [{ ...SNAPSHOT.threads[0]!, settledOverride: "active" as const }],
289 };
290 h.registry.set(h.snapshotAtom(ENVIRONMENT_ID), confirmed);
291 yield* Deferred.succeed(second.reply, { sequence: 2 });
292 yield* Effect.promise(() => unsettle);
293 expect(h.registry.get(h.visibleAtom)).toBe(confirmed);
294 }),
295 );
296
297 it.effect("isolates environments and does not restore a remotely removed thread", () =>
298 Effect.gen(function* () {
299 const h = yield* makeHarness();
300 const otherEnvironment = EnvironmentId.make("other-remote");
301 const result = h.commands.settle.run(h.registry, {
302 environmentId: ENVIRONMENT_ID,
303 input: { threadId: THREAD_ID },
304 });
305 const request = yield* Queue.take(h.requests);
306 expect(h.registry.get(h.commands.snapshotAtom(otherEnvironment))).toBe(SNAPSHOT);
307 const removed = { ...SNAPSHOT, snapshotSequence: 2, threads: [] };
308 h.registry.set(h.snapshotAtom(ENVIRONMENT_ID), removed);
309 expect(h.registry.get(h.visibleAtom)?.threads).toEqual([]);
310 yield* Deferred.fail(request.reply, new Error("Thread removed"));
311 yield* Effect.promise(() => result);
312 expect(h.registry.get(h.visibleAtom)).toBe(removed);
313 }),
314 );
315
316 it.effect("keeps pending approvals visible while a lifecycle request is pending", () =>
317 Effect.gen(function* () {
318 const h = yield* makeHarness();
319 const blocked = {
320 ...SNAPSHOT,
321 threads: [{ ...SNAPSHOT.threads[0]!, pendingRuntimeRequest: APPROVAL }],
322 };
323 h.registry.set(h.snapshotAtom(ENVIRONMENT_ID), blocked);
324 const result = h.commands.settle.run(h.registry, {
325 environmentId: ENVIRONMENT_ID,
326 input: { threadId: THREAD_ID },
327 });
328 const request = yield* Queue.take(h.requests);
329 expect(h.registry.get(h.visibleAtom)?.threads[0]).toBe(blocked.threads[0]);
330 yield* Deferred.fail(request.reply, new Error("Approval pending"));
331 yield* Effect.promise(() => result);
332 }),
333 );
334
335 const undoableActions = ["settle", "snooze"] as const;
336
337 it.effect.each(undoableActions)("restores a confirmed %s when a queued undo fails", (action) =>
338 Effect.gen(function* () {
339 const h = yield* makeHarness();
340 const parked =
341 action === "settle" ? { settledOverride: "settled" as const } : { snoozedUntil: FUTURE };
342 const awake = action === "settle" ? { settledOverride: "active" } : { snoozedUntil: null };
343 const result = h.commands[action].run(h.registry, {
344 environmentId: ENVIRONMENT_ID,
345 input: { threadId: THREAD_ID, snoozedUntil: "2099-01-01T00:00:00.000Z" },
346 });
347 const first = yield* Queue.take(h.requests);
348 const undo = h.commands[action === "settle" ? "unsettle" : "unsnooze"].run(h.registry, {
349 environmentId: ENVIRONMENT_ID,
350 input: { threadId: THREAD_ID, reason: "user" },
351 });
352 expect(h.registry.get(h.visibleAtom)?.threads[0]).toMatchObject(awake);
353 yield* Deferred.succeed(first.reply, { sequence: 2 });
354 expect((yield* Effect.promise(() => result))._tag).toBe("Success");
355 expect(h.registry.get(h.visibleAtom)?.threads[0]).toMatchObject(awake);
356 const confirmed = {
357 ...SNAPSHOT,
358 snapshotSequence: 2,
359 threads: [{ ...SNAPSHOT.threads[0]!, ...parked }],
360 };
361 h.registry.set(h.snapshotAtom(ENVIRONMENT_ID), confirmed);
362 expect(h.registry.get(h.visibleAtom)?.threads[0]).toMatchObject(awake);
363 const second = yield* Queue.take(h.requests);
364 expect(second.command.type).toBe(action === "settle" ? "thread.unsettle" : "thread.unsnooze");
365 yield* Deferred.fail(second.reply, new Error("Undo rejected"));
366 expect((yield* Effect.promise(() => undo))._tag).toBe("Failure");
367 expect(h.registry.get(h.visibleAtom)).toBe(confirmed);
368 }),
369 );
370
371 it.effect.each(undoableActions)(
372 "preserves a newer approval when the %s reply arrives after the shell",
373 (action) =>
374 Effect.gen(function* () {
375 const h = yield* makeHarness();
376 const result = h.commands[action].run(h.registry, {
377 environmentId: ENVIRONMENT_ID,
378 input: { threadId: THREAD_ID, snoozedUntil: "2099-01-01T00:00:00.000Z" },
379 });
380 const request = yield* Queue.take(h.requests);
381 const newer = {
382 ...SNAPSHOT,
383 snapshotSequence: 3,
384 threads: [{ ...SNAPSHOT.threads[0]!, pendingRuntimeRequest: APPROVAL }],
385 };
386 h.registry.set(h.snapshotAtom(ENVIRONMENT_ID), newer);
387 expect(h.registry.get(h.visibleAtom)?.threads[0]).toBe(newer.threads[0]);
388 yield* Deferred.succeed(request.reply, { sequence: 2 });
389 expect((yield* Effect.promise(() => result))._tag).toBe("Success");
390 expect(h.registry.get(h.visibleAtom)).toBe(newer);
391 }),
392 );
393
394 it.effect.each(undoableActions)(
395 "shows an accepted %s while the shell still has an old input request",
396 (action) =>
397 Effect.gen(function* () {
398 const h = yield* makeHarness();
399 const stale = {
400 ...SNAPSHOT,
401 threads: [{ ...SNAPSHOT.threads[0]!, pendingRuntimeRequest: USER_INPUT }],
402 };
403 h.registry.set(h.snapshotAtom(ENVIRONMENT_ID), stale);
404 const result = h.commands[action].run(h.registry, {
405 environmentId: ENVIRONMENT_ID,
406 input: { threadId: THREAD_ID, snoozedUntil: "2099-01-01T00:00:00.000Z" },
407 });
408 const request = yield* Queue.take(h.requests);
409 expect(h.registry.get(h.visibleAtom)?.threads[0]).toBe(stale.threads[0]);
410 yield* Deferred.succeed(request.reply, { sequence: 2 });
411 expect((yield* Effect.promise(() => result))._tag).toBe("Success");
412 expect(h.registry.get(h.visibleAtom)?.threads[0]).toMatchObject(
413 action === "settle" ? { settledOverride: "settled" } : { snoozedUntil: FUTURE },
414 );
415 expect(h.registry.get(h.visibleAtom)?.threads[0]?.pendingRuntimeRequest).toBeNull();
416 expect(h.registry.get(h.snapshotAtom(ENVIRONMENT_ID))).toBe(stale);
417 }),
418 );
419});