mirror of
https://github.com/danny-avila/LibreChat.git
synced 2026-08-28 04:37:37 +00:00
📡 feat: Stream Detached Subagent Activity (#15111)
* feat: stream detached subagent activity * fix: annotate activity stream limits * fix: isolate subagent activity imports * fix: harden detached subagent activity lifecycle * test: cover synchronous activity transport failure * test: include required subagent activity identity * fix: identify and reconnect subagent activity events * fix: bound subagent activity lifecycles * fix: close subagent activity handoff races * fix: bind and synchronize activity subscriptions * fix: detect fresh activity attachment * fix: complete activity synchronization handoff * fix: bind activity sync and failure circuits * fix: expose subscription-bound synchronization * fix: fence activity reconnect publications * test: make detached timeout settlement deterministic * fix: fence Redis activity attachments * fix: close failed activity streams * perf: reuse fenced activity frontier * style: sort subagent thread imports * fix: preserve queued subagent activity * test: type activity publication counter * fix: disconnect subagent activity subscriber * fix: close background activity lifecycle gaps * fix: preserve streamed activity spacing * fix: preserve bounded live subagent activity * fix: merge durable subagent activity safely * fix: model detached activity coverage * fix: type detached activity inputs * fix: order overlapping subagent activity * chore: sort activity test imports * fix: buffer subagent activity handoff gaps * fix: flush activity after parent close * fix: advance closed activity suffixes * fix: preserve detached activity ordering * fix: close detached activity delivery races * fix: bound shared Redis subscriber readiness * fix: expire shared Redis subscription readiness * fix: clean up late Redis subscriptions * fix: preserve late Redis subscription fallback
This commit is contained in:
Danny Avila
GitHub
parent
3ebef4c84e
commit
d3e70159ca
33 changed files with 4125 additions and 110 deletions
|
|
@ -13,12 +13,23 @@ import { initSubagentAggregatorState, initSubagentTickerState } from '~/utils/su
|
|||
import SubagentThreadPanel from './SubagentThreadPanel';
|
||||
|
||||
const mockUseSubagentThreadQuery = jest.fn();
|
||||
const mockUseSubagentActivityStream = jest.fn();
|
||||
const mockApprovalProviderMounted = jest.fn();
|
||||
const mockApprovalProviderUnmounted = jest.fn();
|
||||
let mockIsMobile = false;
|
||||
|
||||
jest.mock('~/data-provider', () => ({
|
||||
useSubagentThreadQuery: (...args: unknown[]) => mockUseSubagentThreadQuery(...args),
|
||||
subagentThreadHasTaskEvidence: (view: SubagentThreadView | undefined, taskId: string): boolean =>
|
||||
view?.messages.some(
|
||||
(message) =>
|
||||
message.messageId === `${taskId}:user` || message.messageId === `${taskId}:assistant`,
|
||||
) === true,
|
||||
}));
|
||||
|
||||
jest.mock('~/data-provider/Subagents/useSubagentActivityStream', () => ({
|
||||
__esModule: true,
|
||||
default: (...args: unknown[]) => mockUseSubagentActivityStream(...args),
|
||||
}));
|
||||
|
||||
jest.mock('~/hooks', () => ({
|
||||
|
|
@ -156,6 +167,7 @@ describe('SubagentThreadPanel', () => {
|
|||
'child-thread',
|
||||
'task',
|
||||
);
|
||||
expect(mockUseSubagentActivityStream).toHaveBeenCalledWith(selection, false);
|
||||
expect(screen.getByText('Research child')).toBeInTheDocument();
|
||||
expect(screen.getByText('Investigate the release.')).toBeInTheDocument();
|
||||
expect(screen.getByText('The release is ready.')).toBeInTheDocument();
|
||||
|
|
@ -205,6 +217,46 @@ describe('SubagentThreadPanel', () => {
|
|||
expect(screen.getByTestId('shared-activity')).toHaveAttribute('data-state', 'ready');
|
||||
});
|
||||
|
||||
it('renders newer detached progress instead of a dispatch-time parent snapshot', () => {
|
||||
mockUseSubagentThreadQuery.mockReturnValue({
|
||||
data: { ...completedView, status: 'running', activity: [] },
|
||||
isLoading: false,
|
||||
isError: false,
|
||||
isReadinessPending: false,
|
||||
});
|
||||
const progressKey = subagentProgressKey(
|
||||
selection.parentMessageId,
|
||||
selection.toolCallId,
|
||||
selection.partIndex,
|
||||
);
|
||||
const detachedSelection: ActiveSubagentPanel = {
|
||||
...selection,
|
||||
persistedContent: [
|
||||
{ type: ContentTypes.TEXT, text: 'Dispatch-time snapshot.' },
|
||||
] as TMessageContentParts[],
|
||||
};
|
||||
|
||||
render(
|
||||
<RecoilRoot
|
||||
initializeState={({ set }) =>
|
||||
set(subagentProgressByToolCallId(progressKey), {
|
||||
subagentRunId: 'child-run',
|
||||
subagentType: 'researcher',
|
||||
status: 'message_delta',
|
||||
contentParts: [{ type: ContentTypes.TEXT, text: 'latest detached text.' }],
|
||||
aggregatorState: initSubagentAggregatorState(),
|
||||
tickerState: initSubagentTickerState(),
|
||||
coverage: 'suffix',
|
||||
})
|
||||
}
|
||||
>
|
||||
<SubagentThreadPanel selection={detachedSelection} />
|
||||
</RecoilRoot>,
|
||||
);
|
||||
|
||||
expect(screen.getByText('Dispatch-time snapshot.latest detached text.')).toBeInTheDocument();
|
||||
});
|
||||
|
||||
it('resets invocation-scoped approval state when the selected card changes', () => {
|
||||
mockUseSubagentThreadQuery.mockReturnValue({
|
||||
data: undefined,
|
||||
|
|
@ -249,6 +301,57 @@ describe('SubagentThreadPanel', () => {
|
|||
|
||||
expect(screen.getByTestId('shared-activity')).toHaveAttribute('data-state', 'loading');
|
||||
expect(screen.getByTestId('shared-activity')).toHaveAttribute('data-status', 'dispatched');
|
||||
expect(mockUseSubagentActivityStream).toHaveBeenLastCalledWith(selection, true);
|
||||
});
|
||||
|
||||
it('opens live activity before the durable child becomes addressable', () => {
|
||||
mockUseSubagentThreadQuery.mockReturnValue({
|
||||
data: undefined,
|
||||
isLoading: false,
|
||||
isError: true,
|
||||
isReadinessPending: true,
|
||||
});
|
||||
|
||||
const { rerender } = render(
|
||||
<RecoilRoot>
|
||||
<SubagentThreadPanel selection={selection} />
|
||||
</RecoilRoot>,
|
||||
);
|
||||
expect(mockUseSubagentActivityStream).toHaveBeenLastCalledWith(selection, true);
|
||||
|
||||
mockUseSubagentThreadQuery.mockReturnValue({
|
||||
data: { ...completedView, status: 'running' },
|
||||
isLoading: false,
|
||||
isError: false,
|
||||
isReadinessPending: false,
|
||||
});
|
||||
rerender(
|
||||
<RecoilRoot>
|
||||
<SubagentThreadPanel selection={selection} />
|
||||
</RecoilRoot>,
|
||||
);
|
||||
|
||||
expect(mockUseSubagentActivityStream).toHaveBeenLastCalledWith(selection, true);
|
||||
});
|
||||
|
||||
it('keeps streaming when terminal thread state belongs to an older task', () => {
|
||||
mockUseSubagentThreadQuery.mockReturnValue({
|
||||
data: {
|
||||
...completedView,
|
||||
messages: [{ ...completedView.messages[1], messageId: 'older-task:assistant' }],
|
||||
},
|
||||
isLoading: false,
|
||||
isError: false,
|
||||
isReadinessPending: false,
|
||||
});
|
||||
|
||||
render(
|
||||
<RecoilRoot>
|
||||
<SubagentThreadPanel selection={selection} />
|
||||
</RecoilRoot>,
|
||||
);
|
||||
|
||||
expect(mockUseSubagentActivityStream).toHaveBeenLastCalledWith(selection, true);
|
||||
});
|
||||
|
||||
it('surfaces a durable read failure after the readiness window', () => {
|
||||
|
|
|
|||
|
|
@ -8,9 +8,10 @@ import {
|
|||
subagentProgressByToolCallId,
|
||||
subagentProgressKey,
|
||||
} from '~/store/subagents';
|
||||
import useSubagentActivityStream from '~/data-provider/Subagents/useSubagentActivityStream';
|
||||
import { subagentThreadHasTaskEvidence, useSubagentThreadQuery } from '~/data-provider';
|
||||
import { adaptDurableThreadActivity, adaptLivePersistedActivity } from './adapters';
|
||||
import ApprovalProvider from '~/components/Chat/Messages/Content/ApprovalContext';
|
||||
import { useSubagentThreadQuery } from '~/data-provider';
|
||||
import { useFocusTrap, useLocalize } from '~/hooks';
|
||||
import SubagentActivity from './SubagentActivity';
|
||||
|
||||
|
|
@ -35,6 +36,13 @@ export default function SubagentThreadPanel({ selection }: { selection: ActiveSu
|
|||
threadId,
|
||||
taskId,
|
||||
);
|
||||
const durableTerminal =
|
||||
subagentThreadHasTaskEvidence(data, taskId) &&
|
||||
(data?.status === 'completed' ||
|
||||
data?.status === 'failed' ||
|
||||
data?.status === 'interrupted' ||
|
||||
data?.status === 'cancelled');
|
||||
useSubagentActivityStream(selection, !durableTerminal);
|
||||
const detachedLiveSubmitting =
|
||||
selection.durable != null &&
|
||||
progress != null &&
|
||||
|
|
@ -73,6 +81,7 @@ export default function SubagentThreadPanel({ selection }: { selection: ActiveSu
|
|||
prompt: selection.prompt,
|
||||
progress,
|
||||
persistedContent: selection.persistedContent,
|
||||
isDetached: selection.durable != null,
|
||||
legacyOutput: selection.legacyOutput,
|
||||
// A detached parent tool step closes as soon as dispatch succeeds;
|
||||
// its terminal status does not describe the still-running child.
|
||||
|
|
|
|||
|
|
@ -45,6 +45,123 @@ describe('child activity adapters', () => {
|
|||
]);
|
||||
});
|
||||
|
||||
it('merges a forward-only detached suffix with the partial parent snapshot', () => {
|
||||
const activity = adaptLivePersistedActivity({
|
||||
title: 'researcher',
|
||||
progress: {
|
||||
subagentRunId: 'run',
|
||||
subagentType: 'researcher',
|
||||
status: 'message_delta',
|
||||
contentParts: [{ type: ContentTypes.TEXT, text: 'latest detached text.' }],
|
||||
aggregatorState: initSubagentAggregatorState(),
|
||||
tickerState: initSubagentTickerState(),
|
||||
coverage: 'suffix',
|
||||
},
|
||||
persistedContent: [
|
||||
{ type: ContentTypes.TEXT, text: 'Dispatch-time snapshot; ' },
|
||||
] as TMessageContentParts[],
|
||||
initialProgress: 1,
|
||||
isSubmitting: true,
|
||||
isDetached: true,
|
||||
});
|
||||
|
||||
expect(activity.items).toEqual([
|
||||
{ type: 'writing', text: 'Dispatch-time snapshot; latest detached text.' },
|
||||
]);
|
||||
});
|
||||
|
||||
it('uses a complete parent-stream projection without duplicating persistence', () => {
|
||||
const activity = adaptLivePersistedActivity({
|
||||
title: 'researcher',
|
||||
progress: {
|
||||
subagentRunId: 'run',
|
||||
subagentType: 'researcher',
|
||||
status: 'message_delta',
|
||||
contentParts: [{ type: ContentTypes.TEXT, text: 'Complete live text.' }],
|
||||
aggregatorState: initSubagentAggregatorState(),
|
||||
tickerState: initSubagentTickerState(),
|
||||
coverage: 'complete',
|
||||
},
|
||||
persistedContent: [{ type: ContentTypes.TEXT, text: 'Complete ' }] as TMessageContentParts[],
|
||||
initialProgress: 1,
|
||||
isSubmitting: true,
|
||||
isDetached: true,
|
||||
});
|
||||
|
||||
expect(activity.items).toEqual([{ type: 'writing', text: 'Complete live text.' }]);
|
||||
});
|
||||
|
||||
it('appends coincident text in a forward-only suffix', () => {
|
||||
const activity = adaptLivePersistedActivity({
|
||||
title: 'researcher',
|
||||
progress: {
|
||||
subagentRunId: 'run',
|
||||
subagentType: 'researcher',
|
||||
status: 'message_delta',
|
||||
contentParts: [{ type: ContentTypes.TEXT, text: 'ha' }],
|
||||
aggregatorState: initSubagentAggregatorState(),
|
||||
tickerState: initSubagentTickerState(),
|
||||
coverage: 'suffix',
|
||||
},
|
||||
persistedContent: [{ type: ContentTypes.TEXT, text: 'ha' }] as TMessageContentParts[],
|
||||
initialProgress: 1,
|
||||
isSubmitting: true,
|
||||
isDetached: true,
|
||||
});
|
||||
|
||||
expect(activity.items).toEqual([{ type: 'writing', text: 'haha' }]);
|
||||
});
|
||||
|
||||
it('preserves persisted tool fields when a sparse completion is the live suffix', () => {
|
||||
const activity = adaptLivePersistedActivity({
|
||||
title: 'researcher',
|
||||
progress: {
|
||||
subagentRunId: 'run',
|
||||
subagentType: 'researcher',
|
||||
status: 'run_step_completed',
|
||||
contentParts: [
|
||||
{
|
||||
type: ContentTypes.TOOL_CALL,
|
||||
tool_call: {
|
||||
id: 'tool-1',
|
||||
name: '',
|
||||
args: '{}',
|
||||
output: 'Found it.',
|
||||
progress: 1,
|
||||
},
|
||||
},
|
||||
],
|
||||
aggregatorState: initSubagentAggregatorState(),
|
||||
tickerState: initSubagentTickerState(),
|
||||
coverage: 'suffix',
|
||||
},
|
||||
persistedContent: [
|
||||
{
|
||||
type: ContentTypes.TOOL_CALL,
|
||||
tool_call: {
|
||||
id: 'tool-1',
|
||||
name: 'search',
|
||||
args: '{"query":"release"}',
|
||||
progress: 0.1,
|
||||
},
|
||||
},
|
||||
] as unknown as TMessageContentParts[],
|
||||
initialProgress: 1,
|
||||
isSubmitting: true,
|
||||
isDetached: true,
|
||||
});
|
||||
|
||||
expect(activity.items).toEqual([
|
||||
expect.objectContaining({
|
||||
type: 'tool',
|
||||
name: 'search',
|
||||
input: '{"query":"release"}',
|
||||
output: 'Found it.',
|
||||
status: 'completed',
|
||||
}),
|
||||
]);
|
||||
});
|
||||
|
||||
it('rehydrates the selected detached task from its sanitized durable activity', () => {
|
||||
const view: SubagentThreadView = {
|
||||
threadId: 'thread',
|
||||
|
|
|
|||
|
|
@ -102,6 +102,50 @@ const publicActivityToChildActivity = (items: SubagentActivityItem[]): ChildActi
|
|||
};
|
||||
});
|
||||
|
||||
/** Merge activity whose transport explicitly declares it is a forward-only
|
||||
* suffix. Complete parent-stream projections bypass this function entirely. */
|
||||
const mergePersistedAndLiveActivity = (
|
||||
persisted: ChildActivityItem[],
|
||||
live: ChildActivityItem[],
|
||||
): ChildActivityItem[] => {
|
||||
if (persisted.length === 0) return live;
|
||||
if (live.length === 0) return persisted;
|
||||
|
||||
const merged = [...persisted];
|
||||
for (const item of live) {
|
||||
if (item.type === 'tool') {
|
||||
const existingIndex = merged.findIndex(
|
||||
(candidate) => candidate.type === 'tool' && candidate.toolCallId === item.toolCallId,
|
||||
);
|
||||
if (existingIndex >= 0) {
|
||||
const existing = merged[existingIndex] as Extract<ChildActivityItem, { type: 'tool' }>;
|
||||
const next = { ...existing, ...item };
|
||||
if (item.name === '') next.name = existing.name;
|
||||
if (
|
||||
existing.input != null &&
|
||||
(item.input == null || item.input === '' || item.input === '{}')
|
||||
) {
|
||||
next.input = existing.input;
|
||||
}
|
||||
merged[existingIndex] = next;
|
||||
} else {
|
||||
merged.push(item);
|
||||
}
|
||||
continue;
|
||||
}
|
||||
|
||||
const previous = merged.at(-1);
|
||||
if (previous?.type !== item.type) {
|
||||
merged.push(item);
|
||||
continue;
|
||||
}
|
||||
const previousText = previous.text ?? '';
|
||||
const nextText = item.text ?? '';
|
||||
merged[merged.length - 1] = { ...item, text: `${previousText}${nextText}` };
|
||||
}
|
||||
return merged;
|
||||
};
|
||||
|
||||
const liveStatus = ({
|
||||
progress,
|
||||
initialProgress,
|
||||
|
|
@ -131,17 +175,22 @@ export function adaptLivePersistedActivity(input: {
|
|||
initialProgress: number;
|
||||
isSubmitting: boolean;
|
||||
runStepStatus?: PartMetadata['runStepStatus'];
|
||||
isDetached?: boolean;
|
||||
reasoningVisibility?: 'visible' | 'marker';
|
||||
approvalVisibility?: 'visible' | 'hidden';
|
||||
}): ChildActivity {
|
||||
const persisted = input.persistedContent ?? [];
|
||||
const live = (input.progress?.contentParts ?? []) as TMessageContentParts[];
|
||||
const parts = persisted.length > 0 ? persisted : live;
|
||||
const items = contentPartsToActivity(
|
||||
parts,
|
||||
input.reasoningVisibility ?? 'visible',
|
||||
input.approvalVisibility ?? 'visible',
|
||||
);
|
||||
const reasoningVisibility = input.reasoningVisibility ?? 'visible';
|
||||
const approvalVisibility = input.approvalVisibility ?? 'visible';
|
||||
const persistedItems = contentPartsToActivity(persisted, reasoningVisibility, approvalVisibility);
|
||||
const liveItems = contentPartsToActivity(live, reasoningVisibility, approvalVisibility);
|
||||
let items = persistedItems.length > 0 ? persistedItems : liveItems;
|
||||
if (input.isDetached === true && input.progress?.coverage === 'suffix') {
|
||||
items = mergePersistedAndLiveActivity(persistedItems, liveItems);
|
||||
} else if (input.isDetached === true && liveItems.length > 0) {
|
||||
items = liveItems;
|
||||
}
|
||||
if (items.length === 0 && input.legacyOutput != null && input.legacyOutput !== '') {
|
||||
items.push({ type: 'writing', text: input.legacyOutput });
|
||||
}
|
||||
|
|
|
|||
|
|
@ -2,6 +2,7 @@ import { renderHook } from '@testing-library/react';
|
|||
import type { SubagentThreadView } from 'librechat-data-provider';
|
||||
import {
|
||||
isSubagentReadinessPending,
|
||||
subagentThreadHasTaskEvidence,
|
||||
subagentThreadRefetchInterval,
|
||||
useSubagentThreadQuery,
|
||||
} from './queries';
|
||||
|
|
@ -46,6 +47,16 @@ describe('subagent thread refresh policy', () => {
|
|||
expect(subagentThreadRefetchInterval(prior, 1_000, 1_000, 'new-task')).toBe(false);
|
||||
});
|
||||
|
||||
it('associates terminal state only with evidence from the selected task', () => {
|
||||
const prior = {
|
||||
...view('completed'),
|
||||
messages: [{ messageId: 'old-task:assistant' }],
|
||||
} as SubagentThreadView;
|
||||
|
||||
expect(subagentThreadHasTaskEvidence(prior, 'new-task')).toBe(false);
|
||||
expect(subagentThreadHasTaskEvidence(prior, 'old-task')).toBe(true);
|
||||
});
|
||||
|
||||
it('stops polling an older API view once the exact task response exists', () => {
|
||||
const rollingDeployView = {
|
||||
...view('running'),
|
||||
|
|
|
|||
|
|
@ -13,20 +13,22 @@ const isTerminal = (status: SubagentThreadView['status']): boolean =>
|
|||
status === 'interrupted' ||
|
||||
status === 'cancelled';
|
||||
|
||||
export const subagentThreadHasTaskEvidence = (
|
||||
view: SubagentThreadView | undefined,
|
||||
taskId: string,
|
||||
): boolean =>
|
||||
view?.messages.some(
|
||||
(message) =>
|
||||
message.messageId === `${taskId}:user` || message.messageId === `${taskId}:assistant`,
|
||||
) === true;
|
||||
|
||||
export const subagentThreadRefetchInterval = (
|
||||
view: SubagentThreadView | undefined,
|
||||
readinessDeadline: number,
|
||||
now = Date.now(),
|
||||
expectedTaskId?: string,
|
||||
): number | false => {
|
||||
if (
|
||||
expectedTaskId != null &&
|
||||
!view?.messages.some(
|
||||
(message) =>
|
||||
message.messageId === `${expectedTaskId}:user` ||
|
||||
message.messageId === `${expectedTaskId}:assistant`,
|
||||
)
|
||||
) {
|
||||
if (expectedTaskId != null && !subagentThreadHasTaskEvidence(view, expectedTaskId)) {
|
||||
return now < readinessDeadline ? ACTIVE_THREAD_REFRESH_MS : false;
|
||||
}
|
||||
// During a rolling deploy, an older replica can return a thread-wide status
|
||||
|
|
|
|||
|
|
@ -0,0 +1,303 @@
|
|||
import React from 'react';
|
||||
import { act, renderHook } from '@testing-library/react';
|
||||
import { RecoilRoot, useRecoilValue, useSetRecoilState } from 'recoil';
|
||||
import { ContentTypes, QueryKeys, StepEvents } from 'librechat-data-provider';
|
||||
import type { ActiveSubagentPanel } from '~/store/subagents';
|
||||
import {
|
||||
subagentParentStreamOpenByToolCallId,
|
||||
subagentProgressByToolCallId,
|
||||
subagentProgressKey,
|
||||
takeRegisteredSubagentProgressKeys,
|
||||
} from '~/store/subagents';
|
||||
import useSubagentActivityStream from './useSubagentActivityStream';
|
||||
|
||||
type Listener = (event: MessageEvent) => void;
|
||||
type MockStream = {
|
||||
url: string;
|
||||
options: { method?: string; headers?: Record<string, string> };
|
||||
listeners: Record<string, Listener>;
|
||||
close: jest.Mock;
|
||||
emit: (type: string, data: unknown) => void;
|
||||
};
|
||||
|
||||
const streams: MockStream[] = [];
|
||||
jest.mock('sse.js', () => ({
|
||||
SSE: jest.fn().mockImplementation((url: string, options: MockStream['options']) => {
|
||||
const listeners: Record<string, Listener> = {};
|
||||
const stream: MockStream = {
|
||||
url,
|
||||
options,
|
||||
listeners,
|
||||
close: jest.fn(),
|
||||
emit: (type, data) => listeners[type]?.({ data: JSON.stringify(data) } as MessageEvent),
|
||||
};
|
||||
streams.push(stream);
|
||||
return {
|
||||
addEventListener: (type: string, listener: Listener) => {
|
||||
listeners[type] = listener;
|
||||
},
|
||||
close: stream.close,
|
||||
};
|
||||
}),
|
||||
}));
|
||||
|
||||
const mockInvalidateQueries = jest.fn();
|
||||
const mockQueryClient = { invalidateQueries: mockInvalidateQueries };
|
||||
jest.mock('@tanstack/react-query', () => ({
|
||||
...jest.requireActual('@tanstack/react-query'),
|
||||
useQueryClient: () => mockQueryClient,
|
||||
}));
|
||||
|
||||
jest.mock('~/hooks/AuthContext', () => ({
|
||||
useAuthContext: () => ({ token: 'token-1', isAuthenticated: true }),
|
||||
}));
|
||||
|
||||
const selection: ActiveSubagentPanel = {
|
||||
host: 'conversation',
|
||||
parentConversationId: 'parent conversation',
|
||||
parentMessageId: 'parent-message',
|
||||
toolCallId: 'tool-call',
|
||||
partIndex: 1,
|
||||
subagentType: 'researcher',
|
||||
initialProgress: 1,
|
||||
isSubmitting: false,
|
||||
durable: { threadId: 'child/thread', taskId: 'task?1' },
|
||||
};
|
||||
|
||||
const wrapper = ({ children }: { children: React.ReactNode }) => (
|
||||
<RecoilRoot>{children}</RecoilRoot>
|
||||
);
|
||||
|
||||
describe('useSubagentActivityStream', () => {
|
||||
beforeEach(() => {
|
||||
streams.length = 0;
|
||||
mockInvalidateQueries.mockClear();
|
||||
takeRegisteredSubagentProgressKeys();
|
||||
});
|
||||
|
||||
it('opens one authorized task stream and closes after terminal delivery', () => {
|
||||
const { result, unmount } = renderHook(
|
||||
() => {
|
||||
useSubagentActivityStream(selection);
|
||||
return useRecoilValue(
|
||||
subagentProgressByToolCallId(
|
||||
subagentProgressKey(
|
||||
selection.parentMessageId,
|
||||
selection.toolCallId,
|
||||
selection.partIndex,
|
||||
),
|
||||
),
|
||||
);
|
||||
},
|
||||
{ wrapper },
|
||||
);
|
||||
|
||||
expect(streams).toHaveLength(1);
|
||||
expect(streams[0]?.url).toContain(
|
||||
'/api/convos/parent%20conversation/subagents/child%2Fthread/tasks/task%3F1/activity',
|
||||
);
|
||||
expect(streams[0]?.options.headers).toEqual({ Authorization: 'Bearer token-1' });
|
||||
|
||||
act(() => {
|
||||
streams[0]?.emit('message', {
|
||||
event: StepEvents.ON_SUBAGENT_UPDATE,
|
||||
data: {
|
||||
runId: 'root',
|
||||
parentRunId: 'parent',
|
||||
subagentRunId: 'child',
|
||||
activityEventId: 'task-1:0',
|
||||
activitySequence: 0,
|
||||
subagentType: 'researcher',
|
||||
subagentKind: 'agent',
|
||||
subagentAgentId: 'agent-1',
|
||||
parentToolCallId: 'tool-call',
|
||||
depth: 1,
|
||||
ancestry: ['parent'],
|
||||
phase: 'message_delta',
|
||||
data: { delta: { content: [{ type: 'text', text: 'Live child output' }] } },
|
||||
timestamp: '2026-08-21T20:00:00.000Z',
|
||||
},
|
||||
});
|
||||
streams[0]?.emit('message', {
|
||||
final: true,
|
||||
subagentActivity: true,
|
||||
status: 'completed',
|
||||
});
|
||||
});
|
||||
|
||||
expect(result.current?.contentParts).toEqual([{ type: 'text', text: 'Live child output' }]);
|
||||
expect(result.current?.coverage).toBe('complete');
|
||||
expect(takeRegisteredSubagentProgressKeys()).toEqual([
|
||||
subagentProgressKey(selection.parentMessageId, selection.toolCallId, selection.partIndex),
|
||||
]);
|
||||
expect(streams[0]?.close).toHaveBeenCalledTimes(1);
|
||||
expect(mockInvalidateQueries).toHaveBeenCalledWith([
|
||||
QueryKeys.subagentThread,
|
||||
'parent conversation',
|
||||
'child/thread',
|
||||
'task?1',
|
||||
]);
|
||||
unmount();
|
||||
expect(streams[0]?.close).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it('accepts an exact task-stream update when older providers omit the optional tool-call id', () => {
|
||||
const { result } = renderHook(
|
||||
() => {
|
||||
useSubagentActivityStream(selection);
|
||||
return useRecoilValue(
|
||||
subagentProgressByToolCallId(
|
||||
subagentProgressKey(
|
||||
selection.parentMessageId,
|
||||
selection.toolCallId,
|
||||
selection.partIndex,
|
||||
),
|
||||
),
|
||||
);
|
||||
},
|
||||
{ wrapper },
|
||||
);
|
||||
|
||||
act(() => {
|
||||
streams[0]?.emit('message', {
|
||||
event: StepEvents.ON_SUBAGENT_UPDATE,
|
||||
data: {
|
||||
runId: 'root',
|
||||
parentRunId: 'parent',
|
||||
subagentRunId: 'child',
|
||||
subagentType: 'researcher',
|
||||
subagentKind: 'agent',
|
||||
depth: 1,
|
||||
ancestry: [],
|
||||
phase: 'message_delta',
|
||||
data: { delta: { content: [{ type: 'text', text: 'Compatible update' }] } },
|
||||
timestamp: '2026-08-21T20:00:00.000Z',
|
||||
},
|
||||
});
|
||||
});
|
||||
|
||||
expect(result.current?.contentParts).toEqual([{ type: 'text', text: 'Compatible update' }]);
|
||||
});
|
||||
|
||||
it('buffers the first detached suffix while the parent stream is still open', () => {
|
||||
const activeSelection = { ...selection, isSubmitting: true };
|
||||
const key = subagentProgressKey(
|
||||
activeSelection.parentMessageId,
|
||||
activeSelection.toolCallId,
|
||||
activeSelection.partIndex,
|
||||
);
|
||||
const { result } = renderHook(
|
||||
() => {
|
||||
useSubagentActivityStream(activeSelection);
|
||||
return {
|
||||
progress: useRecoilValue(subagentProgressByToolCallId(key)),
|
||||
parentOpen: useRecoilValue(subagentParentStreamOpenByToolCallId(key)),
|
||||
closeParent: useSetRecoilState(subagentParentStreamOpenByToolCallId(key)),
|
||||
};
|
||||
},
|
||||
{ wrapper },
|
||||
);
|
||||
|
||||
act(() => {
|
||||
streams[0]?.emit('message', {
|
||||
event: StepEvents.ON_SUBAGENT_UPDATE,
|
||||
data: {
|
||||
runId: 'root',
|
||||
parentRunId: 'parent',
|
||||
subagentRunId: 'child',
|
||||
activityEventId: 'task-1:5',
|
||||
activitySequence: 5,
|
||||
subagentType: 'researcher',
|
||||
subagentKind: 'agent',
|
||||
subagentAgentId: 'agent-1',
|
||||
parentToolCallId: 'tool-call',
|
||||
depth: 1,
|
||||
ancestry: [],
|
||||
phase: 'message_delta',
|
||||
data: { delta: { content: [{ type: ContentTypes.TEXT, text: 'suffix' }] } },
|
||||
timestamp: '2026-08-21T20:00:00.000Z',
|
||||
},
|
||||
});
|
||||
});
|
||||
|
||||
expect(result.current.parentOpen).toBe(true);
|
||||
expect(result.current.progress?.contentParts).toEqual([]);
|
||||
expect(result.current.progress?.pendingSequencedEvents).toHaveLength(1);
|
||||
|
||||
act(() => result.current.closeParent(false));
|
||||
|
||||
expect(result.current.parentOpen).toBe(false);
|
||||
expect(result.current.progress?.contentParts).toEqual([
|
||||
{ type: ContentTypes.TEXT, text: 'suffix' },
|
||||
]);
|
||||
expect(result.current.progress?.pendingSequencedEvents).toBeUndefined();
|
||||
});
|
||||
|
||||
it('reconnects with bounded backoff after a transient stream error', () => {
|
||||
jest.useFakeTimers();
|
||||
const { unmount } = renderHook(() => useSubagentActivityStream(selection), { wrapper });
|
||||
|
||||
act(() => streams[0]?.emit('error', {}));
|
||||
expect(streams[0]?.close).toHaveBeenCalledTimes(1);
|
||||
expect(streams).toHaveLength(1);
|
||||
|
||||
act(() => jest.advanceTimersByTime(500));
|
||||
expect(streams).toHaveLength(2);
|
||||
|
||||
unmount();
|
||||
expect(streams[1]?.close).toHaveBeenCalledTimes(1);
|
||||
jest.useRealTimers();
|
||||
});
|
||||
|
||||
it('preserves reconnect backoff after a stream-unavailable envelope', () => {
|
||||
jest.useFakeTimers();
|
||||
const { unmount } = renderHook(() => useSubagentActivityStream(selection), { wrapper });
|
||||
|
||||
act(() => streams[0]?.emit('error', {}));
|
||||
act(() => jest.advanceTimersByTime(500));
|
||||
expect(streams).toHaveLength(2);
|
||||
|
||||
act(() => {
|
||||
streams[1]?.emit('message', { error: 'Subagent activity stream unavailable' });
|
||||
streams[1]?.emit('error', {});
|
||||
jest.advanceTimersByTime(999);
|
||||
});
|
||||
expect(streams).toHaveLength(2);
|
||||
|
||||
act(() => jest.advanceTimersByTime(1));
|
||||
expect(streams).toHaveLength(3);
|
||||
|
||||
unmount();
|
||||
jest.useRealTimers();
|
||||
});
|
||||
|
||||
it('keeps one forward-only stream across metadata-only selection updates', () => {
|
||||
const { rerender } = renderHook(({ value }) => useSubagentActivityStream(value), {
|
||||
initialProps: { value: selection },
|
||||
wrapper,
|
||||
});
|
||||
expect(streams).toHaveLength(1);
|
||||
|
||||
rerender({
|
||||
value: {
|
||||
...selection,
|
||||
persistedContent: [{ type: ContentTypes.TEXT, text: 'New snapshot.' }],
|
||||
durable: { ...selection.durable! },
|
||||
},
|
||||
});
|
||||
|
||||
expect(streams).toHaveLength(1);
|
||||
expect(streams[0]?.close).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('never opens the private task stream for shares or foreground children', () => {
|
||||
const { rerender } = renderHook(({ value }) => useSubagentActivityStream(value), {
|
||||
initialProps: { value: { ...selection, host: 'share' } as ActiveSubagentPanel },
|
||||
wrapper,
|
||||
});
|
||||
expect(streams).toHaveLength(0);
|
||||
|
||||
rerender({ value: { ...selection, host: 'conversation', durable: undefined } });
|
||||
expect(streams).toHaveLength(0);
|
||||
});
|
||||
});
|
||||
167
client/src/data-provider/Subagents/useSubagentActivityStream.ts
Normal file
167
client/src/data-provider/Subagents/useSubagentActivityStream.ts
Normal file
|
|
@ -0,0 +1,167 @@
|
|||
import { useEffect, useRef } from 'react';
|
||||
import { SSE } from 'sse.js';
|
||||
import { useQueryClient } from '@tanstack/react-query';
|
||||
import { useRecoilValue, useSetRecoilState } from 'recoil';
|
||||
import { QueryKeys, StepEvents, apiBaseUrl } from 'librechat-data-provider';
|
||||
import type { SubagentUpdateEvent } from 'librechat-data-provider';
|
||||
import type { ActiveSubagentPanel } from '~/store/subagents';
|
||||
import {
|
||||
closeParentSubagentProgress,
|
||||
reduceSubagentProgress,
|
||||
registerSubagentProgressKey,
|
||||
subagentParentStreamOpenByToolCallId,
|
||||
subagentProgressByToolCallId,
|
||||
subagentProgressKey,
|
||||
} from '~/store/subagents';
|
||||
import { useAuthContext } from '~/hooks/AuthContext';
|
||||
|
||||
type ActivityEnvelope = {
|
||||
event?: unknown;
|
||||
data?: unknown;
|
||||
final?: unknown;
|
||||
subagentActivity?: unknown;
|
||||
};
|
||||
|
||||
const INITIAL_RECONNECT_MS = 500;
|
||||
const MAX_RECONNECT_MS = 5_000;
|
||||
|
||||
const isSubagentUpdate = (value: unknown): value is SubagentUpdateEvent => {
|
||||
if (value == null || typeof value !== 'object') return false;
|
||||
const event = value as Partial<SubagentUpdateEvent>;
|
||||
return (
|
||||
typeof event.subagentRunId === 'string' &&
|
||||
typeof event.subagentType === 'string' &&
|
||||
(event.activityEventId == null || typeof event.activityEventId === 'string') &&
|
||||
(event.activitySequence == null ||
|
||||
(Number.isSafeInteger(event.activitySequence) && event.activitySequence >= 0)) &&
|
||||
(event.parentToolCallId == null || typeof event.parentToolCallId === 'string') &&
|
||||
typeof event.phase === 'string'
|
||||
);
|
||||
};
|
||||
|
||||
/** Live-only enhancement for the selected durable child; the durable query remains canonical. */
|
||||
export default function useSubagentActivityStream(
|
||||
selection: ActiveSubagentPanel,
|
||||
enabled = true,
|
||||
): void {
|
||||
const { token, isAuthenticated } = useAuthContext();
|
||||
const queryClient = useQueryClient();
|
||||
const key = subagentProgressKey(
|
||||
selection.parentMessageId,
|
||||
selection.toolCallId,
|
||||
selection.partIndex,
|
||||
);
|
||||
const setProgress = useSetRecoilState(subagentProgressByToolCallId(key));
|
||||
const parentStreamOpen = useRecoilValue(subagentParentStreamOpenByToolCallId(key));
|
||||
const setParentStreamOpen = useSetRecoilState(subagentParentStreamOpenByToolCallId(key));
|
||||
const parentStreamOpenRef = useRef(parentStreamOpen);
|
||||
const durable = selection.durable;
|
||||
const threadId = durable?.threadId;
|
||||
const taskId = durable?.taskId;
|
||||
|
||||
useEffect(() => {
|
||||
parentStreamOpenRef.current = parentStreamOpen;
|
||||
if (!parentStreamOpen) {
|
||||
setProgress(closeParentSubagentProgress);
|
||||
}
|
||||
}, [parentStreamOpen, setProgress]);
|
||||
|
||||
useEffect(() => {
|
||||
if (!selection.isSubmitting) return;
|
||||
registerSubagentProgressKey(key);
|
||||
parentStreamOpenRef.current = true;
|
||||
setParentStreamOpen(true);
|
||||
}, [key, selection.isSubmitting, setParentStreamOpen]);
|
||||
|
||||
useEffect(() => {
|
||||
if (
|
||||
selection.host !== 'conversation' ||
|
||||
threadId == null ||
|
||||
taskId == null ||
|
||||
!enabled ||
|
||||
!isAuthenticated ||
|
||||
token == null
|
||||
) {
|
||||
return;
|
||||
}
|
||||
|
||||
const queryKey = [QueryKeys.subagentThread, selection.parentConversationId, threadId, taskId];
|
||||
const endpoint = `${apiBaseUrl()}/api/convos/${encodeURIComponent(selection.parentConversationId)}/subagents/${encodeURIComponent(threadId)}/tasks/${encodeURIComponent(taskId)}/activity`;
|
||||
let stream: SSE | undefined;
|
||||
let retryTimer: ReturnType<typeof setTimeout> | undefined;
|
||||
let retryAttempt = 0;
|
||||
let disposed = false;
|
||||
let terminal = false;
|
||||
|
||||
const closeCurrent = () => {
|
||||
const current = stream;
|
||||
stream = undefined;
|
||||
current?.close();
|
||||
};
|
||||
const connect = () => {
|
||||
retryTimer = undefined;
|
||||
if (disposed || terminal) return;
|
||||
const next = new SSE(endpoint, {
|
||||
method: 'GET',
|
||||
headers: { Authorization: `Bearer ${token}` },
|
||||
});
|
||||
stream = next;
|
||||
|
||||
next.addEventListener('message', (message: MessageEvent) => {
|
||||
if (stream !== next || disposed) return;
|
||||
let envelope: ActivityEnvelope;
|
||||
try {
|
||||
envelope = JSON.parse(message.data) as ActivityEnvelope;
|
||||
} catch {
|
||||
return;
|
||||
}
|
||||
if (envelope.final === true && envelope.subagentActivity === true) {
|
||||
terminal = true;
|
||||
closeCurrent();
|
||||
void queryClient.invalidateQueries(queryKey);
|
||||
return;
|
||||
}
|
||||
const event = envelope.data;
|
||||
if (envelope.event !== StepEvents.ON_SUBAGENT_UPDATE || !isSubagentUpdate(event)) {
|
||||
return;
|
||||
}
|
||||
if (event.parentToolCallId != null && event.parentToolCallId !== selection.toolCallId) {
|
||||
return;
|
||||
}
|
||||
retryAttempt = 0;
|
||||
registerSubagentProgressKey(key);
|
||||
setProgress((previous) =>
|
||||
reduceSubagentProgress(previous, [event], 'detached', parentStreamOpenRef.current),
|
||||
);
|
||||
});
|
||||
next.addEventListener('error', () => {
|
||||
if (stream !== next || disposed || terminal || retryTimer != null) return;
|
||||
closeCurrent();
|
||||
const delay = Math.min(INITIAL_RECONNECT_MS * 2 ** retryAttempt, MAX_RECONNECT_MS);
|
||||
retryAttempt += 1;
|
||||
retryTimer = setTimeout(connect, delay);
|
||||
});
|
||||
};
|
||||
|
||||
connect();
|
||||
return () => {
|
||||
disposed = true;
|
||||
if (retryTimer != null) clearTimeout(retryTimer);
|
||||
closeCurrent();
|
||||
};
|
||||
}, [
|
||||
enabled,
|
||||
isAuthenticated,
|
||||
key,
|
||||
queryClient,
|
||||
selection.host,
|
||||
selection.parentConversationId,
|
||||
selection.parentMessageId,
|
||||
selection.partIndex,
|
||||
selection.toolCallId,
|
||||
setProgress,
|
||||
taskId,
|
||||
threadId,
|
||||
token,
|
||||
]);
|
||||
}
|
||||
|
|
@ -3532,6 +3532,42 @@ describe('useStepHandler', () => {
|
|||
|
||||
expect(getProgress('call_keep')).not.toBeNull();
|
||||
});
|
||||
|
||||
it('uses parent stream closure to release a detached sequence waiting at handoff', () => {
|
||||
const { result, getProgress } = renderStepHandlerWithReader();
|
||||
const { submission } = seedResponseWithSubagentToolCalls(result, ['call_handoff']);
|
||||
|
||||
act(() => {
|
||||
(result.current as any).stepHandler(
|
||||
{
|
||||
event: StepEvents.ON_SUBAGENT_UPDATE,
|
||||
data: makeUpdate({
|
||||
parentToolCallId: 'call_handoff',
|
||||
activityEventId: 'task:5',
|
||||
activitySequence: 5,
|
||||
phase: 'message_delta',
|
||||
data: { delta: { content: [{ type: ContentTypes.TEXT, text: 'suffix' }] } },
|
||||
}),
|
||||
},
|
||||
submission,
|
||||
);
|
||||
});
|
||||
|
||||
expect(getProgress('call_handoff')).toEqual(
|
||||
expect.objectContaining({ contentParts: [], pendingSequencedEvents: [expect.any(Object)] }),
|
||||
);
|
||||
|
||||
act(() => {
|
||||
(result.current as any).clearStepMaps();
|
||||
});
|
||||
|
||||
expect(getProgress('call_handoff')).toEqual(
|
||||
expect.objectContaining({
|
||||
contentParts: [{ type: ContentTypes.TEXT, text: 'suffix' }],
|
||||
lastActivitySequence: 5,
|
||||
}),
|
||||
);
|
||||
});
|
||||
});
|
||||
|
||||
/**
|
||||
|
|
|
|||
|
|
@ -23,14 +23,14 @@ import type {
|
|||
import type { SetterOrUpdater } from 'recoil';
|
||||
import type { AnnounceOptions } from '~/common';
|
||||
import {
|
||||
foldSubagentEvent,
|
||||
foldSubagentEventIntoTicker,
|
||||
initSubagentAggregatorState,
|
||||
initSubagentTickerState,
|
||||
} from '~/utils/subagentContent';
|
||||
import {
|
||||
closeParentSubagentProgress,
|
||||
listRegisteredSubagentProgressKeys,
|
||||
reduceSubagentProgress,
|
||||
registerSubagentProgressKey,
|
||||
subagentParentStreamOpenByToolCallId,
|
||||
subagentProgressByToolCallId,
|
||||
subagentProgressKey,
|
||||
takeRegisteredSubagentProgressKeys,
|
||||
sandboxStartingByToolCallId,
|
||||
} from '~/store';
|
||||
import { isAskUserQuestionPart, isAnsweredAskUserQuestionPart } from '~/utils/approval';
|
||||
|
|
@ -183,13 +183,6 @@ export default function useStepHandler({
|
|||
const pendingSubagentBuffer = useRef(
|
||||
new Map<string, { parentMessageId: string; events: SubagentUpdateEvent[] }>(),
|
||||
);
|
||||
/**
|
||||
* Tracked atom keys so `clearStepMaps` can reset them. Without this, each
|
||||
* subagent invocation leaks an `events: SubagentUpdateEvent[]` array in the
|
||||
* `atomFamily` — atoms persist for the app lifetime.
|
||||
*/
|
||||
const knownSubagentAtomKeys = useRef(new Set<string>());
|
||||
|
||||
const getCurrentMessages = useCallback(
|
||||
(messages: TMessage[]) => {
|
||||
const freshMessages = getMessages();
|
||||
|
|
@ -276,35 +269,11 @@ export default function useStepHandler({
|
|||
}
|
||||
const toApply = pending ? [...pending.events, payload] : [payload];
|
||||
|
||||
knownSubagentAtomKeys.current.add(invocationKey);
|
||||
set(subagentProgressByToolCallId(invocationKey), (prev) => {
|
||||
/** Fold the batch into both aggregators. Pure functions — they
|
||||
* return a new reference only when something actually changed,
|
||||
* so React bails out of unnecessary re-renders downstream. */
|
||||
let contentParts = prev?.contentParts ?? [];
|
||||
let aggregatorState = prev?.aggregatorState ?? initSubagentAggregatorState();
|
||||
let tickerState = prev?.tickerState ?? initSubagentTickerState();
|
||||
for (const event of toApply) {
|
||||
({ parts: contentParts, state: aggregatorState } = foldSubagentEvent(
|
||||
contentParts,
|
||||
aggregatorState,
|
||||
event,
|
||||
));
|
||||
tickerState = foldSubagentEventIntoTicker(tickerState, event);
|
||||
}
|
||||
|
||||
const last = toApply[toApply.length - 1];
|
||||
return {
|
||||
subagentRunId: payload.subagentRunId,
|
||||
subagentType: payload.subagentType,
|
||||
subagentAgentId: payload.subagentAgentId ?? prev?.subagentAgentId,
|
||||
contentParts,
|
||||
aggregatorState,
|
||||
tickerState,
|
||||
status: last.phase,
|
||||
latestLabel: last.label ?? prev?.latestLabel,
|
||||
};
|
||||
});
|
||||
registerSubagentProgressKey(invocationKey);
|
||||
set(subagentParentStreamOpenByToolCallId(invocationKey), true);
|
||||
set(subagentProgressByToolCallId(invocationKey), (prev) =>
|
||||
reduceSubagentProgress(prev, toApply, 'parent', true),
|
||||
);
|
||||
},
|
||||
[resolveSubagentInvocationKey],
|
||||
);
|
||||
|
|
@ -323,10 +292,21 @@ export default function useStepHandler({
|
|||
const resetSubagentAtoms = useRecoilCallback(
|
||||
({ reset }) =>
|
||||
(): void => {
|
||||
for (const invocationKey of knownSubagentAtomKeys.current) {
|
||||
for (const invocationKey of takeRegisteredSubagentProgressKeys()) {
|
||||
reset(subagentProgressByToolCallId(invocationKey));
|
||||
reset(subagentParentStreamOpenByToolCallId(invocationKey));
|
||||
}
|
||||
},
|
||||
[],
|
||||
);
|
||||
|
||||
const closeParentSubagentStreams = useRecoilCallback(
|
||||
({ set }) =>
|
||||
(): void => {
|
||||
for (const invocationKey of listRegisteredSubagentProgressKeys()) {
|
||||
set(subagentParentStreamOpenByToolCallId(invocationKey), false);
|
||||
set(subagentProgressByToolCallId(invocationKey), closeParentSubagentProgress);
|
||||
}
|
||||
knownSubagentAtomKeys.current.clear();
|
||||
},
|
||||
[],
|
||||
);
|
||||
|
|
@ -1437,6 +1417,7 @@ export default function useStepHandler({
|
|||
subagentRunToInvocationKey.current.clear();
|
||||
claimedSubagentInvocationKeys.current.clear();
|
||||
pendingSubagentBuffer.current.clear();
|
||||
closeParentSubagentStreams();
|
||||
/** Unlike subagent atoms below, sandbox-starting flags are transient
|
||||
* status with no audit value — reset them at this boundary so an
|
||||
* interrupted cold boot can't leak a stale "starting" label onto a
|
||||
|
|
@ -1450,7 +1431,7 @@ export default function useStepHandler({
|
|||
* persisted `subagent_content` takes over for historical messages
|
||||
* once the conversation is saved, and we prevent unbounded
|
||||
* atomFamily growth across multi-conversation sessions. */
|
||||
}, [cancelPendingDeltaFlush, resetSandboxAtoms]);
|
||||
}, [cancelPendingDeltaFlush, closeParentSubagentStreams, resetSandboxAtoms]);
|
||||
|
||||
/**
|
||||
* Sync a message into the step handler's messageMap.
|
||||
|
|
|
|||
371
client/src/store/subagents.spec.ts
Normal file
371
client/src/store/subagents.spec.ts
Normal file
|
|
@ -0,0 +1,371 @@
|
|||
import { ContentTypes } from 'librechat-data-provider';
|
||||
import type { SubagentUpdateEvent } from 'librechat-data-provider';
|
||||
import { closeParentSubagentProgress, reduceSubagentProgress } from './subagents';
|
||||
|
||||
const update = (overrides: Partial<SubagentUpdateEvent> = {}): SubagentUpdateEvent => ({
|
||||
runId: 'root-run',
|
||||
parentRunId: 'parent-run',
|
||||
subagentRunId: 'child-run',
|
||||
activityEventId: 'activity-1',
|
||||
subagentType: 'researcher',
|
||||
subagentKind: 'agent',
|
||||
subagentAgentId: 'agent-1',
|
||||
parentToolCallId: 'tool-call',
|
||||
depth: 1,
|
||||
ancestry: [],
|
||||
phase: 'message_delta',
|
||||
data: { delta: { content: [{ type: 'text', text: 'Working.' }] } },
|
||||
label: 'Drafting the report',
|
||||
timestamp: '2026-08-21T20:00:00.000Z',
|
||||
...overrides,
|
||||
});
|
||||
|
||||
describe('reduceSubagentProgress', () => {
|
||||
it('folds an event delivered by both parent and detached streams only once', () => {
|
||||
const event = update();
|
||||
const first = reduceSubagentProgress(null, [event]);
|
||||
const replay = reduceSubagentProgress(first, [event]);
|
||||
|
||||
expect(replay).toBe(first);
|
||||
expect(first?.contentParts).toEqual([{ type: ContentTypes.TEXT, text: 'Working.' }]);
|
||||
expect(first?.tickerState.lines).toHaveLength(1);
|
||||
});
|
||||
|
||||
it('preserves equal chunks that carry distinct host event identities', () => {
|
||||
const progress = reduceSubagentProgress(null, [
|
||||
update({ activityEventId: 'activity-1', activitySequence: 0 }),
|
||||
update({ activityEventId: 'activity-2', activitySequence: 1 }),
|
||||
]);
|
||||
|
||||
expect(progress?.contentParts).toEqual([{ type: ContentTypes.TEXT, text: 'Working.Working.' }]);
|
||||
});
|
||||
|
||||
it('marks an accepted run-start frame complete regardless of its delivery transport', () => {
|
||||
const progress = reduceSubagentProgress(
|
||||
null,
|
||||
[update({ activitySequence: 0 })],
|
||||
'detached',
|
||||
false,
|
||||
);
|
||||
|
||||
expect(progress?.coverage).toBe('complete');
|
||||
});
|
||||
|
||||
it('orders a same-batch overlap by the host sequence before folding', () => {
|
||||
const progress = reduceSubagentProgress(
|
||||
null,
|
||||
[
|
||||
update({
|
||||
activityEventId: 'activity-2',
|
||||
activitySequence: 2,
|
||||
data: { delta: { content: [{ type: ContentTypes.TEXT, text: 'second' }] } },
|
||||
}),
|
||||
update({
|
||||
activityEventId: 'activity-1',
|
||||
activitySequence: 1,
|
||||
data: { delta: { content: [{ type: ContentTypes.TEXT, text: 'first ' }] } },
|
||||
}),
|
||||
],
|
||||
'detached',
|
||||
false,
|
||||
);
|
||||
|
||||
expect(progress?.contentParts).toEqual([{ type: ContentTypes.TEXT, text: 'first second' }]);
|
||||
expect(progress?.lastActivitySequence).toBe(2);
|
||||
});
|
||||
|
||||
it('rejects older overlap events and duplicates beyond the replay-key window', () => {
|
||||
const initial = reduceSubagentProgress(
|
||||
null,
|
||||
[
|
||||
update({
|
||||
activityEventId: 'activity-300',
|
||||
activitySequence: 300,
|
||||
data: { delta: { content: [{ type: ContentTypes.TEXT, text: 'latest' }] } },
|
||||
}),
|
||||
],
|
||||
'detached',
|
||||
false,
|
||||
);
|
||||
const delayed = reduceSubagentProgress(initial, [
|
||||
update({
|
||||
activityEventId: 'activity-1',
|
||||
activitySequence: 1,
|
||||
data: { delta: { content: [{ type: ContentTypes.TEXT, text: 'old' }] } },
|
||||
}),
|
||||
update({
|
||||
activityEventId: 'activity-300',
|
||||
activitySequence: 300,
|
||||
data: { delta: { content: [{ type: ContentTypes.TEXT, text: 'duplicate' }] } },
|
||||
}),
|
||||
]);
|
||||
|
||||
expect(delayed).toBe(initial);
|
||||
expect(delayed?.contentParts).toEqual([{ type: ContentTypes.TEXT, text: 'latest' }]);
|
||||
});
|
||||
|
||||
it('buffers a detached frame until a lagging parent delivers the missing sequence', () => {
|
||||
const detached = reduceSubagentProgress(
|
||||
null,
|
||||
[
|
||||
update({
|
||||
activityEventId: 'activity-1',
|
||||
activitySequence: 1,
|
||||
data: { delta: { content: [{ type: ContentTypes.TEXT, text: 'second' }] } },
|
||||
}),
|
||||
],
|
||||
'detached',
|
||||
true,
|
||||
);
|
||||
expect(detached?.contentParts).toEqual([]);
|
||||
expect(detached?.pendingSequencedEvents).toHaveLength(1);
|
||||
|
||||
const ordered = reduceSubagentProgress(detached, [
|
||||
update({
|
||||
activityEventId: 'activity-0',
|
||||
activitySequence: 0,
|
||||
data: { delta: { content: [{ type: ContentTypes.TEXT, text: 'first ' }] } },
|
||||
}),
|
||||
]);
|
||||
|
||||
expect(ordered?.contentParts).toEqual([{ type: ContentTypes.TEXT, text: 'first second' }]);
|
||||
expect(ordered?.pendingSequencedEvents).toBeUndefined();
|
||||
expect(ordered?.lastActivitySequence).toBe(1);
|
||||
expect(ordered?.coverage).toBe('complete');
|
||||
});
|
||||
|
||||
it('uses parent stream closure as the fence for a detached suffix with no earlier frame', () => {
|
||||
const waiting = reduceSubagentProgress(
|
||||
null,
|
||||
[
|
||||
update({
|
||||
activityEventId: 'activity-5',
|
||||
activitySequence: 5,
|
||||
data: { delta: { content: [{ type: ContentTypes.TEXT, text: 'suffix' }] } },
|
||||
}),
|
||||
],
|
||||
'detached',
|
||||
true,
|
||||
);
|
||||
|
||||
const closed = closeParentSubagentProgress(waiting);
|
||||
|
||||
expect(closed?.contentParts).toEqual([{ type: ContentTypes.TEXT, text: 'suffix' }]);
|
||||
expect(closed?.pendingSequencedEvents).toBeUndefined();
|
||||
expect(closed?.lastActivitySequence).toBe(5);
|
||||
|
||||
const afterMissedFrames = reduceSubagentProgress(
|
||||
closed,
|
||||
[
|
||||
update({
|
||||
activityEventId: 'activity-8',
|
||||
activitySequence: 8,
|
||||
data: { delta: { content: [{ type: ContentTypes.TEXT, text: ' resumed' }] } },
|
||||
}),
|
||||
],
|
||||
'detached',
|
||||
false,
|
||||
);
|
||||
expect(afterMissedFrames?.contentParts).toEqual([
|
||||
{ type: ContentTypes.TEXT, text: 'suffix resumed' },
|
||||
]);
|
||||
expect(afterMissedFrames?.lastActivitySequence).toBe(8);
|
||||
});
|
||||
|
||||
it('bounds future sequence buffering while an earlier parent frame is missing', () => {
|
||||
const waiting = reduceSubagentProgress(
|
||||
null,
|
||||
Array.from({ length: 140 }, (_, index) =>
|
||||
update({
|
||||
activityEventId: `activity-${index + 1}`,
|
||||
activitySequence: index + 1,
|
||||
data: { delta: { content: [{ type: ContentTypes.TEXT, text: 'x'.repeat(2048) }] } },
|
||||
}),
|
||||
),
|
||||
'detached',
|
||||
true,
|
||||
);
|
||||
|
||||
expect(waiting?.pendingSequencedEvents?.length).toBeLessThanOrEqual(100);
|
||||
expect(
|
||||
new TextEncoder().encode(JSON.stringify(waiting?.pendingSequencedEvents)).byteLength,
|
||||
).toBeLessThanOrEqual(128 * 1024);
|
||||
});
|
||||
|
||||
it('accepts the missing expected frame even when the future-frame buffer is full', () => {
|
||||
const waiting = reduceSubagentProgress(
|
||||
null,
|
||||
Array.from({ length: 100 }, (_, index) =>
|
||||
update({
|
||||
activityEventId: `activity-${index + 1}`,
|
||||
activitySequence: index + 1,
|
||||
data: { delta: { content: [{ type: ContentTypes.TEXT, text: 'x' }] } },
|
||||
}),
|
||||
),
|
||||
'detached',
|
||||
true,
|
||||
);
|
||||
|
||||
expect(waiting?.pendingSequencedEvents).toHaveLength(100);
|
||||
|
||||
const ordered = reduceSubagentProgress(waiting, [
|
||||
update({
|
||||
activityEventId: 'activity-0',
|
||||
activitySequence: 0,
|
||||
data: { delta: { content: [{ type: ContentTypes.TEXT, text: 'first-' }] } },
|
||||
}),
|
||||
]);
|
||||
|
||||
expect(ordered?.contentParts).toEqual([
|
||||
{ type: ContentTypes.TEXT, text: `first-${'x'.repeat(100)}` },
|
||||
]);
|
||||
expect(ordered?.pendingSequencedEvents).toBeUndefined();
|
||||
expect(ordered?.lastActivitySequence).toBe(100);
|
||||
expect(ordered?.coverage).toBe('complete');
|
||||
});
|
||||
|
||||
it('preserves legacy unsequenced foreground updates', () => {
|
||||
const progress = reduceSubagentProgress(null, [
|
||||
update({ activityEventId: undefined, activitySequence: undefined }),
|
||||
update({ activityEventId: undefined, activitySequence: undefined }),
|
||||
]);
|
||||
|
||||
expect(progress?.contentParts).toEqual([{ type: ContentTypes.TEXT, text: 'Working.Working.' }]);
|
||||
expect(progress?.lastActivitySequence).toBeUndefined();
|
||||
});
|
||||
|
||||
it('preserves a reasoning activity marker without retaining private reasoning text', () => {
|
||||
const progress = reduceSubagentProgress(
|
||||
null,
|
||||
[
|
||||
update({
|
||||
activitySequence: 0,
|
||||
phase: 'reasoning_delta',
|
||||
data: { delta: { content: [{ type: ContentTypes.THINK, think: 'private' }] } },
|
||||
label: 'Reasoning',
|
||||
}),
|
||||
],
|
||||
'detached',
|
||||
false,
|
||||
);
|
||||
|
||||
expect(progress?.contentParts).toEqual([{ type: ContentTypes.THINK, think: '…' }]);
|
||||
expect(progress?.tickerState.lines).toEqual([
|
||||
expect.objectContaining({ kind: 'reasoning', body: '…' }),
|
||||
]);
|
||||
expect(JSON.stringify(progress)).not.toContain('private');
|
||||
});
|
||||
|
||||
it('preserves visible reasoning on the authoritative parent delivery path', () => {
|
||||
const progress = reduceSubagentProgress(null, [
|
||||
update({
|
||||
activitySequence: 0,
|
||||
phase: 'reasoning_delta',
|
||||
data: { delta: { content: [{ type: ContentTypes.THINK, think: 'Visible reasoning' }] } },
|
||||
label: 'Reasoning',
|
||||
}),
|
||||
]);
|
||||
|
||||
expect(progress?.contentParts).toEqual([
|
||||
{ type: ContentTypes.THINK, think: 'Visible reasoning' },
|
||||
]);
|
||||
});
|
||||
|
||||
it('bounds accumulated live text to the durable activity byte budget', () => {
|
||||
const progress = reduceSubagentProgress(null, [
|
||||
update({
|
||||
activityEventId: 'large-activity',
|
||||
data: { delta: { content: [{ type: 'text', text: 'x'.repeat(96 * 1024) }] } },
|
||||
}),
|
||||
]);
|
||||
|
||||
expect(
|
||||
new TextEncoder().encode(JSON.stringify(progress?.contentParts)).byteLength,
|
||||
).toBeLessThanOrEqual(64 * 1024);
|
||||
expect(progress?.contentParts[0]).toEqual(expect.objectContaining({ type: ContentTypes.TEXT }));
|
||||
});
|
||||
|
||||
it('retains an encoded-byte-bounded singleton containing escaped text', () => {
|
||||
const progress = reduceSubagentProgress(null, [
|
||||
update({
|
||||
activityEventId: 'escaped-activity',
|
||||
data: { delta: { content: [{ type: 'text', text: '\\"'.repeat(48 * 1024) }] } },
|
||||
}),
|
||||
]);
|
||||
|
||||
expect(progress?.contentParts).toHaveLength(1);
|
||||
expect(progress?.contentParts[0]).toEqual(expect.objectContaining({ type: ContentTypes.TEXT }));
|
||||
expect(
|
||||
new TextEncoder().encode(JSON.stringify(progress?.contentParts)).byteLength,
|
||||
).toBeLessThanOrEqual(64 * 1024);
|
||||
});
|
||||
|
||||
it('retains an encoded-byte-bounded singleton tool projection', () => {
|
||||
const progress = reduceSubagentProgress(null, [
|
||||
update({
|
||||
activityEventId: 'escaped-tool-start',
|
||||
phase: 'run_step',
|
||||
data: {
|
||||
stepDetails: {
|
||||
type: 'tool_calls',
|
||||
tool_calls: [{ id: 'tool', name: 'search', args: '\\"'.repeat(48 * 1024) }],
|
||||
},
|
||||
},
|
||||
}),
|
||||
update({
|
||||
activityEventId: 'escaped-tool-complete',
|
||||
phase: 'run_step_completed',
|
||||
data: {
|
||||
result: {
|
||||
type: 'tool_call',
|
||||
tool_call: {
|
||||
id: 'tool',
|
||||
name: 'search',
|
||||
output: '\\\\'.repeat(48 * 1024),
|
||||
progress: 1,
|
||||
},
|
||||
},
|
||||
},
|
||||
}),
|
||||
]);
|
||||
|
||||
expect(progress?.contentParts).toHaveLength(1);
|
||||
expect(progress?.contentParts[0]).toEqual(
|
||||
expect.objectContaining({ type: ContentTypes.TOOL_CALL }),
|
||||
);
|
||||
expect(
|
||||
new TextEncoder().encode(JSON.stringify(progress?.contentParts)).byteLength,
|
||||
).toBeLessThanOrEqual(64 * 1024);
|
||||
});
|
||||
|
||||
it('keeps only the newest bounded activity and continues folding afterward', () => {
|
||||
const toolEvents = Array.from({ length: 120 }, (_, index) =>
|
||||
update({
|
||||
activityEventId: `tool-${index}`,
|
||||
phase: 'run_step',
|
||||
data: {
|
||||
stepDetails: {
|
||||
type: 'tool_calls',
|
||||
tool_calls: [{ id: `call-${index}`, name: 'search', args: { index } }],
|
||||
},
|
||||
},
|
||||
}),
|
||||
);
|
||||
const bounded = reduceSubagentProgress(null, toolEvents);
|
||||
const continued = reduceSubagentProgress(bounded, [
|
||||
update({
|
||||
activityEventId: 'after-bound',
|
||||
phase: 'message_delta',
|
||||
data: { delta: { content: [{ type: 'text', text: 'Final answer.' }] } },
|
||||
}),
|
||||
]);
|
||||
|
||||
expect(bounded?.contentParts).toHaveLength(100);
|
||||
expect(continued?.contentParts).toHaveLength(100);
|
||||
expect(continued?.contentParts.at(-1)).toEqual({
|
||||
type: ContentTypes.TEXT,
|
||||
text: 'Final answer.',
|
||||
});
|
||||
expect(continued?.tickerState.lines.length).toBeLessThanOrEqual(100);
|
||||
});
|
||||
});
|
||||
|
|
@ -1,14 +1,22 @@
|
|||
import { atom, atomFamily } from 'recoil';
|
||||
import { ContentTypes } from 'librechat-data-provider';
|
||||
import type {
|
||||
PartMetadata,
|
||||
SubagentUpdatePhase,
|
||||
TMessageContentParts,
|
||||
SubagentUpdateEvent,
|
||||
} from 'librechat-data-provider';
|
||||
import type {
|
||||
SubagentAggregatorState,
|
||||
SubagentContentPart,
|
||||
SubagentTickerState,
|
||||
} from '~/utils/subagentContent';
|
||||
import {
|
||||
foldSubagentEvent,
|
||||
foldSubagentEventIntoTicker,
|
||||
initSubagentAggregatorState,
|
||||
initSubagentTickerState,
|
||||
} from '~/utils/subagentContent';
|
||||
|
||||
/**
|
||||
* Progress bucket captured per subagent tool call. Populated as
|
||||
|
|
@ -19,8 +27,8 @@ import type {
|
|||
* Both the panel content and the ticker are aggregated *incrementally*
|
||||
* into the atom as each envelope arrives — the atom never keeps the raw
|
||||
* event array. A long-running subagent can emit thousands of deltas
|
||||
* without the state growing past what its structural output (N text
|
||||
* runs + M tool calls + a bounded tail preview) needs.
|
||||
* without retaining the raw event stream. The folded activity is also
|
||||
* capped by item count and encoded size to match the durable public view.
|
||||
*/
|
||||
export interface SubagentProgress {
|
||||
/** Child run id from the SDK — unique per spawn; one tool_call may only have one. */
|
||||
|
|
@ -42,8 +50,203 @@ export interface SubagentProgress {
|
|||
status: SubagentUpdatePhase;
|
||||
/** Convenience: last event's `label` for quick ticker display. */
|
||||
latestLabel?: string;
|
||||
/** Bounded replay fence for events that overlap parent and detached SSE delivery. */
|
||||
recentEventKeys?: string[];
|
||||
/** Highest host sequence folded for this child run. Older overlap frames are ignored. */
|
||||
lastActivitySequence?: number;
|
||||
/** Bounded future frames waiting for an earlier sequence at the parent/detached handoff. */
|
||||
pendingSequencedEvents?: SubagentUpdateEvent[];
|
||||
/** Whether the folded events cover the run from its beginning or only the
|
||||
* forward-only suffix observed after opening a detached task stream. */
|
||||
coverage?: 'complete' | 'suffix';
|
||||
}
|
||||
|
||||
const MAX_RECENT_EVENT_KEYS = 256;
|
||||
const MAX_PENDING_SEQUENCE_EVENTS = 100;
|
||||
const MAX_PENDING_SEQUENCE_BYTES = 128 * 1024;
|
||||
const MAX_LIVE_ACTIVITY_ITEMS = 100;
|
||||
const MAX_LIVE_ACTIVITY_BYTES = 64 * 1024;
|
||||
const MAX_SINGLE_ACTIVITY_ENCODED_BYTES = MAX_LIVE_ACTIVITY_BYTES - 2;
|
||||
const MAX_SINGLE_ACTIVITY_TEXT_BYTES = 60 * 1024;
|
||||
const REDACTED_REASONING_MARKER = '…';
|
||||
|
||||
const encodedBytes = (value: unknown): number =>
|
||||
new TextEncoder().encode(JSON.stringify(value)).byteLength;
|
||||
|
||||
const truncateUtf8 = (value: string, maxBytes: number, keepTail = false): string => {
|
||||
if (new TextEncoder().encode(value).byteLength <= maxBytes) return value;
|
||||
const chars = [...value];
|
||||
let low = 0;
|
||||
let high = chars.length;
|
||||
while (low < high) {
|
||||
const mid = Math.ceil((low + high) / 2);
|
||||
const candidate = keepTail ? chars.slice(-mid).join('') : chars.slice(0, mid).join('');
|
||||
if (new TextEncoder().encode(candidate).byteLength <= maxBytes) low = mid;
|
||||
else high = mid - 1;
|
||||
}
|
||||
return keepTail ? chars.slice(-low).join('') : chars.slice(0, low).join('');
|
||||
};
|
||||
|
||||
const fitStringField = <T>(
|
||||
value: string,
|
||||
candidate: (bounded: string) => T,
|
||||
maxBytes: number,
|
||||
keepTail = false,
|
||||
): T => {
|
||||
const chars = [...value];
|
||||
let low = 0;
|
||||
let high = chars.length;
|
||||
let result = candidate('');
|
||||
while (low < high) {
|
||||
const mid = Math.ceil((low + high) / 2);
|
||||
const bounded = keepTail ? chars.slice(-mid).join('') : chars.slice(0, mid).join('');
|
||||
const next = candidate(bounded);
|
||||
if (encodedBytes(next) <= maxBytes) {
|
||||
low = mid;
|
||||
result = next;
|
||||
} else {
|
||||
high = mid - 1;
|
||||
}
|
||||
}
|
||||
return result;
|
||||
};
|
||||
|
||||
const boundSingletonPart = (part: SubagentContentPart): SubagentContentPart => {
|
||||
if (part.type === ContentTypes.TEXT) {
|
||||
const rawBounded = truncateUtf8(part.text, MAX_SINGLE_ACTIVITY_TEXT_BYTES, true);
|
||||
return fitStringField(
|
||||
rawBounded,
|
||||
(text) => ({ ...part, text }),
|
||||
MAX_SINGLE_ACTIVITY_ENCODED_BYTES,
|
||||
true,
|
||||
);
|
||||
}
|
||||
if (part.type === ContentTypes.THINK) {
|
||||
const rawBounded = truncateUtf8(part.think, MAX_SINGLE_ACTIVITY_TEXT_BYTES, true);
|
||||
return fitStringField(
|
||||
rawBounded,
|
||||
(think) => ({ ...part, think }),
|
||||
MAX_SINGLE_ACTIVITY_ENCODED_BYTES,
|
||||
true,
|
||||
);
|
||||
}
|
||||
|
||||
let bounded: SubagentContentPart = {
|
||||
...part,
|
||||
tool_call: {
|
||||
...part.tool_call,
|
||||
args: truncateUtf8(part.tool_call.args, 24 * 1024),
|
||||
...(part.tool_call.output == null
|
||||
? {}
|
||||
: { output: truncateUtf8(part.tool_call.output, 24 * 1024, true) }),
|
||||
},
|
||||
};
|
||||
if (encodedBytes(bounded) <= MAX_SINGLE_ACTIVITY_ENCODED_BYTES) return bounded;
|
||||
|
||||
const fitToolField = (field: 'output' | 'args' | 'name' | 'id' | 'type', keepTail = false) => {
|
||||
if (bounded.type !== ContentTypes.TOOL_CALL) return;
|
||||
const current = bounded;
|
||||
const value = current.tool_call[field];
|
||||
if (typeof value !== 'string') return;
|
||||
bounded = fitStringField(
|
||||
value,
|
||||
(nextValue) => ({
|
||||
...current,
|
||||
tool_call: { ...current.tool_call, [field]: nextValue },
|
||||
}),
|
||||
MAX_SINGLE_ACTIVITY_ENCODED_BYTES,
|
||||
keepTail,
|
||||
) as SubagentContentPart;
|
||||
};
|
||||
fitToolField('output', true);
|
||||
if (encodedBytes(bounded) > MAX_SINGLE_ACTIVITY_ENCODED_BYTES) fitToolField('args');
|
||||
if (encodedBytes(bounded) > MAX_SINGLE_ACTIVITY_ENCODED_BYTES) fitToolField('name');
|
||||
if (encodedBytes(bounded) > MAX_SINGLE_ACTIVITY_ENCODED_BYTES) fitToolField('id');
|
||||
if (encodedBytes(bounded) > MAX_SINGLE_ACTIVITY_ENCODED_BYTES) fitToolField('type');
|
||||
return bounded;
|
||||
};
|
||||
|
||||
const boundContentParts = (
|
||||
parts: SubagentContentPart[],
|
||||
state: SubagentAggregatorState,
|
||||
): { parts: SubagentContentPart[]; state: SubagentAggregatorState } => {
|
||||
const start = Math.max(0, parts.length - MAX_LIVE_ACTIVITY_ITEMS);
|
||||
let offset = parts.length;
|
||||
let totalBytes = 2;
|
||||
let bounded: SubagentContentPart[] = [];
|
||||
for (let index = parts.length - 1; index >= start; index -= 1) {
|
||||
const partBytes = encodedBytes(parts[index]);
|
||||
const separatorBytes = bounded.length === 0 ? 0 : 1;
|
||||
if (totalBytes + separatorBytes + partBytes > MAX_LIVE_ACTIVITY_BYTES) {
|
||||
if (bounded.length === 0) {
|
||||
bounded = [boundSingletonPart(parts[index])];
|
||||
offset = index;
|
||||
}
|
||||
break;
|
||||
}
|
||||
bounded.unshift(parts[index]);
|
||||
offset = index;
|
||||
totalBytes += separatorBytes + partBytes;
|
||||
}
|
||||
if (encodedBytes(bounded) > MAX_LIVE_ACTIVITY_BYTES) {
|
||||
bounded = [];
|
||||
offset = parts.length;
|
||||
}
|
||||
const rebase = (index: number | null): number | null =>
|
||||
index != null && index >= offset && index - offset < bounded.length ? index - offset : null;
|
||||
const toolCallIndexById = Object.fromEntries(
|
||||
bounded.flatMap((part, index) =>
|
||||
part.type === ContentTypes.TOOL_CALL ? [[part.tool_call.id, index]] : [],
|
||||
),
|
||||
);
|
||||
return {
|
||||
parts: bounded,
|
||||
state: {
|
||||
openTextIdx: rebase(state.openTextIdx),
|
||||
openThinkIdx: rebase(state.openThinkIdx),
|
||||
toolCallIndexById,
|
||||
},
|
||||
};
|
||||
};
|
||||
|
||||
const boundTickerState = (state: SubagentTickerState): SubagentTickerState => {
|
||||
const start = Math.max(0, state.lines.length - MAX_LIVE_ACTIVITY_ITEMS);
|
||||
let offset = state.lines.length;
|
||||
let totalBytes = 2;
|
||||
const lines = [] as SubagentTickerState['lines'];
|
||||
for (let index = state.lines.length - 1; index >= start; index -= 1) {
|
||||
const lineBytes = encodedBytes(state.lines[index]);
|
||||
const separatorBytes = lines.length === 0 ? 0 : 1;
|
||||
if (totalBytes + separatorBytes + lineBytes > MAX_LIVE_ACTIVITY_BYTES) break;
|
||||
lines.unshift(state.lines[index]);
|
||||
offset = index;
|
||||
totalBytes += separatorBytes + lineBytes;
|
||||
}
|
||||
const rebase = (index: number | null): number | null =>
|
||||
index != null && index >= offset ? index - offset : null;
|
||||
return {
|
||||
...state,
|
||||
lines,
|
||||
textLineIdx: rebase(state.textLineIdx),
|
||||
thinkLineIdx: rebase(state.thinkLineIdx),
|
||||
};
|
||||
};
|
||||
|
||||
const hashString = (value: string): string => {
|
||||
let hash = 2166136261;
|
||||
for (let index = 0; index < value.length; index++) {
|
||||
hash ^= value.charCodeAt(index);
|
||||
hash = Math.imul(hash, 16777619);
|
||||
}
|
||||
return (hash >>> 0).toString(36);
|
||||
};
|
||||
|
||||
const eventKey = (event: SubagentUpdateEvent): string | undefined => {
|
||||
const activityEventId = event.activityEventId?.trim();
|
||||
if (!activityEventId) return undefined;
|
||||
return hashString(`${event.subagentRunId}\u0000${activityEventId}`);
|
||||
};
|
||||
|
||||
/** One child invocation selected for the shared read-only activity panel. */
|
||||
export type ActiveSubagentPanel = {
|
||||
host: 'conversation' | 'share';
|
||||
|
|
@ -82,3 +285,237 @@ export const subagentProgressByToolCallId = atomFamily<SubagentProgress | null,
|
|||
key: 'subagentProgressByToolCallId',
|
||||
default: null,
|
||||
});
|
||||
|
||||
/** Parent delivery remains authoritative until its ordered SSE close boundary. */
|
||||
export const subagentParentStreamOpenByToolCallId = atomFamily<boolean, string>({
|
||||
key: 'subagentParentStreamOpenByToolCallId',
|
||||
default: false,
|
||||
});
|
||||
|
||||
/**
|
||||
* Invocation atoms populated by either the parent generation stream or the selected detached
|
||||
* task stream. The conversation host drains this registry on navigation so both transports share
|
||||
* one cleanup boundary instead of leaking detached-only atom-family members for the app lifetime.
|
||||
*/
|
||||
const registeredSubagentProgressKeys = new Set<string>();
|
||||
|
||||
export function registerSubagentProgressKey(key: string): void {
|
||||
registeredSubagentProgressKeys.add(key);
|
||||
}
|
||||
|
||||
export function takeRegisteredSubagentProgressKeys(): string[] {
|
||||
const keys = [...registeredSubagentProgressKeys];
|
||||
registeredSubagentProgressKeys.clear();
|
||||
return keys;
|
||||
}
|
||||
|
||||
export function listRegisteredSubagentProgressKeys(): string[] {
|
||||
return [...registeredSubagentProgressKeys];
|
||||
}
|
||||
|
||||
const validActivitySequence = (value: number | undefined): value is number =>
|
||||
Number.isSafeInteger(value) && value != null && value >= 0;
|
||||
|
||||
const foldAcceptedSubagentEvents = (
|
||||
previous: SubagentProgress | null,
|
||||
events: SubagentUpdateEvent[],
|
||||
source: 'parent' | 'detached',
|
||||
pendingSequencedEvents: SubagentUpdateEvent[],
|
||||
): SubagentProgress | null => {
|
||||
if (events.length === 0) {
|
||||
if (previous == null) {
|
||||
const first = pendingSequencedEvents[0];
|
||||
if (first == null) return null;
|
||||
return {
|
||||
subagentRunId: first.subagentRunId,
|
||||
subagentType: first.subagentType,
|
||||
subagentAgentId: first.subagentAgentId,
|
||||
contentParts: [],
|
||||
aggregatorState: initSubagentAggregatorState(),
|
||||
tickerState: initSubagentTickerState(),
|
||||
status: first.phase,
|
||||
recentEventKeys: [],
|
||||
pendingSequencedEvents,
|
||||
coverage: source === 'detached' ? 'suffix' : 'complete',
|
||||
};
|
||||
}
|
||||
if (
|
||||
(previous.pendingSequencedEvents == null && pendingSequencedEvents.length === 0) ||
|
||||
(previous.pendingSequencedEvents?.length === pendingSequencedEvents.length &&
|
||||
previous.pendingSequencedEvents.every(
|
||||
(event, index) => event === pendingSequencedEvents[index],
|
||||
))
|
||||
) {
|
||||
return previous;
|
||||
}
|
||||
return {
|
||||
...previous,
|
||||
...(pendingSequencedEvents.length === 0 ? {} : { pendingSequencedEvents }),
|
||||
};
|
||||
}
|
||||
const recentEventKeys = [...(previous?.recentEventKeys ?? [])];
|
||||
for (const event of events) {
|
||||
const key = eventKey(event);
|
||||
if (key != null) recentEventKeys.push(key);
|
||||
}
|
||||
const boundedEventKeys = recentEventKeys.slice(-MAX_RECENT_EVENT_KEYS);
|
||||
let contentParts = previous?.contentParts ?? [];
|
||||
let aggregatorState = previous?.aggregatorState ?? initSubagentAggregatorState();
|
||||
let tickerState = previous?.tickerState ?? initSubagentTickerState();
|
||||
for (const event of events) {
|
||||
const foldEvent =
|
||||
event.phase === 'reasoning_delta' &&
|
||||
event.data == null &&
|
||||
aggregatorState.openThinkIdx == null
|
||||
? {
|
||||
...event,
|
||||
data: {
|
||||
delta: {
|
||||
content: [{ type: ContentTypes.THINK, think: REDACTED_REASONING_MARKER }],
|
||||
},
|
||||
},
|
||||
}
|
||||
: event;
|
||||
({ parts: contentParts, state: aggregatorState } = foldSubagentEvent(
|
||||
contentParts,
|
||||
aggregatorState,
|
||||
foldEvent,
|
||||
));
|
||||
tickerState = foldSubagentEventIntoTicker(tickerState, foldEvent);
|
||||
}
|
||||
({ parts: contentParts, state: aggregatorState } = boundContentParts(
|
||||
contentParts,
|
||||
aggregatorState,
|
||||
));
|
||||
tickerState = boundTickerState(tickerState);
|
||||
const last = events[events.length - 1];
|
||||
const lastActivitySequence = [...events]
|
||||
.reverse()
|
||||
.map((event) => event.activitySequence)
|
||||
.find(validActivitySequence);
|
||||
const effectiveActivitySequence = lastActivitySequence ?? previous?.lastActivitySequence;
|
||||
const acceptedRunStart = events.some((event) => event.activitySequence === 0);
|
||||
return {
|
||||
subagentRunId: last.subagentRunId,
|
||||
subagentType: last.subagentType,
|
||||
subagentAgentId: last.subagentAgentId ?? previous?.subagentAgentId,
|
||||
contentParts,
|
||||
aggregatorState,
|
||||
tickerState,
|
||||
status: last.phase,
|
||||
latestLabel: last.label ?? previous?.latestLabel,
|
||||
recentEventKeys: boundedEventKeys,
|
||||
...(effectiveActivitySequence == null
|
||||
? {}
|
||||
: { lastActivitySequence: effectiveActivitySequence }),
|
||||
...(pendingSequencedEvents.length === 0 ? {} : { pendingSequencedEvents }),
|
||||
coverage: acceptedRunStart
|
||||
? 'complete'
|
||||
: (previous?.coverage ?? (source === 'detached' ? 'suffix' : 'complete')),
|
||||
};
|
||||
};
|
||||
|
||||
/** Parent SSE close is an ordering fence: all its earlier frames have already been handled. */
|
||||
export function closeParentSubagentProgress(
|
||||
previous: SubagentProgress | null,
|
||||
): SubagentProgress | null {
|
||||
if (previous?.pendingSequencedEvents == null || previous.pendingSequencedEvents.length === 0) {
|
||||
return previous;
|
||||
}
|
||||
const pending = [...previous.pendingSequencedEvents].sort(
|
||||
(left, right) => (left.activitySequence ?? 0) - (right.activitySequence ?? 0),
|
||||
);
|
||||
return foldAcceptedSubagentEvents(
|
||||
{ ...previous, pendingSequencedEvents: undefined },
|
||||
pending,
|
||||
previous.coverage === 'suffix' ? 'detached' : 'parent',
|
||||
[],
|
||||
);
|
||||
}
|
||||
|
||||
/** Shared reducer for foreground chat SSE and task-scoped detached activity SSE. */
|
||||
export function reduceSubagentProgress(
|
||||
previous: SubagentProgress | null,
|
||||
events: SubagentUpdateEvent[],
|
||||
source: 'parent' | 'detached' = 'parent',
|
||||
waitForEarlierSequences = source === 'parent',
|
||||
): SubagentProgress | null {
|
||||
if (events.length === 0) return previous;
|
||||
const recentEventKeys = [...(previous?.recentEventKeys ?? [])];
|
||||
const seen = new Set(recentEventKeys);
|
||||
const sequenced = events.every(
|
||||
(event) => Number.isSafeInteger(event.activitySequence) && (event.activitySequence ?? -1) >= 0,
|
||||
);
|
||||
const orderedEvents = sequenced
|
||||
? [...events].sort((left, right) =>
|
||||
(left.activitySequence ?? 0) === (right.activitySequence ?? 0)
|
||||
? 0
|
||||
: (left.activitySequence ?? 0) - (right.activitySequence ?? 0),
|
||||
)
|
||||
: events;
|
||||
const sameRun = previous?.subagentRunId === orderedEvents[0]?.subagentRunId;
|
||||
const lastActivitySequence = sameRun ? previous.lastActivitySequence : undefined;
|
||||
const pending = sameRun ? [...(previous.pendingSequencedEvents ?? [])] : [];
|
||||
const pendingSequences = new Set(
|
||||
pending.map((event) => event.activitySequence).filter(validActivitySequence),
|
||||
);
|
||||
const directEvents: SubagentUpdateEvent[] = [];
|
||||
let expected = lastActivitySequence == null ? 0 : lastActivitySequence + 1;
|
||||
if (!waitForEarlierSequences && lastActivitySequence == null) {
|
||||
const firstSequence = [...pending, ...orderedEvents]
|
||||
.map((event) => event.activitySequence)
|
||||
.filter(validActivitySequence)
|
||||
.sort((left, right) => left - right)[0];
|
||||
if (firstSequence != null) expected = firstSequence;
|
||||
}
|
||||
|
||||
const sanitizeSequencedEvent = (event: SubagentUpdateEvent): SubagentUpdateEvent =>
|
||||
source === 'detached' && event.phase === 'reasoning_delta'
|
||||
? { ...event, data: undefined }
|
||||
: event;
|
||||
const drainPending = () => {
|
||||
pending.sort((left, right) => (left.activitySequence ?? 0) - (right.activitySequence ?? 0));
|
||||
while (pending[0]?.activitySequence === expected) {
|
||||
const event = pending.shift();
|
||||
if (event == null) break;
|
||||
pendingSequences.delete(expected);
|
||||
directEvents.push(event);
|
||||
expected += 1;
|
||||
}
|
||||
};
|
||||
|
||||
drainPending();
|
||||
for (const event of orderedEvents) {
|
||||
const sequence = event.activitySequence;
|
||||
const key = eventKey(event);
|
||||
if (key != null && seen.has(key)) continue;
|
||||
if (validActivitySequence(sequence)) {
|
||||
if (sequence < expected || pendingSequences.has(sequence)) continue;
|
||||
const pendingEvent = sanitizeSequencedEvent(event);
|
||||
if (sequence === expected) {
|
||||
directEvents.push(pendingEvent);
|
||||
expected += 1;
|
||||
drainPending();
|
||||
} else if (
|
||||
pending.length < MAX_PENDING_SEQUENCE_EVENTS &&
|
||||
encodedBytes([...pending, pendingEvent]) <= MAX_PENDING_SEQUENCE_BYTES
|
||||
) {
|
||||
pending.push(pendingEvent);
|
||||
pendingSequences.add(sequence);
|
||||
}
|
||||
} else {
|
||||
if (key != null) seen.add(key);
|
||||
directEvents.push(event);
|
||||
}
|
||||
}
|
||||
drainPending();
|
||||
if (
|
||||
!waitForEarlierSequences &&
|
||||
pending[0]?.activitySequence != null &&
|
||||
pending[0].activitySequence > expected
|
||||
) {
|
||||
expected = pending[0].activitySequence;
|
||||
drainPending();
|
||||
}
|
||||
return foldAcceptedSubagentEvents(previous, directEvents, source, pending);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -269,6 +269,25 @@ describe('buildSubagentTickerLines', () => {
|
|||
expect(lines[0]).toEqual({ kind: 'writing', body: 'Hello world' });
|
||||
});
|
||||
|
||||
it('retains meaningful text across a whitespace-heavy stream', () => {
|
||||
const lines = buildSubagentTickerLines([
|
||||
makeEvent({
|
||||
phase: 'message_delta',
|
||||
data: { delta: { content: [{ type: 'text', text: 'Visible' }] } },
|
||||
}),
|
||||
makeEvent({
|
||||
phase: 'message_delta',
|
||||
data: { delta: { content: [{ type: 'text', text: ' \n'.repeat(1200) }] } },
|
||||
}),
|
||||
makeEvent({
|
||||
phase: 'message_delta',
|
||||
data: { delta: { content: [{ type: 'text', text: 'again' }] } },
|
||||
}),
|
||||
]);
|
||||
|
||||
expect(lines[0]).toEqual({ kind: 'writing', body: 'Visible again' });
|
||||
});
|
||||
|
||||
it('truncates the writing body to the tail when it grows past the cap', () => {
|
||||
const longText = 'x'.repeat(1000);
|
||||
const lines = buildSubagentTickerLines([
|
||||
|
|
|
|||
|
|
@ -294,8 +294,8 @@ export interface SubagentTickerState {
|
|||
textLineIdx: number | null;
|
||||
/** Index of the in-flight 'reasoning' line. */
|
||||
thinkLineIdx: number | null;
|
||||
/** Raw message-delta accumulator — truncated into `writing.body` but
|
||||
* preserved so subsequent deltas extend the running preview. */
|
||||
/** Whitespace-normalized message-delta accumulator. A trailing separator is
|
||||
* retained so chunk boundaries still render as one word boundary. */
|
||||
textBuffer: string;
|
||||
thinkBuffer: string;
|
||||
}
|
||||
|
|
@ -317,12 +317,20 @@ export function initSubagentTickerState(): SubagentTickerState {
|
|||
* CSS ellipsis — double-eliding would render a stray dot character
|
||||
* right next to the "Writing:" / "Reasoning:" label. */
|
||||
const PREVIEW_MAX_CHARS = 300;
|
||||
const PREVIEW_BUFFER_MAX_CHARS = PREVIEW_MAX_CHARS * 4;
|
||||
const truncatePreview = (input: string): string => {
|
||||
const normalized = input.replace(/\s+/g, ' ').trim();
|
||||
if (normalized.length <= PREVIEW_MAX_CHARS) return normalized;
|
||||
return normalized.slice(-PREVIEW_MAX_CHARS);
|
||||
};
|
||||
|
||||
const appendPreviewBuffer = (buffer: string, chunk: string): string => {
|
||||
const normalized = `${buffer}${chunk}`.replace(/\s+/g, ' ').trimStart();
|
||||
return normalized.length <= PREVIEW_BUFFER_MAX_CHARS
|
||||
? normalized
|
||||
: normalized.slice(-PREVIEW_BUFFER_MAX_CHARS);
|
||||
};
|
||||
|
||||
const SNIPPET_MAX_CHARS = 48;
|
||||
/** Short head-truncation for tool args/output — caller labels what each
|
||||
* side is. Whitespace collapsed so multi-line outputs stay one line. */
|
||||
|
|
@ -396,7 +404,7 @@ export function foldSubagentEventIntoTicker(
|
|||
state.thinkLineIdx != null || state.thinkBuffer
|
||||
? { ...state, thinkLineIdx: null, thinkBuffer: '' }
|
||||
: state;
|
||||
const textBuffer = afterClose.textBuffer + chunk;
|
||||
const textBuffer = appendPreviewBuffer(afterClose.textBuffer, chunk);
|
||||
const body = truncatePreview(textBuffer);
|
||||
const line: SubagentTickerLine = { kind: 'writing', body };
|
||||
if (afterClose.textLineIdx == null) {
|
||||
|
|
@ -416,7 +424,7 @@ export function foldSubagentEventIntoTicker(
|
|||
state.textLineIdx != null || state.textBuffer
|
||||
? { ...state, textLineIdx: null, textBuffer: '' }
|
||||
: state;
|
||||
const thinkBuffer = afterClose.thinkBuffer + chunk;
|
||||
const thinkBuffer = appendPreviewBuffer(afterClose.thinkBuffer, chunk);
|
||||
const body = truncatePreview(thinkBuffer);
|
||||
const line: SubagentTickerLine = { kind: 'reasoning', body };
|
||||
if (afterClose.thinkLineIdx == null) {
|
||||
|
|
@ -447,7 +455,7 @@ export function foldSubagentEventIntoTicker(
|
|||
typeof tc?.name === 'string' && tc.name.length > 0,
|
||||
);
|
||||
if (named.length === 0) return afterClose;
|
||||
const toolNames = named.map((tc) => tc.name);
|
||||
const toolNames = named.slice(0, 16).map((tc) => truncateSnippet(tc.name));
|
||||
const argsSnippet = named.length === 1 ? summarizeArgs(named[0].args) : undefined;
|
||||
const line: SubagentTickerLine = {
|
||||
kind: 'using_tool',
|
||||
|
|
@ -464,7 +472,7 @@ export function foldSubagentEventIntoTicker(
|
|||
const outputSnippet = tc.output != null ? summarizeOutput(tc.output) : undefined;
|
||||
const line: SubagentTickerLine = {
|
||||
kind: 'tool_complete',
|
||||
toolName: tc.name,
|
||||
toolName: truncateSnippet(tc.name),
|
||||
...(outputSnippet ? { outputSnippet } : {}),
|
||||
};
|
||||
return { ...state, lines: state.lines.concat(line) };
|
||||
|
|
@ -474,7 +482,7 @@ export function foldSubagentEventIntoTicker(
|
|||
const data = event.data as ErrorData | undefined;
|
||||
const line: SubagentTickerLine = {
|
||||
kind: 'error',
|
||||
...(data?.message ? { message: data.message } : {}),
|
||||
...(data?.message ? { message: truncatePreview(data.message) } : {}),
|
||||
};
|
||||
return { ...state, lines: state.lines.concat(line) };
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue