diff --git a/web/mobile/src/features/chat/LiveConversation.tsx b/web/mobile/src/features/chat/LiveConversation.tsx index fdc580abc1c..04bc611e666 100644 --- a/web/mobile/src/features/chat/LiveConversation.tsx +++ b/web/mobile/src/features/chat/LiveConversation.tsx @@ -29,7 +29,13 @@ import { type TurnViewModel, } from "@agenta/chat/model" import {getSessionTurnId} from "@agenta/chat/state" -import {cancelSessionExecution} from "@agenta/entities/session" +import { + cancelSessionExecution, + dropUnacceptedLocalSessionAtom, + isSessionFresh, + markLocalSessionAcceptedAtom, + registerLocalSessionAtom, +} from "@agenta/entities/session" import {invalidateAgentCommittedRevisionCache} from "@agenta/entities/workflow" import {AgentIntroCard} from "@agenta/entity-ui/agent" import {SecretRequestDock} from "@agenta/entity-ui/clientTools" @@ -145,6 +151,12 @@ export const LiveConversation = ({ }, [restoreAttachments, setRejections], ) + + // The rail row for a fresh session follows its first send: admitted keeps it until the server + // lists the session, rejected or refused drops it (nothing will ever list that session). + const registerLocalSession = useSetAtom(registerLocalSessionAtom) + const markLocalSessionAccepted = useSetAtom(markLocalSessionAcceptedAtom) + const dropUnacceptedLocalSession = useSetAtom(dropUnacceptedLocalSessionAtom) const conversation = useAgentConversation({ entityId, sessionId, @@ -152,6 +164,8 @@ export const LiveConversation = ({ sharedReaderRunning: running, sharedReaderLivenessUpdatedAt: livenessUpdatedAt, restoreRefusedSend, + onSendAccepted: () => markLocalSessionAccepted(sessionId), + onSendFailed: () => dropUnacceptedLocalSession(sessionId), }) const canEditSecrets = useProjectPermission(projectId, "edit_secret") const pinRevision = useSetAtom(selectedRevisionAtomFamily(sessionId)) @@ -245,11 +259,46 @@ export const LiveConversation = ({ const sendPendingTask = useSetAtom(sendPendingTaskAtom) const failPendingTask = useSetAtom(failPendingTaskAtom) const pendingTaskError = pendingTask?.delivery === "failed" + const { + isHydrating, + revalidate, + send: sendToConversation, + stop, + voidPendingResume, + } = conversation + // A fresh session becomes real on the server only once this first message is admitted, which + // can take seconds on a cold runner. Note it locally first, so the rail lists it now (#6776). + const send = useCallback( + async (input: Parameters[0]) => { + if (isSessionFresh(sessionId)) { + registerLocalSession({ + sessionId, + projectId, + agentId: agentId ?? null, + name: input.text, + }) + } + try { + await sendToConversation(input) + } catch (error) { + // The durable path reports this through `onSendFailed` too; this covers the rest. + dropUnacceptedLocalSession(sessionId) + throw error + } + }, + [ + agentId, + dropUnacceptedLocalSession, + projectId, + registerLocalSession, + sendToConversation, + sessionId, + ], + ) // A template create asks for its accounts here, on arrival, instead of on a create surface of // its own. Holds the first message while it does; declines silently when there is nothing to // ask, which is every other way into this screen. const setup = useSessionSetupStep(sessionId) - const {isHydrating, revalidate, send, stop, voidPendingResume} = conversation useEffect(() => { if (!pendingTask || pendingTask.delivery) return const decision = pendingTaskDecision({ @@ -299,7 +348,11 @@ export const LiveConversation = ({ readerReady: conversation.readerReady, ownedContinuation: conversation.acceptedRunPending, }) - const showingTurnActivity = streamingHere || remoteTurn.showActivity + // `sendInFlight` covers the gap the other two cannot: the message has left the composer, and + // neither `useChat` (the server-owned path never moves its status) nor liveness (a poll away) + // knows yet. Without it the pulse arrived a runner accept plus a poll after the send (#6778). + const showingTurnActivity = + streamingHere || conversation.sendInFlight || remoteTurn.showActivity const streamingHereRef = useRef(streamingHere) streamingHereRef.current = streamingHere const hitlPendingRef = useRef(conversation.hitlPending) @@ -839,7 +892,7 @@ export const LiveConversation = ({ // An open edit rewrites its held message instead of sending. The // input clears on submit, so the displaced draft goes back after. if (!conversation.editingId) { - await conversation.send({text, parts, stagedFiles}) + await send({text, parts, stagedFiles}) return } const draft = await conversation.commitEdit({ diff --git a/web/mobile/src/features/nav/localSessionRefs.ts b/web/mobile/src/features/nav/localSessionRefs.ts new file mode 100644 index 00000000000..29ccfa9fa87 --- /dev/null +++ b/web/mobile/src/features/nav/localSessionRefs.ts @@ -0,0 +1,71 @@ +import {useEffect} from "react" + +import {sessionStatusAtomFamily} from "@agenta/chat/state" +import {forgetLocalSessionsAtom, localSessionsAtom} from "@agenta/entities/session" +import { + localSessionRefsAtom, + sidebarServerSessionIdsAtomFamily, + type SessionSidebarRef, +} from "@agenta/navigation" +import {pinnedSessionIdsAtom} from "@agenta/sessions/state" +import {projectIdAtom} from "@agenta/shared/state" +import {atom, useAtomValue, useSetAtom} from "jotai" + +/** + * Mobile's binding for `@agenta/navigation`'s local-session seam — the desktop feeds it from its + * playground tab cache; mobile has no tab cache, so it feeds it from the sessions this client + * created and sent into (`localSessionsAtom`, written by the chat screen on a fresh session's + * first send). + * + * The server lists a session only once its first turn is admitted, which on a cold runner takes + * seconds. Until then this is the row's only way into the rail (#6776). `withLocalSessions` lets + * the server row win the moment it exists. + * + * Scoped to the active project: the registry is app-wide, and a row sent from project A must not + * surface in project B's rail after a switch, where the rail would link it under B's URL. + */ +export const localMobileSessionRefsAtom = atom((get) => { + const projectId = get(projectIdAtom) + const pinned = get(pinnedSessionIdsAtom) + const own = Object.values(get(localSessionsAtom)).filter( + (session) => session.projectId === projectId, + ) + return own.map((session) => { + const status = get(sessionStatusAtomFamily(session.sessionId)) + return { + id: session.sessionId, + sessionId: session.sessionId, + name: session.name, + // Mobile rows link by session id alone; `appId` is the open target the server resolves. + appId: null, + agentId: session.agentId, + pinned: pinned.includes(session.sessionId), + alive: false, + activityAt: new Date(session.createdAt).toISOString(), + archived: false, + // A chat you typed into is never a trigger run. + isAutomation: false, + running: status === "running", + waiting: status === "awaiting", + } + }) +}) + +/** + * Mirror the derived rows into the package's writable seam, and retire local copies the server + * has caught up with. Mounted with the rail: a rail that is not rendered has nothing to reconcile. + */ +export const useSyncLocalSessionRefs = (scopeId: string) => { + const refs = useAtomValue(localMobileSessionRefsAtom) + const setLocalRefs = useSetAtom(localSessionRefsAtom) + useEffect(() => { + setLocalRefs(refs) + }, [refs, setLocalRefs]) + + const serverIds = useAtomValue(sidebarServerSessionIdsAtomFamily(scopeId)) + const forget = useSetAtom(forgetLocalSessionsAtom) + useEffect(() => { + const known = refs.filter((ref) => serverIds.has(ref.sessionId)).map((ref) => ref.sessionId) + if (known.length) forget(known) + }, [refs, serverIds, forget]) +} diff --git a/web/mobile/src/features/nav/useMobileNavItems.tsx b/web/mobile/src/features/nav/useMobileNavItems.tsx index e1148928d68..d5febff36ab 100644 --- a/web/mobile/src/features/nav/useMobileNavItems.tsx +++ b/web/mobile/src/features/nav/useMobileNavItems.tsx @@ -39,6 +39,8 @@ import { import {atom, useAtomValue, useSetAtom} from "jotai" import {unwrap} from "jotai/utils" +import {useSyncLocalSessionRefs} from "./localSessionRefs" + /** The drawer's scope id — its open-groups persistence bucket. */ export const MOBILE_NAV_SCOPE_ID = "mobile-main" @@ -109,6 +111,8 @@ const mobileSessionsEntity = defineSidebarEntity( * forking a component. */ export const useMobileNavItems = (projectURL: string): SidebarConfig[] => { + // Sessions this client created ride the shared local seam until the server lists them. + useSyncLocalSessionRefs(MOBILE_NAV_SCOPE_ID) const rawSource = useAtomValue(mobileSessionsEntity.activeSourceAtom) const loadMoreSessions = useSetAtom(loadMoreSidebarSessionsAtomFamily(MOBILE_NAV_SCOPE_ID)) const groups = useAtomValue(sidebarSessionGroupsAtomFamily(MOBILE_NAV_SCOPE_ID)) diff --git a/web/mobile/tests/unit/localSessionRefs.test.ts b/web/mobile/tests/unit/localSessionRefs.test.ts new file mode 100644 index 00000000000..a8023172caa --- /dev/null +++ b/web/mobile/tests/unit/localSessionRefs.test.ts @@ -0,0 +1,70 @@ +/** + * Mobile's binding of the rail's local-session seam (#6776): a session this client sent into + * shows as a row — under its agent, spinning while it runs — until the server lists it. Only in + * the project it was sent from. + */ +import {sessionStatusAtomFamily, setSessionStatusAtom} from "@agenta/chat/state" +import {registerLocalSessionAtom} from "@agenta/entities/session" +import {projectIdAtom} from "@agenta/shared/state" +import {createStore} from "jotai" +import {describe, expect, it} from "vitest" + +import {localMobileSessionRefsAtom} from "@/features/nav/localSessionRefs" + +describe("localMobileSessionRefsAtom", () => { + it("lists nothing until a session is registered", () => { + const store = createStore() + store.set(projectIdAtom, "p1") + expect(store.get(localMobileSessionRefsAtom)).toEqual([]) + }) + + it("turns a registered session into a rail row that follows its run status", () => { + const store = createStore() + store.set(projectIdAtom, "p1") + store.set(registerLocalSessionAtom, { + sessionId: "s1", + projectId: "p1", + agentId: "agent-1", + name: "Write a file", + now: Date.UTC(2026, 8, 12, 10, 0, 0), + }) + store.set(setSessionStatusAtom, {id: "s1", status: "running"}) + expect(store.get(sessionStatusAtomFamily("s1"))).toBe("running") + expect(store.get(localMobileSessionRefsAtom)).toEqual([ + { + id: "s1", + sessionId: "s1", + name: "Write a file", + appId: null, + agentId: "agent-1", + pinned: false, + alive: false, + activityAt: "2026-09-12T10:00:00.000Z", + archived: false, + isAutomation: false, + running: true, + waiting: false, + }, + ]) + + store.set(setSessionStatusAtom, {id: "s1", status: "awaiting"}) + const [row] = store.get(localMobileSessionRefsAtom) + expect(row).toMatchObject({running: false, waiting: true}) + }) + + // The registry is app-wide; a pending row must stay in the project it was sent from + // (#6783 review), or the rail would link it under the other project's URL. + it("keeps a pending row out of another project's rail after a switch", () => { + const store = createStore() + store.set(projectIdAtom, "p1") + store.set(registerLocalSessionAtom, {sessionId: "s1", projectId: "p1", name: "in A"}) + store.set(registerLocalSessionAtom, {sessionId: "s2", projectId: "p2", name: "in B"}) + expect(store.get(localMobileSessionRefsAtom).map((row) => row.sessionId)).toEqual(["s1"]) + + store.set(projectIdAtom, "p2") + expect(store.get(localMobileSessionRefsAtom).map((row) => row.sessionId)).toEqual(["s2"]) + + store.set(projectIdAtom, null) + expect(store.get(localMobileSessionRefsAtom)).toEqual([]) + }) +}) diff --git a/web/packages/agenta-chat/src/assets/pendingSendEchoes.ts b/web/packages/agenta-chat/src/assets/pendingSendEchoes.ts index 61a2b241a5e..dc0502d7f06 100644 --- a/web/packages/agenta-chat/src/assets/pendingSendEchoes.ts +++ b/web/packages/agenta-chat/src/assets/pendingSendEchoes.ts @@ -114,6 +114,18 @@ export const retirePendingSendEchoes = ( } /** Disposable user rows; the id prefix keeps rewind from finding an echo in the AI SDK array. */ +/** + * Is a send still on its way to the runner, or streaming as an echo this client owns? + * + * Every echo that is not refused and not parked in the queue is a turn THIS client started and is + * still waiting on: from the moment it leaves the composer until its durable row retires it. That + * is the local "submitted" signal the AI SDK's `status` carries on the direct path, which the + * server-owned send path never sets — without it the working indicator waited for the next + * liveness poll to notice the run (#6778). + */ +export const pendingSendsInFlight = (pending: readonly PendingSendEcho[]): boolean => + pending.some((echo) => !echo.failed && !echo.parkedInputId) + export const pendingSendEchoMessages = (pending: readonly PendingSendEcho[]): UIMessage[] => pending.map( (item) => diff --git a/web/packages/agenta-chat/src/hooks/useAgentChatQueue.ts b/web/packages/agenta-chat/src/hooks/useAgentChatQueue.ts index 86392a6b96e..f6156a94350 100644 --- a/web/packages/agenta-chat/src/hooks/useAgentChatQueue.ts +++ b/web/packages/agenta-chat/src/hooks/useAgentChatQueue.ts @@ -92,6 +92,10 @@ interface UseAgentChatQueueArgs { * host that reads its own composer back cannot always answer in this one. */ restoreRefusedSend?: (message: QueuedMessage) => boolean | Promise + /** A durable send was admitted: the turn it started, or `null` for a parked input. */ + onSendAccepted?: (message: QueuedMessage, executionId: string | null) => void + /** A durable send will never become a turn: rejected before it left, or refused after. */ + onSendFailed?: (message: QueuedMessage) => void /** Send one released message into the conversation (wraps `useChat`'s `sendMessage`). Must be * referentially stable so the release effect doesn't churn on every streamed token. */ sendQueued: (item: QueuedMessage) => void @@ -139,6 +143,8 @@ export const useAgentChatQueue = ({ continuationExecutionId = null, markRunOwned, restoreRefusedSend, + onSendAccepted, + onSendFailed, sendQueued, sessionId, server, @@ -225,6 +231,29 @@ export const useAgentChatQueue = ({ const restoreRefusedSendRef = useRef(restoreRefusedSend) restoreRefusedSendRef.current = restoreRefusedSend + const onSendAcceptedRef = useRef(onSendAccepted) + onSendAcceptedRef.current = onSendAccepted + const onSendFailedRef = useRef(onSendFailed) + onSendFailedRef.current = onSendFailed + + /** + * Hand a message to the transport on the NON-durable path, and report it as admitted. + * + * The durable path has a server admission event to report; this one has none — `sendQueued` + * is synchronous and answers nothing. Staying silent left the send lifecycle half-driven + * here: a fresh session's optimistic row never left `submitting`, so the `accepted` guard + * that protects it could not hold, and a later message's failure deleted a row the server + * was going to list. Reaching the transport IS the admission this path has, and it is the + * same moment `markRunOwned` already claims the run. + */ + const dispatchUnqueued = useCallback( + (message: QueuedMessage) => { + markRunOwned() + sendQueued(message) + onSendAcceptedRef.current?.(message, null) + }, + [markRunOwned, sendQueued], + ) // Echo rows for durable sends, which the AI SDK chat never receives. Owned by its own hook so // this one keeps to admission, queueing, and editing. @@ -340,9 +369,14 @@ export const useAgentChatQueue = ({ echoes.add(message) return server .submit(message, "queue", { - onAccepted: (executionId) => - echoes.markAccepted(message.id, executionId), - onParked: (inputId) => echoes.markParked(message.id, inputId), + onAccepted: (executionId) => { + echoes.markAccepted(message.id, executionId) + onSendAcceptedRef.current?.(message, executionId) + }, + onParked: (inputId) => { + echoes.markParked(message.id, inputId) + onSendAcceptedRef.current?.(message, null) + }, // One event, one recovery. A refusal that arrives after the promise // resolved goes back to the composer exactly like one that rejected // it, so there is a single place the message lives and a single @@ -360,6 +394,7 @@ export const useAgentChatQueue = ({ // Passing the message re-creates the row when the count rule has // already retired it, so a late refusal always has somewhere to be. echoes.markFailed(message.id, message) + onSendFailedRef.current?.(message) const restoring = restoreRefusedSendRef.current?.(message) if (!restoring) return void Promise.resolve(restoring).then((taken) => { @@ -378,6 +413,7 @@ export const useAgentChatQueue = ({ }) .then(undefined, (error: unknown) => { echoes.drop(message.id) + onSendFailedRef.current?.(message) throw error }) } @@ -400,8 +436,7 @@ export const useAgentChatQueue = ({ ) { releasingRef.current = true lastSentRef.current = message - markRunOwned() - sendQueued(message) + dispatchUnqueued(message) } else { setQueued((q) => [...q, message]) } @@ -410,7 +445,7 @@ export const useAgentChatQueue = ({ ? server.resolveCapabilities().then((capabilities) => admit(capabilities.queue)) : admit(server?.capabilities.queue === true) }, - [canReleaseNow, echoes, recoverable, retryContinuation, markRunOwned, sendQueued, server], + [canReleaseNow, dispatchUnqueued, echoes, recoverable, retryContinuation, server], ) const removeQueued = useCallback( @@ -580,14 +615,17 @@ export const useAgentChatQueue = ({ setQueued(rest) // A released head also needs refusal recovery because it has left the queue. lastSentRef.current = head - markRunOwned() - sendQueued(head) - }, [settled, canReleaseNow, queued, markRunOwned, sendQueued]) + dispatchUnqueued(head) + }, [settled, canReleaseNow, queued, dispatchUnqueued]) return { queued: [...(server?.queued ?? []), ...queued], /** Sent-but-not-yet-saved user rows; merge with `mergePendingSendEchoRows`. */ pendingSendRows: echoes.rows, + /** A send of this mount is on its way: admitted, not yet named by the runner or refused. + * The server-owned path never moves `useChat`'s `status` off "ready", so this is the only + * local evidence a turn was just submitted. */ + sendInFlight: echoes.inFlight, submit, steer, removeQueued, diff --git a/web/packages/agenta-chat/src/hooks/useAgentConversation.ts b/web/packages/agenta-chat/src/hooks/useAgentConversation.ts index 1413031107d..490e63ad7b8 100644 --- a/web/packages/agenta-chat/src/hooks/useAgentConversation.ts +++ b/web/packages/agenta-chat/src/hooks/useAgentConversation.ts @@ -167,6 +167,10 @@ export interface UseAgentConversationArgs { /** Hand a late-refused send back to the composer; return whether it took the text (and the * staged files it carried). See `restoreRefusedSend` in `@agenta/chat/assets`. */ restoreRefusedSend?: (message: QueuedMessage) => boolean | Promise + /** A durable send was admitted: the turn it started, or `null` for a parked input. */ + onSendAccepted?: (message: {text: string}, executionId: string | null) => void + /** A durable send was rejected or refused; no turn will ever carry it. */ + onSendFailed?: (message: {text: string}) => void /** Override the client-tool predicate. Defaults to the package registry's, so a host does not * have to opt IN to elicitation and connect widgets — /m shipped without one for months and * silently folded every client tool into the plain "used N tools" group, leaving the run @@ -189,6 +193,12 @@ export interface AgentConversation { turns: TurnViewModel[] /** Send a user message (routes through the queue: sends now, or holds while busy/paused). */ send: (input: SendInput) => Promise + /** + * A send this mount admitted is still on its way: the runner has neither named its turn nor + * refused it. `status` stays "ready" on the server-owned send path, so a skin that wants to + * show work from the moment the message leaves the composer reads this alongside it. + */ + sendInFlight: boolean /** Prevent an approval decision still being recorded from starting its delayed resume. */ voidPendingResume: () => void /** Abort the in-flight stream and tag the last assistant turn as user-stopped. */ @@ -266,6 +276,8 @@ export const useAgentConversation = ({ sharedReaderRunning = false, sharedReaderLivenessUpdatedAt = 0, restoreRefusedSend, + onSendAccepted, + onSendFailed, isClientToolPart, }: UseAgentConversationArgs): AgentConversation => { // Declared FIRST, so its effect re-arms before any effect below can capture a generation. @@ -543,7 +555,7 @@ export const useAgentConversation = ({ // `messages`/`busy` change every commit; consumers that must stay referentially stable // (`rewind`, the hydration/revalidation adoption guards) read them through refs instead. messagesRef.current = messages - busyRef.current = busy || acceptedRunPending + // `busyRef` is assigned after the queue hook below: `sendInFlight` is part of it. localRenderBusyRef.current = busy && !acceptedRunPending useEffect(() => { @@ -791,6 +803,7 @@ export const useAgentConversation = ({ cancelEdit, commitEdit, pendingSendRows, + sendInFlight, } = useAgentChatQueue({ status, messages, @@ -805,10 +818,15 @@ export const useAgentConversation = ({ // so a late refusal keeps its flagged row instead of restoring the draft. Pass a restorer // in to unify it with the desktop. restoreRefusedSend, + onSendAccepted, + onSendFailed, sendQueued, sessionId, server: serverInputs, }) + // A preserve check that misses `sendInFlight` lets a navigation release and stop the chat in + // the window between the message leaving and the turn being accepted. + busyRef.current = busy || acceptedRunPending || sendInFlight // The server capability chooses one owner. Feature-off servers keep the original ordered row // transition + AI SDK gate release; durable servers own continuation after their 202. @@ -1040,10 +1058,11 @@ export const useAgentConversation = ({ // Publish this session's run state (single source of truth for session-list status dots). // Precedence error > awaiting approval > running > idle. + // `sendInFlight` too: the dot goes live when the message leaves, not when a poll notices. const runStatus = deriveSessionRunStatus({ error: !!errorBoundary.runError, hitlPending, - busy: busy || acceptedRunPending || ownsContinuation, + busy: busy || acceptedRunPending || ownsContinuation || sendInFlight, }) useEffect(() => { setSessionStatus({id: sessionId, status: runStatus}) @@ -1385,6 +1404,7 @@ export const useAgentConversation = ({ connectionWarning: errorBoundary.connectionWarning, turns, send, + sendInFlight, voidPendingResume, stop: handleStop, regenerate: regenerateTurn, diff --git a/web/packages/agenta-chat/src/hooks/usePendingSendEchoes.ts b/web/packages/agenta-chat/src/hooks/usePendingSendEchoes.ts index f415dbb9842..8d7b5b448a1 100644 --- a/web/packages/agenta-chat/src/hooks/usePendingSendEchoes.ts +++ b/web/packages/agenta-chat/src/hooks/usePendingSendEchoes.ts @@ -8,6 +8,7 @@ import { durableUserTurnIds, nextPendingSendCoverage, pendingSendEchoMessages, + pendingSendsInFlight, retirePendingSendEchoes, type PendingSendEcho, } from "../assets/pendingSendEchoes" @@ -21,6 +22,8 @@ export interface PendingSendEchoInput { export interface PendingSendEchoes { /** Disposable user rows to render between the saved transcript and the live answer. */ rows: UIMessage[] + /** A send left the composer and the runner has neither named its turn's row nor refused it. */ + inFlight: boolean /** Show a send immediately, before its request leaves. */ add: (input: PendingSendEchoInput) => void /** The server named the turn this send started; from here it retires on that id alone. */ @@ -152,6 +155,7 @@ export const usePendingSendEchoes = ({ }, []) const rows = useMemo(() => pendingSendEchoMessages(visible), [visible]) + const inFlight = useMemo(() => pendingSendsInFlight(visible), [visible]) - return {rows, add, markAccepted, markParked, markFailed, drop} + return {rows, inFlight, add, markAccepted, markParked, markFailed, drop} } diff --git a/web/packages/agenta-chat/tests/unit/hooks/useAgentChatQueue.test.ts b/web/packages/agenta-chat/tests/unit/hooks/useAgentChatQueue.test.ts index bcb0844d83c..1dff5cc7578 100644 --- a/web/packages/agenta-chat/tests/unit/hooks/useAgentChatQueue.test.ts +++ b/web/packages/agenta-chat/tests/unit/hooks/useAgentChatQueue.test.ts @@ -75,6 +75,8 @@ interface HarnessProps { * than the call, and this stayed synchronous. */ restoreRefusedSend?: Parameters[0]["restoreRefusedSend"] + onSendAccepted?: Parameters[0]["onSendAccepted"] + onSendFailed?: Parameters[0]["onSendFailed"] } const setup = (initial: HarnessProps) => { @@ -102,6 +104,53 @@ describe("useAgentChatQueue", () => { expect(result.current.queued).toHaveLength(0) }) + // The durable path reports admission from the server. This one has no such event, and + // staying silent left a fresh session's optimistic row stuck at `submitting`, where a later + // message's failure could delete a row the server was going to list (Mahmoud, #6783). + it("reports a send handed straight to the transport as admitted", () => { + const onSendAccepted = vi.fn() + const {result, sendQueued} = setup({...settledEmpty, onSendAccepted}) + + act(() => { + result.current.submit({text: "no durable queue here"}) + }) + + expect(sendQueued).toHaveBeenCalledTimes(1) + expect(onSendAccepted).toHaveBeenCalledTimes(1) + expect(onSendAccepted.mock.calls[0][0]).toMatchObject({text: "no durable queue here"}) + // No server admission id exists on this path, and claiming one would be a lie. + expect(onSendAccepted.mock.calls[0][1]).toBeNull() + }) + + it("reports a queued message as admitted when it is released, not when it is queued", () => { + const onSendAccepted = vi.fn() + const {result, rerender, sendQueued} = setup({ + status: "streaming", + messages: [userTurn("u1", "go")], + stopped: false, + onSendAccepted, + }) + + act(() => { + result.current.submit({text: "waits its turn"}) + }) + expect(sendQueued).not.toHaveBeenCalled() + expect(onSendAccepted).not.toHaveBeenCalled() + + act(() => { + rerender({ + status: "ready", + messages: [userTurn("u1", "go")], + stopped: false, + onSendAccepted, + }) + }) + + expect(sendQueued).toHaveBeenCalledTimes(1) + expect(onSendAccepted).toHaveBeenCalledTimes(1) + expect(onSendAccepted.mock.calls[0][0]).toMatchObject({text: "waits its turn"}) + }) + it("queues messages typed while a turn is streaming", () => { const {result, sendQueued} = setup({ status: "streaming", @@ -213,6 +262,62 @@ describe("useAgentChatQueue", () => { expect(sendQueued).not.toHaveBeenCalled() }) + // A host that showed the session the moment the message left needs to hear about every way + // that send can die (#6783 review): rejected before it left, or refused after the 200. + it("reports a rejected durable send as failed, once", async () => { + const onSendFailed = vi.fn() + const onSendAccepted = vi.fn() + const server: ServerQueueAdapter = { + capabilities: {queue: true, steer: true}, + busy: false, + queued: [], + submit: vi.fn().mockRejectedValue(new Error("not ready")), + remove: vi.fn().mockResolvedValue(undefined), + } + const {result} = setup({...settledEmpty, server, onSendFailed, onSendAccepted}) + + await act(async () => { + await expect(result.current.submit({text: "first message"})).rejects.toThrow( + "not ready", + ) + }) + + expect(onSendFailed).toHaveBeenCalledOnce() + expect(onSendFailed.mock.calls[0][0]).toMatchObject({text: "first message"}) + expect(onSendAccepted).not.toHaveBeenCalled() + }) + + it("reports a late refusal as failed and an admitted turn as accepted", async () => { + const onSendFailed = vi.fn() + const onSendAccepted = vi.fn() + const server: ServerQueueAdapter = { + capabilities: {queue: true, steer: true}, + busy: false, + queued: [], + submit: vi.fn(async (message, _policy, watcher) => { + if (message.text === "refused") watcher?.onFailed?.() + else watcher?.onAccepted?.("exec-1") + return "running" as const + }), + remove: vi.fn().mockResolvedValue(undefined), + } + const {result} = setup({...settledEmpty, server, onSendFailed, onSendAccepted}) + + await act(async () => { + await result.current.submit({text: "refused"}) + }) + expect(onSendFailed).toHaveBeenCalledOnce() + expect(onSendAccepted).not.toHaveBeenCalled() + + await act(async () => { + await result.current.submit({text: "admitted"}) + }) + expect(onSendAccepted).toHaveBeenCalledOnce() + expect(onSendAccepted.mock.calls[0][0]).toMatchObject({text: "admitted"}) + expect(onSendAccepted.mock.calls[0][1]).toBe("exec-1") + expect(onSendFailed).toHaveBeenCalledOnce() + }) + it("propagates a refused Steer without inventing a client-only queued message", async () => { const server: ServerQueueAdapter = { capabilities: {queue: true, steer: true}, diff --git a/web/packages/agenta-chat/tests/unit/pendingSendsInFlight.test.ts b/web/packages/agenta-chat/tests/unit/pendingSendsInFlight.test.ts new file mode 100644 index 00000000000..1169d5b761d --- /dev/null +++ b/web/packages/agenta-chat/tests/unit/pendingSendsInFlight.test.ts @@ -0,0 +1,33 @@ +/** + * The local "submitted" signal behind #6778. On the server-owned send path `useChat`'s status + * never leaves "ready", so an echo that is neither refused nor parked is the only evidence that + * a turn of ours is on its way — and the working indicator must read it. + */ +import {describe, expect, it} from "vitest" + +import {pendingSendsInFlight, type PendingSendEcho} from "../../src/assets/pendingSendEchoes" + +const echo = (patch: Partial = {}): PendingSendEcho => ({ + id: "e1", + text: "hello", + coveredAtUserCount: 1, + createdAtUserCount: 0, + ...patch, +}) + +describe("pendingSendsInFlight", () => { + it("is false with nothing pending", () => { + expect(pendingSendsInFlight([])).toBe(false) + }) + + it("is true from the moment a send is admitted, and still after the runner named it", () => { + expect(pendingSendsInFlight([echo()])).toBe(true) + expect(pendingSendsInFlight([echo({executionId: "x1"})])).toBe(true) + }) + + it("ignores refused sends and sends parked in the queue", () => { + expect(pendingSendsInFlight([echo({failed: true})])).toBe(false) + expect(pendingSendsInFlight([echo({parkedInputId: "in1"})])).toBe(false) + expect(pendingSendsInFlight([echo({failed: true}), echo({id: "e2"})])).toBe(true) + }) +}) diff --git a/web/packages/agenta-entities/src/session/core/localSessions.ts b/web/packages/agenta-entities/src/session/core/localSessions.ts new file mode 100644 index 00000000000..549da0f8176 --- /dev/null +++ b/web/packages/agenta-entities/src/session/core/localSessions.ts @@ -0,0 +1,110 @@ +import {atom} from "jotai" + +/** + * A session this client created and the server cannot list yet. + * + * A session becomes real on the backend only once its first turn is admitted — a slow runner can + * take seconds — but the client minted its id and knows the agent and the first message from the + * moment that message left the composer. Session lists read this registry so a brand-new session + * appears the instant it is sent, not when the next list refetch happens to carry it (#6776). + * + * Lifecycle: `submitting` from the send until it is admitted, then `accepted` until the server + * lists the session and `forgetLocalSessionsAtom` retires the copy. A send that fails while still + * `submitting` takes its row with it — nothing on the server will ever list that session. + * + * Sibling of `freshSessions`: that one is a synchronous predicate for "no durable records yet", + * read during render; this one is reactive, so a list can subscribe to it. + */ +export interface LocalSession { + sessionId: string + /** The project it was created in. A list shows only its own project's rows. */ + projectId: string + /** The owning agent's workflow id, where the creating surface knows it. */ + agentId: string | null + /** The first message, until the server names the session itself. */ + name: string | null + /** ms epoch, so the row can take its place in a date-ordered list. */ + createdAt: number + state: "submitting" | "accepted" +} + +export const localSessionsAtom = atom>({}) + +/** Longest a first message can reasonably be a title. */ +const LOCAL_SESSION_NAME_MAX = 120 + +export const localSessionNameFromText = (text: string | null | undefined): string | null => { + const line = (text ?? "").trim().split("\n")[0]?.trim() ?? "" + if (!line) return null + return line.length > LOCAL_SESSION_NAME_MAX ? `${line.slice(0, LOCAL_SESSION_NAME_MAX)}…` : line +} + +/** + * Note a session this client just sent a message into. Idempotent: a repeat keeps the first + * message as the name and only fills in an agent it did not know before. + */ +export const registerLocalSessionAtom = atom( + null, + ( + get, + set, + input: { + sessionId: string + projectId: string + agentId?: string | null + name?: string | null + now?: number + }, + ) => { + const current = get(localSessionsAtom) + const existing = current[input.sessionId] + const name = existing?.name ?? localSessionNameFromText(input.name) + const agentId = existing?.agentId ?? input.agentId ?? null + if (existing && existing.name === name && existing.agentId === agentId) return + set(localSessionsAtom, { + ...current, + [input.sessionId]: { + sessionId: input.sessionId, + projectId: existing?.projectId ?? input.projectId, + agentId, + name, + createdAt: existing?.createdAt ?? input.now ?? Date.now(), + state: existing?.state ?? "submitting", + }, + }) + }, +) + +/** A send into this session was admitted: the server will list it, so the row stays until then. */ +export const markLocalSessionAcceptedAtom = atom(null, (get, set, sessionId: string) => { + const current = get(localSessionsAtom) + const existing = current[sessionId] + if (!existing || existing.state === "accepted") return + set(localSessionsAtom, {...current, [sessionId]: {...existing, state: "accepted"}}) +}) + +/** + * A send into this session failed. If nothing was ever admitted the server will never list the + * session, so the row goes. An accepted session keeps its row: that failure belongs to a later + * message, and the server list is what retires the copy. + */ +export const dropUnacceptedLocalSessionAtom = atom(null, (get, set, sessionId: string) => { + const current = get(localSessionsAtom) + const existing = current[sessionId] + if (!existing || existing.state === "accepted") return + const next = {...current} + delete next[sessionId] + set(localSessionsAtom, next) +}) + +/** The server lists these now (or they were deleted): the local copies have done their job. */ +export const forgetLocalSessionsAtom = atom(null, (get, set, ids: Iterable) => { + const current = get(localSessionsAtom) + let next: Record | null = null + for (const id of ids) { + if (!(id in current)) continue + next ??= {...current} + delete next[id] + } + if (next) set(localSessionsAtom, next) +}) diff --git a/web/packages/agenta-entities/src/session/index.ts b/web/packages/agenta-entities/src/session/index.ts index bd9a3d7ca74..8341b8a16ba 100644 --- a/web/packages/agenta-entities/src/session/index.ts +++ b/web/packages/agenta-entities/src/session/index.ts @@ -211,3 +211,12 @@ export { isSessionFresh, markSessionFresh, } from "./core/freshSessions" +export { + dropUnacceptedLocalSessionAtom, + forgetLocalSessionsAtom, + localSessionNameFromText, + localSessionsAtom, + markLocalSessionAcceptedAtom, + registerLocalSessionAtom, + type LocalSession, +} from "./core/localSessions" diff --git a/web/packages/agenta-entities/tests/unit/session-local-sessions.test.ts b/web/packages/agenta-entities/tests/unit/session-local-sessions.test.ts new file mode 100644 index 00000000000..b75a0c7cfde --- /dev/null +++ b/web/packages/agenta-entities/tests/unit/session-local-sessions.test.ts @@ -0,0 +1,122 @@ +/** + * The client-created session registry behind #6776: a session the server cannot list yet is + * noted at its first send, keeps its first message as its name, follows that send's fate, and + * retires once the server carries it. + */ +import {createStore} from "jotai" +import {describe, expect, it} from "vitest" + +import { + dropUnacceptedLocalSessionAtom, + forgetLocalSessionsAtom, + localSessionNameFromText, + localSessionsAtom, + markLocalSessionAcceptedAtom, + registerLocalSessionAtom, +} from "../../src/session" + +describe("localSessions", () => { + it("registers a session with its first message as the name, in its project", () => { + const store = createStore() + store.set(registerLocalSessionAtom, { + sessionId: "s1", + projectId: "p1", + agentId: "a1", + name: " Write a file named qa.html \n second line", + now: 1_000, + }) + expect(store.get(localSessionsAtom)).toEqual({ + s1: { + sessionId: "s1", + projectId: "p1", + agentId: "a1", + name: "Write a file named qa.html", + createdAt: 1_000, + state: "submitting", + }, + }) + }) + + it("keeps the first name and creation time on a repeat, only filling a missing agent", () => { + const store = createStore() + store.set(registerLocalSessionAtom, { + sessionId: "s1", + projectId: "p1", + name: "first", + now: 1_000, + }) + const before = store.get(localSessionsAtom) + store.set(registerLocalSessionAtom, { + sessionId: "s1", + projectId: "p1", + name: "second", + now: 2_000, + }) + // Nothing changed, so the same object comes back — no spurious list re-render. + expect(store.get(localSessionsAtom)).toBe(before) + store.set(registerLocalSessionAtom, { + sessionId: "s1", + projectId: "p1", + agentId: "a1", + name: "third", + }) + expect(store.get(localSessionsAtom).s1).toEqual({ + sessionId: "s1", + projectId: "p1", + agentId: "a1", + name: "first", + createdAt: 1_000, + state: "submitting", + }) + }) + + // A rejected or refused first send: the server will never list this session, so the row + // must not outlive the send (#6783 review). + it("drops a session whose only send failed", () => { + const store = createStore() + store.set(registerLocalSessionAtom, {sessionId: "s1", projectId: "p1", name: "one"}) + store.set(dropUnacceptedLocalSessionAtom, "s1") + expect(store.get(localSessionsAtom)).toEqual({}) + }) + + it("keeps an accepted session through a later failure, until the server lists it", () => { + const store = createStore() + store.set(registerLocalSessionAtom, {sessionId: "s1", projectId: "p1", name: "one"}) + store.set(markLocalSessionAcceptedAtom, "s1") + expect(store.get(localSessionsAtom).s1.state).toBe("accepted") + const accepted = store.get(localSessionsAtom) + // Accepting twice, or dropping an accepted session, changes nothing. + store.set(markLocalSessionAcceptedAtom, "s1") + store.set(dropUnacceptedLocalSessionAtom, "s1") + expect(store.get(localSessionsAtom)).toBe(accepted) + // Only the server list retires it. + store.set(forgetLocalSessionsAtom, ["s1"]) + expect(store.get(localSessionsAtom)).toEqual({}) + }) + + it("ignores lifecycle events for a session it never registered", () => { + const store = createStore() + const before = store.get(localSessionsAtom) + store.set(markLocalSessionAcceptedAtom, "missing") + store.set(dropUnacceptedLocalSessionAtom, "missing") + expect(store.get(localSessionsAtom)).toBe(before) + }) + + it("forgets only the ids it holds and leaves the map untouched otherwise", () => { + const store = createStore() + store.set(registerLocalSessionAtom, {sessionId: "s1", projectId: "p1", name: "one", now: 1}) + store.set(registerLocalSessionAtom, {sessionId: "s2", projectId: "p1", name: "two", now: 2}) + const before = store.get(localSessionsAtom) + store.set(forgetLocalSessionsAtom, ["missing"]) + expect(store.get(localSessionsAtom)).toBe(before) + store.set(forgetLocalSessionsAtom, ["s1", "missing"]) + expect(Object.keys(store.get(localSessionsAtom))).toEqual(["s2"]) + }) + + it("derives a title from the first line, bounded, or none from blank text", () => { + expect(localSessionNameFromText(" hi \n there")).toBe("hi") + expect(localSessionNameFromText(" \n ")).toBeNull() + expect(localSessionNameFromText(undefined)).toBeNull() + expect(localSessionNameFromText("x".repeat(200))).toBe(`${"x".repeat(120)}…`) + }) +}) diff --git a/web/packages/agenta-navigation/src/dynamic/sessionsSource.ts b/web/packages/agenta-navigation/src/dynamic/sessionsSource.ts index fddf82717c8..944ee87ea07 100644 --- a/web/packages/agenta-navigation/src/dynamic/sessionsSource.ts +++ b/web/packages/agenta-navigation/src/dynamic/sessionsSource.ts @@ -724,6 +724,24 @@ const uniqueBySession = (refs: readonly SessionSidebarRef[]): SessionSidebarRef[ return unique } +/** + * The ids the SERVER lists for this scope, pins and pages included. A host reconciles its local + * rows against it: a session the server carries no longer needs the local seam, and a local row + * that outlived its server twin would resurface the session whenever it aged out of the window. + */ +export const sidebarServerSessionIdsAtomFamily = atomFamily((scopeId: string) => + atom>((get) => { + const ids = new Set() + for (const row of get(sidebarPinnedSessionsQueryAtomFamily(scopeId)).data ?? []) + ids.add(row.session_id) + for (const row of get(sidebarSessionsQueryAtomFamily(scopeId)).data ?? []) + ids.add(row.session_id) + for (const row of get(sidebarSessionsOlderQueryAtomFamily(scopeId)).data?.rows ?? []) + ids.add(row.session_id) + return ids + }), +) + /** * Pinned sessions first, then the rest by activity. * diff --git a/web/packages/agenta-navigation/src/index.ts b/web/packages/agenta-navigation/src/index.ts index f9ad62d1ec8..f5d4b69d205 100644 --- a/web/packages/agenta-navigation/src/index.ts +++ b/web/packages/agenta-navigation/src/index.ts @@ -22,6 +22,7 @@ export { sidebarSessionSearchOpenAtom, sidebarSessionSearchQueryAtom, sidebarSessionSearchResultsAtom, + sidebarServerSessionIdsAtomFamily, withLocalSessions, type SessionSidebarRef, } from "./dynamic/sessionsSource"