diff --git a/apps/mobile/src/App.tsx b/apps/mobile/src/App.tsx index 2033b45d563d2..5a351581da8a9 100644 --- a/apps/mobile/src/App.tsx +++ b/apps/mobile/src/App.tsx @@ -1,6 +1,8 @@ import { installPermissionUpdateNotice } from "./features/connection/permissionUpdateNotice"; import { installComposerAttachmentUploads } from "./state/composer-attachment-uploads"; import { installThreadOutboxDelivery } from "./state/threadOutboxDelivery"; +import { installThreadOfflineSync } from "./state/thread-prefetch"; +import { installThreadCacheProtection } from "./state/thread-cache-protection"; import * as Linking from "expo-linking"; import * as SplashScreen from "expo-splash-screen"; import { useEffect } from "react"; @@ -90,7 +92,9 @@ declare const module: { readonly hot?: FoundationHotModule } | undefined; const appServiceDisposers = [ installPermissionUpdateNotice(), + installThreadCacheProtection(), installThreadOutboxDelivery(), + installThreadOfflineSync(), installComposerAttachmentUploads(), ]; disposeOnFoundationReplace(typeof module === "undefined" ? undefined : module.hot, () => { diff --git a/apps/mobile/src/connection/environment-cache-store.test.ts b/apps/mobile/src/connection/environment-cache-store.test.ts index d8031de9e7a39..d4491a9bec3e7 100644 --- a/apps/mobile/src/connection/environment-cache-store.test.ts +++ b/apps/mobile/src/connection/environment-cache-store.test.ts @@ -232,6 +232,9 @@ function makeDatabase() { values.set(id, payload); schemaVersions.set(id, schemaVersion); }), + listThreadCacheIndex: () => Effect.succeed([]), + enforceThreadCacheBudget: () => Effect.succeed({ evicted: 0, bytes: 0 }), + setThreadCacheProtection: () => Effect.succeed(undefined), removeCache: (environmentId, kind, cacheKey) => Effect.sync(() => { const id = cacheId(environmentId, kind, cacheKey); diff --git a/apps/mobile/src/connection/environment-cache-store.ts b/apps/mobile/src/connection/environment-cache-store.ts index b55af4718d15f..76b0d2f132826 100644 --- a/apps/mobile/src/connection/environment-cache-store.ts +++ b/apps/mobile/src/connection/environment-cache-store.ts @@ -7,6 +7,7 @@ import { makeStoredShellSnapshotEncoder, } from "@t3tools/client-runtime/platform"; import { type EnvironmentId, ServerConfig, VcsListRefsResult } from "@t3tools/contracts"; +import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; @@ -147,7 +148,14 @@ export const make = Effect.fn("MobileEnvironmentCacheStore.make")(function* () { snapshot, }).pipe(Effect.mapError((cause) => persistenceError("save-thread", cause))); yield* database - .saveCache(environmentId, "thread", threadId, COMPACT_THREAD_CACHE_SCHEMA_VERSION, payload) + .saveCache( + environmentId, + "thread", + threadId, + COMPACT_THREAD_CACHE_SCHEMA_VERSION, + payload, + DateTime.toEpochMillis(snapshot.projection.thread.updatedAt), + ) .pipe(Effect.mapError(mapDatabaseError("save-thread"))); }), removeThread: Effect.fn("MobileEnvironmentCache.removeThread")((environmentId, threadId) => diff --git a/apps/mobile/src/features/threads/ThreadDetailScreen.tsx b/apps/mobile/src/features/threads/ThreadDetailScreen.tsx index 1820baa8095ca..b376eb5608524 100644 --- a/apps/mobile/src/features/threads/ThreadDetailScreen.tsx +++ b/apps/mobile/src/features/threads/ThreadDetailScreen.tsx @@ -174,7 +174,8 @@ import { FLOATING_WORKING_CONTROL_COVERAGE, FloatingWorkingControl, } from "./floating-working-control"; -import { connectionFloatingStatus, type FloatingWorkingStatus } from "./floating-working-status"; +import type { FloatingWorkingStatus } from "./floating-working-status"; +import { useThreadConnectionChip } from "./use-thread-connection-chip"; import { derivePendingUserInputMaxHeight, ESTIMATED_KEYBOARD_HEIGHT, @@ -610,16 +611,15 @@ function ThreadDetailBody(props: ThreadDetailScreenProps) { const pendingBackgroundWork = presentPendingBackgroundWork( props.selectedThread.pendingBackgroundTasks, ); + const connectionChip = useThreadConnectionChip({ + environmentId: props.environmentId, + threadId: props.selectedThread.id, + connectionState: props.connectionStateLabel, + connectionError: props.connectionError, + onReconnect: props.onReconnectEnvironment, + queuedCount: props.queuedMessages.length, + }); const floatingStatus = ((): FloatingWorkingStatus | null => { - const connectionStatus = connectionFloatingStatus({ - connectionError: props.connectionError, - connectionState: props.connectionStateLabel, - environmentLabel: props.environmentLabel, - onReconnect: props.onReconnectEnvironment, - }); - if (connectionStatus !== null && !props.draftMode) { - return connectionStatus; - } if (props.activePendingApproval !== null || props.activePendingUserInput !== null) { return null; } @@ -651,7 +651,11 @@ function ThreadDetailBody(props: ThreadDetailScreenProps) { if (props.isCompacting && contentPresentationKind === "ready") { return { kind: "compacting" }; } - if (props.activeWorkStartedAt !== null && contentPresentationKind === "ready") { + if ( + props.activeWorkStartedAt !== null && + contentPresentationKind === "ready" && + (props.connectionStateLabel === "connected" || props.threadSyncStatus !== "cached") + ) { return { kind: "working", startedAt: props.activeWorkStartedAt }; } if (pendingBackgroundWork !== null && contentPresentationKind === "ready") { @@ -721,7 +725,7 @@ function ThreadDetailBody(props: ThreadDetailScreenProps) { agentsSegment !== null || devicePreviews.length > 0 || browserTabs.tabs.length > 0 || - props.connectionStateLabel !== "connected" || + connectionChip !== null || props.queuedMessages.length > 0 || props.selectedThreadFeed.some( (entry) => "acknowledged" in entry && entry.acknowledged === true, @@ -1298,6 +1302,7 @@ function ThreadDetailBody(props: ThreadDetailScreenProps) { worktreeSetup={props.worktreeSetup} setupWorkingStartedAt={props.setupWorkingStartedAt} queuedMessages={props.queuedMessages} + connectionState={props.connectionStateLabel} dispatchingMessageId={props.dispatchingMessageId} // A native subagent has no composer to edit a pending message in; // Cancel on the edit banner would discard it. @@ -1374,9 +1379,6 @@ function ThreadDetailBody(props: ThreadDetailScreenProps) { composerWidthStyle, ]} > - {/* No paddingTop here: the overlay's measured height becomes the - list's bottom inset, so any padding above the pill/composer - pushes the resting content floor up by the same amount. */} 0 diff --git a/apps/mobile/src/features/threads/ThreadFeed.tsx b/apps/mobile/src/features/threads/ThreadFeed.tsx index 36fe66a026493..1aef5ee0fc7ce 100644 --- a/apps/mobile/src/features/threads/ThreadFeed.tsx +++ b/apps/mobile/src/features/threads/ThreadFeed.tsx @@ -53,6 +53,7 @@ import { type ScheduledTaskId, type ScopedThreadRef, } from "@t3tools/contracts"; +import type { EnvironmentConnectionPhase } from "@t3tools/client-runtime/connection"; import { renderAssistantCitationsAsText } from "@t3tools/shared/assistantCitations"; import { encodeComposerContextFragment } from "@t3tools/shared/composerContextClipboard"; import { parseThreadLinkHref, resolveThreadLinkRef } from "@t3tools/shared/threadLinks"; @@ -329,6 +330,7 @@ export interface ThreadFeedProps { readonly worktreeSetup?: WorktreeSetupCardProps | null; readonly setupWorkingStartedAt?: string | null; readonly queuedMessages: ReadonlyArray; + readonly connectionState?: EnvironmentConnectionPhase; readonly dispatchingMessageId: MessageId | null; readonly onEditPendingMessage: ((message: QueuedThreadMessage) => void) | null; readonly environmentId: EnvironmentId; @@ -1240,6 +1242,7 @@ function renderFeedEntry( | "onUseArtifactTemplate" | "skills" | "dispatchingMessageId" + | "connectionState" | "onEditPendingMessage" | "threadId" | "workspaceRoot" @@ -1700,7 +1703,13 @@ function renderFeedEntry( ) : null} - {entry.pendingMessage && !entry.acknowledged ? "Pending" : timestampLabel} + {entry.pendingMessage && !entry.acknowledged + ? props.connectionState !== "connected" + ? "Sends when connected" + : entry.pendingMessage.messageId === props.dispatchingMessageId + ? "Sending…" + : "Queued" + : timestampLabel} {props.onEditPendingMessage !== null && entry.pendingMessage && @@ -2635,6 +2644,7 @@ export const ThreadFeed = memo(function ThreadFeed({ worktreeSetup: props.worktreeSetup, setupWorkingStartedAt: props.setupWorkingStartedAt, dispatchingMessageId: props.dispatchingMessageId, + connected: props.connectionState === "connected", unsettledTurnId, copiedRowId, expandedWorkGroups, @@ -3075,6 +3085,7 @@ export const ThreadFeed = memo(function ThreadFeed({ {renderFeedEntry(info, { environmentId: props.environmentId, + connectionState: props.connectionState, dispatchingMessageId: props.dispatchingMessageId, onEditPendingMessage: props.onEditPendingMessage, onUseArtifactTemplate: props.onUseArtifactTemplate, diff --git a/apps/mobile/src/features/threads/floating-working-control.tsx b/apps/mobile/src/features/threads/floating-working-control.tsx index 24322aa46d43d..1db7fefd3c4b5 100644 --- a/apps/mobile/src/features/threads/floating-working-control.tsx +++ b/apps/mobile/src/features/threads/floating-working-control.tsx @@ -48,7 +48,6 @@ const CONTROL_SEPARATION = (CONTROL_ROW_GAP + CONTROL_HEIGHT) / 2; const LABEL_ENTERING = FadeIn.duration(160).delay(80).reduceMotion(ReduceMotion.System); const LABEL_EXITING = FadeOut.duration(100).reduceMotion(ReduceMotion.System); -// Expo reapplies glass after layout or reattachment, restoring material UIKit can lose behind visible labels. const UniwindGlassView = withUniwind(GlassView, { style: { fromClassName: "className" }, }); @@ -57,7 +56,6 @@ const UniwindGlassContainer = withUniwind(GlassContainer, { }); const AnimatedGlassView = Animated.createAnimatedComponent(UniwindGlassView); -// Segment widths feed back into labelWidth; ignore sub-point changes to avoid rounding-driven ellipsis loops. function settleWidth(next: number) { return (current: number) => (Math.abs(current - next) < 1 ? current : next); } @@ -68,6 +66,7 @@ export const FLOATING_WORKING_CONTROL_COVERAGE = CONTROL_OVERLAY_OFFSET + CONTRO export function FloatingWorkingControl(props: { readonly colorScheme: "light" | "dark"; readonly status: FloatingWorkingStatus | null; + readonly connectionChip: { readonly label: string; readonly onPress: () => void } | null; readonly devicePreview: { readonly count: number; readonly onPress: () => void } | null; readonly browserPreview: { readonly count: number; readonly onPress: () => void } | null; readonly showScrollToEnd: boolean; @@ -81,6 +80,7 @@ export function FloatingWorkingControl(props: { const { width: windowWidth } = useWindowDimensions(); const [overlayWidth, setOverlayWidth] = useState(windowWidth); const [queueWidth, setQueueWidth] = useState(0); + const [connectionWidth, setConnectionWidth] = useState(0); const [actionWidth, setActionWidth] = useState(0); const action = props.status?.kind === "background" || props.status?.kind === "lifecycle" @@ -88,6 +88,7 @@ export function FloatingWorkingControl(props: { : null; const [agentsWidth, setAgentsWidth] = useState(0); const hasQueue = props.queuedCount > 0; + const hasConnectionChip = props.connectionChip !== null; const hasPreview = props.devicePreview !== null || props.browserPreview !== null; const [previewWidth, setPreviewWidth] = useState(0); const agents = props.agents; @@ -99,6 +100,7 @@ export function FloatingWorkingControl(props: { 32 - (action !== null ? actionWidth : 0) - (hasQueue ? queueWidth : 0) - + (hasConnectionChip ? connectionWidth : 0) - (hasAgents ? agentsWidth : 0) - (hasPreview ? previewWidth : 0), ); @@ -119,7 +121,6 @@ export function FloatingWorkingControl(props: { opacity: separationProgress.value, })); - // Animate an in-flow sizer for native glass layout. Measure labels at fixed width to avoid repeated animation retargeting. const capsuleWidth = useSharedValue(null); const measuredWidthRef = useRef(null); const handleLabelLayout = (event: LayoutChangeEvent) => { @@ -132,7 +133,7 @@ export function FloatingWorkingControl(props: { capsuleWidth.set(first ? width : withTiming(width, CONTROL_TIMING)); }; const hasStatus = props.status !== null; - const hasCapsule = hasStatus || hasQueue || hasAgents || hasPreview; + const hasCapsule = hasStatus || hasConnectionChip || hasQueue || hasAgents || hasPreview; useEffect(() => { if (!hasStatus) { measuredWidthRef.current = null; @@ -156,11 +157,14 @@ export function FloatingWorkingControl(props: { hasQueue || (props.devicePreview !== null && props.browserPreview !== null); const statusInteractive = - props.status?.kind === "connection" || - props.status?.kind === "child-input" || - props.status?.kind === "background"; + props.status?.kind === "child-input" || props.status?.kind === "background"; const capsuleInteractive = - statusInteractive || action !== null || hasQueue || hasAgents || hasPreview; + statusInteractive || + hasConnectionChip || + action !== null || + hasQueue || + hasAgents || + hasPreview; const statusContent = props.status !== null ? ( + {props.connectionChip !== null ? ( + setConnectionWidth(settleWidth(event.nativeEvent.layout.width))} + className="h-11 shrink flex-row items-center px-3 active:opacity-70" + style={{ maxWidth: labelWidth + connectionWidth }} + > + + + {props.connectionChip.label} + + + ) : null} {statusContent} {action !== null ? ( ); } - if (props.status.kind === "connection") { - return ( - - {props.status.tone === "reconnecting" ? ( - - ) : ( - - )} - - {props.status.label} - - - ); - } if (props.status.kind === "lifecycle") { return ( void; } +export interface ConnectionChipStatus { + readonly label: string; + readonly onPress: () => void; +} + export type FloatingWorkingStatus = | { readonly kind: "working"; readonly startedAt: string } | { readonly kind: "syncing"; readonly label: string } @@ -32,53 +37,33 @@ export type FloatingWorkingStatus = readonly action: FloatingStatusAction; } | { readonly kind: "goal"; readonly label: string; readonly accessibilityLabel: string } - | { readonly kind: "preparing"; readonly label: string } - | { - readonly kind: "connection"; - readonly tone: "reconnecting" | "unavailable"; - readonly label: string; - readonly onPress: () => void; - }; + | { readonly kind: "preparing"; readonly label: string }; -export function connectionFloatingStatus(input: { - readonly connectionError: string | null; +export function connectionChipStatus(input: { readonly connectionState: EnvironmentConnectionPhase; - readonly environmentLabel: string | null; + readonly connectionError?: string | null; readonly onReconnect: () => void; -}): FloatingWorkingStatus | null { - const environmentLabel = input.environmentLabel ?? "Environment"; - const unavailable = (label: string): FloatingWorkingStatus => ({ - kind: "connection", - tone: "unavailable", - label, - onPress: input.onReconnect, - }); - - switch (input.connectionState) { - case "connecting": - case "reconnecting": - return { - kind: "connection", - tone: "reconnecting", - label: - input.connectionError === null - ? `Reconnecting to ${environmentLabel}...` - : `Failed to connect. Retrying ${environmentLabel}...`, - onPress: input.onReconnect, - }; - case "offline": - return unavailable("You are offline"); - case "unsupported": - return unavailable("Client not supported"); - case "error": - return unavailable( - input.connectionError - ? `Failed to connect to ${environmentLabel}: ${input.connectionError}` - : `Failed to connect to ${environmentLabel}`, - ); - case "available": - return unavailable(`${environmentLabel} is not connected`); - case "connected": - return null; + readonly queuedCount: number; +}): ConnectionChipStatus | null { + if (input.connectionState === "connected") { + return input.queuedCount > 0 + ? { label: `${input.queuedCount} queued`, onPress: input.onReconnect } + : null; } + const label = + input.connectionState === "offline" + ? "Offline" + : input.connectionState === "connecting" || input.connectionState === "reconnecting" + ? "Reconnecting" + : input.connectionState === "error" + ? input.connectionError + ? `Connection failed: ${input.connectionError}` + : "Connection failed" + : input.connectionState === "unsupported" + ? "Client not supported" + : "Not connected"; + return { + label: input.queuedCount > 0 ? `${label} · ${input.queuedCount} queued` : label, + onPress: input.onReconnect, + }; } diff --git a/apps/mobile/src/features/threads/threadContentPresentation.test.ts b/apps/mobile/src/features/threads/threadContentPresentation.test.ts index 3990799d85925..c7e3773fc4459 100644 --- a/apps/mobile/src/features/threads/threadContentPresentation.test.ts +++ b/apps/mobile/src/features/threads/threadContentPresentation.test.ts @@ -24,7 +24,7 @@ describe("thread content presentation", () => { }), ).toEqual({ kind: "unavailable", - title: "Messages not cached", + title: "Not downloaded yet", detail: "Reconnect this environment to load the conversation.", }); }); diff --git a/apps/mobile/src/features/threads/threadContentPresentation.ts b/apps/mobile/src/features/threads/threadContentPresentation.ts index a180ed7594db0..32d12ac314565 100644 --- a/apps/mobile/src/features/threads/threadContentPresentation.ts +++ b/apps/mobile/src/features/threads/threadContentPresentation.ts @@ -41,7 +41,10 @@ export function projectThreadContentPresentation(input: { } return { kind: "unavailable", - title: "Messages not cached", - detail: "Reconnect this environment to load the conversation.", + title: "Not downloaded yet", + detail: + input.connectionState === "error" || input.connectionState === "unsupported" + ? "Reconnect this environment to load the conversation." + : "It will load when you’re back online.", }; } diff --git a/apps/mobile/src/features/threads/use-thread-connection-chip.ts b/apps/mobile/src/features/threads/use-thread-connection-chip.ts new file mode 100644 index 0000000000000..5eb484e098790 --- /dev/null +++ b/apps/mobile/src/features/threads/use-thread-connection-chip.ts @@ -0,0 +1,34 @@ +import { useFocusEffect } from "@react-navigation/native"; +import type { EnvironmentConnectionPhase } from "@t3tools/client-runtime/connection"; +import { useCallback } from "react"; + +import { useDelayedStatus } from "../../lib/useDelayedStatus"; +import { setOpenThreadCacheKey } from "../../state/thread-cache-protection"; +import { connectionChipStatus } from "./floating-working-status"; + +export function useThreadConnectionChip(input: { + readonly environmentId: string; + readonly threadId: string; + readonly connectionState: EnvironmentConnectionPhase; + readonly connectionError: string | null; + readonly onReconnect: () => void; + readonly queuedCount: number; +}) { + const key = `${input.environmentId}:${input.threadId}`; + useFocusEffect( + useCallback(() => { + setOpenThreadCacheKey(key); + return () => setOpenThreadCacheKey(null); + }, [key]), + ); + const status = connectionChipStatus(input); + const label = useDelayedStatus(key, status?.label ?? null, { + showDelayMs: + input.connectionState === "offline" || + input.connectionState === "connecting" || + input.connectionState === "reconnecting" + ? 4_000 + : 0, + }); + return label === null ? null : { label, onPress: input.onReconnect }; +} diff --git a/apps/mobile/src/lib/useDelayedStatus.ts b/apps/mobile/src/lib/useDelayedStatus.ts index 2bf3aaa5538b8..b27b0e4cfd8c7 100644 --- a/apps/mobile/src/lib/useDelayedStatus.ts +++ b/apps/mobile/src/lib/useDelayedStatus.ts @@ -7,12 +7,16 @@ import { useState } from "react"; import { useDeadlinePassed } from "./clock"; -export function useDelayedStatus(key: string, value: A | null): A | null { +export function useDelayedStatus( + key: string, + value: A | null, + timing?: { readonly showDelayMs?: number; readonly minVisibleMs?: number }, +): A | null { const [state, setState] = useState(() => - stepDelayedStatus(null, { key, value, deadlinePassed: false }, Date.now), + stepDelayedStatus(null, { key, value, deadlinePassed: false, timing }, Date.now), ); const deadlinePassed = useDeadlinePassed(delayedStatusDeadline(state)); - const next = stepDelayedStatus(state, { key, value, deadlinePassed }, Date.now); + const next = stepDelayedStatus(state, { key, value, deadlinePassed, timing }, Date.now); if (next !== state) setState(next); return shownDelayedStatus(next, key); } diff --git a/apps/mobile/src/persistence/mobile-database.ts b/apps/mobile/src/persistence/mobile-database.ts index e6549c2229629..c784ddc4a1438 100644 --- a/apps/mobile/src/persistence/mobile-database.ts +++ b/apps/mobile/src/persistence/mobile-database.ts @@ -3,11 +3,20 @@ import * as Context from "effect/Context"; 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 Schema from "effect/Schema"; +import * as Semaphore from "effect/Semaphore"; import type { SQLiteDatabase } from "expo-sqlite"; +import { + enforceThreadCacheBudget, + THREAD_CACHE_MAX_BYTES, + type ThreadCacheBudgetInput, + type ThreadCacheIndexRow, +} from "./thread-cache-budget"; + const DATABASE_NAME = "t3code-client.db"; -const DATABASE_SCHEMA_VERSION = 1; +const DATABASE_SCHEMA_VERSION = 2; const LEGACY_CACHE_DIRECTORIES = [ "connection-shell-snapshots", "shell-snapshots", @@ -52,6 +61,8 @@ const MobileDatabaseOperation = Schema.Literals([ "migrate", "load-cache", "list-cache", + "list-thread-cache-index", + "enforce-thread-cache-budget", "save-cache", "remove-cache", "clear-cache-kind", @@ -166,18 +177,30 @@ async function migrateLegacyFileCaches(database: SQLiteDatabase): Promise { + await transaction.runAsync( + `INSERT INTO client_cache + (environment_id, kind, cache_key, schema_version, payload, updated_at) + VALUES (?, ?, ?, ?, ?, ?) + ON CONFLICT (environment_id, kind, cache_key) DO NOTHING`, + record.environmentId, + record.kind, + record.cacheKey, + record.schemaVersion, + record.payload, + Date.now(), + ); + await transaction.runAsync( + `INSERT OR IGNORE INTO client_cache_meta + (environment_id, kind, cache_key, byte_size, accessed_at, source_updated_at) + VALUES (?, ?, ?, ?, ?, NULL)`, + record.environmentId, + record.kind, + record.cacheKey, + new TextEncoder().encode(record.payload).length, + Date.now(), + ); + }); } directory.delete(); } catch (cause) { @@ -209,7 +232,18 @@ export class MobileDatabase extends Context.Service< cacheKey: string, schemaVersion: number, payload: string, + sourceUpdatedAt?: number | null, ) => Effect.Effect; + readonly listThreadCacheIndex: ( + environmentId: EnvironmentId, + ) => Effect.Effect, MobileDatabaseError>; + readonly enforceThreadCacheBudget: ( + input: ThreadCacheBudgetInput, + ) => Effect.Effect<{ evicted: number; bytes: number }, MobileDatabaseError>; + readonly setThreadCacheProtection: (input: { + readonly loaded: boolean; + readonly protectedThreadKeys: ReadonlySet; + }) => Effect.Effect; readonly removeCache: ( environmentId: EnvironmentId, kind: ClientCacheKind, @@ -253,9 +287,6 @@ const makeAvailable = Effect.gen(function* () { yield* Effect.tryPromise({ try: async () => { await database.execAsync("PRAGMA journal_mode = WAL; PRAGMA foreign_keys = ON;"); - const schema = await database.getFirstAsync<{ readonly user_version: number }>( - "PRAGMA user_version", - ); await database.withExclusiveTransactionAsync(async (transaction) => { await transaction.execAsync(` CREATE TABLE IF NOT EXISTS client_cache ( @@ -271,6 +302,20 @@ const makeAvailable = Effect.gen(function* () { CREATE INDEX IF NOT EXISTS client_cache_environment_updated ON client_cache (environment_id, updated_at DESC); + CREATE TABLE IF NOT EXISTS client_cache_meta ( + environment_id TEXT NOT NULL, + kind TEXT NOT NULL, + cache_key TEXT NOT NULL, + byte_size INTEGER NOT NULL, + accessed_at INTEGER NOT NULL, + source_updated_at INTEGER, + PRIMARY KEY (environment_id, kind, cache_key), + FOREIGN KEY (environment_id, kind, cache_key) + REFERENCES client_cache(environment_id, kind, cache_key) ON DELETE CASCADE + ) WITHOUT ROWID; + CREATE INDEX IF NOT EXISTS client_cache_meta_thread_accessed + ON client_cache_meta (accessed_at, environment_id, cache_key) WHERE kind = 'thread'; + CREATE TABLE IF NOT EXISTS client_preferences ( singleton INTEGER PRIMARY KEY NOT NULL CHECK (singleton = 1), payload TEXT NOT NULL, @@ -278,7 +323,16 @@ const makeAvailable = Effect.gen(function* () { ); `); }); + const schema = await database.getFirstAsync<{ readonly user_version: number }>( + "PRAGMA user_version", + ); if ((schema?.user_version ?? 0) < DATABASE_SCHEMA_VERSION) { + await database.runAsync( + `INSERT OR IGNORE INTO client_cache_meta + (environment_id, kind, cache_key, byte_size, accessed_at, source_updated_at) + SELECT environment_id, kind, cache_key, LENGTH(CAST(payload AS BLOB)), updated_at, NULL + FROM client_cache`, + ); const migrated = await migrateLegacyFileCaches(database); if (migrated) { await database.execAsync(`PRAGMA user_version = ${DATABASE_SCHEMA_VERSION};`); @@ -288,23 +342,69 @@ const makeAvailable = Effect.gen(function* () { catch: databaseError("migrate"), }); + // Exclusive transactions get their own connection with no busy handler; overlapping writes fail as locked. + const writeLock = yield* Semaphore.make(1); + const budgetRequests = yield* Queue.sliding(1); + let budgetOutboxLoaded = false; + const budgetProtectedThreadKeys = new Set(); + yield* Effect.gen(function* () { + while (true) { + yield* Queue.take(budgetRequests); + yield* Effect.sleep("30 seconds"); + while ((yield* Queue.poll(budgetRequests))._tag === "Some") {} + yield* Effect.tryPromise({ + try: () => + budgetOutboxLoaded + ? enforceThreadCacheBudget(database, { + maxBytes: THREAD_CACHE_MAX_BYTES, + protectedThreadKeys: budgetProtectedThreadKeys, + }) + : Promise.resolve({ evicted: 0, bytes: 0 }), + catch: databaseError("enforce-thread-cache-budget"), + }).pipe( + writeLock.withPermits(1), + Effect.catch((error) => Effect.logWarning(error)), + ); + } + }).pipe(Effect.forkScoped); + return MobileDatabase.of({ - loadCache: Effect.fn("MobileDatabase.loadCache")((environmentId, kind, cacheKey) => - Effect.tryPromise({ + loadCache: Effect.fn("MobileDatabase.loadCache")(function* (environmentId, kind, cacheKey) { + const now = Date.now(); + const row = yield* Effect.tryPromise({ try: () => - database.getFirstAsync<{ readonly payload: string }>( - `SELECT payload + database.getFirstAsync<{ + readonly payload: string; + readonly accessedAt: number; + }>( + `SELECT client_cache.payload, client_cache_meta.accessed_at AS accessedAt FROM client_cache - WHERE environment_id = ? AND kind = ? AND cache_key = ? - AND (kind != 'pull-request' OR updated_at >= ?)`, + JOIN client_cache_meta USING (environment_id, kind, cache_key) + WHERE client_cache.environment_id = ? AND client_cache.kind = ? AND client_cache.cache_key = ? + AND (client_cache.kind != 'pull-request' OR client_cache.updated_at >= ?)`, environmentId, kind, cacheKey, - Date.now() - 30 * 24 * 60 * 60_000, + now - 30 * 24 * 60 * 60_000, ), catch: databaseError("load-cache"), - }).pipe(Effect.map((row) => Option.fromNullishOr(row?.payload))), - ), + }); + if (kind === "thread" && row && row.accessedAt <= now - 60_000) { + yield* Effect.tryPromise({ + try: () => + database.runAsync( + `UPDATE client_cache_meta SET accessed_at = ? + WHERE environment_id = ? AND kind = 'thread' AND cache_key = ? AND accessed_at <= ?`, + now, + environmentId, + cacheKey, + now - 60_000, + ), + catch: databaseError("load-cache"), + }).pipe(writeLock.withPermits(1)); + } + return Option.fromNullishOr(row?.payload); + }), listCache: Effect.fn("MobileDatabase.listCache")((kind) => Effect.tryPromise({ try: () => @@ -315,11 +415,37 @@ const makeAvailable = Effect.gen(function* () { catch: databaseError("list-cache"), }).pipe(Effect.map((rows) => rows.map((row) => row.payload))), ), + listThreadCacheIndex: Effect.fn("MobileDatabase.listThreadCacheIndex")((environmentId) => + Effect.tryPromise({ + try: () => + database.getAllAsync( + `SELECT cache_key AS threadId, source_updated_at AS sourceUpdatedAt, + byte_size AS bytes, accessed_at AS accessedAt + FROM client_cache_meta WHERE environment_id = ? AND kind = 'thread'`, + environmentId, + ), + catch: databaseError("list-thread-cache-index"), + }), + ), + enforceThreadCacheBudget: Effect.fn("MobileDatabase.enforceThreadCacheBudget")((input) => + Effect.tryPromise({ + try: () => enforceThreadCacheBudget(database, input), + catch: databaseError("enforce-thread-cache-budget"), + }).pipe(writeLock.withPermits(1)), + ), + setThreadCacheProtection: Effect.fn("MobileDatabase.setThreadCacheProtection")((input) => + Effect.sync(() => { + budgetOutboxLoaded = input.loaded; + budgetProtectedThreadKeys.clear(); + for (const key of input.protectedThreadKeys) budgetProtectedThreadKeys.add(key); + }).pipe( + Effect.andThen(() => (input.loaded ? Queue.offer(budgetRequests, undefined) : Effect.void)), + ), + ), saveCache: Effect.fn("MobileDatabase.saveCache")( - (environmentId, kind, cacheKey, schemaVersion, payload) => + (environmentId, kind, cacheKey, schemaVersion, payload, sourceUpdatedAt) => Effect.tryPromise({ try: async () => { - // Bound offline PR caches to 32 MiB and 256 responses per environment; reject huge diffs that would evict everything. if ( kind === "pull-request" && new TextEncoder().encode(payload).length > 4 * 1024 * 1024 @@ -341,10 +467,23 @@ const makeAvailable = Effect.gen(function* () { payload, Date.now(), ); + await transaction.runAsync( + `INSERT INTO client_cache_meta + (environment_id, kind, cache_key, byte_size, accessed_at, source_updated_at) + VALUES (?, ?, ?, ?, ?, ?) + ON CONFLICT (environment_id, kind, cache_key) DO UPDATE SET + byte_size = excluded.byte_size, + accessed_at = excluded.accessed_at, + source_updated_at = excluded.source_updated_at`, + environmentId, + kind, + cacheKey, + new TextEncoder().encode(payload).length, + Date.now(), + kind === "thread" ? (sourceUpdatedAt ?? null) : null, + ); if (kind === "pull-request") { - await transaction.runAsync( - `DELETE FROM client_cache WHERE environment_id = ? AND kind = 'pull-request' - AND cache_key IN ( + const pruned = `cache_key IN ( SELECT cache_key FROM ( SELECT cache_key, updated_at, ROW_NUMBER() OVER (ORDER BY updated_at DESC, cache_key) AS position, @@ -353,17 +492,29 @@ const makeAvailable = Effect.gen(function* () { ) AS total_bytes FROM client_cache WHERE environment_id = ? AND kind = 'pull-request' ) WHERE position > 256 OR total_bytes > 33554432 OR updated_at < ? - )`, - environmentId, - environmentId, - Date.now() - 30 * 24 * 60 * 60_000, - ); + )`; + const cutoff = Date.now() - 30 * 24 * 60 * 60_000; + // The transaction connection does not enforce foreign keys, so the cascade never runs. + for (const table of ["client_cache_meta", "client_cache"]) { + await transaction.runAsync( + `DELETE FROM ${table} WHERE environment_id = ? AND kind = 'pull-request' AND ${pruned}`, + environmentId, + environmentId, + cutoff, + ); + } } }; - await save(database); + await database.withExclusiveTransactionAsync(save); }, catch: databaseError("save-cache"), - }).pipe(Effect.asVoid), + }).pipe( + writeLock.withPermits(1), + Effect.andThen(() => + kind === "thread" ? Queue.offer(budgetRequests, undefined) : Effect.void, + ), + Effect.asVoid, + ), ), removeCache: Effect.fn("MobileDatabase.removeCache")((environmentId, kind, cacheKey) => Effect.tryPromise({ @@ -376,7 +527,7 @@ const makeAvailable = Effect.gen(function* () { cacheKey, ), catch: databaseError("remove-cache"), - }).pipe(Effect.asVoid), + }).pipe(writeLock.withPermits(1), Effect.asVoid), ), clearCacheKind: Effect.fn("MobileDatabase.clearCacheKind")((environmentId, kind) => Effect.tryPromise({ @@ -387,19 +538,19 @@ const makeAvailable = Effect.gen(function* () { kind, ), catch: databaseError("clear-cache-kind"), - }).pipe(Effect.asVoid), + }).pipe(writeLock.withPermits(1), Effect.asVoid), ), clearEnvironmentCache: Effect.fn("MobileDatabase.clearEnvironmentCache")((environmentId) => Effect.tryPromise({ try: () => database.runAsync("DELETE FROM client_cache WHERE environment_id = ?", environmentId), catch: databaseError("clear-environment-cache"), - }).pipe(Effect.asVoid), + }).pipe(writeLock.withPermits(1), Effect.asVoid), ), clearAllCaches: Effect.tryPromise({ try: () => database.runAsync("DELETE FROM client_cache"), catch: databaseError("clear-all-caches"), - }).pipe(Effect.asVoid), + }).pipe(writeLock.withPermits(1), Effect.asVoid), inspectCaches: Effect.tryPromise({ try: () => database.getAllAsync(` @@ -447,7 +598,7 @@ const makeAvailable = Effect.gen(function* () { updatedAt, ), catch: databaseError("save-preferences"), - }).pipe(Effect.asVoid), + }).pipe(writeLock.withPermits(1), Effect.asVoid), ), }); }); @@ -457,6 +608,9 @@ function makeUnavailable(error: MobileDatabaseError): MobileDatabase["Service"] return MobileDatabase.of({ loadCache: () => fail, listCache: () => fail, + listThreadCacheIndex: () => fail, + enforceThreadCacheBudget: () => fail, + setThreadCacheProtection: () => fail, saveCache: () => fail, removeCache: () => fail, clearCacheKind: () => fail, diff --git a/apps/mobile/src/persistence/thread-cache-budget.ts b/apps/mobile/src/persistence/thread-cache-budget.ts new file mode 100644 index 0000000000000..193414e86803a --- /dev/null +++ b/apps/mobile/src/persistence/thread-cache-budget.ts @@ -0,0 +1,57 @@ +import type { SQLiteDatabase } from "expo-sqlite"; + +export const THREAD_CACHE_MAX_BYTES = 150 * 1024 * 1024; + +export interface ThreadCacheIndexRow { + readonly threadId: string; + readonly sourceUpdatedAt: number | null; + readonly bytes: number; + readonly accessedAt: number; +} + +export interface ThreadCacheBudgetInput { + readonly maxBytes: number; + readonly protectedThreadKeys: ReadonlySet; +} + +export async function enforceThreadCacheBudget( + database: SQLiteDatabase, + input: ThreadCacheBudgetInput, +): Promise<{ evicted: number; bytes: number }> { + if (!Number.isFinite(input.maxBytes) || input.maxBytes < 0) { + throw new RangeError("Thread cache budget must be a nonnegative finite number."); + } + let evicted = 0; + let bytes = 0; + await database.withExclusiveTransactionAsync(async (transaction) => { + const total = await transaction.getFirstAsync<{ readonly bytes: number }>( + "SELECT COALESCE(SUM(byte_size), 0) AS bytes FROM client_cache_meta WHERE kind = 'thread'", + ); + bytes = total?.bytes ?? 0; + if (bytes <= input.maxBytes) return; + const rows = await transaction.getAllAsync<{ + readonly environmentId: string; + readonly threadId: string; + readonly bytes: number; + }>(`SELECT environment_id AS environmentId, cache_key AS threadId, byte_size AS bytes + FROM client_cache_meta WHERE kind = 'thread' + ORDER BY accessed_at, environment_id, cache_key`); + for (const row of rows) { + if (bytes <= input.maxBytes) break; + if (input.protectedThreadKeys.has(`${row.environmentId}:${row.threadId}`)) continue; + const metadata = await transaction.runAsync( + "DELETE FROM client_cache_meta WHERE environment_id = ? AND kind = 'thread' AND cache_key = ?", + row.environmentId, + row.threadId, + ); + const result = await transaction.runAsync( + "DELETE FROM client_cache WHERE environment_id = ? AND kind = 'thread' AND cache_key = ?", + row.environmentId, + row.threadId, + ); + evicted += result.changes; + bytes -= row.bytes * metadata.changes; + } + }); + return { evicted, bytes }; +} diff --git a/apps/mobile/src/state/persist-composer-message.ts b/apps/mobile/src/state/persist-composer-message.ts new file mode 100644 index 0000000000000..aa3e3883c98f8 --- /dev/null +++ b/apps/mobile/src/state/persist-composer-message.ts @@ -0,0 +1,38 @@ +import { enqueueThreadOutboxMessage } from "./thread-outbox"; +import type { QueuedThreadMessage } from "./thread-outbox-model"; +import { setThreadComposerError } from "./thread-composer-error"; +import { + clearComposerDraftContent, + getComposerDraftSnapshot, + scheduleUnusedComposerAttachmentCleanup, + type ComposerDraft, +} from "./use-composer-drafts"; + +export async function persistComposerMessage( + threadKey: string, + draft: ComposerDraft, + message: QueuedThreadMessage, +) { + try { + await enqueueThreadOutboxMessage(message); + const current = getComposerDraftSnapshot(threadKey); + if ( + current.text === draft.text && + JSON.stringify(current.context) === JSON.stringify(draft.context) && + current.attachments.length === draft.attachments.length && + current.attachments.every( + (attachment, index) => attachment.id === draft.attachments[index]?.id, + ) + ) { + clearComposerDraftContent(threadKey, { deferAttachmentCleanup: true }); + } + } catch (error) { + setThreadComposerError( + threadKey, + error instanceof Error ? error.message : "Failed to save the queued message.", + ); + return null; + } + scheduleUnusedComposerAttachmentCleanup(draft.attachments); + return message.messageId; +} diff --git a/apps/mobile/src/state/thread-cache-protection.ts b/apps/mobile/src/state/thread-cache-protection.ts new file mode 100644 index 0000000000000..30ed24d14be04 --- /dev/null +++ b/apps/mobile/src/state/thread-cache-protection.ts @@ -0,0 +1,49 @@ +import * as Effect from "effect/Effect"; + +import { runtime } from "../lib/runtime"; +import { MobileDatabase } from "../persistence/mobile-database"; +import { appAtomRegistry } from "./atom-registry"; +import { threadOutboxManager } from "./thread-outbox"; + +let openThreadCacheKey: string | null = null; +let refreshProtection: (() => void) | undefined; + +export function getOpenThreadCacheKey(): string | null { + return openThreadCacheKey; +} + +export function setOpenThreadCacheKey(key: string | null): void { + openThreadCacheKey = key; + refreshProtection?.(); +} + +export function installThreadCacheProtection(): () => void { + const refresh = () => { + void runtime.runPromise( + Effect.gen(function* () { + const database = yield* MobileDatabase; + const loaded = appAtomRegistry.get(threadOutboxManager.loadedAtom); + const protectedThreadKeys = new Set( + Object.entries(appAtomRegistry.get(threadOutboxManager.queuedMessagesByThreadKeyAtom)) + .filter(([, messages]) => messages.length > 0) + .map(([key]) => key), + ); + if (openThreadCacheKey !== null) protectedThreadKeys.add(openThreadCacheKey); + yield* database.setThreadCacheProtection({ loaded, protectedThreadKeys }); + }).pipe(Effect.catch(() => Effect.void)), + ); + }; + refreshProtection = refresh; + const stopLoaded = appAtomRegistry.subscribe(threadOutboxManager.loadedAtom, refresh, { + immediate: true, + }); + const stopQueued = appAtomRegistry.subscribe( + threadOutboxManager.queuedMessagesByThreadKeyAtom, + refresh, + ); + return () => { + stopLoaded(); + stopQueued(); + refreshProtection = undefined; + }; +} diff --git a/apps/mobile/src/state/thread-outbox-manager.ts b/apps/mobile/src/state/thread-outbox-manager.ts index 732db0159b7f7..44338ebdb4b47 100644 --- a/apps/mobile/src/state/thread-outbox-manager.ts +++ b/apps/mobile/src/state/thread-outbox-manager.ts @@ -41,6 +41,10 @@ export function createThreadOutboxManager(options: ThreadOutboxManagerOptions) { const queuedMessagesByThreadKeyAtom = Atom.make< Record> >({}).pipe(Atom.keepAlive, Atom.withLabel("mobile:thread-outbox:queued-messages")); + const loadedAtom = Atom.make(false).pipe( + Atom.keepAlive, + Atom.withLabel("mobile:thread-outbox:loaded"), + ); const warn = options.warn ?? ((message: string, error: unknown) => { @@ -48,6 +52,8 @@ export function createThreadOutboxManager(options: ThreadOutboxManagerOptions) { }); let loadPromise: Promise | null = null; let mutationQueue: Promise = Promise.resolve(); + let nextEnqueueOrdinal = 0; + let ordinalReady = false; const revisions = new Map(); const bumpRevision = (messageId: MessageId): void => { revisions.set(messageId, (revisions.get(messageId) ?? 0) + 1); @@ -76,12 +82,30 @@ export function createThreadOutboxManager(options: ThreadOutboxManagerOptions) { } loadPromise = serialize(async () => { const result = await options.storage.load(); - const current = currentMessages(); - const currentIds = new Set(current.map((message) => message.messageId)); + const currentIds = new Set(currentMessages().map((message) => message.messageId)); const recovered = result.messages.filter( (message) => !currentIds.has(message.messageId) && !revisions.has(message.messageId), ); - if (recovered.length > 0) setMessages([...recovered, ...current]); + nextEnqueueOrdinal = Math.max( + nextEnqueueOrdinal, + ...result.messages.map( + (message) => message.enqueueOrdinal ?? Date.parse(message.createdAt), + ), + ); + if (recovered.length > 0) { + const latest = currentMessages(); + const latestIds = new Set(latest.map((message) => message.messageId)); + setMessages([ + ...recovered.filter( + (message) => !latestIds.has(message.messageId) && !revisions.has(message.messageId), + ), + ...latest, + ]); + } + if (result.errors.length === 0) { + options.registry.set(loadedAtom, true); + ordinalReady = result.errors.length === 0 && result.messages.length > 0; + } if (result.errors.length > 0) { throw new AggregateError(result.errors, "Some queued messages could not be read."); } @@ -103,29 +127,40 @@ export function createThreadOutboxManager(options: ThreadOutboxManagerOptions) { return loadPromise; }; - const enqueue = (message: QueuedThreadMessage): Promise => { - bumpRevision(message.messageId); + const enqueue = async (message: QueuedThreadMessage): Promise => { + if (message.enqueueOrdinal === undefined && !ordinalReady) { + await load(); + } + const queuedMessage = + message.enqueueOrdinal !== undefined || !ordinalReady + ? message + : { + ...message, + enqueueOrdinal: Math.max(Date.now(), (nextEnqueueOrdinal ?? 0) + 1), + }; + nextEnqueueOrdinal = Math.max(nextEnqueueOrdinal ?? 0, queuedMessage.enqueueOrdinal ?? 0); + bumpRevision(queuedMessage.messageId); setMessages([ - ...currentMessages().filter((candidate) => candidate.messageId !== message.messageId), - message, + ...currentMessages().filter((candidate) => candidate.messageId !== queuedMessage.messageId), + queuedMessage, ]); return serialize(async () => { try { - await options.storage.write(message); + await options.storage.write(queuedMessage); } catch (cause) { // Rollback by reference: a retry with the same id may already own a newer optimistic entry. - setMessages(currentMessages().filter((candidate) => candidate !== message)); + setMessages(currentMessages().filter((candidate) => candidate !== queuedMessage)); // Remove compensating persisted copies after rollback unless a same-id entry survives, or restart resurrects rejected messages. if (!currentMessages().some((candidate) => candidate.messageId === message.messageId)) { try { - await options.storage.remove(message); + await options.storage.remove(queuedMessage); } catch {} } throw new ThreadOutboxManagerError({ operation: "enqueue", - environmentId: message.environmentId, - threadId: message.threadId, - messageId: message.messageId, + environmentId: queuedMessage.environmentId, + threadId: queuedMessage.threadId, + messageId: queuedMessage.messageId, cause, }); } @@ -341,6 +376,7 @@ export function createThreadOutboxManager(options: ThreadOutboxManagerOptions) { return { queuedMessagesByThreadKeyAtom, + loadedAtom, serialize, load, enqueue, diff --git a/apps/mobile/src/state/thread-outbox-model.ts b/apps/mobile/src/state/thread-outbox-model.ts index 9b765fd4a40b4..f097106909227 100644 --- a/apps/mobile/src/state/thread-outbox-model.ts +++ b/apps/mobile/src/state/thread-outbox-model.ts @@ -30,8 +30,7 @@ import type { DraftComposerAttachment } from "../lib/composerImages"; import { scopedThreadKey } from "../lib/scopedEntities"; import { resolveProviderInteractionMode } from "./legacy-plan-mode"; -// Keep current writes until a compatible native baseline includes the v4 reader. -const THREAD_OUTBOX_SCHEMA_VERSION = 3; +const THREAD_OUTBOX_SCHEMA_VERSION = 5; const THREAD_OUTBOX_MAX_RETRY_DELAY_MS = 16_000; const QueuedThreadCreationSchema = Schema.Struct({ @@ -46,7 +45,7 @@ const QueuedThreadCreationSchema = Schema.Struct({ }); export const QueuedThreadMessageSchema = Schema.Struct({ - schemaVersion: Schema.Literals([1, 2, THREAD_OUTBOX_SCHEMA_VERSION, 4]), + schemaVersion: Schema.Literals([1, 2, 3, 4, THREAD_OUTBOX_SCHEMA_VERSION]), environmentId: EnvironmentId, threadId: ThreadId, messageId: MessageId, @@ -60,6 +59,7 @@ export const QueuedThreadMessageSchema = Schema.Struct({ interactionMode: Schema.optional(ProviderInteractionMode), creation: Schema.optional(QueuedThreadCreationSchema), createdAt: IsoDateTime, + enqueueOrdinal: Schema.optional(Schema.Number), }); const decodeStoredQueuedThreadMessage = Schema.decodeUnknownSync(QueuedThreadMessageSchema); @@ -91,6 +91,7 @@ export interface QueuedThreadMessage { readonly dispatchMode?: ComposerDispatchMode; readonly creation?: QueuedThreadCreation; readonly createdAt: string; + readonly enqueueOrdinal?: number; } export interface ThreadSettingsSnapshot { @@ -138,6 +139,11 @@ export function decodeQueuedThreadMessage(value: unknown): QueuedThreadMessage { return message; } +// Ordinals are enqueue timestamps, so messages saved before ordinals existed sort by creation time. +function queuedMessageOrder(message: QueuedThreadMessage): number { + return message.enqueueOrdinal ?? Date.parse(message.createdAt); +} + export function groupQueuedThreadMessages( messages: ReadonlyArray, ): Record> { @@ -152,9 +158,19 @@ export function groupQueuedThreadMessages( (grouped[threadKey] ??= []).push(message); } for (const queue of Object.values(grouped)) { - queue.sort((left, right) => left.createdAt.localeCompare(right.createdAt)); + queue.sort( + (left, right) => + queuedMessageOrder(left) - queuedMessageOrder(right) || + left.createdAt.localeCompare(right.createdAt), + ); } - return grouped; + return Object.fromEntries( + Object.entries(grouped).sort( + ([, left], [, right]) => + (left[0] === undefined ? Number.NaN : queuedMessageOrder(left[0])) - + (right[0] === undefined ? Number.NaN : queuedMessageOrder(right[0])), + ), + ); } export function flattenQueuedThreadMessages( diff --git a/apps/mobile/src/state/thread-outbox.test.ts b/apps/mobile/src/state/thread-outbox.test.ts index 66bd21b2d9bf5..a1f801ab682c1 100644 --- a/apps/mobile/src/state/thread-outbox.test.ts +++ b/apps/mobile/src/state/thread-outbox.test.ts @@ -108,6 +108,7 @@ function queuedMessage(input: { text: input.messageId, attachments: [], createdAt: input.createdAt, + enqueueOrdinal: 0, }; } diff --git a/apps/mobile/src/state/thread-prefetch.ts b/apps/mobile/src/state/thread-prefetch.ts new file mode 100644 index 0000000000000..dfb72d93b39da --- /dev/null +++ b/apps/mobile/src/state/thread-prefetch.ts @@ -0,0 +1,249 @@ +import { persistEnvironmentThreadSnapshots } from "@t3tools/client-runtime/state/threads"; +import { threadRuntimeIsActive } from "@t3tools/client-runtime/state/models"; +import type { EnvironmentId, ThreadId } from "@t3tools/contracts"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as Option from "effect/Option"; +import { AsyncResult, Atom } from "effect/reactivity"; +import { AppState } from "react-native"; + +import { runtime } from "../lib/runtime"; +import { MobileDatabase } from "../persistence/mobile-database"; +import { appAtomRegistry } from "./atom-registry"; +import { environmentShell } from "./shell"; +import { getOpenThreadCacheKey } from "./thread-cache-protection"; +import { threadOutboxManager } from "./thread-outbox"; +import { environmentThreads, environmentThreadShells } from "./threads"; +import { workspaceConnections } from "./workspace"; + +const PREFETCH_CONCURRENCY = 2; +const PREFETCH_WAIT_MS = 30_000; +const PREFETCH_DEBOUNCE_MS = 250; + +const prefetchInputsAtom = Atom.make((get) => { + const environments = get(workspaceConnections.environmentsAtom); + return { + environments, + threads: get(environmentThreadShells.quietThreadShellsAtom), + loaded: get(threadOutboxManager.loadedAtom), + queued: get(threadOutboxManager.queuedMessagesByThreadKeyAtom), + live: new Set( + environments + .filter( + ({ environmentId, connectionState }) => + connectionState === "connected" && + get(environmentShell.stateValueAtom(environmentId)).status === "live", + ) + .map(({ environmentId }) => environmentId), + ), + }; +}).pipe(Atom.withLabel("mobile:thread-prefetch:inputs")); + +type PrefetchTarget = { + readonly environmentId: EnvironmentId; + readonly threadId: ThreadId; + readonly updatedAt: string; + readonly running: boolean; +}; + +function waitForThread(target: PrefetchTarget, signal: AbortSignal): Promise { + if (signal.aborted) return Promise.resolve(false); + const atom = environmentThreads.stateAtom(target.environmentId, target.threadId); + const release = appAtomRegistry.mount(atom); + return new Promise((resolve) => { + let settled = false; + let unsubscribe: (() => void) | undefined; + const timeout = setTimeout(finish, PREFETCH_WAIT_MS); + const aborted = () => finish(false); + function finish(synchronized = false) { + if (settled) return; + settled = true; + clearTimeout(timeout); + unsubscribe?.(); + release(); + signal.removeEventListener("abort", aborted); + resolve(synchronized); + } + const check = () => { + const result = appAtomRegistry.get(atom); + const value = AsyncResult.value(result); + if (AsyncResult.isFailure(result)) { + finish(); + } else if ( + value._tag === "Some" && + (value.value.status === "deleted" || + (value.value.status === "live" && + Option.isSome(value.value.data) && + DateTime.toEpochMillis(value.value.data.value.thread.updatedAt) >= + Date.parse(target.updatedAt))) + ) { + finish(true); + } + }; + unsubscribe = appAtomRegistry.subscribe(atom, check); + signal.addEventListener("abort", aborted, { once: true }); + if (signal.aborted) finish(); + else check(); + }); +} + +export function installThreadOfflineSync(): () => void { + let active = true; + let foreground = AppState.currentState === "active"; + let running = false; + let requested = false; + let timer: ReturnType | undefined; + let abort: AbortController | undefined; + const refreshed = new Map(); + + const eligible = (target: PrefetchTarget): boolean => { + const current = appAtomRegistry.get(prefetchInputsAtom); + return ( + active && + foreground && + current.loaded && + current.live.has(target.environmentId) && + !Object.values(current.queued) + .flat() + .some((message) => message.environmentId === target.environmentId) && + getOpenThreadCacheKey() !== `${target.environmentId}:${target.threadId}` + ); + }; + + const run = async () => { + timer = undefined; + if (!active || !foreground || running) return; + requested = false; + const current = appAtomRegistry.get(prefetchInputsAtom); + if (!current.loaded) return; + const candidates = current.threads + .filter( + (thread) => + current.live.has(thread.environmentId) && + thread.archivedAt === null && + thread.deletedAt === null && + thread.lineage.parentThreadId === null, + ) + .sort((left, right) => Date.parse(right.updatedAt) - Date.parse(left.updatedAt)); + const eligibleCandidates = candidates.filter( + (thread) => + !Object.values(current.queued) + .flat() + .some((message) => message.environmentId === thread.environmentId) && + getOpenThreadCacheKey() !== `${thread.environmentId}:${thread.id}`, + ); + const recent = eligibleCandidates + .filter((thread) => !threadRuntimeIsActive(thread.runtime)) + .slice(0, 50); + const runningThreads = eligibleCandidates + .filter((thread) => threadRuntimeIsActive(thread.runtime)) + .slice(0, 10); + const targets = [ + ...new Map( + [...runningThreads, ...recent].map((thread) => [ + `${thread.environmentId}:${thread.id}`, + { + environmentId: thread.environmentId, + threadId: thread.id, + updatedAt: thread.updatedAt, + running: threadRuntimeIsActive(thread.runtime), + }, + ]), + ).values(), + ].filter(eligible); + running = true; + const controller = new AbortController(); + abort = controller; + try { + const indexes = await runtime.runPromise( + MobileDatabase.pipe( + Effect.flatMap((database) => + Effect.forEach( + [...new Set(targets.map((target) => target.environmentId))], + (environmentId) => database.listThreadCacheIndex(environmentId), + { concurrency: "unbounded" }, + ), + ), + ), + ); + const environmentIds = [...new Set(targets.map((target) => target.environmentId))]; + const cached = new Map( + indexes.flatMap((rows, index) => + rows.map((row) => [`${environmentIds[index]}:${row.threadId}`, row.sourceUpdatedAt]), + ), + ); + const pending = targets.filter((target) => { + const key = `${target.environmentId}:${target.threadId}`; + if (refreshed.get(key) === target.updatedAt) return false; + const cachedAt = cached.get(key); + return target.running || cachedAt == null || cachedAt < Date.parse(target.updatedAt); + }); + let next = 0; + const worker = async () => { + while (!controller.signal.aborted) { + const target = pending[next++]; + if (!target) return; + if (!eligible(target)) continue; + const synchronized = await waitForThread(target, controller.signal); + if (synchronized && !controller.signal.aborted) + refreshed.set(`${target.environmentId}:${target.threadId}`, target.updatedAt); + } + }; + await Promise.all(Array.from({ length: PREFETCH_CONCURRENCY }, worker)); + } catch (error) { + console.warn("[thread-prefetch] could not refresh cached threads", error); + } finally { + running = false; + abort = undefined; + if (requested) schedule(); + } + }; + + function schedule() { + if (!active || !foreground) return; + requested = true; + if (running) return; + clearTimeout(timer); + timer = setTimeout(() => void run(), PREFETCH_DEBOUNCE_MS); + } + + let liveEnvironments = new Set(); + const stop = appAtomRegistry.subscribe( + prefetchInputsAtom, + (inputs) => { + const reconnected = [...inputs.live].some((id) => !liveEnvironments.has(id)); + if (reconnected) refreshed.clear(); + liveEnvironments = inputs.live; + if ( + abort && + (!inputs.loaded || + Object.values(inputs.queued).some((queue) => queue.length > 0) || + inputs.live.size === 0) + ) + abort.abort(); + schedule(); + }, + { immediate: true }, + ); + const subscription = AppState.addEventListener("change", (state) => { + const wasForeground = foreground; + foreground = state === "active"; + if (!foreground) { + abort?.abort(); + clearTimeout(timer); + void runtime.runPromise( + persistEnvironmentThreadSnapshots.pipe(Effect.catch(() => Effect.void)), + ); + } else if (!wasForeground) { + refreshed.clear(); + schedule(); + } + }); + return () => { + active = false; + stop(); + subscription.remove(); + abort?.abort(); + clearTimeout(timer); + }; +} diff --git a/apps/mobile/src/state/use-thread-composer-state.ts b/apps/mobile/src/state/use-thread-composer-state.ts index 74206557b0ad2..fd04bc434db7a 100644 --- a/apps/mobile/src/state/use-thread-composer-state.ts +++ b/apps/mobile/src/state/use-thread-composer-state.ts @@ -75,7 +75,6 @@ import { clearComposerDraftContent, composerContextImportsAtom, getComposerDraftSnapshot, - mergeComposerDraftContent, removeComposerDraftAttachment, scheduleUnusedComposerAttachmentCleanup, setComposerDraftText, @@ -104,13 +103,14 @@ import { useQueuedRunEdit, } from "./queued-run-edit"; import { setPendingConnectionError } from "../state/use-remote-environment-registry"; -import { clearThreadComposerError, setThreadComposerError } from "./thread-composer-error"; +import { clearThreadComposerError } from "./thread-composer-error"; import { useSelectedThreadProjection, useSelectedThreadVisibleTurnItems, } from "../state/use-thread-detail"; import { useThreadSelection } from "../state/use-thread-selection"; -import { enqueueThreadOutboxMessage } from "./thread-outbox"; +import { persistComposerMessage } from "./persist-composer-message"; +import type { QueuedThreadMessage } from "./thread-outbox-model"; import { dispatchingQueuedMessageIdAtom, useThreadOutboxMessages } from "./use-thread-outbox"; import { threadEnvironment } from "./threads"; import { useAtomCommand } from "./use-atom-command"; @@ -214,6 +214,7 @@ export function useThreadComposerState() { }); const [isSavingQueuedEdit, setIsSavingQueuedEdit] = useState(false); const savingQueuedEditRef = useRef(false); + const sendingMessageRef = useRef(false); const pastedTextFileNamesRef = useRef<{ threadKey: string | null; names: Set }>({ threadKey: null, names: new Set(), @@ -507,6 +508,20 @@ export function useThreadComposerState() { text: string, hasAttachmentsOrContext: boolean, ) => Promise, + ) => { + if (sendingMessageRef.current) return null; + sendingMessageRef.current = true; + return sendMessageImpl(followUpOverride, beforeDispatch).finally(() => { + sendingMessageRef.current = false; + }); + }; + + const sendMessageImpl = async ( + followUpOverride?: ActiveTurnComposerAction, + beforeDispatch?: ( + text: string, + hasAttachmentsOrContext: boolean, + ) => Promise, ) => { if ( selectedThreadShell && @@ -668,7 +683,7 @@ export function useThreadComposerState() { const metadata = makeQueuedMessageMetadata(); const messageId = MessageId.make(metadata.messageId); clearThreadComposerError(threadKey); - const enqueuePromise = enqueueThreadOutboxMessage({ + const queuedMessage: QueuedThreadMessage = { environmentId: selectedThreadShell.environmentId, threadId: selectedThreadShell.id, messageId, @@ -684,25 +699,8 @@ export function useThreadComposerState() { ), ...(followUpDispatchMode === null ? {} : { dispatchMode: followUpDispatchMode }), createdAt: metadata.createdAt, - }); - clearComposerDraftContent(threadKey, { deferAttachmentCleanup: true }); - enqueuePromise.then( - () => scheduleUnusedComposerAttachmentCleanup(attachments), - (error: unknown) => { - // Restore with uncapped append; capped merge could drop attachments added during the failed write. - void mergeComposerDraftContent(threadKey, { - text, - context: draft.context, - attachments: [], - }); - appendComposerDraftAttachments(threadKey, attachments, { allowOverflow: true }); - setThreadComposerError( - threadKey, - error instanceof Error ? error.message : "Failed to save the queued message.", - ); - }, - ); - return messageId; + }; + return persistComposerMessage(threadKey, draft, queuedMessage); }; const onChangeDraftMessage = (value: string) => { diff --git a/packages/client-runtime/src/delayedStatus.test.ts b/packages/client-runtime/src/delayedStatus.test.ts index f49e1ad78a810..e9bf4d6ea1ff6 100644 --- a/packages/client-runtime/src/delayedStatus.test.ts +++ b/packages/client-runtime/src/delayedStatus.test.ts @@ -39,7 +39,7 @@ describe("stepDelayedStatus", () => { it("shows the latest value when the show delay passes", () => { const pending = step(null, "connecting", 0); const changed = step(pending, "syncing", 200); - expect(changed).toBe(pending); + expect(changed).toMatchObject({ phase: "pending", value: "syncing", showAt: 400 }); expect(shownDelayedStatus(step(changed, "syncing", 400), "thread-1")).toBe("syncing"); }); diff --git a/packages/client-runtime/src/delayedStatus.ts b/packages/client-runtime/src/delayedStatus.ts index 7de7962ef641d..972103ea91d6b 100644 --- a/packages/client-runtime/src/delayedStatus.ts +++ b/packages/client-runtime/src/delayedStatus.ts @@ -1,6 +1,11 @@ const STATUS_SHOW_DELAY_MS = 400; const STATUS_MIN_VISIBLE_MS = 400; +export interface DelayedStatusTiming { + readonly showDelayMs?: number; + readonly minVisibleMs?: number; +} + /** * A status shows only after it has stayed set for a moment, then stays shown * for a minimum time so a quick flicker never flashes on screen. @@ -8,7 +13,13 @@ const STATUS_MIN_VISIBLE_MS = 400; */ export type DelayedStatusState = | { readonly phase: "idle"; readonly key: string } - | { readonly phase: "pending"; readonly key: string; readonly showAt: number } + | { + readonly phase: "pending"; + readonly key: string; + readonly value: A; + readonly showDelayMs: number; + readonly showAt: number; + } | { readonly phase: "shown"; readonly key: string; @@ -21,6 +32,7 @@ export interface DelayedStatusInput { /** Compared by identity, so use strings or other stable values. */ readonly value: A | null; readonly deadlinePassed: boolean; + readonly timing?: DelayedStatusTiming; } /** @@ -30,29 +42,41 @@ export interface DelayedStatusInput { */ export function stepDelayedStatus( state: DelayedStatusState | null, - { key, value, deadlinePassed }: DelayedStatusInput, + { key, value, deadlinePassed, timing: inputTiming }: DelayedStatusInput, now: () => number, ): DelayedStatusState { const idle = (): DelayedStatusState => ({ phase: "idle", key }); - const pending = (): DelayedStatusState => ({ - phase: "pending", - key, - showAt: now() + STATUS_SHOW_DELAY_MS, - }); + const showDelayMs = inputTiming?.showDelayMs ?? STATUS_SHOW_DELAY_MS; + const pending = (pendingValue: A): DelayedStatusState => + showDelayMs <= 0 + ? shown(pendingValue) + : { + phase: "pending", + key, + value: pendingValue, + showDelayMs, + showAt: now() + showDelayMs, + }; const shown = (shownValue: A): DelayedStatusState => ({ phase: "shown", key, value: shownValue, - holdUntil: now() + STATUS_MIN_VISIBLE_MS, + holdUntil: now() + (inputTiming?.minVisibleMs ?? STATUS_MIN_VISIBLE_MS), }); - if (state === null || state.key !== key) return value === null ? idle() : pending(); + if (state === null || state.key !== key) return value === null ? idle() : pending(value); switch (state.phase) { case "idle": - return value === null ? state : pending(); + return value === null ? state : pending(value); case "pending": if (value === null) return idle(); - return deadlinePassed ? shown(value) : state; + if (showDelayMs <= 0) return shown(value); + if (deadlinePassed) return shown(value); + if (value !== state.value || showDelayMs !== state.showDelayMs) { + const showAt = Math.min(state.showAt, now() + showDelayMs); + return { phase: "pending", key, value, showDelayMs, showAt }; + } + return state; case "shown": if (value === null) return deadlinePassed ? idle() : state; return value === state.value ? state : shown(value); diff --git a/packages/client-runtime/src/state/threadCachePolicy.ts b/packages/client-runtime/src/state/threadCachePolicy.ts new file mode 100644 index 0000000000000..4a098089e1705 --- /dev/null +++ b/packages/client-runtime/src/state/threadCachePolicy.ts @@ -0,0 +1,34 @@ +import type { + OrchestrationV2ThreadDetailSnapshot, + OrchestrationV2ThreadProjection, +} from "@t3tools/contracts"; + +import type { ThreadHistoryMeta } from "./threadHistoryMerge.ts"; + +export function isActiveThread(thread: OrchestrationV2ThreadProjection): boolean { + return thread.runs.some( + (run) => run.status === "preparing" || run.status === "starting" || run.status === "running", + ); +} + +export function snapshotToPersist( + snapshotSequence: number, + projection: OrchestrationV2ThreadProjection, + history: ThreadHistoryMeta, + acceptsBoundedSnapshots: boolean, +): OrchestrationV2ThreadDetailSnapshot { + if (acceptsBoundedSnapshots || history.hasMoreHistory || history.historyCursor !== null) { + return { + snapshotSequence, + projection, + historyCursor: history.historyCursor, + hasMoreHistory: history.hasMoreHistory, + latestLocalTurnOrdinal: history.latestLocalTurnOrdinal, + }; + } + return { snapshotSequence, projection }; +} + +export function shouldPersistThread(history: ThreadHistoryMeta): boolean { + return !history.expanded; +} diff --git a/packages/client-runtime/src/state/threads.ts b/packages/client-runtime/src/state/threads.ts index 693af5329fe1b..63caad4a7b5ca 100644 --- a/packages/client-runtime/src/state/threads.ts +++ b/packages/client-runtime/src/state/threads.ts @@ -36,6 +36,7 @@ import { request, subscribeDynamic } from "../rpc/client.ts"; import { parseThreadKey, threadKey } from "./entities.ts"; import { applyOrchestrationV2ProjectionEvent } from "./orchestrationV2Projection.ts"; import { THREAD_SNAPSHOT_IDLE_TTL_MS } from "./threadRetention.ts"; +import { isActiveThread, snapshotToPersist, shouldPersistThread } from "./threadCachePolicy.ts"; import { followStreamInEnvironment } from "./runtime.ts"; import * as ThreadHistoryController from "./threadHistoryController.ts"; import { fetchEnvironmentThreadHistoryPage } from "./threadHistoryHttp.ts"; @@ -55,6 +56,14 @@ import { type EnvironmentThreadStatus, } from "./threadState.ts"; +const environmentThreadBackgroundPersistors = new Set>(); + +export const persistEnvironmentThreadSnapshots: Effect.Effect = Effect.suspend(() => + Effect.forEach(Array.from(environmentThreadBackgroundPersistors), (persist) => persist, { + concurrency: "unbounded", + }).pipe(Effect.asVoid), +); + function statusWithoutLiveData( data: Option.Option, ): EnvironmentThreadStatus { @@ -90,40 +99,6 @@ function historyMetaFromCachedSnapshot( }; } -function snapshotToPersist( - snapshotSequence: number, - projection: OrchestrationV2ThreadProjection, - history: ThreadHistoryMeta, - acceptsBoundedSnapshots: boolean, -): OrchestrationV2ThreadDetailSnapshot { - // A complete bounded snapshot still proves paging support. Retain that evidence - // so a thread that grows while closed can resume with a bounded fallback. - if (acceptsBoundedSnapshots || history.hasMoreHistory || history.historyCursor !== null) { - return { - snapshotSequence, - projection, - historyCursor: history.historyCursor, - hasMoreHistory: history.hasMoreHistory, - latestLocalTurnOrdinal: history.latestLocalTurnOrdinal, - }; - } - return { snapshotSequence, projection }; -} - -function shouldPersistThread( - thread: OrchestrationV2ThreadProjection, - history: ThreadHistoryMeta, -): boolean { - // After the user loads older pages the in-memory timeline can grow large. - // Keep those expanded projections out of the monolithic cache. - if (history.expanded) { - return false; - } - return !thread.runs.some( - (run) => run.status === "preparing" || run.status === "starting" || run.status === "running", - ); -} - // Only still-active local copies take the server's item: events applied meanwhile win, // and inherited rows belong to another thread. function settleTurnItems( @@ -329,28 +304,30 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make ); }); + const persistCurrent = Effect.suspend(() => { + const { state: current, sequence: snapshotSequence } = committed; + return Option.match(current.data, { + onNone: () => Effect.void, + onSome: (projection) => + shouldPersistThread(current.history) + ? persist( + snapshotToPersist( + snapshotSequence, + projection, + current.history, + committed.acceptsBoundedSnapshots === true, + ), + ) + : queuedSnapshot !== undefined && queuedSnapshot !== savedSnapshot + ? persist(queuedSnapshot) + : Effect.void, + }); + }); + environmentThreadBackgroundPersistors.add(persistCurrent); yield* Effect.addFinalizer(() => - Effect.suspend(() => { - const { state: current, sequence: snapshotSequence } = committed; - return Option.match(current.data, { - onNone: () => Effect.void, - onSome: (projection) => - shouldPersistThread(projection, current.history) - ? persist( - snapshotToPersist( - snapshotSequence, - projection, - current.history, - committed.acceptsBoundedSnapshots === true, - ), - ) - : // The consumer may still be waiting out its delay; an expanded thread queues - // only cache repairs, which must survive leaving the thread. - queuedSnapshot !== undefined && queuedSnapshot !== savedSnapshot - ? persist(queuedSnapshot) - : Effect.void, - }); - }), + Effect.sync(() => environmentThreadBackgroundPersistors.delete(persistCurrent)).pipe( + Effect.andThen(persistCurrent), + ), ); yield* runCachePersistence(persistence, persist).pipe(Effect.forkScoped); @@ -379,6 +356,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make ...current, status: current.status === "deleted" ? current.status : statusWithoutLiveData(current.data), })); + yield* persistCurrent; }); const setStreamError = (message: string) => Ref.set(awaitingCompletion, false).pipe( @@ -425,7 +403,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make // Active projections can update many times per second and retain large tool // payloads. Persist once the run settles so cache encoding stays off the // streaming path. Progressive meta rides along when the window is incomplete. - if (shouldPersistThread(thread, next.history)) { + if (shouldPersistThread(next.history) && !isActiveThread(thread)) { const snapshotSequence = yield* SubscriptionRef.get(lastSequence); yield* queuePersist( snapshotToPersist( @@ -579,7 +557,11 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make yield* setDeleted(); return; } - if (result._tag === "applied" && shouldPersistThread(result.projection, result.history)) { + if ( + result._tag === "applied" && + shouldPersistThread(result.history) && + !isActiveThread(result.projection) + ) { const snapshotSequence = yield* SubscriptionRef.get(lastSequence); yield* queuePersist( snapshotToPersist( @@ -804,7 +786,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make // An expanded timeline is never saved, so reload would bring the stale row back. // Repair the snapshot last handed to the cache instead, keeping its window and cursor. // A resumed state skipped the cache load, so read the saved snapshot then. - if (shouldPersistThread(thread, latest.history)) return; + if (shouldPersistThread(latest.history)) return; // A terminal event that landed during the lookup is newer than any reply, and the // expanded live state never saves it, whatever the lookup returned. const latestItems = new Map(settled); @@ -814,8 +796,14 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make } } if (latestItems.size === 0) return; + const handed = + queuedSnapshot === undefined || + (savedSnapshot !== undefined && + savedSnapshot.snapshotSequence > queuedSnapshot.snapshotSequence) + ? savedSnapshot + : queuedSnapshot; const saved = - queuedSnapshot ?? + handed ?? Option.getOrUndefined( yield* cache .loadThread(environmentId, threadId)