authorgravatar for 62337003+Bil0000@users.noreply.github.comBilal Bakr <62337003+Bil0000@users.noreply.github.com> 2026-09-15 09:54:38+03:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-09-14 23:54:38-07:00
log9ea892e3b365faff31eed294ac38f70d122656b2
tree3bc1abf0ad98a029177952727634f942b614a965
parenta37b85279d71f958482cb8b6fef1886cdb43f364
signature Signed by PGP key B5690EEEBB952194

fix(client): show thread state changes before remote replies (#11408)

8 files changed, 646 insertions(+), 9 deletions(-)

apps/mobile/src/state/threads.ts+5-2
...@@ -15,14 +15,17 @@ import { environmentCatalog } from "../connection/catalog";...@@ -15,14 +15,17 @@ import { environmentCatalog } from "../connection/catalog";
15import { connectionAtomRuntime } from "../connection/runtime";15import { connectionAtomRuntime } from "../connection/runtime";
16import { environmentSnapshotAtom } from "./shell";16import { environmentSnapshotAtom } from "./shell";
1717
18export const threadEnvironment = createThreadEnvironmentAtoms(connectionAtomRuntime);18export const threadEnvironment = createThreadEnvironmentAtoms(
19 connectionAtomRuntime,
20 environmentSnapshotAtom,
21);
19export const environmentThreads = createEnvironmentThreadStateAtoms(connectionAtomRuntime);22export const environmentThreads = createEnvironmentThreadStateAtoms(connectionAtomRuntime);
20export const environmentThreadDetails = createEnvironmentThreadDetailAtoms(23export const environmentThreadDetails = createEnvironmentThreadDetailAtoms(
21 environmentThreads.stateAtom,24 environmentThreads.stateAtom,
22);25);
23export const environmentThreadShells = createEnvironmentThreadShellAtoms({26export const environmentThreadShells = createEnvironmentThreadShellAtoms({
24 catalogValueAtom: environmentCatalog.catalogValueAtom,27 catalogValueAtom: environmentCatalog.catalogValueAtom,
25 snapshotAtom: environmentSnapshotAtom,28 snapshotAtom: threadEnvironment.snapshotAtom,
26});29});
2730
28const EMPTY_THREAD_STATE_ATOM = Atom.make(AsyncResult.success(EMPTY_ENVIRONMENT_THREAD_STATE)).pipe(31const EMPTY_THREAD_STATE_ATOM = Atom.make(AsyncResult.success(EMPTY_ENVIRONMENT_THREAD_STATE)).pipe(
apps/web/src/components/Sidebar.logic.test.ts+42
...@@ -44,6 +44,7 @@ import {...@@ -44,6 +44,7 @@ import {
44 sortProjectsForSidebar,44 sortProjectsForSidebar,
45 sortScopedProjectsForSidebar,45 sortScopedProjectsForSidebar,
46 shouldCreateNewThreadInCurrentProject,46 shouldCreateNewThreadInCurrentProject,
47 shouldNavigateAfterThreadPark,
47 THREAD_JUMP_HINT_SHOW_DELAY_MS,48 THREAD_JUMP_HINT_SHOW_DELAY_MS,
48 type SidebarListItem,49 type SidebarListItem,
49 type SidebarListMarker,50 type SidebarListMarker,
...@@ -2500,3 +2501,44 @@ describe("resolveSidebarDropVerb", () => {...@@ -2500,3 +2501,44 @@ describe("resolveSidebarDropVerb", () => {
2500 expect(resolveSidebarDropVerb("active", "snoozed")).toBeNull();2501 expect(resolveSidebarDropVerb("active", "snoozed")).toBeNull();
2501 });2502 });
2502});2503});
2504
2505describe("navigation after parking a thread", () => {
2506 it.each([
2507 ["settle", "settled", null, "thread", true],
2508 ["settle", "active", null, "thread", false],
2509 ["settle", "settled", null, "other-thread", false],
2510 ["snooze", null, "2099-01-01T00:00:00.000Z", "thread", true],
2511 ["snooze", null, null, "thread", false],
2512 ["snooze", null, "2026-09-12T09:00:00.000Z", "thread", false],
2513 ["snooze", null, "2099-01-01T00:00:00.000Z", "thread", false, true],
2514 ["snooze", null, "2099-01-01T00:00:00.000Z", "other-thread", false],
2515 ] as const)(
2516 "%s with state %s / %s on %s navigates: %s",
2517 (
2518 action,
2519 settledOverride,
2520 snoozedUntil,
2521 currentThreadKey,
2522 expected,
2523 hasPendingApprovals: boolean = false,
2524 ) => {
2525 expect(
2526 shouldNavigateAfterThreadPark({
2527 threadKey: "thread",
2528 currentThreadKey,
2529 action,
2530 now: "2026-09-12T10:00:00.000Z",
2531 thread: {
2532 settledOverride,
2533 snoozedUntil,
2534 snoozedAt: null,
2535 session: null,
2536 latestTurn: null,
2537 hasPendingApprovals,
2538 hasPendingUserInput: false,
2539 },
2540 }),
2541 ).toBe(expected);
2542 },
2543 );
2544});
apps/web/src/components/Sidebar.logic.ts+20
...@@ -9,6 +9,10 @@ import type { ContextMenuItem } from "@t3tools/contracts";...@@ -9,6 +9,10 @@ import type { ContextMenuItem } from "@t3tools/contracts";
9import type { SidebarProjectSortOrder, SidebarThreadSortOrder } from "@t3tools/contracts/settings";9import type { SidebarProjectSortOrder, SidebarThreadSortOrder } from "@t3tools/contracts/settings";
10import type { AsyncResult } from "effect/unstable/reactivity";10import type { AsyncResult } from "effect/unstable/reactivity";
11import { planPinnedReorder } from "@t3tools/client-runtime/state/thread-sort";11import { planPinnedReorder } from "@t3tools/client-runtime/state/thread-sort";
12import {
13 effectiveSnoozed,
14 type ThreadSnoozeShell,
15} from "@t3tools/client-runtime/state/thread-settled";
12import {16import {
13 getThreadSortTimestamp,17 getThreadSortTimestamp,
14 resolveSettledThreadTimestamp,18 resolveSettledThreadTimestamp,
...@@ -21,6 +25,22 @@ import type { SidebarThreadSummary, Thread } from "../types";...@@ -21,6 +25,22 @@ import type { SidebarThreadSummary, Thread } from "../types";
21import { cn } from "../lib/utils";25import { cn } from "../lib/utils";
22import { isLatestTurnSettled } from "../session-logic";26import { isLatestTurnSettled } from "../session-logic";
2327
28export function shouldNavigateAfterThreadPark(input: {
29 readonly threadKey: string;
30 readonly currentThreadKey: string | null;
31 readonly action: "settle" | "snooze";
32 readonly now: string;
33 readonly thread: (ThreadSnoozeShell & Pick<SidebarThreadSummary, "settledOverride">) | null;
34}): boolean {
35 return (
36 input.threadKey === input.currentThreadKey &&
37 input.thread !== null &&
38 (input.action === "settle"
39 ? input.thread.settledOverride === "settled"
40 : effectiveSnoozed(input.thread, { now: input.now }))
41 );
42}
43
24const THREAD_SELECTION_SAFE_SELECTOR = "[data-thread-item], [data-thread-selection-safe]";44const THREAD_SELECTION_SAFE_SELECTOR = "[data-thread-item], [data-thread-selection-safe]";
25export const THREAD_JUMP_HINT_SHOW_DELAY_MS = 200;45export const THREAD_JUMP_HINT_SHOW_DELAY_MS = 200;
26// Visible sidebar rows are prewarmed into the thread-detail cache so opening a46// Visible sidebar rows are prewarmed into the thread-detail cache so opening a
apps/web/src/components/Sidebar.tsx+30-3
...@@ -169,6 +169,7 @@ import {...@@ -169,6 +169,7 @@ import {
169 resolveSidebarThreadStatus,169 resolveSidebarThreadStatus,
170 searchSidebarThreads,170 searchSidebarThreads,
171 shouldCreateNewThreadInCurrentProject,171 shouldCreateNewThreadInCurrentProject,
172 shouldNavigateAfterThreadPark,
172 shouldRecedeSidebarThread,173 shouldRecedeSidebarThread,
173 resolveWorkingStartedAt,174 resolveWorkingStartedAt,
174 sidebarListItemId,175 sidebarListItemId,
...@@ -3053,7 +3054,15 @@ export default function Sidebar() {...@@ -3053,7 +3054,15 @@ export default function Sidebar() {
3053 }3054 }
3054 // Only move forward if the user is still on the settled thread —3055 // Only move forward if the user is still on the settled thread —
3055 // a navigation made during the await wins over ours.3056 // a navigation made during the await wins over ours.
3056 if (routeThreadKeyRef.current === threadKey) {3057 if (
3058 shouldNavigateAfterThreadPark({
3059 threadKey,
3060 currentThreadKey: routeThreadKeyRef.current,
3061 action: "settle",
3062 now: new Date().toISOString(),
3063 thread: readThreadShell(threadRef),
3064 })
3065 ) {
3057 navigateAfterSettle?.();3066 navigateAfterSettle?.();
3058 }3067 }
3059 } finally {3068 } finally {
...@@ -3582,7 +3591,17 @@ export default function Sidebar() {...@@ -3582,7 +3591,17 @@ export default function Sidebar() {
3582 const settled = await run(settleThread(threadRef), "Failed to settle thread").finally(3591 const settled = await run(settleThread(threadRef), "Failed to settle thread").finally(
3583 () => settlingThreadKeysRef.current.delete(activeKey),3592 () => settlingThreadKeysRef.current.delete(activeKey),
3584 );3593 );
3585 if (settled && routeThreadKeyRef.current === activeKey) navigateAfterSettle?.();3594 if (
3595 settled &&
3596 shouldNavigateAfterThreadPark({
3597 threadKey: activeKey,
3598 currentThreadKey: routeThreadKeyRef.current,
3599 action: "settle",
3600 now: new Date().toISOString(),
3601 thread: readThreadShell(threadRef),
3602 })
3603 )
3604 navigateAfterSettle?.();
3586 return;3605 return;
3587 }3606 }
3588 case "move-active":3607 case "move-active":
...@@ -3679,7 +3698,15 @@ export default function Sidebar() {...@@ -3679,7 +3698,15 @@ export default function Sidebar() {
3679 }3698 }
3680 // Only move forward if the user is still on the snoozed thread —3699 // Only move forward if the user is still on the snoozed thread —
3681 // a navigation made during the await wins over ours.3700 // a navigation made during the await wins over ours.
3682 if (routeThreadKeyRef.current === threadKey) {3701 if (
3702 shouldNavigateAfterThreadPark({
3703 threadKey,
3704 currentThreadKey: routeThreadKeyRef.current,
3705 action: "snooze",
3706 now: new Date().toISOString(),
3707 thread: readThreadShell(threadRef),
3708 })
3709 ) {
3683 navigateAfterSnooze?.();3710 navigateAfterSnooze?.();
3684 }3711 }
3685 return { status: "success" } as const;3712 return { status: "success" } as const;
apps/web/src/state/threads.ts+5-2
...@@ -15,14 +15,17 @@ import { environmentCatalog } from "../connection/catalog";...@@ -15,14 +15,17 @@ import { environmentCatalog } from "../connection/catalog";
15import { connectionAtomRuntime } from "../connection/runtime";15import { connectionAtomRuntime } from "../connection/runtime";
16import { environmentSnapshotAtom } from "./shell";16import { environmentSnapshotAtom } from "./shell";
1717
18export const threadEnvironment = createThreadEnvironmentAtoms(connectionAtomRuntime);18export const threadEnvironment = createThreadEnvironmentAtoms(
19 connectionAtomRuntime,
20 environmentSnapshotAtom,
21);
19const environmentThreads = createEnvironmentThreadStateAtoms(connectionAtomRuntime);22const environmentThreads = createEnvironmentThreadStateAtoms(connectionAtomRuntime);
20export const environmentThreadDetails = createEnvironmentThreadDetailAtoms(23export const environmentThreadDetails = createEnvironmentThreadDetailAtoms(
21 environmentThreads.stateAtom,24 environmentThreads.stateAtom,
22);25);
23export const environmentThreadShells = createEnvironmentThreadShellAtoms({26export const environmentThreadShells = createEnvironmentThreadShellAtoms({
24 catalogValueAtom: environmentCatalog.catalogValueAtom,27 catalogValueAtom: environmentCatalog.catalogValueAtom,
25 snapshotAtom: environmentSnapshotAtom,28 snapshotAtom: threadEnvironment.snapshotAtom,
26});29});
2730
28const EMPTY_THREAD_STATE_ATOM = Atom.make(AsyncResult.success(EMPTY_ENVIRONMENT_THREAD_STATE)).pipe(31const EMPTY_THREAD_STATE_ATOM = Atom.make(AsyncResult.success(EMPTY_ENVIRONMENT_THREAD_STATE)).pipe(
packages/client-runtime/src/state/threadCommands.test.ts created+357
...@@ -0,0 +1,357 @@
1import { describe, expect, it } from "@effect/vitest";
2import {
3 CommandId,
4 EnvironmentId,
5 ORCHESTRATION_WS_METHODS,
6 ProjectId,
7 ProviderInstanceId,
8 ThreadId,
9 type ClientOrchestrationCommand,
10 type OrchestrationShellSnapshot,
11} from "@t3tools/contracts";
12import * as Crypto from "effect/Crypto";
13import * as Deferred from "effect/Deferred";
14import * as Effect from "effect/Effect";
15import * as Layer from "effect/Layer";
16import * as Option from "effect/Option";
17import * as Queue from "effect/Queue";
18import * as SubscriptionRef from "effect/SubscriptionRef";
19import { Atom, AtomRegistry } from "effect/unstable/reactivity";
20
21import { EnvironmentRegistry } from "../connection/registry.ts";
22import { EnvironmentSupervisor } from "../connection/supervisor.ts";
23import type { RpcSession } from "../rpc/session.ts";
24import { createThreadEnvironmentAtoms } from "./threadCommands.ts";
25
26const ENVIRONMENT_ID = EnvironmentId.make("remote");
27const THREAD_ID = ThreadId.make("thread");
28const NOW = "2026-09-12T10:00:00.000Z";
29const SNAPSHOT: OrchestrationShellSnapshot = {
30 snapshotSequence: 1,
31 updatedAt: NOW,
32 projects: [],
33 threads: [
34 {
35 id: THREAD_ID,
36 projectId: ProjectId.make("project"),
37 title: "Remote thread",
38 modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" },
39 runtimeMode: "full-access",
40 interactionMode: "default",
41 branch: null,
42 worktreePath: null,
43 latestTurn: null,
44 createdAt: NOW,
45 updatedAt: NOW,
46 archivedAt: null,
47 settledOverride: null,
48 settledAt: null,
49 pullRequests: [],
50 session: null,
51 latestUserMessageAt: null,
52 hasPendingApprovals: false,
53 hasPendingUserInput: false,
54 hasActionableProposedPlan: false,
55 },
56 ],
57};
58
59const makeHarness = Effect.fn("TestThreadCommands.makeHarness")(function* () {
60 const requests = yield* Queue.unbounded<{
61 command: ClientOrchestrationCommand;
62 reply: Deferred.Deferred<{ sequence: number }, Error>;
63 }>();
64 const supervisor = EnvironmentSupervisor.of({
65 target: { environmentId: ENVIRONMENT_ID },
66 session: yield* SubscriptionRef.make(
67 Option.some({
68 client: {
69 [ORCHESTRATION_WS_METHODS.dispatchCommand]: (command: ClientOrchestrationCommand) =>
70 Effect.gen(function* () {
71 const reply = yield* Deferred.make<{ sequence: number }, Error>();
72 yield* Queue.offer(requests, { command, reply });
73 return yield* Deferred.await(reply);
74 }),
75 },
76 } as unknown as RpcSession),
77 ),
78 } as EnvironmentSupervisor["Service"]);
79 const runtime = Atom.runtime(
80 Layer.mergeAll(
81 Layer.succeed(EnvironmentRegistry, {
82 run: (_environmentId, effect) =>
83 Effect.provideService(effect, EnvironmentSupervisor, supervisor),
84 } as EnvironmentRegistry["Service"]),
85 Layer.succeed(
86 Crypto.Crypto,
87 Crypto.make({
88 randomBytes: (size) => new Uint8Array(size),
89 digest: (_algorithm, data) => Effect.succeed(data),
90 }),
91 ),
92 ),
93 );
94 const snapshotAtom = Atom.family((_environmentId: EnvironmentId) => Atom.make(SNAPSHOT));
95 const commands = createThreadEnvironmentAtoms(runtime, snapshotAtom);
96 const registry = AtomRegistry.make();
97 yield* Effect.addFinalizer(() => Effect.sync(() => registry.dispose()));
98 const visibleAtom = commands.snapshotAtom(ENVIRONMENT_ID);
99 registry.mount(visibleAtom);
100 return { registry, commands, snapshotAtom, visibleAtom, requests };
101});
102
103describe("remote thread lifecycle commands", () => {
104 const actions = [
105 ["settle", {}, { settledOverride: "settled", pinnedAt: null, snoozedUntil: null }],
106 ["unsettle", { reason: "user" }, { settledOverride: "active", settledAt: null }],
107 [
108 "snooze",
109 { snoozedUntil: "2099-01-01T00:00:00.000Z" },
110 { snoozedUntil: "2099-01-01T00:00:00.000Z" },
111 ],
112 ["unsnooze", { reason: "user" }, { snoozedUntil: null, snoozedAt: null }],
113 ["pin", { orderKey: "a" }, { pinnedAt: expect.any(String), pinOrderKey: "a" }],
114 ["unpin", {}, { pinnedAt: null, pinOrderKey: null }],
115 ["reorderPin", { orderKey: "b" }, { pinOrderKey: "b" }],
116 ["reorderActive", { orderKey: "b" }, { activeOrderKey: "b" }],
117 ] as const;
118
119 for (const [action, input, expected] of actions) {
120 it.effect(`shows ${action} before a delayed remote reply and rolls back a rejection`, () =>
121 Effect.gen(function* () {
122 const h = yield* makeHarness();
123 const source = h.snapshotAtom(ENVIRONMENT_ID);
124 const initial = {
125 ...SNAPSHOT,
126 threads: [
127 {
128 ...SNAPSHOT.threads[0]!,
129 ...(action === "unsettle" || action === "pin"
130 ? { settledOverride: "settled" as const, settledAt: NOW }
131 : {}),
132 ...(action === "unsnooze" || action === "settle" || action === "pin"
133 ? { snoozedUntil: "2099-01-01T00:00:00.000Z", snoozedAt: NOW }
134 : {}),
135 ...(action === "unpin" || action === "settle"
136 ? { pinnedAt: NOW, pinOrderKey: "a" }
137 : {}),
138 },
139 ],
140 };
141 h.registry.set(source, initial);
142 const result = h.commands[action].run(h.registry, {
143 environmentId: ENVIRONMENT_ID,
144 input: {
145 threadId: THREAD_ID,
146 commandId: CommandId.make(action),
147 reason: "user",
148 orderKey: "a",
149 snoozedUntil: "2099-01-01T00:00:00.000Z",
150 ...input,
151 },
152 });
153 expect(h.registry.get(h.visibleAtom)?.threads[0]).toMatchObject(expected);
154 const request = yield* Queue.take(h.requests);
155 expect(h.registry.get(source)).toBe(initial);
156 yield* Deferred.fail(request.reply, new Error("Remote rejected the action"));
157 expect((yield* Effect.promise(() => result))._tag).toBe("Failure");
158 expect(h.registry.get(h.visibleAtom)).toBe(initial);
159 }),
160 );
161 }
162
163 it.effect("keeps the preview after acknowledgement until the matching shell update arrives", () =>
164 Effect.gen(function* () {
165 const h = yield* makeHarness();
166 const result = h.commands.settle.run(h.registry, {
167 environmentId: ENVIRONMENT_ID,
168 input: { threadId: THREAD_ID },
169 });
170 const request = yield* Queue.take(h.requests);
171 yield* Deferred.succeed(request.reply, { sequence: 3 });
172 expect((yield* Effect.promise(() => result))._tag).toBe("Success");
173 const changed = {
174 ...SNAPSHOT,
175 snapshotSequence: 2,
176 threads: [{ ...SNAPSHOT.threads[0]!, title: "Renamed remotely" }],
177 };
178 h.registry.set(h.snapshotAtom(ENVIRONMENT_ID), changed);
179 expect(h.registry.get(h.visibleAtom)?.threads[0]).toMatchObject({
180 title: "Renamed remotely",
181 settledOverride: "settled",
182 });
183 const confirmed = {
184 ...changed,
185 snapshotSequence: 3,
186 threads: [
187 {
188 ...changed.threads[0]!,
189 settledOverride: "settled" as const,
190 settledAt: "2026-09-12T12:00:00.000Z",
191 },
192 ],
193 };
194 h.registry.set(h.snapshotAtom(ENVIRONMENT_ID), confirmed);
195 expect(h.registry.get(h.visibleAtom)).toBe(confirmed);
196 h.registry.set(h.snapshotAtom(ENVIRONMENT_ID), { ...SNAPSHOT, snapshotSequence: 4 });
197 expect(h.registry.get(h.visibleAtom)?.threads[0]?.settledOverride).toBeNull();
198 }),
199 );
200
201 it.effect(
202 "shows a queued reverse action immediately and preserves it if the earlier action fails",
203 () =>
204 Effect.gen(function* () {
205 const h = yield* makeHarness();
206 const settle = h.commands.settle.run(h.registry, {
207 environmentId: ENVIRONMENT_ID,
208 input: { threadId: THREAD_ID },
209 });
210 const first = yield* Queue.take(h.requests);
211 const unsettle = h.commands.unsettle.run(h.registry, {
212 environmentId: ENVIRONMENT_ID,
213 input: { threadId: THREAD_ID, reason: "user" },
214 });
215 expect(h.registry.get(h.visibleAtom)?.threads[0]?.settledOverride).toBe("active");
216 yield* Deferred.fail(first.reply, new Error("Settle rejected"));
217 yield* Effect.promise(() => settle);
218 expect(h.registry.get(h.visibleAtom)?.threads[0]?.settledOverride).toBe("active");
219 const second = yield* Queue.take(h.requests);
220 expect(second.command.type).toBe("thread.unsettle");
221 const confirmed = {
222 ...SNAPSHOT,
223 snapshotSequence: 2,
224 threads: [{ ...SNAPSHOT.threads[0]!, settledOverride: "active" as const }],
225 };
226 h.registry.set(h.snapshotAtom(ENVIRONMENT_ID), confirmed);
227 yield* Deferred.succeed(second.reply, { sequence: 2 });
228 yield* Effect.promise(() => unsettle);
229 expect(h.registry.get(h.visibleAtom)).toBe(confirmed);
230 }),
231 );
232
233 it.effect("isolates environments and does not restore a remotely removed thread", () =>
234 Effect.gen(function* () {
235 const h = yield* makeHarness();
236 const otherEnvironment = EnvironmentId.make("other-remote");
237 const result = h.commands.settle.run(h.registry, {
238 environmentId: ENVIRONMENT_ID,
239 input: { threadId: THREAD_ID },
240 });
241 const request = yield* Queue.take(h.requests);
242 expect(h.registry.get(h.commands.snapshotAtom(otherEnvironment))).toBe(SNAPSHOT);
243 const removed = { ...SNAPSHOT, snapshotSequence: 2, threads: [] };
244 h.registry.set(h.snapshotAtom(ENVIRONMENT_ID), removed);
245 expect(h.registry.get(h.visibleAtom)?.threads).toEqual([]);
246 yield* Deferred.fail(request.reply, new Error("Thread removed"));
247 yield* Effect.promise(() => result);
248 expect(h.registry.get(h.visibleAtom)).toBe(removed);
249 }),
250 );
251
252 it.effect("keeps pending approvals visible while a lifecycle request is pending", () =>
253 Effect.gen(function* () {
254 const h = yield* makeHarness();
255 const blocked = {
256 ...SNAPSHOT,
257 threads: [{ ...SNAPSHOT.threads[0]!, hasPendingApprovals: true }],
258 };
259 h.registry.set(h.snapshotAtom(ENVIRONMENT_ID), blocked);
260 const result = h.commands.settle.run(h.registry, {
261 environmentId: ENVIRONMENT_ID,
262 input: { threadId: THREAD_ID },
263 });
264 const request = yield* Queue.take(h.requests);
265 expect(h.registry.get(h.visibleAtom)?.threads[0]).toBe(blocked.threads[0]);
266 yield* Deferred.fail(request.reply, new Error("Approval pending"));
267 yield* Effect.promise(() => result);
268 }),
269 );
270
271 for (const action of ["settle", "snooze"] as const) {
272 it.effect(`restores a confirmed ${action} when a queued undo fails`, () =>
273 Effect.gen(function* () {
274 const h = yield* makeHarness();
275 const parked =
276 action === "settle"
277 ? { settledOverride: "settled" as const }
278 : { snoozedUntil: "2099-01-01T00:00:00.000Z" };
279 const awake = action === "settle" ? { settledOverride: "active" } : { snoozedUntil: null };
280 const result = h.commands[action].run(h.registry, {
281 environmentId: ENVIRONMENT_ID,
282 input: { threadId: THREAD_ID, snoozedUntil: "2099-01-01T00:00:00.000Z" },
283 });
284 const first = yield* Queue.take(h.requests);
285 const undo = h.commands[action === "settle" ? "unsettle" : "unsnooze"].run(h.registry, {
286 environmentId: ENVIRONMENT_ID,
287 input: { threadId: THREAD_ID, reason: "user" },
288 });
289 expect(h.registry.get(h.visibleAtom)?.threads[0]).toMatchObject(awake);
290 yield* Deferred.succeed(first.reply, { sequence: 2 });
291 expect((yield* Effect.promise(() => result))._tag).toBe("Success");
292 expect(h.registry.get(h.visibleAtom)?.threads[0]).toMatchObject(awake);
293 const confirmed = {
294 ...SNAPSHOT,
295 snapshotSequence: 2,
296 threads: [{ ...SNAPSHOT.threads[0]!, ...parked }],
297 };
298 h.registry.set(h.snapshotAtom(ENVIRONMENT_ID), confirmed);
299 expect(h.registry.get(h.visibleAtom)?.threads[0]).toMatchObject(awake);
300 const second = yield* Queue.take(h.requests);
301 expect(second.command.type).toBe(
302 action === "settle" ? "thread.unsettle" : "thread.unsnooze",
303 );
304 yield* Deferred.fail(second.reply, new Error("Undo rejected"));
305 expect((yield* Effect.promise(() => undo))._tag).toBe("Failure");
306 expect(h.registry.get(h.visibleAtom)).toBe(confirmed);
307 }),
308 );
309
310 it.effect(`preserves a newer approval when the ${action} reply arrives after the shell`, () =>
311 Effect.gen(function* () {
312 const h = yield* makeHarness();
313 const result = h.commands[action].run(h.registry, {
314 environmentId: ENVIRONMENT_ID,
315 input: { threadId: THREAD_ID, snoozedUntil: "2099-01-01T00:00:00.000Z" },
316 });
317 const request = yield* Queue.take(h.requests);
318 const newer = {
319 ...SNAPSHOT,
320 snapshotSequence: 3,
321 threads: [{ ...SNAPSHOT.threads[0]!, hasPendingApprovals: true }],
322 };
323 h.registry.set(h.snapshotAtom(ENVIRONMENT_ID), newer);
324 expect(h.registry.get(h.visibleAtom)?.threads[0]).toBe(newer.threads[0]);
325 yield* Deferred.succeed(request.reply, { sequence: 2 });
326 expect((yield* Effect.promise(() => result))._tag).toBe("Success");
327 expect(h.registry.get(h.visibleAtom)).toBe(newer);
328 }),
329 );
330
331 it.effect(`shows an accepted ${action} while the shell still has an old input request`, () =>
332 Effect.gen(function* () {
333 const h = yield* makeHarness();
334 const stale = {
335 ...SNAPSHOT,
336 threads: [{ ...SNAPSHOT.threads[0]!, hasPendingUserInput: true }],
337 };
338 h.registry.set(h.snapshotAtom(ENVIRONMENT_ID), stale);
339 const result = h.commands[action].run(h.registry, {
340 environmentId: ENVIRONMENT_ID,
341 input: { threadId: THREAD_ID, snoozedUntil: "2099-01-01T00:00:00.000Z" },
342 });
343 const request = yield* Queue.take(h.requests);
344 expect(h.registry.get(h.visibleAtom)?.threads[0]).toBe(stale.threads[0]);
345 yield* Deferred.succeed(request.reply, { sequence: 2 });
346 expect((yield* Effect.promise(() => result))._tag).toBe("Success");
347 expect(h.registry.get(h.visibleAtom)?.threads[0]).toMatchObject(
348 action === "settle"
349 ? { settledOverride: "settled" }
350 : { snoozedUntil: "2099-01-01T00:00:00.000Z" },
351 );
352 expect(h.registry.get(h.visibleAtom)?.threads[0]?.hasPendingUserInput).toBe(false);
353 expect(h.registry.get(h.snapshotAtom(ENVIRONMENT_ID))).toBe(stale);
354 }),
355 );
356 }
357});
packages/client-runtime/src/state/threadCommands.ts+85-2
...@@ -1,6 +1,13 @@...@@ -1,6 +1,13 @@
1import * as Crypto from "effect/Crypto";1import * as Crypto from "effect/Crypto";
2import { Atom } from "effect/unstable/reactivity";2import { Atom } from "effect/unstable/reactivity";
3import { WS_METHODS } from "@t3tools/contracts";3import {
4 WS_METHODS,
5 type EnvironmentId,
6 type OrchestrationShellSnapshot,
7} from "@t3tools/contracts";
8
9import { createOptimisticThreadLifecycle } from "./threadLifecycle.ts";
10import { canSnooze } from "./threadSettled.ts";
411
5import {12import {
6 createAtomCommandScheduler,13 createAtomCommandScheduler,
...@@ -88,6 +95,7 @@ export type {...@@ -88,6 +95,7 @@ export type {
8895
89export function createThreadEnvironmentAtoms<R, E>(96export function createThreadEnvironmentAtoms<R, E>(
90 runtime: Atom.AtomRuntime<EnvironmentRegistry | Crypto.Crypto | R, E>,97 runtime: Atom.AtomRuntime<EnvironmentRegistry | Crypto.Crypto | R, E>,
98 snapshotAtom: (environmentId: EnvironmentId) => Atom.Atom<OrchestrationShellSnapshot | null>,
91) {99) {
92 const scheduler = createAtomCommandScheduler();100 const scheduler = createAtomCommandScheduler();
93 const concurrency = {101 const concurrency = {
...@@ -95,7 +103,7 @@ export function createThreadEnvironmentAtoms<R, E>(...@@ -95,7 +103,7 @@ export function createThreadEnvironmentAtoms<R, E>(
95 key: ({ environmentId, input }: { environmentId: string; input: { threadId: string } }) =>103 key: ({ environmentId, input }: { environmentId: string; input: { threadId: string } }) =>
96 JSON.stringify([environmentId, input.threadId]),104 JSON.stringify([environmentId, input.threadId]),
97 };105 };
98 return {106 const commands = {
99 create: createEnvironmentCommand(runtime, {107 create: createEnvironmentCommand(runtime, {
100 label: "environment-data:commands:thread:create",108 label: "environment-data:commands:thread:create",
101 execute: (input: CreateThreadInput) => createThread(input),109 execute: (input: CreateThreadInput) => createThread(input),
...@@ -247,4 +255,79 @@ export function createThreadEnvironmentAtoms<R, E>(...@@ -247,4 +255,79 @@ export function createThreadEnvironmentAtoms<R, E>(
247 concurrency,255 concurrency,
248 }),256 }),
249 };257 };
258 const optimistic = createOptimisticThreadLifecycle(snapshotAtom);
259 return {
260 ...commands,
261 snapshotAtom: optimistic.snapshotAtom,
262 settle: optimistic.wrap(commands.settle, (thread, _input, now, accepted) =>
263 !accepted &&
264 (!canSnooze(thread, { now }) ||
265 thread.session?.status === "starting" ||
266 thread.session?.status === "running")
267 ? thread
268 : {
269 ...thread,
270 hasPendingApprovals: false,
271 hasPendingUserInput: false,
272 settledOverride: "settled",
273 settledAt: thread.settledOverride === "settled" ? (thread.settledAt ?? now) : now,
274 unsettledAt: null,
275 activeOrderKey: null,
276 pinnedAt: null,
277 pinOrderKey: null,
278 snoozedAt: null,
279 snoozedUntil: null,
280 },
281 ),
282 unsettle: optimistic.wrap(commands.unsettle, (thread, input, now) => ({
283 ...thread,
284 settledOverride: input.reason === "user" ? "active" : null,
285 settledAt: null,
286 unsettledAt: thread.settledOverride === "active" ? (thread.unsettledAt ?? null) : now,
287 })),
288 snooze: optimistic.wrap(commands.snooze, (thread, input, now, accepted) =>
289 (!accepted && !canSnooze(thread, { now })) ||
290 !(Date.parse(input.snoozedUntil) > Date.parse(now))
291 ? thread
292 : {
293 ...thread,
294 hasPendingApprovals: false,
295 hasPendingUserInput: false,
296 snoozedUntil: input.snoozedUntil,
297 snoozedAt: thread.snoozedUntil === input.snoozedUntil ? (thread.snoozedAt ?? now) : now,
298 },
299 ),
300 unsnooze: optimistic.wrap(commands.unsnooze, (thread) => ({
301 ...thread,
302 snoozedUntil: null,
303 snoozedAt: null,
304 })),
305 pin: optimistic.wrap(commands.pin, (thread, input, now) => ({
306 ...thread,
307 pinnedAt: thread.pinnedAt ?? now,
308 pinOrderKey: thread.pinnedAt == null ? (input.orderKey ?? null) : thread.pinOrderKey,
309 ...(thread.settledOverride === "settled"
310 ? {
311 settledOverride: "active" as const,
312 settledAt: null,
313 unsettledAt: now,
314 }
315 : {}),
316 snoozedUntil: null,
317 snoozedAt: null,
318 })),
319 unpin: optimistic.wrap(commands.unpin, (thread) => ({
320 ...thread,
321 pinnedAt: null,
322 pinOrderKey: null,
323 })),
324 reorderPin: optimistic.wrap(commands.reorderPin, (thread, input) => ({
325 ...thread,
326 pinOrderKey: input.orderKey,
327 })),
328 reorderActive: optimistic.wrap(commands.reorderActive, (thread, input) => ({
329 ...thread,
330 activeOrderKey: input.orderKey,
331 })),
332 };
250}333}
packages/client-runtime/src/state/threadLifecycle.ts created+102
...@@ -0,0 +1,102 @@
1import type {
2 EnvironmentId,
3 OrchestrationShellSnapshot,
4 OrchestrationThreadShell,
5 ThreadId,
6} from "@t3tools/contracts";
7import * as DateTime from "effect/DateTime";
8import { Atom } from "effect/unstable/reactivity";
9
10import type { AtomCommand } from "./runtime.ts";
11
12interface PendingThreadUpdate {
13 readonly threadId: ThreadId;
14 readonly apply: (thread: OrchestrationThreadShell) => OrchestrationThreadShell;
15 sequence?: number;
16}
17
18export function createOptimisticThreadLifecycle(
19 sourceSnapshotAtom: (
20 environmentId: EnvironmentId,
21 ) => Atom.Atom<OrchestrationShellSnapshot | null>,
22) {
23 const pendingAtom = Atom.family((_environmentId: EnvironmentId) =>
24 Atom.make<ReadonlyArray<PendingThreadUpdate>>([]).pipe(Atom.keepAlive),
25 );
26 const snapshotAtom = Atom.family((environmentId: EnvironmentId) =>
27 Atom.make((get) => {
28 const snapshot = get(sourceSnapshotAtom(environmentId));
29 const pending = get(pendingAtom(environmentId));
30 if (snapshot === null || pending.length === 0) return snapshot;
31 const byThread = new Map<ThreadId, PendingThreadUpdate[]>();
32 for (const update of pending) {
33 if (update.sequence !== undefined && update.sequence <= snapshot.snapshotSequence) continue;
34 const updates = byThread.get(update.threadId) ?? [];
35 updates.push(update);
36 byThread.set(update.threadId, updates);
37 }
38 if (byThread.size === 0) return snapshot;
39 return {
40 ...snapshot,
41 threads: snapshot.threads.map((thread) =>
42 (byThread.get(thread.id) ?? []).reduce(
43 (current, update) => update.apply(current),
44 thread,
45 ),
46 ),
47 };
48 }),
49 );
50
51 function wrap<Input extends { readonly threadId: ThreadId }, E>(
52 command: AtomCommand<
53 { readonly environmentId: EnvironmentId; readonly input: Input },
54 { readonly sequence: number },
55 E
56 >,
57 apply: (
58 thread: OrchestrationThreadShell,
59 input: Input,
60 now: string,
61 accepted: boolean,
62 ) => OrchestrationThreadShell,
63 ): typeof command {
64 return {
65 label: command.label,
66 run: async (registry, target) => {
67 const now = DateTime.formatIso(DateTime.nowUnsafe());
68 const pending = pendingAtom(target.environmentId);
69 const source = sourceSnapshotAtom(target.environmentId);
70 const update: PendingThreadUpdate = {
71 threadId: target.input.threadId,
72 apply: (thread) => apply(thread, target.input, now, update.sequence !== undefined),
73 };
74 const remove = () =>
75 registry.update(pending, (current) => current.filter((item) => item !== update));
76 registry.update(pending, (current) => [...current, update]);
77 let confirmed = false;
78 try {
79 const result = await command.run(registry, target);
80 if (result._tag === "Success") {
81 update.sequence = result.value.sequence;
82 registry.update(pending, (current) => [...current]);
83 const reconcile = (snapshot: OrchestrationShellSnapshot | null) => {
84 if (snapshot === null || snapshot.snapshotSequence >= result.value.sequence) {
85 remove();
86 unsubscribe();
87 }
88 };
89 const unsubscribe = registry.subscribe(source, reconcile);
90 reconcile(registry.get(source));
91 confirmed = true;
92 }
93 return result;
94 } finally {
95 if (!confirmed) remove();
96 }
97 },
98 };
99 }
100
101 return { snapshotAtom, wrap };
102}