diff --git a/web/oss/src/components/AgentChatSlice/AgentConversation.tsx b/web/oss/src/components/AgentChatSlice/AgentConversation.tsx index 3f8bc4617b..561c0da698 100644 --- a/web/oss/src/components/AgentChatSlice/AgentConversation.tsx +++ b/web/oss/src/components/AgentChatSlice/AgentConversation.tsx @@ -149,6 +149,7 @@ const AgentConversation = ({ markLiveGate, resumeOrphaned, isSeen, + runningElsewhere, } = useAgentChatSession({entityId, sessionId, initialMessages, intent: scrollIntent}) // Turn Inspector: open state, the focused turn, and the assistant → turn-number mapping. @@ -652,6 +653,7 @@ const AgentConversation = ({ entityId={entityId} messages={messages} busy={busy} + runningElsewhere={runningElsewhere} hitlPending={hitlPending} queue={{queued, removeQueued, clearQueue}} modelKey={modelKey} diff --git a/web/oss/src/components/AgentChatSlice/assets/loadSession.ts b/web/oss/src/components/AgentChatSlice/assets/loadSession.ts index e7590fba71..c81a9ac04f 100644 --- a/web/oss/src/components/AgentChatSlice/assets/loadSession.ts +++ b/web/oss/src/components/AgentChatSlice/assets/loadSession.ts @@ -21,10 +21,20 @@ import {transcriptToMessages} from "./transcriptToMessages" * authoritative). Because this return is a one-shot copy, `onRefreshed` re-delivers the * transcript when that revalidation lands — callers apply it behind their own adoption guards. */ +export interface SessionTranscript { + messages: UIMessage[] + /** + * How many durable records this transcript was built from. The log is append-only and ordered, + * so this is an EXACT "has the server moved on?" watermark — unlike a message count, which + * `transcriptToMessages` deliberately holds flat while a turn grows (issue #5530). + */ + recordCount: number +} + export const loadSessionMessages = async ( sessionId: string, - onRefreshed?: (messages: UIMessage[]) => void, -): Promise => { + onRefreshed?: (transcript: SessionTranscript) => void, +): Promise => { // Fetch through the shared records query cache (same key as `sessionRecordsQueryFamily`) so // hydration, revalidation, and the Inspector's atom subscribers share ONE network flight per // stale window instead of each issuing a raw duplicate request. A failure resolves to `null` @@ -36,11 +46,14 @@ export const loadSessionMessages = async ( void refreshed.then((fresh) => { if (!fresh || fresh.length === 0) return const freshMsgs = transcriptToMessages(fresh) - if (freshMsgs && freshMsgs.length > 0) onRefreshed(freshMsgs) + if (freshMsgs && freshMsgs.length > 0) { + onRefreshed({messages: freshMsgs, recordCount: fresh.length}) + } }) } if (!records || records.length === 0) return null - return transcriptToMessages(records) + const messages = transcriptToMessages(records) + return messages ? {messages, recordCount: records.length} : null } catch (err) { console.warn("[loadSessionMessages] hydration fetch failed:", err) return null diff --git a/web/oss/src/components/AgentChatSlice/assets/transcriptToMessages.test.ts b/web/oss/src/components/AgentChatSlice/assets/transcriptToMessages.test.ts index 8b09a0d99a..a07f630839 100644 --- a/web/oss/src/components/AgentChatSlice/assets/transcriptToMessages.test.ts +++ b/web/oss/src/components/AgentChatSlice/assets/transcriptToMessages.test.ts @@ -304,3 +304,50 @@ describe("transcriptToMessages paused end-marker", () => { expect((messages?.[0].metadata as {paused?: boolean} | undefined)?.paused).toBeUndefined() }) }) + +/** + * Regression guard for issue #5530. The adoption guard must not go back to comparing message + * counts: a turn grows IN PLACE here, so the count is identical before and after it completes. + * If a mapper change ever makes these counts differ, revisit `shouldAdoptServerTranscript` — + * do not "simplify" it back to a count comparison on the strength of that change. + */ +describe("transcriptToMessages turn growth is invisible to a message count", () => { + const midTurn = (): SessionRecord[] => [ + ...approvalRecords(), + record("record-done-paused", {type: "done", stopReason: "paused"}), + ] + + const completed = (): SessionRecord[] => [ + ...midTurn(), + record("record-response", { + type: "interaction_response", + id: "approval-1", + kind: "user_approval", + payload: {toolCallId: "tool-1", approved: true}, + }), + record("record-result", { + type: "tool_result", + id: "tool-1", + output: {stdout: "a.txt\nb.txt"}, + }), + record("record-answer", {type: "message", text: "There are two files."}), + record("record-done", {type: "done"}), + ] + + it("renders the same number of messages before and after the turn finishes", () => { + const partial = transcriptToMessages(midTurn()) + const full = transcriptToMessages(completed()) + + expect(partial).toHaveLength(1) + expect(full).toHaveLength(1) + // Same messages, far more content — and far more records behind it. + expect(completed().length).toBeGreaterThan(midTurn().length) + }) + + it("carries strictly more content in the completed turn", () => { + const partial = transcriptToMessages(midTurn()) + const full = transcriptToMessages(completed()) + + expect(full?.[0].parts.length).toBeGreaterThan(partial?.[0].parts.length ?? 0) + }) +}) diff --git a/web/oss/src/components/AgentChatSlice/components/AgentComposerDock.tsx b/web/oss/src/components/AgentChatSlice/components/AgentComposerDock.tsx index 31cea63e54..e36e9335ab 100644 --- a/web/oss/src/components/AgentChatSlice/components/AgentComposerDock.tsx +++ b/web/oss/src/components/AgentChatSlice/components/AgentComposerDock.tsx @@ -32,6 +32,7 @@ import MicPermissionNotice from "./MicPermissionNotice" import QueuedMessages from "./QueuedMessages" import RecordingBar from "./RecordingBar" import RevealCollapse from "./RevealCollapse" +import RunningElsewhereStrip from "./RunningElsewhereStrip" import VoiceInputButton from "./VoiceInputButton" // The composer carries Lexical — the heaviest dependency of this chunk — out of the @@ -51,6 +52,7 @@ const AgentComposerDock = ({ entityId, messages, busy, + runningElsewhere, hitlPending, queue, modelKey, @@ -78,6 +80,8 @@ const AgentComposerDock = ({ entityId: string messages: UIMessage[] busy: boolean + /** The backend reports a live run for this session that this browser is not driving. */ + runningElsewhere: boolean hitlPending: boolean queue: { queued: QueuedMessage[] @@ -201,6 +205,11 @@ const AgentComposerDock = ({
+ {/* Sits with the other docked strips so a session running in another browser reads + as busy instead of frozen (#5530). */} + {runningElsewhere && !chromeHidden ? ( + + ) : null} ( +
+ + + + + + This session is running somewhere else — the transcript updates as the turn progresses. + +
+) + +export default RunningElsewhereStrip diff --git a/web/oss/src/components/AgentChatSlice/hooks/useAgentChatSession.ts b/web/oss/src/components/AgentChatSlice/hooks/useAgentChatSession.ts index 11e3e2667c..027addfa4a 100644 --- a/web/oss/src/components/AgentChatSlice/hooks/useAgentChatSession.ts +++ b/web/oss/src/components/AgentChatSlice/hooks/useAgentChatSession.ts @@ -34,6 +34,7 @@ import {expandedKeysForMessages, pruneExpandedAtom} from "../state/expandState" import { persistSessionMessagesAtom, sessionMessagesAtom, + sessionRecordCountsReadAtom, stampMessagesCreatedAtAtom, } from "../state/sessions" import {captureTurnRequestAtom} from "../state/turnCaptures" @@ -74,6 +75,15 @@ export const useAgentChatSession = ({ // Immutable snapshot of the restored ids (seenIdsRef grows) — the first-seen stamping // effect below skips these so a reload can't masquerade as the turns' send time. const restoredIdsRef = useRef>(new Set(initialMessages.map((m) => m.id))) + // How many durable records the transcript we're RENDERING was built from — the exact test for + // "has the server moved on?" (issue #5530). Message counts can't see a turn growing in place: + // `transcriptToMessages` folds a paused turn into its resume and only closes a message on + // `done`, so a mid-turn snapshot and the finished turn have the SAME count and a count-based + // guard rejects the finished server copy forever. Cleared the moment a live turn starts, since + // we can't know what the server logged for it — the next open then re-syncs from the log. + const recordWatermarkRef = useRef( + store.get(sessionRecordCountsReadAtom)[sessionId], + ) // Whether the LAST assistant turn was user-stopped. You can only cancel the in-flight (last) turn, // so this is a single boolean gated on position at render time — independent of message ids (which // can be missing/duplicated in restore/error paths and would otherwise smear the tag onto every @@ -177,13 +187,15 @@ export const useAgentChatSession = ({ // throttle-revalidate the drives) as the turn streams, not just at onFinish. useFileActivityDetector({sessionId, messages}) - const {isHydrating, hydratedEmpty} = useSessionHydration({ + const {isHydrating, hydratedEmpty, runningElsewhere} = useSessionHydration({ sessionId, initialMessages, messagesRef, busyRef, seenIdsRef, restoredIdsRef, + recordWatermarkRef, + busy, setMessages, persistMessages, intent, @@ -266,10 +278,19 @@ export const useAgentChatSession = ({ }) }, [error, setMessages]) + // A live turn makes the transcript no longer a copy of the server's, and we can't know how many + // records the runner logged for it — so drop the watermark and let the next open re-sync from + // the durable log. MUST stay declared above the persist effect: on the commit where `status` + // flips to "submitted", effects run in declaration order, so clearing here is what stops the + // persist below from filing a locally-extended transcript under a server watermark. + useEffect(() => { + if (status === "submitted" || status === "streaming") recordWatermarkRef.current = undefined + }, [status]) + // Persist the conversation whenever its stream settles (skip mid-stream). useEffect(() => { if (status === "streaming") return - persistMessages({id: sessionId, messages}) + persistMessages({id: sessionId, messages, recordCount: recordWatermarkRef.current}) }, [messages, status, sessionId, persistMessages]) // Bound the in-message expand-state store: on settle, drop entries whose owning message is gone @@ -398,6 +419,7 @@ export const useAgentChatSession = ({ busyRef, isHydrating, hydratedEmpty, + runningElsewhere, stopped, setStopped, handleStop, diff --git a/web/oss/src/components/AgentChatSlice/hooks/useSessionHydration.ts b/web/oss/src/components/AgentChatSlice/hooks/useSessionHydration.ts index 428df0d964..92815dfdf0 100644 --- a/web/oss/src/components/AgentChatSlice/hooks/useSessionHydration.ts +++ b/web/oss/src/components/AgentChatSlice/hooks/useSessionHydration.ts @@ -1,18 +1,32 @@ -import {type MutableRefObject, useEffect, useState} from "react" +import {type MutableRefObject, useCallback, useEffect, useRef, useState} from "react" +import {shouldAdoptServerTranscript} from "@agenta/entities/session" import {type UIMessage} from "ai" +import {useAtomValue} from "jotai" -import {loadSessionMessages} from "../assets/loadSession" -import {getPendingApprovals} from "../components/ApprovalDock" +import {loadSessionMessages, type SessionTranscript} from "../assets/loadSession" +import {sessionLivenessAtomFamily} from "../state/liveness" import {isSessionFresh} from "../state/sessionEphemera" import {type ScrollIntent} from "./useScrollIntent" +/** Catch-up cadence while a session runs elsewhere — matches the records query's own staleTime and + * the liveness poll, so a tick refetches instead of resolving from a still-fresh cache. */ +const REMOTE_RUN_POLL_MS = 15_000 + +/** Ceiling once the log goes quiet. `is_running` is Redis-TTL'd at an hour (the runner clears it at + * turn end; the TTL is only the crash backstop), so a runner that dies mid-turn leaves the flag set + * for a long time — without a backoff that is a full-log refetch every 15s for an hour. Real growth + * resets to the fast cadence, so a long turn that is simply quiet (a slow tool call emits no + * records until it returns) is still followed. */ +const REMOTE_RUN_POLL_MAX_MS = 60_000 + /** * Hybrid history for one session tab. localStorage holds only the session INDEX; the durable * conversation CONTENT lives in the backend record log — so a tab either paints from cache and * revalidates (SWR), or hydrates from the server once (cache miss). Both paths adopt the server - * transcript under the same guards: never mid-stream, and never behind what's on screen. + * transcript through the SAME guard: never mid-stream, and only when the record log has grown past + * what's on screen. */ export const useSessionHydration = ({ sessionId, @@ -21,6 +35,8 @@ export const useSessionHydration = ({ busyRef, seenIdsRef, restoredIdsRef, + recordWatermarkRef, + busy, setMessages, persistMessages, intent, @@ -31,8 +47,12 @@ export const useSessionHydration = ({ busyRef: MutableRefObject seenIdsRef: MutableRefObject> restoredIdsRef: MutableRefObject> + /** Records the rendered transcript was built from; `undefined` once a live turn supersedes it. */ + recordWatermarkRef: MutableRefObject + /** THIS browser is streaming the turn — reactive, so the catch-up poll can start/stop on it. */ + busy: boolean setMessages: (messages: UIMessage[]) => void - persistMessages: (args: {id: string; messages: UIMessage[]}) => void + persistMessages: (args: {id: string; messages: UIMessage[]; recordCount?: number}) => void intent: ScrollIntent }) => { // Cache-first — when this tab opens with no locally-cached messages (a session this browser @@ -48,6 +68,67 @@ export const useSessionHydration = ({ // durable history was pruned by retention or never persisted. Drives the "history unavailable" // notice so a wiped session isn't mistaken for a brand-new chat. const [hydratedEmpty, setHydratedEmpty] = useState(false) + + /** + * The ONE adoption guard, shared by both paths below (they used to carry divergent copies, and + * the hydration one had the same #5530 blind spot). Adopts the durable transcript when the + * record log has grown past what we're rendering. Returns whether it adopted. + */ + const adoptServerTranscript = useCallback( + (transcript: SessionTranscript | null, {armJump = true} = {}): boolean => { + if (!transcript) return false + const {messages: serverMsgs, recordCount} = transcript + const adopt = shouldAdoptServerTranscript({ + serverRecordCount: recordCount, + serverMessageCount: serverMsgs.length, + localMessageCount: messagesRef.current.length, + watermark: recordWatermarkRef.current, + busy: busyRef.current, + }) + if (!adopt) return false + // Restored history renders settled (no live fade-in) and pinned to the bottom. + serverMsgs.forEach((m) => { + seenIdsRef.current.add(m.id) + restoredIdsRef.current.add(m.id) + }) + // Opening a session jumps to the live edge; a background catch-up must NOT — it would + // yank a reader who scrolled up. Following the growth is `stickRef`'s call, the same + // rule the live stream uses. + if (armJump || intent.stickRef.current) intent.armJump() + // Written synchronously, before any React commit. `messagesRef` lags a commit behind, + // so two deliveries landing back-to-back (disk-restored result + background refetch) + // can both see the pre-adoption transcript — it is this watermark, not the on-screen + // length, that keeps the guard order-independent and stops an older snapshot from + // clobbering a newer one. + recordWatermarkRef.current = recordCount + setMessages(serverMsgs) + persistMessages({id: sessionId, messages: serverMsgs, recordCount}) + return true + }, + // `intent`'s MEMBERS, not `intent`: `useScrollIntent` returns a fresh object every render, + // so the object itself would recreate this callback each render and churn everything keyed + // on it. `armJump` (useCallback []) and `stickRef` (useRef) are stable for the life of the + // conversation. + [ + sessionId, + messagesRef, + busyRef, + seenIdsRef, + restoredIdsRef, + recordWatermarkRef, + setMessages, + persistMessages, + intent.armJump, + intent.stickRef, + ], + ) + // The remote-run poll below must NOT re-arm its timer on re-renders: the liveness query alone + // re-renders this hook ~every 15s while a run is live elsewhere, and restarting a fresh 15s + // timer on each of those starves the poll and resets its backoff. The effect reads the CURRENT + // adopter through this ref and keys only on the poll's real inputs. + const adoptServerTranscriptRef = useRef(adoptServerTranscript) + adoptServerTranscriptRef.current = adoptServerTranscript + useEffect(() => { // A session created brand-new in this browser and not yet run has no backend records — // skip the guaranteed-empty query (cleared on first send; after a reload it re-hydrates). @@ -59,40 +140,31 @@ export const useSessionHydration = ({ // StrictMode's mount→unmount→mount cycle re-runs the fetch (the first run is cancelled) // instead of latching a ref that leaves the transcript blank. let cancelled = false - // How long a transcript this effect has already handed to `setMessages`. `messagesRef` only - // catches up on the next React commit, so two adoptions landing before that commit would - // both read the pre-adoption length and let the older snapshot win. The high-water mark - // makes the guard order-independent. - let adoptedLength = initialMessages.length - // ONE adopter for both deliveries: the initial result (disk-restored log, paints instantly) - // and the guaranteed background refetch. Never mid-stream, only when strictly ahead. - const adopt = (msgs: UIMessage[]) => { - if (cancelled || busyRef.current) return - if (msgs.length <= Math.max(adoptedLength, messagesRef.current.length)) return - adoptedLength = msgs.length - // Restored history renders settled (no live fade-in) and pinned to the bottom. - msgs.forEach((m) => { - seenIdsRef.current.add(m.id) - restoredIdsRef.current.add(m.id) - }) - intent.armJump() - // The restore may have said "no records" while the server has some — clear the notice. - setHydratedEmpty(false) - setMessages(msgs) - persistMessages({id: sessionId, messages: msgs}) - } - loadSessionMessages(sessionId, adopt) - .then((msgs) => { + // Post-restore revalidation: the first result may be the disk-restored log (paints + // instantly); adopt the background refetch when it lands. The refetch can land BEFORE the + // promise handler below runs (both are microtasks racing), and `messagesRef` only catches + // up on the next React commit — so record here, not via what's on screen, that real + // history was already adopted. + let adopted = false + loadSessionMessages(sessionId, (fresh) => { + if (cancelled) return + // The restore said "no records" but the server has some — clear the notice. + if (adoptServerTranscript(fresh)) { + adopted = true + setHydratedEmpty(false) + } + }) + .then((transcript) => { if (cancelled) return - if (!msgs || msgs.length === 0) { + if (!transcript || transcript.messages.length === 0) { // Known session, but the server has no records for it → history was pruned or // never persisted. Flag it so the transcript shows the "unavailable" notice. // Only when nothing has been adopted yet — a refetch that already landed is // real history, and this stale first result must not blank it out. - if (adoptedLength === initialMessages.length) setHydratedEmpty(true) + if (!adopted) setHydratedEmpty(true) return } - adopt(msgs) + adoptServerTranscript(transcript) }) .finally(() => { if (!cancelled) setIsHydrating(false) @@ -104,55 +176,21 @@ export const useSessionHydration = ({ }, [sessionId]) // SWR revalidate-on-open: a cached session paints instantly from localStorage; in the background - // we refetch the durable records ONCE (low-priority) and adopt the server transcript ONLY IF it's - // strictly ahead of what we're showing (a turn finished on another device). We never clobber a - // transcript that's live (`busyRef`), or that the server isn't strictly ahead of — so a local - // optimistic/unsent tail is safe. Cache-MISS sessions are hydrated by the effect above; fresh - // never-run sessions have no server records. Reconciliation is by message COUNT, not content: - // detecting a same-length server-side edit/regenerate is deferred, as is focus/interval - // revalidation. FOLLOWUP(sessions,swr): see docs/designs/sessions/frontend-integration.md. + // we refetch the durable records ONCE (low-priority) and adopt the server transcript when the + // record log has grown past ours. Cache-MISS sessions are hydrated by the effect above; fresh + // never-run sessions have no server records. + // + // The old count-based guard also carried a special case for "local tail paused, server tail + // complete" — a resume that finished on another device, invisible to a message count. The + // record watermark sees that growth directly, so the special case is gone rather than extended. useEffect(() => { if (initialMessages.length === 0 || isSessionFresh(sessionId)) return // As with the hydration effect above: no persistent ref, so StrictMode's double-mount // re-runs the revalidation rather than latching it out. let cancelled = false - // `messagesRef` only catches up on the next React commit, so a transcript this effect just - // adopted is invisible to the guard until then. Hold it here too and compare against - // whichever is longer — the checks below read the tail, not just the count, so this keeps - // the whole snapshot rather than a length. - let adopted: UIMessage[] | null = null - const adopt = (serverMsgs: UIMessage[] | null) => { - if (cancelled || !serverMsgs || serverMsgs.length === 0) return - const onScreen = messagesRef.current - const prev = adopted && adopted.length > onScreen.length ? adopted : onScreen - if (busyRef.current) return - // Adopt the server transcript when it is strictly ahead by count, OR when our LOCAL tail - // is stuck paused (mid-approval) while the server has moved past it to a terminal turn — a - // resume that completed on another device. Count alone misses the latter (same bubble - // count) and was silently propped up by the now-removed duplicate user row; the server's - // `paused` flag rides the runner's `done.stopReason` through `transcriptToMessages`. - const serverAheadByCount = serverMsgs.length > prev.length - const localTailPaused = getPendingApprovals(prev).length > 0 - const serverTail = serverMsgs[serverMsgs.length - 1] as - | {role?: string; metadata?: {paused?: boolean}} - | undefined - // The paused-tail exception adopts a resume that completed elsewhere, so the server must - // NOT be behind (>= guards against a lagging snapshot discarding newer local approval - // state) and its tail must be a finished assistant turn — not a shorter, older stream. - const serverTailComplete = - serverMsgs.length >= prev.length && - serverTail?.role === "assistant" && - !serverTail.metadata?.paused && - getPendingApprovals(serverMsgs).length === 0 - if (!serverAheadByCount && !(localTailPaused && serverTailComplete)) return - adopted = serverMsgs - serverMsgs.forEach((m) => { - seenIdsRef.current.add(m.id) - restoredIdsRef.current.add(m.id) - }) - intent.armJump() - setMessages(serverMsgs) - persistMessages({id: sessionId, messages: serverMsgs}) + const adopt = (transcript: SessionTranscript | null) => { + if (cancelled) return + adoptServerTranscript(transcript) } // The first result may itself be the disk-restored records log; the callback re-applies // the same guarded adoption when the guaranteed background revalidation lands. @@ -163,5 +201,47 @@ export const useSessionHydration = ({ // Once per mounted session tab; `sessionId` is stable for this instance. }, [sessionId]) - return {isHydrating, hydratedEmpty} + // ── Follow a run happening somewhere else (#5530) ────────────────────────── + // There is no push channel to browsers: the runner publishes every event to Redis, but the only + // consumer is the ingest worker that writes them to the DB. So a session driven from another tab + // or device is followed by re-reading the durable log on a timer, and the adoption guard above + // decides whether anything actually changed. `isRunning` also covers OUR stream, so exclude the + // case where this browser is the one driving. + const {nest} = useAtomValue(sessionLivenessAtomFamily(sessionId)) + const runningElsewhere = nest.isRunning && !busy + + useEffect(() => { + if (!runningElsewhere) return + let cancelled = false + let timer: ReturnType | undefined + let delay = REMOTE_RUN_POLL_MS + const poll = async () => { + let grew = false + try { + const adopt = adoptServerTranscriptRef.current + const transcript = await loadSessionMessages(sessionId, (fresh) => { + if (!cancelled && adopt(fresh, {armJump: false})) grew = true + }) + if (!cancelled && adopt(transcript, {armJump: false})) grew = true + } catch { + // `loadSessionMessages` already swallows + logs; keep polling regardless. + } finally { + // Chained, not `setInterval`: the log is ~200KB and backend-slow, so a slow fetch + // must never stack requests on top of itself. + if (!cancelled) { + delay = grew ? REMOTE_RUN_POLL_MS : Math.min(delay * 2, REMOTE_RUN_POLL_MAX_MS) + timer = setTimeout(poll, delay) + } + } + } + timer = setTimeout(poll, delay) + return () => { + cancelled = true + if (timer) clearTimeout(timer) + } + // Deliberately NOT keyed on `adoptServerTranscript` — the poll reads it through the ref + // above, so a re-render can't cancel a pending tick or reset the backoff. + }, [runningElsewhere, sessionId]) + + return {isHydrating, hydratedEmpty, runningElsewhere} } diff --git a/web/oss/src/components/AgentChatSlice/state/sessions.ts b/web/oss/src/components/AgentChatSlice/state/sessions.ts index ca270857ae..13adf06d4b 100644 --- a/web/oss/src/components/AgentChatSlice/state/sessions.ts +++ b/web/oss/src/components/AgentChatSlice/state/sessions.ts @@ -7,7 +7,7 @@ import { import {generateId} from "@agenta/shared/utils" import type {UIMessage} from "ai" import {atom, type Getter, type Setter} from "jotai" -import {atomFamily, atomWithStorage, selectAtom} from "jotai/utils" +import {atomFamily, atomWithStorage, createJSONStorage, selectAtom} from "jotai/utils" import {routerAppIdAtom} from "@/oss/state/app/atoms/fetcher" import {projectIdAtom} from "@/oss/state/project" @@ -91,11 +91,25 @@ export const defaultScopeKeyAtom = atom((get) => { // effect sees an empty list in that window and creates a stray session on every reload/HMR. const STORAGE_OPTS = {getOnInit: true} as const +/** + * localStorage WITHOUT jotai's cross-browser-tab sync. The default storage subscribes to the + * `storage` event, so a write in one browser tab replaced these records live in every other one — + * and since the open-tab list drives the antd `Tabs` items, an incoming replacement UNMOUNTED a + * streaming conversation, orphaning its `useChat` stream mid-turn (the in-flight transcript is not + * persisted until the stream settles, so it was lost). Each browser tab now owns its view; storage + * is still shared, so a reload picks up whatever was last written. + */ +const tabLocalStorage = () => { + const storage = createJSONStorage() + delete storage.subscribe + return storage +} + /** Full per-scope session history (open AND closed). */ const sessionsByAppAtom = atomWithStorage>( "agenta:agent-chat:sessions", {}, - undefined, + tabLocalStorage(), STORAGE_OPTS, ) @@ -109,14 +123,14 @@ const sessionsByAppAtom = atomWithStorage>( const openIdsByAppAtom = atomWithStorage>( "agenta:agent-chat:open-sessions", {}, - undefined, + tabLocalStorage(), STORAGE_OPTS, ) const activeByAppAtom = atomWithStorage>( "agenta:agent-chat:active-session", {}, - undefined, + tabLocalStorage(), STORAGE_OPTS, ) @@ -125,10 +139,28 @@ const activeByAppAtom = atomWithStorage>( export const sessionMessagesAtom = atomWithStorage>( "agenta:agent-chat:messages", {}, - undefined, + tabLocalStorage(), + STORAGE_OPTS, +) + +/** + * Per-session adoption watermark: how many durable records the CACHED transcript above was built + * from. Absent means "not server-derived" (a locally-streamed turn, or a pre-#5530 cache) and reads + * as 0, so the next open re-syncs from the server once. + * + * Private on purpose — it must only ever move together with `sessionMessagesAtom`, so every write + * goes through `persistSessionMessagesAtom` and every delete through `dropSessionMessages`. + */ +const sessionRecordCountsAtom = atomWithStorage>( + "agenta:agent-chat:record-counts", + {}, + tabLocalStorage(), STORAGE_OPTS, ) +/** Read-only view of the watermarks; the guards read one non-reactively via `store.get`. */ +export const sessionRecordCountsReadAtom = atom((get) => get(sessionRecordCountsAtom)) + /** Open tab ids for a scope, with the pre-upgrade fallback (everything open). Pure read helper * for the writers below — never mutates. */ const currentOpenIds = (get: Getter, key: string): string[] => { @@ -314,11 +346,7 @@ export const deleteSessionAtomFamily = atomFamily((key: string) => set(activeByAppAtom, {...active, [key]: open.filter((x) => x !== id)[0] ?? ""}) } - const messages = {...get(sessionMessagesAtom)} - if (id in messages) { - delete messages[id] - set(sessionMessagesAtom, messages) - } + dropSessionMessages(get, set, [id]) clearSessionEphemera(id) @@ -470,16 +498,8 @@ export const reconcileServerSessionsAtomFamily = atomFamily((key: string) => if (active[key] && droppedSet.has(active[key])) { set(activeByAppAtom, {...active, [key]: nextOpen[0] ?? ""}) } - const messages = {...get(sessionMessagesAtom)} - let msgsChanged = false - for (const id of dropped) { - if (id in messages) { - delete messages[id] - msgsChanged = true - } - clearSessionEphemera(id) - } - if (msgsChanged) set(sessionMessagesAtom, messages) + dropSessionMessages(get, set, dropped) + for (const id of dropped) clearSessionEphemera(id) } }), ) @@ -553,18 +573,8 @@ export const resetScopeAtomFamily = atomFamily((key: string) => delete next[key] set(activeByAppAtom, next) } - if (ids.length) { - const messages = {...get(sessionMessagesAtom)} - let changed = false - for (const id of ids) { - if (id in messages) { - delete messages[id] - changed = true - } - clearSessionEphemera(id) - } - if (changed) set(sessionMessagesAtom, messages) - } + dropSessionMessages(get, set, ids) + for (const id of ids) clearSessionEphemera(id) }), ) @@ -658,12 +668,13 @@ const writeMessagesWithQuotaGuard = ( set: Setter, next: Record, keepId: string, -): void => { +): {evicted: string[]; persisted: boolean} => { let candidate = next + const evicted: string[] = [] for (;;) { try { set(sessionMessagesAtom, candidate) - return + return {evicted, persisted: true} } catch (e) { if (!isQuotaExceeded(e)) throw e // Object keys keep insertion order, so the first non-active id is the oldest. @@ -671,19 +682,69 @@ const writeMessagesWithQuotaGuard = ( if (oldest === undefined) { // Even the active session alone won't fit — keep it in memory, skip persistence. console.warn("[agent-chat] message store over quota; skipping persistence") - return + return {evicted, persisted: false} } + evicted.push(oldest) candidate = {...candidate} delete candidate[oldest] } } } -/** Write a session's messages to the persisted store (called when its stream settles). */ +/** + * Drop cached transcripts AND their watermarks for `ids` — the two stores must never diverge, or a + * re-adopted session would be judged against a watermark belonging to a transcript that's gone. + * The single deletion path for both; every caller that forgets a session routes through here. + */ +const dropSessionMessages = (get: Getter, set: Setter, ids: string[]): void => { + if (ids.length === 0) return + const messages = {...get(sessionMessagesAtom)} + const counts = {...get(sessionRecordCountsAtom)} + let messagesChanged = false + let countsChanged = false + for (const id of ids) { + if (id in messages) { + delete messages[id] + messagesChanged = true + } + if (id in counts) { + delete counts[id] + countsChanged = true + } + } + if (messagesChanged) set(sessionMessagesAtom, messages) + if (countsChanged) set(sessionRecordCountsAtom, counts) +} + +/** + * Write a session's messages to the persisted store (called when its stream settles), together with + * the record watermark the transcript reflects. Pass `recordCount: undefined` for a locally-streamed + * transcript — we can't know how many records the server logged for it, and clearing the watermark + * makes the next open re-sync from the durable log rather than trust a stale number. + */ export const persistSessionMessagesAtom = atom( null, - (get, set, {id, messages}: {id: string; messages: UIMessage[]}) => { - writeMessagesWithQuotaGuard(set, {...get(sessionMessagesAtom), [id]: messages}, id) + ( + get, + set, + {id, messages, recordCount}: {id: string; messages: UIMessage[]; recordCount?: number}, + ) => { + const {evicted, persisted} = writeMessagesWithQuotaGuard( + set, + {...get(sessionMessagesAtom), [id]: messages}, + id, + ) + const counts = {...get(sessionRecordCountsAtom)} + // If the transcript write itself was skipped (a single session over quota), the persisted + // store still holds the OLD messages — filing the NEW watermark against them would make + // `shouldAdoptServerTranscript` reject the complete server log as "not newer" on every + // future open, freezing the stale cache. The stores must never diverge (see + // `dropSessionMessages`), so drop the watermark and let the next open re-sync. + if (!persisted || recordCount === undefined) delete counts[id] + else counts[id] = recordCount + // A quota eviction dropped those transcripts, so their watermarks go too. + for (const evictedId of evicted) delete counts[evictedId] + set(sessionRecordCountsAtom, counts) }, ) diff --git a/web/packages/agenta-entities/src/session/core/transcriptAdoption.ts b/web/packages/agenta-entities/src/session/core/transcriptAdoption.ts new file mode 100644 index 0000000000..06aec815f5 --- /dev/null +++ b/web/packages/agenta-entities/src/session/core/transcriptAdoption.ts @@ -0,0 +1,47 @@ +/** + * When a client should replace the transcript it is rendering with the server's durable one. + * + * The durable record log is append-only and ordered, so "the server has more RECORDS than my + * transcript was built from" is an exact test for "the server moved on". Message counts are not: + * the agent-chat mapper only closes a message on a `done` record and deliberately folds a paused + * turn into its resume, so a turn that grows in place — tool results landing, an approval + * round-trip completing — keeps the same message count from start to finish. + * + * Issue #5530: a browser that snapshotted a session mid-turn (while another browser drove it) was + * left with a partial transcript whose message count already matched the finished one. The + * count-based guard concluded the server was not ahead and kept the partial copy — on every + * subsequent reload, permanently. + */ +export interface TranscriptAdoptionInput { + /** Durable records the SERVER transcript was built from. */ + serverRecordCount: number + /** Messages in the server transcript. */ + serverMessageCount: number + /** Messages currently rendered locally. */ + localMessageCount: number + /** Records the rendered transcript was built from; `undefined` = not server-derived. */ + watermark: number | undefined + /** This client is streaming the turn, so it — not the log — is the authority. */ + busy: boolean +} + +export const shouldAdoptServerTranscript = ({ + serverRecordCount, + serverMessageCount, + localMessageCount, + watermark, + busy, +}: TranscriptAdoptionInput): boolean => { + // Nothing to adopt. + if (serverMessageCount === 0) return false + // A live local stream outranks the log until it settles. + if (busy) return false + // THE trigger: the log grew past what we render. An absent watermark reads as 0, so a + // locally-streamed or pre-#5530 cache re-syncs from the server once on its next open. + if (serverRecordCount <= (watermark ?? 0)) return false + // Floor, never a trigger: ingest lag can serve a snapshot shorter than what we render, and + // trading down would drop the tail. Equal length is fine — that is the #5530 case, where the + // same messages carry far more content. + if (serverMessageCount < localMessageCount) return false + return true +} diff --git a/web/packages/agenta-entities/src/session/index.ts b/web/packages/agenta-entities/src/session/index.ts index 380f7d0454..f6b91d39b2 100644 --- a/web/packages/agenta-entities/src/session/index.ts +++ b/web/packages/agenta-entities/src/session/index.ts @@ -67,6 +67,7 @@ export { type SessionStreamNest, type SandboxLiveness, } from "./core/liveness" +export {shouldAdoptServerTranscript, type TranscriptAdoptionInput} from "./core/transcriptAdoption" export {deriveMountRows, mountBreadcrumbs, type MountRow} from "./core/mountBrowser" export { sessionRecordsQueryFamily, diff --git a/web/packages/agenta-entities/tests/unit/session-transcript-adoption.test.ts b/web/packages/agenta-entities/tests/unit/session-transcript-adoption.test.ts new file mode 100644 index 0000000000..aadfb0c4f0 --- /dev/null +++ b/web/packages/agenta-entities/tests/unit/session-transcript-adoption.test.ts @@ -0,0 +1,66 @@ +/** + * Pins the transcript-adoption rule (`core/transcriptAdoption.ts`) — the whole of issue #5530, + * where a browser that snapshotted a session mid-turn stayed stuck on the partial copy forever + * because the guard compared MESSAGE counts and a turn grows in place. + * + * The rule has one trigger (the record log grew past our watermark) and two vetoes (a live local + * stream; a server snapshot shorter than what we render). These lock all three. + */ +import {describe, expect, it} from "vitest" + +import { + shouldAdoptServerTranscript, + type TranscriptAdoptionInput, +} from "../../src/session/core/transcriptAdoption" + +/** A finished server turn (40 records) against a snapshot taken mid-turn (12 records). */ +const input = (overrides: Partial = {}): TranscriptAdoptionInput => ({ + serverRecordCount: 40, + serverMessageCount: 2, + localMessageCount: 2, + watermark: 12, + busy: false, + ...overrides, +}) + +describe("shouldAdoptServerTranscript", () => { + it("adopts a turn that grew IN PLACE — the issue #5530 case", () => { + // Identical message counts, far more records. A count-based guard rejects this forever. + expect(shouldAdoptServerTranscript(input())).toBe(true) + }) + + it("rejects when the log has not grown past the rendered transcript", () => { + expect(shouldAdoptServerTranscript(input({serverRecordCount: 12}))).toBe(false) + expect(shouldAdoptServerTranscript(input({serverRecordCount: 11}))).toBe(false) + }) + + it("re-syncs a transcript with no watermark, then stops repeating", () => { + // A locally-streamed turn, or a cache written before the watermark existed. + expect(shouldAdoptServerTranscript(input({watermark: undefined}))).toBe(true) + // Once adopted, the watermark equals the log and the next pass is a no-op. + expect(shouldAdoptServerTranscript(input({watermark: 40, serverRecordCount: 40}))).toBe( + false, + ) + }) + + it("never trades a longer local transcript for a lagging server snapshot", () => { + // Ingest lag: ahead in records, not yet caught up in messages. + expect( + shouldAdoptServerTranscript( + input({serverMessageCount: 2, localMessageCount: 3, watermark: undefined}), + ), + ).toBe(false) + }) + + it("never clobbers a live local stream", () => { + expect(shouldAdoptServerTranscript(input({busy: true}))).toBe(false) + // Even a far-ahead server copy waits for the stream to settle. + expect(shouldAdoptServerTranscript(input({busy: true, serverRecordCount: 9999}))).toBe( + false, + ) + }) + + it("ignores an empty server transcript", () => { + expect(shouldAdoptServerTranscript(input({serverMessageCount: 0}))).toBe(false) + }) +})