diff --git a/.changeset/quiet-agent-session-latency.md b/.changeset/quiet-agent-session-latency.md new file mode 100644 index 00000000000..a241e8af8a4 --- /dev/null +++ b/.changeset/quiet-agent-session-latency.md @@ -0,0 +1,6 @@ +--- +"@nocobase/acp-bridge": patch +"@nocobase/app-plugin-agent-sessions": patch +--- + +Return agent settings before process cleanup, preserve editable drafts and model selections during discovery, display confirmed messages immediately, and bound realtime refresh rates with recovery reads and optional timing diagnostics. diff --git a/packages/libs/acp-bridge/src/bridge.ts b/packages/libs/acp-bridge/src/bridge.ts index 9c41ff27d3f..c79b06acc47 100644 --- a/packages/libs/acp-bridge/src/bridge.ts +++ b/packages/libs/acp-bridge/src/bridge.ts @@ -237,6 +237,7 @@ export async function runBridge(options: BridgeOptions): Promise { ); if (signal.aborted) return; for (const info of infos) agents.set(info.id, info); + // Registration also negotiates protocol upgrades; polls do not carry that handshake. await register(); })() .catch((error: unknown) => { @@ -276,6 +277,19 @@ export async function runBridge(options: BridgeOptions): Promise { const discover = async ( command: Extract, ): Promise => { + const started = Date.now(); + const timing = (phase: string): void => { + if (env.NOCOBASE_ACP_DIAGNOSTICS === '1') + log( + JSON.stringify({ + phase, + at: Date.now(), + durationMs: Date.now() - started, + requestId: command.requestId, + }), + ); + }; + timing('discoveryReceived'); const discoverySignal = AbortSignal.any([ signal, discoveryShutdown.signal, @@ -320,13 +334,23 @@ export async function runBridge(options: BridgeOptions): Promise { : message(error) ).slice(0, 2000), }; - } finally { - await session.close(); } - if (!signal.aborted) - await post(BRIDGE_ROUTES.reply, reply).catch((error: unknown) => - log(`Could not answer the application: ${message(error)}`), - ); + // Start cleanup immediately, but do not hold the answer behind the process-group grace period. + // This discovery remains tracked until both delivery and cleanup settle, including on shutdown. + timing('discoveryReady'); + const closing = session.close().catch((error: unknown) => { + log(`Could not close the discovery process: ${message(error)}`); + }); + try { + if (!signal.aborted) + await post(BRIDGE_ROUTES.reply, reply).catch((error: unknown) => + log(`Could not answer the application: ${message(error)}`), + ); + timing('discoveryReplied'); + } finally { + await closing; + timing('discoveryClosed'); + } }; const reporter = new Reporter({ send: (request) => post(BRIDGE_ROUTES.report, request), @@ -384,13 +408,30 @@ export async function runBridge(options: BridgeOptions): Promise { const sessionId = command.session.id; let session = sessions.get(sessionId); if (!session) { + const diagnostics = env.NOCOBASE_ACP_DIAGNOSTICS === '1'; + let promptId: string | undefined; + let firstToken = false; + const timing = (phase: string): void => { + log(JSON.stringify({phase, at:Date.now(), sessionId, promptId})); + }; session = new BridgeSession(command.session, { client: options.client, server: options.server, env, roots, resolve, - emit: (event) => reporter.add(sessionId, event), + emit: (event) => { + if (diagnostics) { + if (event.kind === 'turnStarted') { + promptId = event.promptId; firstToken = false; timing('agentPromptStarted'); + } else if (event.kind === 'update' && event.update.sessionUpdate === 'agent_message_chunk' && promptId && !firstToken) { + firstToken = true; timing('agentFirstToken'); + } else if (event.kind === 'turnEnded') { + timing('agentTurnEnded'); promptId = undefined; + } + } + reporter.add(sessionId,event); + }, stateChanged: () => reporter.touch(sessionId), }); sessions.set(sessionId, session); diff --git a/packages/libs/acp-bridge/tests/bridge.test.ts b/packages/libs/acp-bridge/tests/bridge.test.ts index 602a90ffdb6..8585eed4090 100644 --- a/packages/libs/acp-bridge/tests/bridge.test.ts +++ b/packages/libs/acp-bridge/tests/bridge.test.ts @@ -8,6 +8,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import { BRIDGE_HEADERS, BRIDGE_PROTOCOL_VERSION } from '../src/protocol.js'; import * as agents from '../src/agents.js'; +import { BridgeSession } from '../src/session.js'; import { AgentProcess } from '../src/agent-process.js'; import type { BridgeCommand } from '../src/protocol.js'; @@ -110,6 +111,47 @@ const registers = (): Seen[] => seen.filter((request) => request.path.endsWith('/register')); describe('runBridge', () => { + it('delivers discovered choices before slow cleanup and awaits cleanup during shutdown', async () => { + let release: (() => void) | undefined; + vi.spyOn(BridgeSession.prototype, 'discover').mockResolvedValue({ + status: 'available', + choices: [], + }); + vi.spyOn(BridgeSession.prototype, 'close').mockImplementation( + () => + new Promise((resolve) => { + release = resolve; + }), + ); + answer = () => ({ data: { computerId: 'c1', key: 'k1' } }); + commands.push({ + type: 'discoverChoices', + requestId: 'd1', + agentId: 'opencode', + cwd: root, + settings: {}, + }); + const { done, stop } = await run(); + try { + await until(() => seen.some((one) => one.path.endsWith('/reply'))); + expect(release).toBeDefined(); + expect( + seen.find((one) => one.path.endsWith('/reply'))?.body, + ).toMatchObject({ requestId: 'd1', discovery: { status: 'available' } }); + let settled = false; + void done.then(() => { + settled = true; + }); + stop(); + await new Promise((resolve) => setTimeout(resolve, 50)); + expect(settled).toBe(false); + } finally { + release?.(); + stop(); + await done; + } + }); + it('refreshes with separate probe processes while an active session keeps running', async () => { vi.useFakeTimers({ toFake: ['setInterval', 'clearInterval'] }); vi.spyOn(agents, 'resolveAgent').mockImplementation(async (spec) => diff --git a/packages/plugins/app-plugin-agent-sessions/client/api.ts b/packages/plugins/app-plugin-agent-sessions/client/api.ts index d2c21228447..a334f7ec121 100644 --- a/packages/plugins/app-plugin-agent-sessions/client/api.ts +++ b/packages/plugins/app-plugin-agent-sessions/client/api.ts @@ -14,6 +14,7 @@ import { useQuery, useQueryClient, type InfiniteData, + type QueryClient, type UseInfiniteQueryResult, type UseMutationResult, type UseQueryResult, @@ -92,6 +93,8 @@ export function useSessions( }), initialPageParam: '', getNextPageParam: (last) => last.meta.nextPageToken, + refetchInterval: 30_000, + refetchOnWindowFocus: true, }); } @@ -101,8 +104,7 @@ export function useSession(sessionId: string): UseQueryResult { queryKey: keys.session(sessionId), queryFn: () => data(api, { path: path('sessions', sessionId) }), // Linked records may change without a session event (for example, an issue accepted elsewhere). - refetchInterval: (query) => - query.state.data?.links.length ? 5_000 : false, + refetchInterval: 15_000, refetchOnWindowFocus: true, }); } @@ -118,7 +120,9 @@ export function useTranscript( const client = useQueryClient(); return useQuery({ queryKey: keys.transcript(sessionId), - queryFn: async () => { + refetchInterval: 15_000, + refetchOnWindowFocus: true, + queryFn: async ({ signal }) => { const held = client.getQueryData(keys.transcript(sessionId)) ?? []; @@ -129,10 +133,17 @@ export function useTranscript( const page = await api.request>({ path: path('sessions', sessionId, 'events'), query: { after }, + signal, }); events.push(...page.data); after = page.data.at(-1)?.seq ?? after; - if (!page.meta.hasMore) return events; + if (!page.meta.hasMore) + return mergeTranscript( + client.getQueryData( + keys.transcript(sessionId), + ) ?? [], + events, + ); } }, }); @@ -179,32 +190,112 @@ export function useMarkSeen(sessionId: string, seq: number | undefined): void { }, [api, sessionId, seq]); } +const liveSubscriptions = new WeakMap< + QueryClient, + { refs: number; release: () => void } +>(); + /** Refreshes what the viewer sees when the server says something of theirs changed, or the connection comes back. */ export function useLiveSessions(): void { const realtime = useService(realtimeClientToken); const client = useQueryClient(); useEffect(() => { - // A streaming answer changes its session several times a second; the list follows at a calmer pace. - let listTimer: ReturnType | undefined; - const refreshList = (): void => { - listTimer ??= setTimeout(() => { - listTimer = undefined; - void client.invalidateQueries({ - queryKey: ['agentSessions', 'sessions'], - }); - }, 2_000); + const existing = liveSubscriptions.get(client); + const detach = (): void => { + const shared = liveSubscriptions.get(client); + if (shared && --shared.refs === 0) { + shared.release(); + liveSubscriptions.delete(client); + } + }; + if (existing) { + existing.refs += 1; + return detach; + } + type Slot = { + readonly id: string; + readonly key: readonly unknown[]; + readonly interval: number; + pending: boolean; + running: boolean; + started: number; + timer: ReturnType | undefined; + }; + const slots = new Map(); + let disposed = false; + const arm = (slot: Slot): void => { + if (disposed || slot.timer || slot.running) return; + const delay = slot.started + ? Math.max(0, slot.started + slot.interval - Date.now()) + : slot.interval; + const timer = setTimeout(() => { + slot.timer = undefined; + void flush(slot); + }, delay); + slot.timer = timer; + }; + const flush = async (slot: Slot): Promise => { + slot.pending = false; + slot.running = true; + slot.started = Date.now(); + // An existing read may predate this notification. Follow it with a fresh read after it settles. + if (client.getQueryState(slot.key)?.fetchStatus === 'fetching') + slot.pending = true; + try { + // Never keep aborting a slow read during a continuous stream. + await client.invalidateQueries( + { queryKey: slot.key }, + { cancelRefetch: false }, + ); + } finally { + slot.running = false; + if (slot.pending) arm(slot); + else slots.delete(slot.id); + } + }; + const schedule = ( + id: string, + key: readonly unknown[], + interval: number, + ): void => { + const slot = slots.get(id) ?? { + id, + key, + interval, + pending: false, + running: false, + started: 0, + timer: undefined, + }; + slots.set(id, slot); + slot.pending = true; + arm(slot); + }; + const invalidateDiscovery = (): void => { + void client.cancelQueries({ queryKey: ['agentSessions', 'discovery'] }); + void client.invalidateQueries({ + queryKey: ['agentSessions', 'discovery'], + }); }; const refresh = (change?: AgentSessionsChange): void => { if (change?.kind === 'session') { + const id = change.sessionId; + schedule(`events:${id}`, keys.transcript(id), 250); + schedule(`session:${id}`, keys.session(id), 1_000); + schedule('list', ['agentSessions', 'sessions'], 2_000); + } else if (change?.kind === 'computers') { + invalidateDiscovery(); + void client.invalidateQueries({ queryKey: keys.computers }); void client.invalidateQueries({ - queryKey: keys.session(change.sessionId), + queryKey: ['agentSessions', 'session'], }); - void client.invalidateQueries({ - queryKey: keys.transcript(change.sessionId), - }); - refreshList(); } else { - void client.invalidateQueries({ queryKey: keys.all }); + invalidateDiscovery(); + void client.invalidateQueries({ + predicate: (query) => + query.queryKey[0] === 'agentSessions' && + query.queryKey[1] !== 'discovery', + }); } }; const unsubscribe = realtime.subscribe( @@ -212,14 +303,33 @@ export function useLiveSessions(): void { ({ payload }) => refresh(payload), ); const unsubscribeOpen = realtime.onOpen(() => refresh()); - return () => { - clearTimeout(listTimer); + const release = (): void => { + disposed = true; + for (const slot of slots.values()) clearTimeout(slot.timer); + // No bridge generation is exposed to the browser. Do not reuse discovery across an unobserved interval. + void client.invalidateQueries({ + queryKey: ['agentSessions', 'discovery'], + refetchType: 'none', + }); unsubscribe(); unsubscribeOpen(); }; + liveSubscriptions.set(client, { refs: 1, release }); + refresh(); + return detach; }, [client, realtime]); } +/** Preserve confirmed events missing from an older GET; fetched updates may grow existing streamed events. */ +export function mergeTranscript( + held: readonly TranscriptEvent[], + fetched: readonly TranscriptEvent[], +): TranscriptEvent[] { + const events = new Map(held.map((event) => [event.seq, event])); + for (const event of fetched) events.set(event.seq, event); + return [...events.values()].sort((left, right) => left.seq - right.seq); +} + type Mutation = UseMutationResult; /** A write whose answer is the session as it is now. */ @@ -264,7 +374,14 @@ export function useCreateSession(): Mutation< path: path('sessions'), method: 'POST', json: input, + signal: AbortSignal.timeout(30_000), }), + retry: false, + onError: () => { + void client.invalidateQueries({ + queryKey: ['agentSessions', 'sessions'], + }); + }, onSuccess: (session) => { client.setQueryData(keys.session(session.id), session); void client.invalidateQueries({ @@ -276,17 +393,35 @@ export function useCreateSession(): Mutation< export function usePrompt( sessionId: string, -): Mutation<{ content: PromptContent[]; mentions: Mention[] }> { - return useSessionWrite( - sessionId, - async (api, json: { content: PromptContent[]; mentions: Mention[] }) => { - await api.request({ +): Mutation< + { content: PromptContent[]; mentions: Mention[] }, + TranscriptEvent +> { + const api = useApiClient(); + const client = useQueryClient(); + return useMutation({ + retry: false, + mutationFn: (json) => + data(api, { path: path('sessions', sessionId, 'prompt'), method: 'POST', json, - }); + signal: AbortSignal.timeout(30_000), + }), + onSuccess: async (event) => { + await client.cancelQueries({ queryKey: keys.transcript(sessionId) }); + client.setQueryData( + keys.transcript(sessionId), + (held = []) => mergeTranscript(held, [event]), + ); + // Realtime supplies the bounded session/list refresh. Transcript recovery also polls and refetches on focus. }, - ); + onError: () => { + // A failed transport may have committed. Reconcile; never automatically resend the prompt. + void client.invalidateQueries({ queryKey: keys.transcript(sessionId) }); + void client.invalidateQueries({ queryKey: keys.session(sessionId) }); + }, + }); } export function useCancel(sessionId: string): Mutation { diff --git a/packages/plugins/app-plugin-agent-sessions/client/components/choice-menus.tsx b/packages/plugins/app-plugin-agent-sessions/client/components/choice-menus.tsx index a024f963ad8..2cff1c97ff8 100644 --- a/packages/plugins/app-plugin-agent-sessions/client/components/choice-menus.tsx +++ b/packages/plugins/app-plugin-agent-sessions/client/components/choice-menus.tsx @@ -54,12 +54,14 @@ export function ChoiceMenus({ choices, values, disabled, + loading, onChoose, }: { readonly choices: readonly SessionChoice[] | undefined; /** The values chosen so far, by choice id; a choice left out shows its current value. */ readonly values?: Readonly>; readonly disabled?: boolean; + readonly loading?: boolean; readonly onChoose: (choiceId: string, value: string) => void; }): ReactElement { const { t } = useTranslation(); @@ -88,7 +90,8 @@ export function ChoiceMenus({ .map((choice) => ( @@ -122,7 +125,8 @@ export function ChoiceMenus({ {more.some((choice) => choice.options.length > 1) && ( export function Composer({ placeholder, disabled, + sendDisabled, busy, sending, commands = [], @@ -201,8 +204,11 @@ export function Composer({ ); }; + const sendingRef = useRef(false); const send = async (interrupt: boolean): Promise => { - if (empty || disabled || sending) return; + if (empty || disabled || sendDisabled || sending || sendingRef.current) + return; + sendingRef.current = true; const content: PromptContent[] = [ ...(text.trim() ? [{ type: 'text' as const, text: text.trim() }] : []), ...images.map(({ mimeType, data }) => ({ @@ -219,8 +225,10 @@ export function Composer({ await onSend(content, interrupt, kept); } catch { // The caller said what went wrong; the draft stays for another try. + sendingRef.current = false; return; } + sendingRef.current = false; setText(''); setCaret(0); setImages([]); @@ -377,7 +385,7 @@ export function Composer({ value={text} rows={large ? 4 : compact ? 1 : 2} autoFocus={autoFocus} - disabled={disabled} + disabled={disabled || sending} placeholder={placeholder} aria-label={placeholder} onChange={(event) => { @@ -412,7 +420,7 @@ export function Composer({ > filesRef.current?.click()} > @@ -421,7 +429,7 @@ export function Composer({ { setExpanded(!expanded); inputRef.current?.focus(); @@ -449,7 +457,7 @@ export function Composer({ {mentions && ( insert('#')} > @@ -458,7 +466,7 @@ export function Composer({ {commands.length > 0 && ( type('/', 1)} > @@ -466,6 +474,11 @@ export function Composer({ )}
{toolbar} + {sending && ( + + {t('composer.sending')} + + )}
{trailing} @@ -473,7 +486,7 @@ export function Composer({ type='button' aria-label={busy ? t('composer.queue') : t('composer.send')} title={busy ? t('composer.queue') : t('composer.send')} - disabled={empty || disabled || sending} + disabled={empty || disabled || sendDisabled || sending} onClick={() => void send(false)} className={cn( 'inline-flex size-9 shrink-0 items-center justify-center rounded-full transition-colors', diff --git a/packages/plugins/app-plugin-agent-sessions/client/components/new-session.tsx b/packages/plugins/app-plugin-agent-sessions/client/components/new-session.tsx index 9335b01662f..dd513cb9e28 100644 --- a/packages/plugins/app-plugin-agent-sessions/client/components/new-session.tsx +++ b/packages/plugins/app-plugin-agent-sessions/client/components/new-session.tsx @@ -5,7 +5,7 @@ * by browsing the computer, never typed. The computer, agent, folder and project last used are remembered in this * browser. Without a connected computer it says how to connect one. */ -import { useToaster } from '@nocobase/app-client'; +import { ApiClientError, useToaster } from '@nocobase/app-client'; import { useTranslation } from '@nocobase/i18n/client'; import { cn } from 'cn'; import { @@ -80,7 +80,9 @@ export function NewSession(): ReactElement { const all = computers.data ?? []; const online = all.filter((computer) => computer.online); const computer = - all.find((candidate) => candidate.id === choice.computerId) ?? online[0]; + all.find((candidate) => candidate.id === choice.computerId) ?? + online[0] ?? + all[0]; if (computer && choice.computerId !== computer.id) setChoice({ ...choice, computerId: computer.id }); const installed = computer ? installedAgents(computer) : []; @@ -108,8 +110,10 @@ export function NewSession(): ReactElement { computer?.id, agent, cwd, - computer?.online, computer?.folders, + computer?.roots, + computer?.home, + computer?.host, ]), { computerId: computer?.id ?? '', @@ -118,9 +122,12 @@ export function NewSession(): ReactElement { enabled: Boolean(computer?.online && agent && cwd), }, ); - const offered = discovery.result?.choices; + const offered = discovery.displayed?.choices; const values = Object.fromEntries( - (offered ?? []).map((offer) => [offer.id, offer.currentValue]), + (offered ?? []).map((offer) => [ + offer.id, + discovery.settings[offer.id] ?? offer.currentValue, + ]), ); const ready = Boolean(computer?.online && agent && cwd && discovery.result); @@ -191,11 +198,14 @@ export function NewSession(): ReactElement { {t('newSession.subtitle')}

+ {computer?.outdatedBridge && ( + + )} @@ -328,7 +339,16 @@ export function NewSession(): ReactElement { } onSend={async (content, _interrupt, mentions) => { - if (!computer || !agent || !ready || sendingRef.current) return; + if ( + !computer || + !agent || + !ready || + !discovery.isCurrent() || + sendingRef.current + ) { + discovery.retry(); + throw new Error(t('newSession.settingsLoading')); + } sendingRef.current = true; try { const session = await create.mutateAsync({ @@ -349,7 +369,10 @@ export function NewSession(): ReactElement { } catch (error) { toaster.show({ type: 'error', - title: errorText(error, t('errors.generic')), + title: + error instanceof ApiClientError && error.status < 500 + ? errorText(error, t('errors.generic')) + : t('newSession.sendUnconfirmed'), }); throw error; } finally { diff --git a/packages/plugins/app-plugin-agent-sessions/client/components/session-view.tsx b/packages/plugins/app-plugin-agent-sessions/client/components/session-view.tsx index 0fe73f5ce7c..bbed029e3d0 100644 --- a/packages/plugins/app-plugin-agent-sessions/client/components/session-view.tsx +++ b/packages/plugins/app-plugin-agent-sessions/client/components/session-view.tsx @@ -3,7 +3,7 @@ * choices, and beside it that work in detail, where the session runs and what it has used. A session whose computer is * offline, or that is archived, reads as before and takes no new messages. */ -import { useToaster } from '@nocobase/app-client'; +import { ApiClientError, useToaster } from '@nocobase/app-client'; import { useTranslation } from '@nocobase/i18n/client'; import { cn } from 'cn'; import { @@ -264,7 +264,12 @@ function Conversation({ await prompt.mutateAsync({ content, mentions }); pinnedRef.current = true; } catch (error) { - fail(error); + if (error instanceof ApiClientError && error.status < 500) fail(error); + else + void toaster.show({ + type: 'error', + title: t('composer.sendUnconfirmed'), + }); throw error; } }; diff --git a/packages/plugins/app-plugin-agent-sessions/client/locales/en-US.ts b/packages/plugins/app-plugin-agent-sessions/client/locales/en-US.ts index 5f549f5e7e8..7c31cc98d6d 100644 --- a/packages/plugins/app-plugin-agent-sessions/client/locales/en-US.ts +++ b/packages/plugins/app-plugin-agent-sessions/client/locales/en-US.ts @@ -70,6 +70,8 @@ const enUS = { unknown: 'Unknown', }, newSession: { + sendUnconfirmed: + 'Send was not confirmed. Check recent sessions before trying again; a session may already have been created.', settingsOutdated: 'This computer\'s command line is too old for this application. Update it there, as "My computers" shows.', settingsOffline: @@ -155,6 +157,9 @@ const enUS = { recordBody: 'The full record is kept here, and only you can see it.', }, composer: { + sendUnconfirmed: + 'Send was not confirmed. Check the conversation before trying again; it may already have been sent.', + sending: 'Sending…', expand: 'Expand input', collapse: 'Collapse input', placeholder: 'Reply, or ask for the next thing…', diff --git a/packages/plugins/app-plugin-agent-sessions/client/locales/zh-CN.ts b/packages/plugins/app-plugin-agent-sessions/client/locales/zh-CN.ts index 251cb89302d..9ddc64eac7a 100644 --- a/packages/plugins/app-plugin-agent-sessions/client/locales/zh-CN.ts +++ b/packages/plugins/app-plugin-agent-sessions/client/locales/zh-CN.ts @@ -63,6 +63,7 @@ const zhCN: AgentSessionsResource = { unknown: '未知', }, newSession: { + sendUnconfirmed: '发送尚未确认。请先检查最近的会话再重试；会话可能已创建。', settingsOutdated: '这台电脑上的命令行版本过旧，请按“我的电脑”里的提示在那台电脑上更新。', settingsOffline: '此电脑已离线。请重新连接桥接程序以加载设置和发送消息。', @@ -142,6 +143,8 @@ const zhCN: AgentSessionsResource = { recordBody: '完整记录保存在这里，只有你能看到。', }, composer: { + sendUnconfirmed: '发送尚未确认。请先检查对话再尝试发送；消息可能已发送。', + sending: '正在发送…', expand: '展开输入', collapse: '收起输入', placeholder: '回复，或者说下一件事…', diff --git a/packages/plugins/app-plugin-agent-sessions/client/use-choice-discovery.ts b/packages/plugins/app-plugin-agent-sessions/client/use-choice-discovery.ts index 2d086d7885b..63d8f3894e0 100644 --- a/packages/plugins/app-plugin-agent-sessions/client/use-choice-discovery.ts +++ b/packages/plugins/app-plugin-agent-sessions/client/use-choice-discovery.ts @@ -1,27 +1,30 @@ -import { - realtimeClientToken, - useApiClient, - useService, -} from '@nocobase/app-client'; +import { useApiClient } from '@nocobase/app-client'; +import { useQuery, useQueryClient } from '@tanstack/react-query'; import { useEffect, useState } from 'react'; -import { - AGENT_SESSIONS_TOPIC, - type AgentSessionsChange, - type DiscoveredChoices, -} from '../shared/api.js'; +import { type DiscoveredChoices } from '../shared/api.js'; const EMPTY: Readonly> = {}; +export const DISCOVERY_CACHE_MS: number = 5 * 60_000; +export const DISCOVERY_WAIT_MS: number = 30_000; +export const discoveryKey: readonly ['agentSessions', 'discovery'] = [ + 'agentSessions', + 'discovery', +]; export interface ChoiceDiscovery { readonly loading: boolean; + /** Only a fresh answer for the final selection can be used to send. */ readonly result: DiscoveredChoices | undefined; + /** Previous choices in this target, for display while dependent settings load. */ + readonly displayed: DiscoveredChoices | undefined; + readonly settings: Readonly>; readonly error: Error | undefined; + readonly isCurrent: () => boolean; readonly retry: () => void; readonly choose: (id: string, value: string) => void; } -/** Each effect is a request generation, including A → B → A and retries of the same selection. */ export function useChoiceDiscovery( key: string, target: { @@ -32,92 +35,121 @@ export function useChoiceDiscovery( }, ): ChoiceDiscovery { const api = useApiClient(); - const realtime = useService(realtimeClientToken); - const [selection, setSelection] = useState<{ + const client = useQueryClient(); + const [selection, setSelection] = useState({ key, settings: EMPTY }); + const [previous, setPrevious] = useState<{ key: string; - settings: Readonly>; - }>({ key: '', settings: EMPTY }); - const [attempt, setAttempt] = useState(0); - const settings = selection.key === key ? selection.settings : EMPTY; - const [answer, setAnswer] = useState<{ - key: string; - settings: Readonly>; - attempt: number; - result?: DiscoveredChoices; - error?: Error; + result: DiscoveredChoices; }>(); - const { computerId, agentId, cwd, enabled } = target; + const settings = selection.key === key ? selection.settings : EMPTY; if (selection.key !== key) { setSelection({ key, settings: EMPTY }); - setAnswer(undefined); + setPrevious(undefined); } + const { computerId, agentId, cwd, enabled } = target; + const query = useQuery({ + queryKey: [...discoveryKey, key, settings], + enabled, + staleTime: DISCOVERY_CACHE_MS, + retry: false, + refetchOnWindowFocus: true, + queryFn: async ({ signal }) => { + const controller = new AbortController(); + const abort = (): void => controller.abort(); + signal.addEventListener('abort', abort, { once: true }); + if (signal.aborted) abort(); + let timer: ReturnType | undefined; + try { + // The explicit race also bounds a transport that fails to settle on abort. + return await Promise.race([ + api + .request<{ data: DiscoveredChoices }>({ + path: `agentSessions/computers/${encodeURIComponent(computerId)}/discoverChoices`, + method: 'POST', + json: { agentId, cwd, settings }, + signal: controller.signal, + }) + .then(({ data }) => data), + new Promise((_, reject) => { + timer = setTimeout(() => { + reject( + new Error( + 'Agent settings discovery timed out. Retry settings.', + ), + ); + controller.abort(); + }, DISCOVERY_WAIT_MS); + }), + ]); + } finally { + clearTimeout(timer); + signal.removeEventListener('abort', abort); + } + }, + }); + if (query.data && previous?.result !== query.data) + setPrevious({ key, result: query.data }); + useEffect(() => { - const invalidate = (): void => { - setSelection((current) => ({ key: current.key, settings: EMPTY })); - setAttempt((current) => current + 1); - }; - const unsubscribe = realtime.subscribe( - AGENT_SESSIONS_TOPIC, - ({ payload }) => { - if (payload.kind === 'computers') invalidate(); + if (!enabled) { + void client.cancelQueries({ queryKey: discoveryKey }); + void client.invalidateQueries({ + queryKey: discoveryKey, + refetchType: 'none', + }); + } + }, [client, enabled]); + useEffect(() => { + if (!enabled || !query.dataUpdatedAt || query.isError) return; + const timer = setTimeout( + () => { + void client.invalidateQueries({ + queryKey: [...discoveryKey, key, settings], + exact: true, + }); }, + Math.max(1, query.dataUpdatedAt + DISCOVERY_CACHE_MS - Date.now()), ); - const unsubscribeOpen = realtime.onOpen(invalidate); - return () => { - unsubscribe(); - unsubscribeOpen(); - }; - }, [realtime]); - useEffect(() => { - if (!enabled) return; - let current = true; - api - .request<{ data: DiscoveredChoices }>({ - path: `agentSessions/computers/${encodeURIComponent(computerId)}/discoverChoices`, - method: 'POST', - json: { agentId, cwd, settings }, - }) - .then( - ({ data: result }) => { - if (current) setAnswer({ key, settings, attempt, result }); - }, - (error: unknown) => { - if (current) - setAnswer({ - key, - settings, - attempt, - error: error instanceof Error ? error : new Error(String(error)), - }); - }, - ); - return () => { - current = false; - }; - }, [api, key, computerId, agentId, cwd, enabled, settings, attempt]); + return () => clearTimeout(timer); + }, [client, enabled, key, settings, query.dataUpdatedAt, query.isError]); + const fresh = - selection.key === key && - answer?.key === key && - answer.settings === settings && - answer.attempt === attempt; - const result = fresh && enabled ? answer.result : undefined; + enabled && !query.isFetching && !query.isStale && !query.isError; + const displayed = + enabled && previous?.key === key ? previous.result : undefined; return { - loading: enabled && !fresh, - result, - error: fresh && enabled ? answer.error : undefined, + loading: enabled && (query.isPending || query.isFetching), + result: fresh ? query.data : undefined, + displayed, + settings, + error: enabled && query.isError ? query.error : undefined, + isCurrent: () => { + const state = client.getQueryState([ + ...discoveryKey, + key, + settings, + ]); + return Boolean( + enabled && + state?.data && + state.data === query.data && + state.status === 'success' && + !state.isInvalidated && + state.fetchStatus === 'idle' && + Date.now() - state.dataUpdatedAt < DISCOVERY_CACHE_MS, + ); + }, retry: () => { - setSelection({ key, settings: EMPTY }); - setAttempt((value) => value + 1); + void query.refetch(); }, choose: (id, value) => { - // A different model can replace dependent choices; rediscover their defaults instead of carrying stale values. const model = - result?.choices.find((choice) => choice.id === id)?.category === + displayed?.choices.find((choice) => choice.id === id)?.category === 'model'; const kept = model ? Object.fromEntries( Object.entries(settings).filter(([choiceId]) => - result?.choices.some( + displayed?.choices.some( (choice) => choice.id === choiceId && choice.category === 'mode', ), diff --git a/packages/plugins/app-plugin-agent-sessions/docs/performance.md b/packages/plugins/app-plugin-agent-sessions/docs/performance.md new file mode 100644 index 00000000000..6d8d6061548 --- /dev/null +++ b/packages/plugins/app-plugin-agent-sessions/docs/performance.md @@ -0,0 +1,19 @@ +# Agent session latency and recovery + +Discovery returns before disposable process cleanup completes. The bridge still tracks cleanup and waits for it at shutdown. Periodic registration remains necessary for protocol upgrade negotiation. An unchanged keyed registration skips the computer write and notification; polling maintains the heartbeat and publishes recovery from offline. Registration still publishes actual computer changes, bridge restart and removal of an upgrade notice. + +Discovery results are cached for five minutes by the complete observable target (computer, agent information, directory, allowed roots and folders, home and bridge host) and selected overrides. Changing a model refreshes its dependent options, while the model label changes immediately. Draft editing remains available during discovery. Sending requires a fresh result for the final selection. Cached or previous options may be displayed while revalidating, but cannot authorize sending. Retry and reconnect preserve overrides; target changes clear them. Offline, computers notifications, reconnect and expiry invalidate discoveries. The browser waits at most 30 seconds. Aborting HTTP ends the server's reply wait when the adapter propagates the abort; it does not cancel a queued bridge command. The bridge retains its own discovery deadline and process cleanup. + +The browser API does not expose the stored bridge generation. Therefore discovery cache entries are invalidated when the last live page subscription leaves: they cannot safely survive an interval in which restart notifications were unobserved. Cross-page reuse needs a separate decision to expose the existing `bridgeStartedAt` value or another authoritative revision. Agent version alone is insufficient. Agent-local configuration files may change without any advertised version change; discovery expiry and strict settings validation before prompting remain necessary. + +The composer shows Sending while a prompt awaits confirmation, retains the draft on failure and serializes submission. A confirmed event is merged by sequence immediately. Transcript reads consume abort signals and retain confirmed events missing from older reads. Mutations never automatically retry a prompt. A lost confirmation can mean the message was persisted; the UI warns to check the conversation before retrying. Exactly-once retry is not implemented: the server allocates a new prompt ID for every submission and exposes no caller-supplied idempotency key. First-session creation has the same ambiguity. A protocol decision is required for safe automatic retry. + +One shared subscription per query client batches transcript refreshes with a fixed 250 ms deadline, session refreshes with a fixed one-second deadline and list refreshes with a fixed two-second deadline. Continuous events cannot postpone these deadlines. Slow reads finish without repeated cancellation; changes received during a read trigger a compensation read after it settles. Network and server response time remain outside the scheduling deadline. Focus and reconnect refresh the data. Visible pages also poll transcript and session every 15 seconds and lists every 30 seconds to recover from missed notifications. Hidden pages use React Query's default background-polling suppression. Linked records refresh on the same 15-second interval. + +## Optional timing diagnostics + +Enable DEBUG for the `agent-sessions` logger using the application's logging configuration. Request logs record receipt and response times, duration and event loop utilization, without bodies, headers or individual directory paths. Service logs record prompt persistence, confirmation and event publication request, with sequence and opaque prompt ID. The provider samples event loop delay only at DEBUG and stops sampling at shutdown; its reported maximum is cumulative since sampling started, not a per-request database-wait estimate. Event publication is asynchronous: publication request does not establish delivery. + +Set `NOCOBASE_ACP_DIAGNOSTICS=1` in the bridge environment for discovery receipt, readiness, reply and cleanup timings. This is off by default. Logs include only timestamps, durations and an opaque discovery request ID. + +For a slow production send, collect sanitized browser HAR timestamps for the prompt POST and transcript GET, browser render timestamps, DEBUG service logs and bridge RPC phase timestamps on synchronized clocks. Compare the health endpoint and event loop delay concurrently. Measure database connection acquisition and transaction duration through the production driver's existing tracing; neither is inferred from HTTP duration. Record deployment version, database dialect, pool size, instance count, runner count, streaming-session count and open-page count. Measure actual model first token and completion separately from persistence and realtime delivery. Local fake-agent measurements cannot establish the cause of a production ten-second delay. diff --git a/packages/plugins/app-plugin-agent-sessions/server/providers/agent-sessions.ts b/packages/plugins/app-plugin-agent-sessions/server/providers/agent-sessions.ts index 07e60a18928..bf216617771 100644 --- a/packages/plugins/app-plugin-agent-sessions/server/providers/agent-sessions.ts +++ b/packages/plugins/app-plugin-agent-sessions/server/providers/agent-sessions.ts @@ -1,3 +1,4 @@ +import { monitorEventLoopDelay, type IntervalHistogram } from 'node:perf_hooks'; import { authenticationToken } from '@nocobase/app-plugin-authentication'; import { serverFileRepositoryManagerToken } from '@nocobase/app-plugin-file/server'; import { cachingToken } from '@nocobase/app-server/caching'; @@ -9,6 +10,7 @@ import { } from '@nocobase/app-server/realtime'; import { apiDocsToken } from '@nocobase/app-server/router'; import { secretsServiceToken } from '@nocobase/app-server/secrets'; +import { loggingToken } from '@nocobase/app-server/logging'; import { databaseManagerToken } from '@nocobase/db'; import { ServiceProvider } from '@nocobase/service-provider'; @@ -27,6 +29,7 @@ export class AgentSessionsProvider extends ServiceProvider #topic: RealtimeUserTopic | undefined; readonly #releases: (() => void)[] = []; + #loopDelay: IntervalHistogram | undefined; public override register(): void { this.app.container.singleton(agentSessionsToken, (container) => { @@ -41,7 +44,27 @@ export class AgentSessionsProvider extends ServiceProvider const secrets = container.has(secretsServiceToken) ? container.resolve(secretsServiceToken) : undefined; + const logger = container.has(loggingToken) + ? container.resolve(loggingToken).getLogger('agent-sessions') + : undefined; return new AgentSessionsService({ + ...(logger?.isLevelEnabled('debug') + ? { + diagnose: (phase, fields) => { + if (logger.isLevelEnabled('debug')) + logger.debug( + { + phase, + ...fields, + eventLoopDelayMaxMs: this.#loopDelay + ? this.#loopDelay.max / 1e6 + : undefined, + }, + 'Agent sessions timing', + ); + }, + } + : {}), database: container.resolve(databaseManagerToken), // Without a secrets key there is nothing to sign tool tokens with, and sessions get no tools. ...(secrets?.ready @@ -84,6 +107,17 @@ export class AgentSessionsProvider extends ServiceProvider } /** Teaches the application's authentication and its API document the computers' keys. */ + public override start(): Promise { + const logger = this.app.container.has(loggingToken) + ? this.app.container.resolve(loggingToken).getLogger('agent-sessions') + : undefined; + if (logger?.isLevelEnabled('debug')) { + this.#loopDelay = monitorEventLoopDelay({ resolution: 20 }); + this.#loopDelay.enable(); + } + return Promise.resolve(); + } + public override boot(): Promise { const { container } = this.app; const service = container.resolve(agentSessionsToken); @@ -101,6 +135,7 @@ export class AgentSessionsProvider extends ServiceProvider } public override shutdown(): Promise { + this.#loopDelay?.disable(); for (const release of this.#releases.splice(0)) release(); this.#topic?.close(); this.#topic = undefined; diff --git a/packages/plugins/app-plugin-agent-sessions/server/routes/index.ts b/packages/plugins/app-plugin-agent-sessions/server/routes/index.ts index 4a3c04c781d..b7484b816e4 100644 --- a/packages/plugins/app-plugin-agent-sessions/server/routes/index.ts +++ b/packages/plugins/app-plugin-agent-sessions/server/routes/index.ts @@ -2,12 +2,15 @@ import { authenticationToken } from '@nocobase/app-plugin-authentication'; import type { AppPluginApplication } from '@nocobase/app-server/plugins'; import { defineApiRoutes, + getRequestId, defineRootRoutes, type AppApiRouteContribution, type AppRootRouteContribution, type AppRouteContribution, } from '@nocobase/app-server/router'; import { Hono, type Context } from 'hono'; +import { performance } from 'node:perf_hooks'; +import { loggingToken } from '@nocobase/app-server/logging'; import { agentSessionsErrorHandler } from '../errors.js'; import { agentSessionsToken } from '../tokens.js'; @@ -27,6 +30,38 @@ export const apiRoutes: AppApiRouteContribution = router.onError(agentSessionsErrorHandler); const bridge = (context: Context): boolean => context.req.path.includes('/agentSessions/bridge/'); + const logger = container.has(loggingToken) + ? container.resolve(loggingToken).getLogger('agent-sessions') + : undefined; + router.use('/agentSessions/*', async (context: Context, next) => { + if (!logger?.isLevelEnabled('debug')) return next(); + const candidate = getRequestId(context); + const requestId = /^[0-9a-f-]{36}$/iu.test(candidate) ? candidate : undefined; + const start = performance.now(); + const utilization = performance.eventLoopUtilization(); + logger.debug( + { phase: 'received', requestId, at: Date.now(), method: context.req.method }, + 'Agent sessions timing', + ); + try { + await next(); + } finally { + logger.debug( + { + phase: 'response', + requestId, + at: Date.now(), + durationMs: performance.now() - start, + method: context.req.method, + route: context.req.routePath, + status: context.res.status, + eventLoopUtilization: + performance.eventLoopUtilization(utilization).utilization, + }, + 'Agent sessions timing', + ); + } + }); router.use('/agentSessions/*', auth.required({ skip: bridge })); router.use('/agentSessions/bridge/*', auth.required({ scopedKeys: true })); router.route('/agentSessions/bridge', bridgeRoutes(service)); diff --git a/packages/plugins/app-plugin-agent-sessions/server/routes/sessions.ts b/packages/plugins/app-plugin-agent-sessions/server/routes/sessions.ts index 38948f0613c..ed60d798dda 100644 --- a/packages/plugins/app-plugin-agent-sessions/server/routes/sessions.ts +++ b/packages/plugins/app-plugin-agent-sessions/server/routes/sessions.ts @@ -90,6 +90,7 @@ export function sessionRoutes(service: AgentSessionsService): Hono { userId(context), context.req.valid('param').computerId, context.req.valid('json'), + context.req.raw.signal, ), }), ); diff --git a/packages/plugins/app-plugin-agent-sessions/server/service.ts b/packages/plugins/app-plugin-agent-sessions/server/service.ts index 1df2e884417..f89544e0370 100644 --- a/packages/plugins/app-plugin-agent-sessions/server/service.ts +++ b/packages/plugins/app-plugin-agent-sessions/server/service.ts @@ -149,6 +149,11 @@ export interface AgentSessionsServiceOptions { /** Runs `task` after `delayMs`, as owners are told of what they did not see (`notices.ts`); a timer by default. */ readonly schedule?: (task: () => Promise, delayMs: number) => void; readonly now?: () => number; + /** Optional phase timing; excludes message bodies, credentials and user metadata. */ + readonly diagnose?: ( + phase: string, + fields: Readonly>, + ) => void; } export interface SessionPage { @@ -168,6 +173,7 @@ export class AgentSessionsService { readonly #database: DatabaseManager; readonly #publish: (userId: string, change: AgentSessionsChange) => void; readonly #now: () => number; + readonly #diagnose: AgentSessionsServiceOptions['diagnose']; readonly #waiters = new Map void>>(); readonly #linkKinds = new Map(); readonly #tools: SessionTool[] = []; @@ -189,6 +195,7 @@ export class AgentSessionsService { this.#schedule = options.schedule ?? later; this.#publish = options.publish ?? (() => undefined); this.#now = options.now ?? Date.now; + this.#diagnose = options.diagnose; this.#tools.push(this.#workTool()); } @@ -425,7 +432,9 @@ export class AgentSessionsService { readonly cwd: string; readonly settings: Readonly>; }, + signal?: AbortSignal, ): Promise { + signal?.throwIfAborted(); const computer = await this.#computer(userId, computerId); if (!isOnline(computer, this.#now())) throw new AgentSessionsError('computerOffline'); @@ -458,6 +467,7 @@ export class AgentSessionsService { const reply = await this.#awaitReply( `${computer.id}:${requestId}`, DISCOVERY_TIMEOUT_MS + 10_000, + signal, ); if (!reply.discovery || reply.error) throw new AgentSessionsError( @@ -522,12 +532,22 @@ export class AgentSessionsService { : undefined; if (keyed && !known) throw new AgentSessionsError('computerNotFound'); const computerId = known?.id ?? randomUUID(); - if (known) + const restarted = known?.bridgeStartedAt !== request.bridgeStartedAt; + const viewChanged = + !known || + restarted || + !isOnline(known, this.#now()) || + Object.entries(request.computer).some( + ([field, value]) => + JSON.stringify(known[field as keyof typeof known]) !== + JSON.stringify(value), + ); + if (known && (viewChanged || !keyed)) await computers(this.#conn()).updateMany({ filter: { id: computerId }, values, }); - else + else if (!known) await computers(this.#conn()).createOne({ values: { id: computerId, @@ -538,7 +558,6 @@ export class AgentSessionsService { }, }); - const restarted = known?.bridgeStartedAt !== request.bridgeStartedAt; const interrupted = !restarted ? [] : await sessions(this.#conn()).findMany({ @@ -569,8 +588,8 @@ export class AgentSessionsService { }); if (changed) this.#changed(changed); } - await this.#replies.take(outdatedKey(computerId)); - this.#publish(userId, { kind: 'computers' }); + const outdated = await this.#replies.take(outdatedKey(computerId)); + if (viewChanged || outdated) this.#publish(userId, { kind: 'computers' }); return { computerId, ...(key ? { key } : {}) }; } @@ -1017,8 +1036,20 @@ export class AgentSessionsService { ); return result; }); + this.#diagnose?.('promptPersisted', { + sessionId, + at: Date.now(), + seq: changed.lastSeq, + promptId, + }); this.#wake(session.computerId); this.#changed(changed); + this.#diagnose?.('promptConfirmed', { + sessionId, + at: Date.now(), + seq: changed.lastSeq, + promptId, + }); return { kind: 'prompt', at: changed.session.lastActivityAt, @@ -1173,17 +1204,25 @@ export class AgentSessionsService { async #awaitReply( key: string, timeout: number = REPLY_WAIT_MS, + signal?: AbortSignal, ): Promise { - for (let waited = 0; waited < timeout; waited += REPLY_CHECK_MS) { + const deadline = Date.now() + timeout; + while (Date.now() < deadline) { + signal?.throwIfAborted(); const reply = await this.#replies.take(key); if (reply) return reply; await new Promise((resolve) => { - const timer = setTimeout(done, REPLY_CHECK_MS); + const timer = setTimeout( + done, + Math.min(REPLY_CHECK_MS, deadline - Date.now()), + ); + signal?.addEventListener('abort', done, { once: true }); const waiters = this.#replyWaiters; waiters.set(key, done); function done(): void { clearTimeout(timer); waiters.delete(key); + signal?.removeEventListener('abort', done); resolve(); } }); @@ -1526,6 +1565,11 @@ export class AgentSessionsService { readonly session: SessionRecord; readonly lastSeq: number; }): void { + this.#diagnose?.('eventPublishRequested', { + sessionId: changed.session.id, + at: Date.now(), + seq: changed.lastSeq, + }); this.#publish(changed.session.userId, { kind: 'session', sessionId: changed.session.id, diff --git a/packages/plugins/app-plugin-agent-sessions/tests/client/api.test.ts b/packages/plugins/app-plugin-agent-sessions/tests/client/api.test.ts new file mode 100644 index 00000000000..f5eb16cfd92 --- /dev/null +++ b/packages/plugins/app-plugin-agent-sessions/tests/client/api.test.ts @@ -0,0 +1,216 @@ +import { answerApi, renderWithApp } from '@nocobase/app-testing/client'; +import { QueryClient, QueryClientProvider } from '@tanstack/react-query'; +import { act, fireEvent, screen, waitFor } from '@testing-library/react'; +import { createElement } from 'react'; +import plugin from '../../client/plugin.js'; +import { realtimeClientToken, type RealtimeClient } from '@nocobase/app-client'; +import { useLiveSessions, usePrompt, useTranscript } from '../../client/api.js'; +import { describe, expect, it, vi } from 'vitest'; +import { mergeTranscript } from '../../client/api.js'; +import type { TranscriptEvent } from '../../shared/api.js'; +const prompt = (seq: number): TranscriptEvent => ({ + kind: 'prompt', + seq, + at: '2026-10-14T12:00:00Z', + promptId: `p${seq}`, + content: [{ type: 'text', text: 'test' }], +}); +describe('confirmed transcript events', () => { + it('retains a confirmation absent from an older GET, merges repeated seq and sorts out of order events', () => { + expect( + mergeTranscript([prompt(3), prompt(1)], [prompt(1), prompt(2)]).map( + (event) => event.seq, + ), + ).toEqual([1, 2, 3]); + expect(mergeTranscript([prompt(1)], [prompt(1)])).toHaveLength(1); + }); +}); + +function ConversationProbe() { + const events = useTranscript('test'); + const send = usePrompt('test'); + return createElement( + 'section', + null, + createElement( + 'button', + { + onClick: () => + send.mutate({ + content: [{ type: 'text', text: 'test' }], + mentions: [], + }), + }, + 'Send', + ), + createElement( + 'output', + null, + (events.data ?? []).map((event) => event.seq).join(','), + ), + ); +} +it('shows a confirmation while GET is still pending and retains it when that older GET finishes', async () => { + let finishRead: ((answer: unknown) => void) | undefined; + await renderWithApp( + createElement( + QueryClientProvider, + { client: new QueryClient() }, + createElement(ConversationProbe), + ), + { + plugins: [plugin({ routes: false })], + namespace: '@nocobase/app-plugin-agent-sessions', + fetch: answerApi(({ method }) => + method === 'GET' + ? new Promise((resolve) => { + finishRead = resolve; + }) + : { data: prompt(1) }, + ), + }, + ); + await waitFor(() => expect(finishRead).toBeDefined()); + fireEvent.click(screen.getByText('Send')); + await screen.findByText('1'); + await act(async () => { + finishRead?.({ data: [], meta: { hasMore: false } }); + }); + expect(screen.getByText('1')).toBeInTheDocument(); +}); + +function LiveProbe() { + useLiveSessions(); + return null; +} +it('shares nested subscriptions and invalidates inactive discovery results after reconnect', async () => { + const client = new QueryClient(); + const opened = new Set<() => void>(); + let subscriptions = 0; + const realtime: RealtimeClient = { + connected: true, + subscribe: () => { + subscriptions += 1; + return () => { + subscriptions -= 1; + }; + }, + onOpen: (listener) => { + opened.add(listener); + return () => { + opened.delete(listener); + }; + }, + onError: () => () => undefined, + reconnect: () => undefined, + close: () => undefined, + }; + await renderWithApp( + createElement( + QueryClientProvider, + { client }, + createElement( + 'section', + null, + createElement(LiveProbe), + createElement(LiveProbe), + ), + ), + { + plugins: [plugin({ routes: false })], + namespace: '@nocobase/app-plugin-agent-sessions', + services: (app) => { + const actual = app.container.resolve(realtimeClientToken); + vi.spyOn(actual, 'subscribe').mockImplementation(realtime.subscribe); + vi.spyOn(actual, 'onOpen').mockImplementation(realtime.onOpen); + }, + fetch: answerApi(() => ({ data: [] })), + }, + ); + expect(subscriptions).toBe(1); + expect(opened.size).toBe(1); + const key = ['agentSessions', 'discovery', 'inactive']; + client.setQueryData(key, { status: 'available', choices: [] }); + expect(client.getQueryState(key)?.isInvalidated).toBe(false); + await act(async () => { + for (const listener of opened) listener(); + }); + expect(client.getQueryState(key)?.isInvalidated).toBe(true); +}); + +function StreamingProbe() { + useLiveSessions(); + const events = useTranscript('test'); + return createElement( + 'output', + null, + (events.data ?? []).map((event) => event.seq).join(','), + ); +} +it('does not cancel slow reads during a continuous stream and follows them with a compensation read', async () => { + const client = new QueryClient(); + client.setQueryData(['agentSessions', 'transcript', 'test'], [prompt(1)]); + let emit: (() => void) | undefined; + const releases: ((answer: unknown) => void)[] = []; + let allowFresh = false; + const signals: AbortSignal[] = []; + let reads = 0; + const response = answerApi(() => { + reads += 1; + return !allowFresh + ? new Promise((resolve) => { + releases.push(resolve); + }) + : { data: [prompt(2)], meta: { hasMore: false } }; + }); + await renderWithApp( + createElement( + QueryClientProvider, + { client }, + createElement(StreamingProbe), + ), + { + plugins: [plugin({ routes: false })], + namespace: '@nocobase/app-plugin-agent-sessions', + services: (app) => { + const actual = app.container.resolve(realtimeClientToken); + vi.spyOn(actual, 'subscribe').mockImplementation((topic, listener) => { + emit = () => + listener({ + type: 'event', + topic, + publishedAt: '2026-10-14T12:00:00Z', + payload: { + kind: 'session', + sessionId: 'test', + lastSeq: 2, + } as never, + }); + return () => undefined; + }); + }, + fetch: (request) => { + signals.push(request.signal); + return response(request); + }, + }, + ); + await waitFor(() => expect(releases.length).toBeGreaterThan(0)); + const signal = signals.at(-1)!; + const interval = setInterval(() => emit?.(), 50); + try { + await act(async () => { + await new Promise((resolve) => setTimeout(resolve, 450)); + }); + expect(signal.aborted).toBe(false); + await act(async () => { + allowFresh = true; + for (const release of releases) + release({ data: [prompt(1)], meta: { hasMore: false } }); + }); + await screen.findByText('1,2'); + expect(reads).toBeGreaterThan(1); + } finally { + clearInterval(interval); + } +}); diff --git a/packages/plugins/app-plugin-agent-sessions/tests/client/choice-discovery.test.tsx b/packages/plugins/app-plugin-agent-sessions/tests/client/choice-discovery.test.tsx index f28e88a330a..4be638a1a50 100644 --- a/packages/plugins/app-plugin-agent-sessions/tests/client/choice-discovery.test.tsx +++ b/packages/plugins/app-plugin-agent-sessions/tests/client/choice-discovery.test.tsx @@ -5,7 +5,8 @@ import { } from '@nocobase/app-testing/client'; import { act, fireEvent, screen, waitFor } from '@testing-library/react'; import { useState, type ReactElement } from 'react'; -import { describe, expect, it } from 'vitest'; +import { QueryClient, QueryClientProvider } from '@tanstack/react-query'; +import { describe, expect, it, vi } from 'vitest'; import plugin from '../../client/plugin.js'; import { ChoiceMenus } from '../../client/components/choice-menus.js'; @@ -33,8 +34,9 @@ function Probe(): ReactElement { const [agentId, setAgentId] = useState('codex'); const [cwd, setCwd] = useState('/work'); const [online, setOnline] = useState(true); + const [sent, setSent] = useState(''); const discovery = useChoiceDiscovery( - JSON.stringify([computerId, agentId, cwd, online]), + JSON.stringify([computerId, agentId, cwd]), { computerId, agentId, cwd, enabled: online }, ); return ( @@ -52,7 +54,13 @@ function Probe(): ReactElement { - + + {sent} {online ? discovery.loading @@ -63,7 +71,9 @@ function Probe(): ReactElement { : 'Offline'} @@ -72,26 +82,72 @@ function Probe(): ReactElement { async function setup() { const requests: { call: ApiCall; resolve: (result: unknown) => void }[] = []; - await renderWithApp(, { - namespace: '@nocobase/app-plugin-agent-sessions', - plugins: [plugin({ routes: false })], - fetch: answerApi( - (call) => - new Promise((resolve) => { - requests.push({ call, resolve }); - }), - ), + const client = new QueryClient({ + defaultOptions: { queries: { retry: false } }, }); + await renderWithApp( + + + , + { + namespace: '@nocobase/app-plugin-agent-sessions', + plugins: [plugin({ routes: false })], + fetch: answerApi( + (call) => + new Promise((resolve) => { + requests.push({ call, resolve }); + }), + ), + }, + ); await waitFor(() => expect(requests).toHaveLength(1)); const answer = async (index: number, result: unknown): Promise => { await act(async () => { requests[index]!.resolve(result); }); }; - return { requests, answer }; + return { requests, answer, client }; } describe('choice discovery', () => { + it('checks expiration again at submission even before a stale timer rerenders', async () => { + const { answer } = await setup(); + await answer(0, { data: choice('First') }); + await screen.findByText('First'); + const clock = vi + .spyOn(Date, 'now') + .mockReturnValue(Date.now() + 6 * 60_000); + try { + fireEvent.click(screen.getByRole('button', { name: 'Send' })); + expect(screen.getByText('Blocked')).toBeInTheDocument(); + } finally { + clock.mockRestore(); + } + }); + + it('reuses a valid cached target and disables sending during revalidation', async () => { + const { requests, answer, client } = await setup(); + await answer(0, { data: choice('First') }); + await screen.findByText('First'); + fireEvent.click(screen.getByText('Computer')); + await waitFor(() => expect(requests).toHaveLength(2)); + await answer(1, { data: choice('Second') }); + await screen.findByText('Second'); + fireEvent.click(screen.getByText('Computer')); + await screen.findByText('First'); + expect(requests).toHaveLength(2); + expect(screen.getByRole('button', { name: 'Send' })).not.toBeDisabled(); + await act(async () => { + void client.invalidateQueries({ + queryKey: ['agentSessions', 'discovery'], + }); + }); + await waitFor(() => expect(requests).toHaveLength(3)); + expect(screen.getByRole('button', { name: 'Send' })).toBeDisabled(); + await answer(2, { data: choice('Refreshed') }); + await screen.findByText('Refreshed'); + }); + it('ignores delayed responses across A → B → A and clears settings on computer, agent and folder changes', async () => { const { requests, answer } = await setup(); expect(screen.getByRole('button', { name: 'Send' })).toBeDisabled(); @@ -146,6 +202,7 @@ describe('choice discovery', () => { fireEvent.click(screen.getByText('Retry')); await waitFor(() => expect(requests).toHaveLength(2)); await answer(1, { data: { status: 'available', choices: [] } }); + await screen.findByText('available'); expect(screen.getByRole('button', { name: 'Send' })).not.toBeDisabled(); fireEvent.click(screen.getByText('Connection')); expect(screen.getByText('Offline')).toBeInTheDocument(); diff --git a/packages/plugins/app-plugin-agent-sessions/tests/client/new-session.test.tsx b/packages/plugins/app-plugin-agent-sessions/tests/client/new-session.test.tsx index 563843d65c7..26d15f08f2d 100644 --- a/packages/plugins/app-plugin-agent-sessions/tests/client/new-session.test.tsx +++ b/packages/plugins/app-plugin-agent-sessions/tests/client/new-session.test.tsx @@ -184,7 +184,8 @@ describe('new-session composer', () => { expect( screen.getByRole('button', { name: 'Fresh computer' }), ).toBeInTheDocument(); - expect(screen.getByRole('textbox')).toBeDisabled(); + expect(screen.getByRole('textbox')).not.toBeDisabled(); + expect(screen.getByRole('button', { name: 'Send' })).toBeDisabled(); expect(screen.queryByText('Initial model')).toBeNull(); expect(requests).toEqual(['agentSessions/computers/fresh/discoverChoices']); fireEvent.click(screen.getByRole('button', { name: 'Fresh computer' })); @@ -232,7 +233,8 @@ describe('new-session composer', () => { }, ); await screen.findByText('Loading settings…'); - expect(screen.getByRole('textbox')).toBeDisabled(); + expect(screen.getByRole('textbox')).not.toBeDisabled(); + expect(screen.getByRole('button', { name: 'Send' })).toBeDisabled(); await waitFor(() => expect(discovered).toHaveLength(1)); await act(async () => discovered[0]!.resolve({ data: choices })); await screen.findByText('Reasoning: High'); @@ -246,7 +248,8 @@ describe('new-session composer', () => { cwd: '/work', settings: { model: 'chosen' }, }); - expect(screen.getByRole('textbox')).toBeDisabled(); + expect(screen.getByRole('textbox')).not.toBeDisabled(); + expect(screen.getByRole('button', { name: 'Send' })).toBeDisabled(); await act(async () => discovered[1]!.resolve({ data: { diff --git a/packages/plugins/app-plugin-agent-sessions/tests/server/service.test.ts b/packages/plugins/app-plugin-agent-sessions/tests/server/service.test.ts index fb1b0af1540..6b0c1ca9101 100644 --- a/packages/plugins/app-plugin-agent-sessions/tests/server/service.test.ts +++ b/packages/plugins/app-plugin-agent-sessions/tests/server/service.test.ts @@ -35,30 +35,38 @@ function setup(database: DatabaseManager) { now: () => clock.now, publish: (userId, change) => changes.push([userId, change]), }); - const register = (userId = 'u1', overrides: Partial = {}) => - service.register(userId, { - protocol: BRIDGE_PROTOCOL_VERSION, - computerId: null, - bridgeStartedAt: 'bridge-1', - computer: { - name: 'Laptop', - platform: 'darwin', - arch: 'arm64', - host: 'nb-studio 1.0.0', - home: '/home/zhou', - roots: ['/home/zhou'], - agents: [ - { id: 'codex', name: 'Codex', version: '2.2.2' }, - { - id: 'opencode', - name: 'OpenCode', - version: null, - problem: 'notInstalled', - }, - ], + const register = ( + userId = 'u1', + overrides: Partial = {}, + keyed = false, + ) => + service.register( + userId, + { + protocol: BRIDGE_PROTOCOL_VERSION, + computerId: null, + bridgeStartedAt: 'bridge-1', + computer: { + name: 'Laptop', + platform: 'darwin', + arch: 'arm64', + host: 'nb-studio 1.0.0', + home: '/home/zhou', + roots: ['/home/zhou'], + agents: [ + { id: 'codex', name: 'Codex', version: '2.2.2' }, + { + id: 'opencode', + name: 'OpenCode', + version: null, + problem: 'notInstalled', + }, + ], + }, + ...overrides, }, - ...overrides, - }); + keyed, + ); let batch = 0; const report = ( computerId: string, @@ -289,6 +297,51 @@ describe('AgentSessionsService', () => { }); } + test('publishes registration changes and recovery, but not an unchanged scan', async ({ + database, + }) => { + const { service, register, clock, changes } = setup(database); + const { computerId } = await register(); + changes.length = 0; + await register('u1', { computerId }, true); + expect(changes).toEqual([]); + await register('u1', { computerId, bridgeStartedAt: 'bridge-2' }); + expect( + changes.filter(([, change]) => change.kind === 'computers'), + ).toHaveLength(1); + changes.length = 0; + clock.now += 120_000; + await register('u1', { computerId, bridgeStartedAt: 'bridge-2' }); + expect( + changes.filter(([, change]) => change.kind === 'computers'), + ).toHaveLength(1); + expect((await service.listComputers('u1'))[0]?.online).toBe(true); + }); + + test('stops waiting after discovery cancellation without assuming the queued command was cancelled', async ({ + database, + }) => { + const { service, register, take } = setup(database); + const { computerId } = await register(); + const stop = new AbortController(); + const pending = service.discoverChoices( + 'u1', + computerId, + { agentId: 'codex', cwd: '/home/zhou', settings: {} }, + stop.signal, + ); + const rejected = pending.catch((error: unknown) => error); + let queued: + import('@nocobase/acp-bridge/protocol').BridgeCommand | undefined; + await vi.waitFor(async () => { + [queued] = await take(computerId); + expect(queued?.type).toBe('discoverChoices'); + }); + stop.abort(); + expect(await rejected).toMatchObject({ name: 'AbortError' }); + expect(queued?.type).toBe('discoverChoices'); + }); + test('discovers a fresh computer without a Studio session and carries settings to the first prompt', async ({ database, }) => {