diff --git a/apps/mobile/src/state/acknowledged-thread-messages.test.ts b/apps/mobile/src/state/acknowledged-thread-messages.test.ts new file mode 100644 index 0000000000000..7dadf26312410 --- /dev/null +++ b/apps/mobile/src/state/acknowledged-thread-messages.test.ts @@ -0,0 +1,60 @@ +import { CommandId, EnvironmentId, MessageId, ThreadId } from "@t3tools/contracts"; +import { describe, expect, it, vi } from "vite-plus/test"; + +import { unsettledAcknowledgedThreadMessages } from "./acknowledged-thread-messages"; + +vi.mock("./threads", () => ({ environmentThreadDetails: {} })); + +const environmentId = EnvironmentId.make("environment-1"); +const threadId = ThreadId.make("thread-1"); + +function acknowledged(messageId: string, acknowledgedSequence?: number) { + return { + environmentId, + threadId, + messageId: MessageId.make(messageId), + commandId: CommandId.make(`command-${messageId}`), + text: messageId, + attachments: [], + // Client time can be ahead of the server, or equal for different sends. + createdAt: "2099-01-01T00:00:00.000Z", + ...(acknowledgedSequence === undefined ? {} : { acknowledgedSequence }), + }; +} + +describe("acknowledged message reconciliation", () => { + it("settles an unloaded echo only after a bounded snapshot includes its receipt", () => { + const old = acknowledged("accepted-outside-window", 40); + const inFlight = acknowledged("accepted-after-snapshot", 51); + const unknown = acknowledged("no-ordering-evidence"); + const otherThread = { ...old, threadId: ThreadId.make("thread-2") }; + const otherEnvironment = { ...old, environmentId: EnvironmentId.make("environment-2") }; + const echoed = acknowledged("loaded-echo", 52); + const messages = [old, inFlight, unknown, otherThread, otherEnvironment, echoed]; + const echoes = new Set([echoed.messageId]); + const history = { environmentId, threadId, hasMoreHistory: true, snapshotSequence: 40 }; + + expect(unsettledAcknowledgedThreadMessages(messages, echoes, history)).toEqual([ + inFlight, + unknown, + otherThread, + otherEnvironment, + ]); + expect( + unsettledAcknowledgedThreadMessages(messages, echoes, { ...history, snapshotSequence: 39 }), + ).toEqual([old, inFlight, unknown, otherThread, otherEnvironment]); + }); + + it("keeps missing echoes without evidence of a bounded, committed history window", () => { + const message = acknowledged("still-waiting", 40); + const history = { environmentId, threadId, hasMoreHistory: false, snapshotSequence: 50 }; + expect(unsettledAcknowledgedThreadMessages([message], new Set(), history)).toEqual([message]); + expect( + unsettledAcknowledgedThreadMessages([message], new Set(), { + environmentId, + threadId, + hasMoreHistory: true, + }), + ).toEqual([message]); + }); +}); diff --git a/apps/mobile/src/state/acknowledged-thread-messages.ts b/apps/mobile/src/state/acknowledged-thread-messages.ts index ab9f26dd7d412..9bfc083a0288d 100644 --- a/apps/mobile/src/state/acknowledged-thread-messages.ts +++ b/apps/mobile/src/state/acknowledged-thread-messages.ts @@ -1,12 +1,44 @@ +import type { MessageId, ScopedThreadRef } from "@t3tools/contracts"; import * as Option from "effect/Option"; import { Atom } from "effect/reactivity"; import { appAtomRegistry } from "./atom-registry"; import type { QueuedThreadMessage } from "./thread-outbox-model"; import { environmentThreadDetails } from "./threads"; +interface AcknowledgedThreadMessage extends QueuedThreadMessage { + readonly acknowledgedSequence?: number; +} + +export function unsettledAcknowledgedThreadMessages( + messages: ReadonlyArray, + echoedIds: ReadonlySet, + history?: ScopedThreadRef & { + readonly hasMoreHistory: boolean; + readonly snapshotSequence?: number; + }, +): ReadonlyArray { + return messages.filter((message) => { + if ( + history !== undefined && + (message.environmentId !== history.environmentId || message.threadId !== history.threadId) + ) { + return true; + } + if (echoedIds.has(message.messageId)) return false; + // A bounded snapshot taken after the send committed can omit its echo. + // Client timestamps cannot establish this ordering, especially after offline sends. + return !( + history?.hasMoreHistory && + history.snapshotSequence !== undefined && + message.acknowledgedSequence !== undefined && + message.acknowledgedSequence <= history.snapshotSequence + ); + }); +} + // Keep acknowledged rows until their subscribed projection arrives; command receipts can precede messages. // Render hides rows the projection has echoed, so dropping them here only frees memory. -export const acknowledgedThreadMessagesAtom = Atom.make>( +export const acknowledgedThreadMessagesAtom = Atom.make>( [], ).pipe(Atom.keepAlive); @@ -25,7 +57,9 @@ function isEchoedInMountedThread(message: QueuedThreadMessage): boolean { } function writeAcknowledgedThreadMessages( - update: (current: ReadonlyArray) => ReadonlyArray, + update: ( + current: ReadonlyArray, + ) => ReadonlyArray, ) { const current = appAtomRegistry.get(acknowledgedThreadMessagesAtom); const next = update(current).filter((entry) => !isEchoedInMountedThread(entry)); @@ -35,11 +69,16 @@ function writeAcknowledgedThreadMessages( appAtomRegistry.set(acknowledgedThreadMessagesAtom, next); } -export function retainAcknowledgedThreadMessage(message: QueuedThreadMessage) { +export function retainAcknowledgedThreadMessage( + message: QueuedThreadMessage, + acknowledgedSequence?: number, +) { + const acknowledgedMessage = + acknowledgedSequence === undefined ? message : { ...message, acknowledgedSequence }; writeAcknowledgedThreadMessages((current) => current.some((entry) => entry.messageId === message.messageId) ? current - : [...current, message], + : [...current, acknowledgedMessage], ); } diff --git a/apps/mobile/src/state/thread-task-permissions.test.ts b/apps/mobile/src/state/thread-task-permissions.test.ts index 5697ca63e5ea5..bf7f6747e60a6 100644 --- a/apps/mobile/src/state/thread-task-permissions.test.ts +++ b/apps/mobile/src/state/thread-task-permissions.test.ts @@ -1,6 +1,9 @@ import { AuthOrchestrationOperateScope, DEFAULT_SERVER_SETTINGS, + CommandId, + EnvironmentId, + ThreadId, MessageId, ProjectId, ProviderDriverKind, @@ -32,6 +35,7 @@ const state = vi.hoisted(() => ({ }, serverConfig: null as ServerConfig | null, projection: null as OrchestrationV2ThreadProjection | null, + history: { hasMoreHistory: true, snapshotSequence: 39 }, pendingRequests: { approvals: [], userInputs: [] } as PendingThreadRequests, enqueue: vi.fn(async () => undefined), clearDraft: vi.fn(), @@ -94,6 +98,7 @@ vi.mock("./use-thread-selection", () => ({ useThreadSelection: () => ({ selectedThread: state.thread, selectedThreadCreation: null, + selectedThreadDetailRef: { environmentId: "secondary", threadId: "thread" }, selectedEnvironmentRuntime: { connectionState: state.connectionState, serverConfig: state.serverConfig, @@ -113,6 +118,7 @@ vi.mock("./threads", async () => { environmentThreadDetails: { queueWorkflowAtom: () => Atom.make(null), checkpointSummariesAtom: () => Atom.make([]), + historyAtom: () => Atom.make(state.history), pendingRequestsAtom: () => Atom.make(state.pendingRequests), }, threadEnvironment: { @@ -174,6 +180,9 @@ vi.mock("../lib/commandMetadata", () => ({ vi.mock("../lib/copyTextWithHaptic", () => ({ copyTextWithHaptic: vi.fn() })); vi.mock("./use-remote-environment-registry", () => ({ setPendingConnectionError: vi.fn() })); +import { appAtomRegistry } from "./atom-registry"; +import { acknowledgedThreadMessagesAtom } from "./acknowledged-thread-messages"; + import { useThreadComposerState } from "./use-thread-composer-state"; import { useSelectedThreadRequests } from "./use-selected-thread-requests"; @@ -182,6 +191,8 @@ beforeEach(() => { state.connectionState = "connected"; state.draft = { text: "Keep my draft", attachments: [], modelSelection: undefined }; state.projection = null; + state.history = { hasMoreHistory: true, snapshotSequence: 39 }; + appAtomRegistry.set(acknowledgedThreadMessagesAtom, []); state.serverConfig = { providers: [{ instanceId: "codex", driver: "codex", enabled: true }], environment: { capabilities: {} }, @@ -428,3 +439,27 @@ describe("mobile handoff preview and submission", () => { expect(state.clearDraft).toHaveBeenCalledTimes(1); }); }); + +it("reopening bounded history removes an accepted row only after its receipt", () => { + const message = { + environmentId: EnvironmentId.make("secondary"), + threadId: ThreadId.make("thread"), + messageId: MessageId.make("accepted-outside-window"), + commandId: CommandId.make("accepted-command"), + text: "old accepted message", + attachments: [], + createdAt: "2099-01-01T00:00:00.000Z", + acknowledgedSequence: 40, + }; + appAtomRegistry.set(acknowledgedThreadMessagesAtom, [message]); + state.projection = { ...v2Projection, messages: [] }; + expect(useThreadComposerState().selectedThreadFeed.map((entry) => entry.id)).toEqual([ + message.messageId, + ]); + state.history = { hasMoreHistory: true, snapshotSequence: 40 }; + expect(useThreadComposerState().selectedThreadFeed).toEqual([]); + state.history = { hasMoreHistory: false, snapshotSequence: 40 }; + expect(useThreadComposerState().selectedThreadFeed.map((entry) => entry.id)).toEqual([ + message.messageId, + ]); +}); diff --git a/apps/mobile/src/state/threadOutboxDelivery.permissions.test.ts b/apps/mobile/src/state/threadOutboxDelivery.permissions.test.ts index edc80c5007024..fce22de75e9c2 100644 --- a/apps/mobile/src/state/threadOutboxDelivery.permissions.test.ts +++ b/apps/mobile/src/state/threadOutboxDelivery.permissions.test.ts @@ -224,6 +224,7 @@ beforeEach(() => { ]) { command.mockReset().mockResolvedValue(AsyncResult.success(undefined)); } + state.start.mockResolvedValue(AsyncResult.success({ sequence: 1 })); state.prepare.mockReset().mockImplementation(async ({ attachments }) => ({ status: "ready", attachments: [], @@ -427,7 +428,8 @@ describe("queued task operation access", () => { ); await state.manager.enqueue(queued); const started = Promise.withResolvers(); - const delivery = Promise.withResolvers>(); + const delivery = + Promise.withResolvers>(); state.start.mockImplementationOnce(() => { started.resolve(); return delivery.promise; @@ -437,7 +439,7 @@ describe("queued task operation access", () => { setOperateGrant("secondary", false); delivery.resolve( accepted - ? AsyncResult.success(undefined) + ? AsyncResult.success({ sequence: 1 }) : AsyncResult.failure(Cause.fail(new Error("This connection cannot control this task."))), ); await drained; diff --git a/apps/mobile/src/state/threadOutboxDelivery.test.ts b/apps/mobile/src/state/threadOutboxDelivery.test.ts index 901ffaa9e8291..e8555f15423f4 100644 --- a/apps/mobile/src/state/threadOutboxDelivery.test.ts +++ b/apps/mobile/src/state/threadOutboxDelivery.test.ts @@ -458,10 +458,14 @@ describe("thread outbox drain delivery cleanup", () => { await harness.manager.enqueue(message); const deliveryRevision = harness.manager.revisionOf(message.messageId); - await expect(completeQueuedMessageDelivery(message, deliveryRevision)).resolves.toBe("removed"); + await expect(completeQueuedMessageDelivery(message, deliveryRevision, 40)).resolves.toBe( + "removed", + ); expect(remainingMessages()).toEqual([]); - expect(appAtomRegistry.get(acknowledgedThreadMessagesAtom)).toEqual([message]); + expect(appAtomRegistry.get(acknowledgedThreadMessagesAtom)).toEqual([ + { ...message, acknowledgedSequence: 40 }, + ]); }); it("keeps a delivered message when its editor opens during storage removal", async () => { diff --git a/apps/mobile/src/state/threadOutboxDelivery.ts b/apps/mobile/src/state/threadOutboxDelivery.ts index 89c6f13358952..0fcb5a54c2105 100644 --- a/apps/mobile/src/state/threadOutboxDelivery.ts +++ b/apps/mobile/src/state/threadOutboxDelivery.ts @@ -254,6 +254,7 @@ function isQueuedMessagePayloadCurrent( export async function completeQueuedMessageDelivery( queuedMessage: QueuedThreadMessage, deliveryRevision: number, + acknowledgedSequence?: number, ): Promise<"removed" | "edited" | "failed"> { clearThreadComposerError( scopedThreadKey(queuedMessage.environmentId, queuedMessage.threadId), @@ -270,7 +271,7 @@ export async function completeQueuedMessageDelivery( if (appAtomRegistry.get(editingQueuedMessageIdsAtom)[queuedMessage.messageId]) { return "edited"; } - retainAcknowledgedThreadMessage(queuedMessage); + retainAcknowledgedThreadMessage(queuedMessage, acknowledgedSequence); const removed = await removeThreadOutboxMessage( queuedMessage, deliveryRevision, @@ -904,7 +905,13 @@ export function createThreadOutboxDelivery(): { } acknowledgedExistingThreadMessageIds.add(persistedMessage.messageId); const delivered = - (await completeQueuedMessageDelivery(persistedMessage, deliveryRevision)) === "removed"; + (await completeQueuedMessageDelivery( + persistedMessage, + deliveryRevision, + AsyncResult.isSuccess(deliveryResult) && "sequence" in deliveryResult.value + ? deliveryResult.value.sequence + : undefined, + )) === "removed"; if (delivered) { acknowledgedExistingThreadMessageIds.delete(persistedMessage.messageId); } diff --git a/apps/mobile/src/state/use-thread-composer-state.ts b/apps/mobile/src/state/use-thread-composer-state.ts index 74206557b0ad2..ce236d2746af6 100644 --- a/apps/mobile/src/state/use-thread-composer-state.ts +++ b/apps/mobile/src/state/use-thread-composer-state.ts @@ -13,7 +13,7 @@ import { deriveThreadRuntime, threadRuntimeHasInterruptibleRun, } from "@t3tools/client-runtime/state/thread-execution"; -import { useRef, useState } from "react"; +import { useEffect, useRef, useState } from "react"; import { CommandId, @@ -30,6 +30,7 @@ import { import { clampFileAttachmentUploadBytes } from "@t3tools/client-runtime/state/attachments"; import { nextPastedTextFileName, pastedTextDisposition } from "@t3tools/client-runtime/text-paste"; import { + EMPTY_THREAD_HISTORY_META, parseCodexFeedbackCommand, submitCodexFeedback, type CodexFeedbackSubmission, @@ -53,7 +54,10 @@ import { import type { DraftComposerImageAttachment } from "../lib/composerImages"; import { scopedThreadKey } from "../lib/scopedEntities"; import { buildThreadFeed, createThreadFeedCache } from "../lib/threadActivity"; -import { acknowledgedThreadMessagesAtom } from "./acknowledged-thread-messages"; +import { + acknowledgedThreadMessagesAtom, + unsettledAcknowledgedThreadMessages, +} from "./acknowledged-thread-messages"; import { appendPendingThreadMessages, selectPendingAcknowledgedMessages, @@ -119,6 +123,7 @@ const EMPTY_QUEUE_WORKFLOW_ATOM = Atom.make(null).pipe( Atom.withLabel("mobile-thread-queue-workflow:empty"), ); const EMPTY_CHECKPOINT_SUMMARIES_ATOM = Atom.make>([]); +const EMPTY_THREAD_HISTORY_ATOM = Atom.make(EMPTY_THREAD_HISTORY_META); export function appendReviewCommentToDraft(input: { readonly environmentId: EnvironmentId; @@ -250,6 +255,16 @@ export function useThreadComposerState() { const selectedThreadMessages = selectedThreadProjection?.projection.messages; const selectedThreadAttempts = selectedThreadProjection?.projection.attempts; const selectedThreadNodes = selectedThreadProjection?.projection.nodes; + const selectedThreadHistory = useAtomValue( + selectedThreadDetailRef === null + ? EMPTY_THREAD_HISTORY_ATOM + : environmentThreadDetails.historyAtom(selectedThreadDetailRef), + ); + const echoedIds = new Set(selectedThreadMessages?.map((message) => message.id)); + const acknowledgmentHistory = + selectedThreadDetailRef === null + ? undefined + : { ...selectedThreadDetailRef, ...selectedThreadHistory }; const checkpointSummaries = useAtomValue( selectedThreadDetailRef === null ? EMPTY_CHECKPOINT_SUMMARIES_ATOM @@ -274,16 +289,39 @@ export function useThreadComposerState() { checkpointSummaries, cache: feedCache, }); - const pendingAcknowledgments = selectPendingAcknowledgedMessages(acknowledgedMessages, { - threadKey: selectedThreadKey, - queuedMessages: selectedThreadQueuedMessages, - projectedMessages: selectedThreadMessages, - }); + const pendingAcknowledgments = selectPendingAcknowledgedMessages( + unsettledAcknowledgedThreadMessages(acknowledgedMessages, echoedIds, acknowledgmentHistory), + { + threadKey: selectedThreadKey, + queuedMessages: selectedThreadQueuedMessages, + projectedMessages: selectedThreadMessages, + }, + ); if (pendingAcknowledgments.length === 0) return feed; return appendPendingThreadMessages(feed, feed, pendingAcknowledgments).map((entry) => entry.pendingMessage ? { ...entry, acknowledged: true } : entry, ); })(); + // oxlint-disable-next-line t3code/react-effect-budget -- Retire acknowledged rows when the subscribed history window advances. + useEffect(() => { + if (acknowledgedMessages.length === 0) return; + const current = appAtomRegistry.get(acknowledgedThreadMessagesAtom); + const unsettled = unsettledAcknowledgedThreadMessages( + current, + new Set(selectedThreadMessages?.map((message) => message.id)), + selectedThreadDetailRef === null + ? undefined + : { ...selectedThreadDetailRef, ...selectedThreadHistory }, + ); + if (unsettled.length !== current.length) { + appAtomRegistry.set(acknowledgedThreadMessagesAtom, unsettled); + } + }, [ + acknowledgedMessages, + selectedThreadMessages, + selectedThreadDetailRef, + selectedThreadHistory, + ]); const preferencesResult = useAtomValue(mobilePreferencesAtom); const followUpBehavior = AsyncResult.isSuccess(preferencesResult) diff --git a/packages/client-runtime/src/state/threadHistoryMerge.ts b/packages/client-runtime/src/state/threadHistoryMerge.ts index 24840ee28420a..b653ecf50a591 100644 --- a/packages/client-runtime/src/state/threadHistoryMerge.ts +++ b/packages/client-runtime/src/state/threadHistoryMerge.ts @@ -5,6 +5,8 @@ import type { import { isOrchestrationV2TurnItemVisible } from "@t3tools/shared/orchestrationV2Timeline"; export type ThreadHistoryMeta = { + /** Server sequence of the snapshot that established this history window. */ + readonly snapshotSequence?: number; readonly historyCursor: string | null; readonly hasMoreHistory: boolean; readonly loading: boolean; @@ -119,12 +121,12 @@ export function applyHistoryPageMeta( }, ): ThreadHistoryMeta { return { + ...current, historyCursor: page.nextCursor, hasMoreHistory: page.hasMoreHistory, loading: false, error: null, expanded: true, - latestLocalTurnOrdinal: current.latestLocalTurnOrdinal, }; } diff --git a/packages/client-runtime/src/state/threads-sync.test.ts b/packages/client-runtime/src/state/threads-sync.test.ts index 93bee9f46dd09..b002606ead154 100644 --- a/packages/client-runtime/src/state/threads-sync.test.ts +++ b/packages/client-runtime/src/state/threads-sync.test.ts @@ -1198,6 +1198,7 @@ describe("EnvironmentThreads", () => { value.status === "live" && value.history.historyCursor === "socket-history-cursor", ); expect(state.history).toMatchObject({ + snapshotSequence: 14, historyCursor: "socket-history-cursor", hasMoreHistory: true, loading: false, @@ -1222,6 +1223,7 @@ describe("EnvironmentThreads", () => { Option.getOrThrow(expanded.data).turnItems.some((item) => item.id === olderItem.id), ).toBe(true); expect(expanded.history.hasMoreHistory).toBe(false); + expect(expanded.history.snapshotSequence).toBe(14); }), ); @@ -1488,6 +1490,7 @@ describe("EnvironmentThreads", () => { ); expect(state.history).toMatchObject({ + snapshotSequence: 11, historyCursor: "opaque-cursor", hasMoreHistory: true, loading: false, @@ -1505,6 +1508,7 @@ describe("EnvironmentThreads", () => { const harness = yield* makeHarness({ cached: BASE_PROJECTION }); const state = yield* awaitThreadState(harness.observed, (value) => Option.isSome(value.data)); expect(state.history).toMatchObject({ + snapshotSequence: CACHED_SNAPSHOT_SEQUENCE, historyCursor: null, hasMoreHistory: false, expanded: false, diff --git a/packages/client-runtime/src/state/threads.ts b/packages/client-runtime/src/state/threads.ts index 693af5329fe1b..268e26b60a76f 100644 --- a/packages/client-runtime/src/state/threads.ts +++ b/packages/client-runtime/src/state/threads.ts @@ -81,6 +81,7 @@ function historyMetaFromCachedSnapshot( const historyCursor = snapshot.historyCursor ?? null; const hasMoreHistory = snapshot.hasMoreHistory ?? false; return { + snapshotSequence: snapshot.snapshotSequence, historyCursor, hasMoreHistory, loading: false, @@ -616,6 +617,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make hasProgressiveHistory ? { history: { + snapshotSequence: item.snapshotSequence, historyCursor: item.historyCursor ?? null, hasMoreHistory: item.hasMoreHistory ?? false, loading: false, @@ -929,6 +931,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make const history: ThreadHistoryMeta = httpResult.history !== undefined ? { + snapshotSequence: httpResult.snapshot.snapshotSequence, historyCursor: httpResult.history.historyCursor, hasMoreHistory: httpResult.history.hasMoreHistory, loading: false,