diff --git a/apps/mobile/src/state/threads.ts b/apps/mobile/src/state/threads.ts index 7f2471230510..ce0097635ac4 100644 --- a/apps/mobile/src/state/threads.ts +++ b/apps/mobile/src/state/threads.ts @@ -15,14 +15,17 @@ import { environmentCatalog } from "../connection/catalog"; import { connectionAtomRuntime } from "../connection/runtime"; import { environmentSnapshotAtom } from "./shell"; -export const threadEnvironment = createThreadEnvironmentAtoms(connectionAtomRuntime); +export const threadEnvironment = createThreadEnvironmentAtoms( + connectionAtomRuntime, + environmentSnapshotAtom, +); export const environmentThreads = createEnvironmentThreadStateAtoms(connectionAtomRuntime); export const environmentThreadDetails = createEnvironmentThreadDetailAtoms( environmentThreads.stateAtom, ); export const environmentThreadShells = createEnvironmentThreadShellAtoms({ catalogValueAtom: environmentCatalog.catalogValueAtom, - snapshotAtom: environmentSnapshotAtom, + snapshotAtom: threadEnvironment.snapshotAtom, }); const EMPTY_THREAD_STATE_ATOM = Atom.make(AsyncResult.success(EMPTY_ENVIRONMENT_THREAD_STATE)).pipe( diff --git a/apps/web/src/components/Sidebar.logic.test.ts b/apps/web/src/components/Sidebar.logic.test.ts index 33157e7b4b4b..eabfac0331da 100644 --- a/apps/web/src/components/Sidebar.logic.test.ts +++ b/apps/web/src/components/Sidebar.logic.test.ts @@ -44,6 +44,7 @@ import { sortProjectsForSidebar, sortScopedProjectsForSidebar, shouldCreateNewThreadInCurrentProject, + shouldNavigateAfterThreadPark, THREAD_JUMP_HINT_SHOW_DELAY_MS, type SidebarListItem, type SidebarListMarker, @@ -2492,3 +2493,44 @@ describe("resolveSidebarDropVerb", () => { expect(resolveSidebarDropVerb("active", "snoozed")).toBeNull(); }); }); + +describe("navigation after parking a thread", () => { + it.each([ + ["settle", "settled", null, "thread", true], + ["settle", "active", null, "thread", false], + ["settle", "settled", null, "other-thread", false], + ["snooze", null, "2099-01-01T00:00:00.000Z", "thread", true], + ["snooze", null, null, "thread", false], + ["snooze", null, "2026-09-12T09:00:00.000Z", "thread", false], + ["snooze", null, "2099-01-01T00:00:00.000Z", "thread", false, true], + ["snooze", null, "2099-01-01T00:00:00.000Z", "other-thread", false], + ] as const)( + "%s with state %s / %s on %s navigates: %s", + ( + action, + settledOverride, + snoozedUntil, + currentThreadKey, + expected, + hasPendingApprovals: boolean = false, + ) => { + expect( + shouldNavigateAfterThreadPark({ + threadKey: "thread", + currentThreadKey, + action, + now: "2026-09-12T10:00:00.000Z", + thread: { + settledOverride, + snoozedUntil, + snoozedAt: null, + session: null, + latestTurn: null, + hasPendingApprovals, + hasPendingUserInput: false, + }, + }), + ).toBe(expected); + }, + ); +}); diff --git a/apps/web/src/components/Sidebar.logic.ts b/apps/web/src/components/Sidebar.logic.ts index 0150eb473d51..c9e119f30076 100644 --- a/apps/web/src/components/Sidebar.logic.ts +++ b/apps/web/src/components/Sidebar.logic.ts @@ -9,6 +9,10 @@ import type { ContextMenuItem } from "@t3tools/contracts"; import type { SidebarProjectSortOrder, SidebarThreadSortOrder } from "@t3tools/contracts/settings"; import type { AsyncResult } from "effect/unstable/reactivity"; import { planPinnedReorder } from "@t3tools/client-runtime/state/thread-sort"; +import { + effectiveSnoozed, + type ThreadSnoozeShell, +} from "@t3tools/client-runtime/state/thread-settled"; import { getThreadSortTimestamp, resolveSettledThreadTimestamp, @@ -21,6 +25,22 @@ import type { SidebarThreadSummary, Thread } from "../types"; import { cn } from "../lib/utils"; import { isLatestTurnSettled } from "../session-logic"; +export function shouldNavigateAfterThreadPark(input: { + readonly threadKey: string; + readonly currentThreadKey: string | null; + readonly action: "settle" | "snooze"; + readonly now: string; + readonly thread: (ThreadSnoozeShell & Pick) | null; +}): boolean { + return ( + input.threadKey === input.currentThreadKey && + input.thread !== null && + (input.action === "settle" + ? input.thread.settledOverride === "settled" + : effectiveSnoozed(input.thread, { now: input.now })) + ); +} + const THREAD_SELECTION_SAFE_SELECTOR = "[data-thread-item], [data-thread-selection-safe]"; export const THREAD_JUMP_HINT_SHOW_DELAY_MS = 200; // Visible sidebar rows are prewarmed into the thread-detail cache so opening a diff --git a/apps/web/src/components/Sidebar.tsx b/apps/web/src/components/Sidebar.tsx index c951689230e9..d76e5b062869 100644 --- a/apps/web/src/components/Sidebar.tsx +++ b/apps/web/src/components/Sidebar.tsx @@ -167,6 +167,7 @@ import { resolveSidebarThreadStatus, searchSidebarThreads, shouldCreateNewThreadInCurrentProject, + shouldNavigateAfterThreadPark, shouldRecedeSidebarThread, resolveWorkingStartedAt, sidebarListItemId, @@ -3038,7 +3039,15 @@ export default function Sidebar() { } // Only move forward if the user is still on the settled thread — // a navigation made during the await wins over ours. - if (routeThreadKeyRef.current === threadKey) { + if ( + shouldNavigateAfterThreadPark({ + threadKey, + currentThreadKey: routeThreadKeyRef.current, + action: "settle", + now: new Date().toISOString(), + thread: readThreadShell(threadRef), + }) + ) { navigateAfterSettle?.(); } } finally { @@ -3567,7 +3576,17 @@ export default function Sidebar() { const settled = await run(settleThread(threadRef), "Failed to settle thread").finally( () => settlingThreadKeysRef.current.delete(activeKey), ); - if (settled && routeThreadKeyRef.current === activeKey) navigateAfterSettle?.(); + if ( + settled && + shouldNavigateAfterThreadPark({ + threadKey: activeKey, + currentThreadKey: routeThreadKeyRef.current, + action: "settle", + now: new Date().toISOString(), + thread: readThreadShell(threadRef), + }) + ) + navigateAfterSettle?.(); return; } case "move-active": @@ -3664,7 +3683,15 @@ export default function Sidebar() { } // Only move forward if the user is still on the snoozed thread — // a navigation made during the await wins over ours. - if (routeThreadKeyRef.current === threadKey) { + if ( + shouldNavigateAfterThreadPark({ + threadKey, + currentThreadKey: routeThreadKeyRef.current, + action: "snooze", + now: new Date().toISOString(), + thread: readThreadShell(threadRef), + }) + ) { navigateAfterSnooze?.(); } return { status: "success" } as const; diff --git a/apps/web/src/state/threads.ts b/apps/web/src/state/threads.ts index c7caaa6a35a7..deda3ca29e9a 100644 --- a/apps/web/src/state/threads.ts +++ b/apps/web/src/state/threads.ts @@ -15,14 +15,17 @@ import { environmentCatalog } from "../connection/catalog"; import { connectionAtomRuntime } from "../connection/runtime"; import { environmentSnapshotAtom } from "./shell"; -export const threadEnvironment = createThreadEnvironmentAtoms(connectionAtomRuntime); +export const threadEnvironment = createThreadEnvironmentAtoms( + connectionAtomRuntime, + environmentSnapshotAtom, +); const environmentThreads = createEnvironmentThreadStateAtoms(connectionAtomRuntime); export const environmentThreadDetails = createEnvironmentThreadDetailAtoms( environmentThreads.stateAtom, ); export const environmentThreadShells = createEnvironmentThreadShellAtoms({ catalogValueAtom: environmentCatalog.catalogValueAtom, - snapshotAtom: environmentSnapshotAtom, + snapshotAtom: threadEnvironment.snapshotAtom, }); const EMPTY_THREAD_STATE_ATOM = Atom.make(AsyncResult.success(EMPTY_ENVIRONMENT_THREAD_STATE)).pipe( diff --git a/packages/client-runtime/src/state/threadCommands.test.ts b/packages/client-runtime/src/state/threadCommands.test.ts new file mode 100644 index 000000000000..dc70ac5548b1 --- /dev/null +++ b/packages/client-runtime/src/state/threadCommands.test.ts @@ -0,0 +1,357 @@ +import { describe, expect, it } from "@effect/vitest"; +import { + CommandId, + EnvironmentId, + ORCHESTRATION_WS_METHODS, + ProjectId, + ProviderInstanceId, + ThreadId, + type ClientOrchestrationCommand, + type OrchestrationShellSnapshot, +} from "@t3tools/contracts"; +import * as Crypto from "effect/Crypto"; +import * as Deferred from "effect/Deferred"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Queue from "effect/Queue"; +import * as SubscriptionRef from "effect/SubscriptionRef"; +import { Atom, AtomRegistry } from "effect/unstable/reactivity"; + +import { EnvironmentRegistry } from "../connection/registry.ts"; +import { EnvironmentSupervisor } from "../connection/supervisor.ts"; +import type { RpcSession } from "../rpc/session.ts"; +import { createThreadEnvironmentAtoms } from "./threadCommands.ts"; + +const ENVIRONMENT_ID = EnvironmentId.make("remote"); +const THREAD_ID = ThreadId.make("thread"); +const NOW = "2026-09-12T10:00:00.000Z"; +const SNAPSHOT: OrchestrationShellSnapshot = { + snapshotSequence: 1, + updatedAt: NOW, + projects: [], + threads: [ + { + id: THREAD_ID, + projectId: ProjectId.make("project"), + title: "Remote thread", + modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" }, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + latestTurn: null, + createdAt: NOW, + updatedAt: NOW, + archivedAt: null, + settledOverride: null, + settledAt: null, + pullRequests: [], + session: null, + latestUserMessageAt: null, + hasPendingApprovals: false, + hasPendingUserInput: false, + hasActionableProposedPlan: false, + }, + ], +}; + +const makeHarness = Effect.fn("TestThreadCommands.makeHarness")(function* () { + const requests = yield* Queue.unbounded<{ + command: ClientOrchestrationCommand; + reply: Deferred.Deferred<{ sequence: number }, Error>; + }>(); + const supervisor = EnvironmentSupervisor.of({ + target: { environmentId: ENVIRONMENT_ID }, + session: yield* SubscriptionRef.make( + Option.some({ + client: { + [ORCHESTRATION_WS_METHODS.dispatchCommand]: (command: ClientOrchestrationCommand) => + Effect.gen(function* () { + const reply = yield* Deferred.make<{ sequence: number }, Error>(); + yield* Queue.offer(requests, { command, reply }); + return yield* Deferred.await(reply); + }), + }, + } as unknown as RpcSession), + ), + } as EnvironmentSupervisor["Service"]); + const runtime = Atom.runtime( + Layer.mergeAll( + Layer.succeed(EnvironmentRegistry, { + run: (_environmentId, effect) => + Effect.provideService(effect, EnvironmentSupervisor, supervisor), + } as EnvironmentRegistry["Service"]), + Layer.succeed( + Crypto.Crypto, + Crypto.make({ + randomBytes: (size) => new Uint8Array(size), + digest: (_algorithm, data) => Effect.succeed(data), + }), + ), + ), + ); + const snapshotAtom = Atom.family((_environmentId: EnvironmentId) => Atom.make(SNAPSHOT)); + const commands = createThreadEnvironmentAtoms(runtime, snapshotAtom); + const registry = AtomRegistry.make(); + yield* Effect.addFinalizer(() => Effect.sync(() => registry.dispose())); + const visibleAtom = commands.snapshotAtom(ENVIRONMENT_ID); + registry.mount(visibleAtom); + return { registry, commands, snapshotAtom, visibleAtom, requests }; +}); + +describe("remote thread lifecycle commands", () => { + const actions = [ + ["settle", {}, { settledOverride: "settled", pinnedAt: null, snoozedUntil: null }], + ["unsettle", { reason: "user" }, { settledOverride: "active", settledAt: null }], + [ + "snooze", + { snoozedUntil: "2099-01-01T00:00:00.000Z" }, + { snoozedUntil: "2099-01-01T00:00:00.000Z" }, + ], + ["unsnooze", { reason: "user" }, { snoozedUntil: null, snoozedAt: null }], + ["pin", { orderKey: "a" }, { pinnedAt: expect.any(String), pinOrderKey: "a" }], + ["unpin", {}, { pinnedAt: null, pinOrderKey: null }], + ["reorderPin", { orderKey: "b" }, { pinOrderKey: "b" }], + ["reorderActive", { orderKey: "b" }, { activeOrderKey: "b" }], + ] as const; + + for (const [action, input, expected] of actions) { + it.effect(`shows ${action} before a delayed remote reply and rolls back a rejection`, () => + Effect.gen(function* () { + const h = yield* makeHarness(); + const source = h.snapshotAtom(ENVIRONMENT_ID); + const initial = { + ...SNAPSHOT, + threads: [ + { + ...SNAPSHOT.threads[0]!, + ...(action === "unsettle" || action === "pin" + ? { settledOverride: "settled" as const, settledAt: NOW } + : {}), + ...(action === "unsnooze" || action === "settle" || action === "pin" + ? { snoozedUntil: "2099-01-01T00:00:00.000Z", snoozedAt: NOW } + : {}), + ...(action === "unpin" || action === "settle" + ? { pinnedAt: NOW, pinOrderKey: "a" } + : {}), + }, + ], + }; + h.registry.set(source, initial); + const result = h.commands[action].run(h.registry, { + environmentId: ENVIRONMENT_ID, + input: { + threadId: THREAD_ID, + commandId: CommandId.make(action), + reason: "user", + orderKey: "a", + snoozedUntil: "2099-01-01T00:00:00.000Z", + ...input, + }, + }); + expect(h.registry.get(h.visibleAtom)?.threads[0]).toMatchObject(expected); + const request = yield* Queue.take(h.requests); + expect(h.registry.get(source)).toBe(initial); + yield* Deferred.fail(request.reply, new Error("Remote rejected the action")); + expect((yield* Effect.promise(() => result))._tag).toBe("Failure"); + expect(h.registry.get(h.visibleAtom)).toBe(initial); + }), + ); + } + + it.effect("keeps the preview after acknowledgement until the matching shell update arrives", () => + Effect.gen(function* () { + const h = yield* makeHarness(); + const result = h.commands.settle.run(h.registry, { + environmentId: ENVIRONMENT_ID, + input: { threadId: THREAD_ID }, + }); + const request = yield* Queue.take(h.requests); + yield* Deferred.succeed(request.reply, { sequence: 3 }); + expect((yield* Effect.promise(() => result))._tag).toBe("Success"); + const changed = { + ...SNAPSHOT, + snapshotSequence: 2, + threads: [{ ...SNAPSHOT.threads[0]!, title: "Renamed remotely" }], + }; + h.registry.set(h.snapshotAtom(ENVIRONMENT_ID), changed); + expect(h.registry.get(h.visibleAtom)?.threads[0]).toMatchObject({ + title: "Renamed remotely", + settledOverride: "settled", + }); + const confirmed = { + ...changed, + snapshotSequence: 3, + threads: [ + { + ...changed.threads[0]!, + settledOverride: "settled" as const, + settledAt: "2026-09-12T12:00:00.000Z", + }, + ], + }; + h.registry.set(h.snapshotAtom(ENVIRONMENT_ID), confirmed); + expect(h.registry.get(h.visibleAtom)).toBe(confirmed); + h.registry.set(h.snapshotAtom(ENVIRONMENT_ID), { ...SNAPSHOT, snapshotSequence: 4 }); + expect(h.registry.get(h.visibleAtom)?.threads[0]?.settledOverride).toBeNull(); + }), + ); + + it.effect( + "shows a queued reverse action immediately and preserves it if the earlier action fails", + () => + Effect.gen(function* () { + const h = yield* makeHarness(); + const settle = h.commands.settle.run(h.registry, { + environmentId: ENVIRONMENT_ID, + input: { threadId: THREAD_ID }, + }); + const first = yield* Queue.take(h.requests); + const unsettle = h.commands.unsettle.run(h.registry, { + environmentId: ENVIRONMENT_ID, + input: { threadId: THREAD_ID, reason: "user" }, + }); + expect(h.registry.get(h.visibleAtom)?.threads[0]?.settledOverride).toBe("active"); + yield* Deferred.fail(first.reply, new Error("Settle rejected")); + yield* Effect.promise(() => settle); + expect(h.registry.get(h.visibleAtom)?.threads[0]?.settledOverride).toBe("active"); + const second = yield* Queue.take(h.requests); + expect(second.command.type).toBe("thread.unsettle"); + const confirmed = { + ...SNAPSHOT, + snapshotSequence: 2, + threads: [{ ...SNAPSHOT.threads[0]!, settledOverride: "active" as const }], + }; + h.registry.set(h.snapshotAtom(ENVIRONMENT_ID), confirmed); + yield* Deferred.succeed(second.reply, { sequence: 2 }); + yield* Effect.promise(() => unsettle); + expect(h.registry.get(h.visibleAtom)).toBe(confirmed); + }), + ); + + it.effect("isolates environments and does not restore a remotely removed thread", () => + Effect.gen(function* () { + const h = yield* makeHarness(); + const otherEnvironment = EnvironmentId.make("other-remote"); + const result = h.commands.settle.run(h.registry, { + environmentId: ENVIRONMENT_ID, + input: { threadId: THREAD_ID }, + }); + const request = yield* Queue.take(h.requests); + expect(h.registry.get(h.commands.snapshotAtom(otherEnvironment))).toBe(SNAPSHOT); + const removed = { ...SNAPSHOT, snapshotSequence: 2, threads: [] }; + h.registry.set(h.snapshotAtom(ENVIRONMENT_ID), removed); + expect(h.registry.get(h.visibleAtom)?.threads).toEqual([]); + yield* Deferred.fail(request.reply, new Error("Thread removed")); + yield* Effect.promise(() => result); + expect(h.registry.get(h.visibleAtom)).toBe(removed); + }), + ); + + it.effect("keeps pending approvals visible while a lifecycle request is pending", () => + Effect.gen(function* () { + const h = yield* makeHarness(); + const blocked = { + ...SNAPSHOT, + threads: [{ ...SNAPSHOT.threads[0]!, hasPendingApprovals: true }], + }; + h.registry.set(h.snapshotAtom(ENVIRONMENT_ID), blocked); + const result = h.commands.settle.run(h.registry, { + environmentId: ENVIRONMENT_ID, + input: { threadId: THREAD_ID }, + }); + const request = yield* Queue.take(h.requests); + expect(h.registry.get(h.visibleAtom)?.threads[0]).toBe(blocked.threads[0]); + yield* Deferred.fail(request.reply, new Error("Approval pending")); + yield* Effect.promise(() => result); + }), + ); + + for (const action of ["settle", "snooze"] as const) { + it.effect(`restores a confirmed ${action} when a queued undo fails`, () => + Effect.gen(function* () { + const h = yield* makeHarness(); + const parked = + action === "settle" + ? { settledOverride: "settled" as const } + : { snoozedUntil: "2099-01-01T00:00:00.000Z" }; + const awake = action === "settle" ? { settledOverride: "active" } : { snoozedUntil: null }; + const result = h.commands[action].run(h.registry, { + environmentId: ENVIRONMENT_ID, + input: { threadId: THREAD_ID, snoozedUntil: "2099-01-01T00:00:00.000Z" }, + }); + const first = yield* Queue.take(h.requests); + const undo = h.commands[action === "settle" ? "unsettle" : "unsnooze"].run(h.registry, { + environmentId: ENVIRONMENT_ID, + input: { threadId: THREAD_ID, reason: "user" }, + }); + expect(h.registry.get(h.visibleAtom)?.threads[0]).toMatchObject(awake); + yield* Deferred.succeed(first.reply, { sequence: 2 }); + expect((yield* Effect.promise(() => result))._tag).toBe("Success"); + expect(h.registry.get(h.visibleAtom)?.threads[0]).toMatchObject(awake); + const confirmed = { + ...SNAPSHOT, + snapshotSequence: 2, + threads: [{ ...SNAPSHOT.threads[0]!, ...parked }], + }; + h.registry.set(h.snapshotAtom(ENVIRONMENT_ID), confirmed); + expect(h.registry.get(h.visibleAtom)?.threads[0]).toMatchObject(awake); + const second = yield* Queue.take(h.requests); + expect(second.command.type).toBe( + action === "settle" ? "thread.unsettle" : "thread.unsnooze", + ); + yield* Deferred.fail(second.reply, new Error("Undo rejected")); + expect((yield* Effect.promise(() => undo))._tag).toBe("Failure"); + expect(h.registry.get(h.visibleAtom)).toBe(confirmed); + }), + ); + + it.effect(`preserves a newer approval when the ${action} reply arrives after the shell`, () => + Effect.gen(function* () { + const h = yield* makeHarness(); + const result = h.commands[action].run(h.registry, { + environmentId: ENVIRONMENT_ID, + input: { threadId: THREAD_ID, snoozedUntil: "2099-01-01T00:00:00.000Z" }, + }); + const request = yield* Queue.take(h.requests); + const newer = { + ...SNAPSHOT, + snapshotSequence: 3, + threads: [{ ...SNAPSHOT.threads[0]!, hasPendingApprovals: true }], + }; + h.registry.set(h.snapshotAtom(ENVIRONMENT_ID), newer); + expect(h.registry.get(h.visibleAtom)?.threads[0]).toBe(newer.threads[0]); + yield* Deferred.succeed(request.reply, { sequence: 2 }); + expect((yield* Effect.promise(() => result))._tag).toBe("Success"); + expect(h.registry.get(h.visibleAtom)).toBe(newer); + }), + ); + + it.effect(`shows an accepted ${action} while the shell still has an old input request`, () => + Effect.gen(function* () { + const h = yield* makeHarness(); + const stale = { + ...SNAPSHOT, + threads: [{ ...SNAPSHOT.threads[0]!, hasPendingUserInput: true }], + }; + h.registry.set(h.snapshotAtom(ENVIRONMENT_ID), stale); + const result = h.commands[action].run(h.registry, { + environmentId: ENVIRONMENT_ID, + input: { threadId: THREAD_ID, snoozedUntil: "2099-01-01T00:00:00.000Z" }, + }); + const request = yield* Queue.take(h.requests); + expect(h.registry.get(h.visibleAtom)?.threads[0]).toBe(stale.threads[0]); + yield* Deferred.succeed(request.reply, { sequence: 2 }); + expect((yield* Effect.promise(() => result))._tag).toBe("Success"); + expect(h.registry.get(h.visibleAtom)?.threads[0]).toMatchObject( + action === "settle" + ? { settledOverride: "settled" } + : { snoozedUntil: "2099-01-01T00:00:00.000Z" }, + ); + expect(h.registry.get(h.visibleAtom)?.threads[0]?.hasPendingUserInput).toBe(false); + expect(h.registry.get(h.snapshotAtom(ENVIRONMENT_ID))).toBe(stale); + }), + ); + } +}); diff --git a/packages/client-runtime/src/state/threadCommands.ts b/packages/client-runtime/src/state/threadCommands.ts index 1f10a0dff7ec..93e22cfd0c70 100644 --- a/packages/client-runtime/src/state/threadCommands.ts +++ b/packages/client-runtime/src/state/threadCommands.ts @@ -1,6 +1,13 @@ import * as Crypto from "effect/Crypto"; import { Atom } from "effect/unstable/reactivity"; -import { WS_METHODS } from "@t3tools/contracts"; +import { + WS_METHODS, + type EnvironmentId, + type OrchestrationShellSnapshot, +} from "@t3tools/contracts"; + +import { createOptimisticThreadLifecycle } from "./threadLifecycle.ts"; +import { canSnooze } from "./threadSettled.ts"; import { createAtomCommandScheduler, @@ -88,6 +95,7 @@ export type { export function createThreadEnvironmentAtoms( runtime: Atom.AtomRuntime, + snapshotAtom: (environmentId: EnvironmentId) => Atom.Atom, ) { const scheduler = createAtomCommandScheduler(); const concurrency = { @@ -95,7 +103,7 @@ export function createThreadEnvironmentAtoms( key: ({ environmentId, input }: { environmentId: string; input: { threadId: string } }) => JSON.stringify([environmentId, input.threadId]), }; - return { + const commands = { create: createEnvironmentCommand(runtime, { label: "environment-data:commands:thread:create", execute: (input: CreateThreadInput) => createThread(input), @@ -247,4 +255,79 @@ export function createThreadEnvironmentAtoms( concurrency, }), }; + const optimistic = createOptimisticThreadLifecycle(snapshotAtom); + return { + ...commands, + snapshotAtom: optimistic.snapshotAtom, + settle: optimistic.wrap(commands.settle, (thread, _input, now, accepted) => + !accepted && + (!canSnooze(thread, { now }) || + thread.session?.status === "starting" || + thread.session?.status === "running") + ? thread + : { + ...thread, + hasPendingApprovals: false, + hasPendingUserInput: false, + settledOverride: "settled", + settledAt: thread.settledOverride === "settled" ? (thread.settledAt ?? now) : now, + unsettledAt: null, + activeOrderKey: null, + pinnedAt: null, + pinOrderKey: null, + snoozedAt: null, + snoozedUntil: null, + }, + ), + unsettle: optimistic.wrap(commands.unsettle, (thread, input, now) => ({ + ...thread, + settledOverride: input.reason === "user" ? "active" : null, + settledAt: null, + unsettledAt: thread.settledOverride === "active" ? (thread.unsettledAt ?? null) : now, + })), + snooze: optimistic.wrap(commands.snooze, (thread, input, now, accepted) => + (!accepted && !canSnooze(thread, { now })) || + !(Date.parse(input.snoozedUntil) > Date.parse(now)) + ? thread + : { + ...thread, + hasPendingApprovals: false, + hasPendingUserInput: false, + snoozedUntil: input.snoozedUntil, + snoozedAt: thread.snoozedUntil === input.snoozedUntil ? (thread.snoozedAt ?? now) : now, + }, + ), + unsnooze: optimistic.wrap(commands.unsnooze, (thread) => ({ + ...thread, + snoozedUntil: null, + snoozedAt: null, + })), + pin: optimistic.wrap(commands.pin, (thread, input, now) => ({ + ...thread, + pinnedAt: thread.pinnedAt ?? now, + pinOrderKey: thread.pinnedAt == null ? (input.orderKey ?? null) : thread.pinOrderKey, + ...(thread.settledOverride === "settled" + ? { + settledOverride: "active" as const, + settledAt: null, + unsettledAt: now, + } + : {}), + snoozedUntil: null, + snoozedAt: null, + })), + unpin: optimistic.wrap(commands.unpin, (thread) => ({ + ...thread, + pinnedAt: null, + pinOrderKey: null, + })), + reorderPin: optimistic.wrap(commands.reorderPin, (thread, input) => ({ + ...thread, + pinOrderKey: input.orderKey, + })), + reorderActive: optimistic.wrap(commands.reorderActive, (thread, input) => ({ + ...thread, + activeOrderKey: input.orderKey, + })), + }; } diff --git a/packages/client-runtime/src/state/threadLifecycle.ts b/packages/client-runtime/src/state/threadLifecycle.ts new file mode 100644 index 000000000000..f5d98ffe3be3 --- /dev/null +++ b/packages/client-runtime/src/state/threadLifecycle.ts @@ -0,0 +1,102 @@ +import type { + EnvironmentId, + OrchestrationShellSnapshot, + OrchestrationThreadShell, + ThreadId, +} from "@t3tools/contracts"; +import * as DateTime from "effect/DateTime"; +import { Atom } from "effect/unstable/reactivity"; + +import type { AtomCommand } from "./runtime.ts"; + +interface PendingThreadUpdate { + readonly threadId: ThreadId; + readonly apply: (thread: OrchestrationThreadShell) => OrchestrationThreadShell; + sequence?: number; +} + +export function createOptimisticThreadLifecycle( + sourceSnapshotAtom: ( + environmentId: EnvironmentId, + ) => Atom.Atom, +) { + const pendingAtom = Atom.family((_environmentId: EnvironmentId) => + Atom.make>([]).pipe(Atom.keepAlive), + ); + const snapshotAtom = Atom.family((environmentId: EnvironmentId) => + Atom.make((get) => { + const snapshot = get(sourceSnapshotAtom(environmentId)); + const pending = get(pendingAtom(environmentId)); + if (snapshot === null || pending.length === 0) return snapshot; + const byThread = new Map(); + for (const update of pending) { + if (update.sequence !== undefined && update.sequence <= snapshot.snapshotSequence) continue; + const updates = byThread.get(update.threadId) ?? []; + updates.push(update); + byThread.set(update.threadId, updates); + } + if (byThread.size === 0) return snapshot; + return { + ...snapshot, + threads: snapshot.threads.map((thread) => + (byThread.get(thread.id) ?? []).reduce( + (current, update) => update.apply(current), + thread, + ), + ), + }; + }), + ); + + function wrap( + command: AtomCommand< + { readonly environmentId: EnvironmentId; readonly input: Input }, + { readonly sequence: number }, + E + >, + apply: ( + thread: OrchestrationThreadShell, + input: Input, + now: string, + accepted: boolean, + ) => OrchestrationThreadShell, + ): typeof command { + return { + label: command.label, + run: async (registry, target) => { + const now = DateTime.formatIso(DateTime.nowUnsafe()); + const pending = pendingAtom(target.environmentId); + const source = sourceSnapshotAtom(target.environmentId); + const update: PendingThreadUpdate = { + threadId: target.input.threadId, + apply: (thread) => apply(thread, target.input, now, update.sequence !== undefined), + }; + const remove = () => + registry.update(pending, (current) => current.filter((item) => item !== update)); + registry.update(pending, (current) => [...current, update]); + let confirmed = false; + try { + const result = await command.run(registry, target); + if (result._tag === "Success") { + update.sequence = result.value.sequence; + registry.update(pending, (current) => [...current]); + const reconcile = (snapshot: OrchestrationShellSnapshot | null) => { + if (snapshot === null || snapshot.snapshotSequence >= result.value.sequence) { + remove(); + unsubscribe(); + } + }; + const unsubscribe = registry.subscribe(source, reconcile); + reconcile(registry.get(source)); + confirmed = true; + } + return result; + } finally { + if (!confirmed) remove(); + } + }, + }; + } + + return { snapshotAtom, wrap }; +}