diff --git a/api/server/routes/agents/index.js b/api/server/routes/agents/index.js index 3ab83f6613..8768fde8f5 100644 --- a/api/server/routes/agents/index.js +++ b/api/server/routes/agents/index.js @@ -7,6 +7,7 @@ const { buildAbortedResponseMetadata, isPendingActionStale, toClientPendingAction, + getGenerationElapsedMs, isHITLEnabled, captureAgentCheckpointGeneration, deleteAgentCheckpoint, @@ -524,6 +525,7 @@ router.get('/chat/status/:conversationId', async (req, res) => { status: job.status, aggregatedContent: resumeState?.aggregatedContent ?? [], createdAt: job.createdAt, + elapsedMs: getGenerationElapsedMs(job), resumeState, // Surface the live pending approval so a client rebuilding from /chat/status // (reload / cross-replica) has the action id + payload to render and submit diff --git a/client/src/data-provider/SSE/queries.ts b/client/src/data-provider/SSE/queries.ts index d7d0813e75..ca5834c2f1 100644 --- a/client/src/data-provider/SSE/queries.ts +++ b/client/src/data-provider/SSE/queries.ts @@ -17,6 +17,9 @@ export interface StreamStatusResponse { status?: 'running' | 'complete' | 'error' | 'aborted' | 'requires_action'; aggregatedContent?: Array<{ type: string; text?: string }>; createdAt?: number; + /** Generation age computed on the server's clock, so a reattaching client + * can rebuild a clock-local elapsed baseline free of cross-machine skew. */ + elapsedMs?: number; resumeState?: Agents.ResumeState; /** Live pending approval when `status === 'requires_action'`; mirrors * `resumeState.pendingAction`, surfaced top-level for the resume-on-load path. */ diff --git a/client/src/hooks/SSE/__tests__/useResumeOnLoad.spec.tsx b/client/src/hooks/SSE/__tests__/useResumeOnLoad.spec.tsx index 1af5beda97..b19721b91c 100644 --- a/client/src/hooks/SSE/__tests__/useResumeOnLoad.spec.tsx +++ b/client/src/hooks/SSE/__tests__/useResumeOnLoad.spec.tsx @@ -75,6 +75,8 @@ function renderUseResumeOnLoad({ pendingSteers, onPendingSteers, onQueuedMessages, + submissionStart, + onSubmissionStart, }: { messages?: TMessage[]; getMessages?: () => TMessage[] | undefined; @@ -87,6 +89,8 @@ function renderUseResumeOnLoad({ pendingSteers?: PendingSteer[]; onPendingSteers?: (steers: PendingSteer[]) => void; onQueuedMessages?: (queued: QueuedMessage[]) => void; + submissionStart?: number; + onSubmissionStart?: (submissionStart: number | null) => void; }) { const getMessages = jest.fn(getMessagesOverride ?? (() => messages)); const queryClient = new QueryClient({ @@ -96,6 +100,9 @@ function renderUseResumeOnLoad({ const initializeState = (snapshot: MutableSnapshot) => { snapshot.set(store.conversationByIndex(0), buildConversation(conversationId)); snapshot.set(store.submissionByIndex(0), submission); + if (submissionStart != null) { + snapshot.set(store.submissionStartFamily(0), submissionStart); + } if (pendingSteers) { snapshot.set(store.pendingSteersByConvoId(conversationId), pendingSteers); } @@ -107,6 +114,11 @@ function renderUseResumeOnLoad({ onSubmission?.(currentSubmission); return null; }; + const SubmissionStartProbe = () => { + const currentStart = useRecoilValue(store.submissionStartFamily(0)); + onSubmissionStart?.(currentStart); + return null; + }; const PendingSteersProbe = () => { const steers = useRecoilValue(store.pendingSteersByConvoId(conversationId)); onPendingSteers?.(steers); @@ -129,6 +141,7 @@ function renderUseResumeOnLoad({ + @@ -290,6 +303,97 @@ describe('useResumeOnLoad', () => { expect(attached?.resumeGenerationCreatedAt).toBe(4242); }); + /** The elapsed indicator's baseline must be the generation's real start: + * an attach with no surviving anchor (a reload, or a run another client + * started) rebuilds it clock-locally from the server-computed generation + * age, subtracted from the moment the status response ARRIVED — so + * neither cross-machine clock skew nor a slow history fetch between + * receipt and apply can distort the reading. */ + it('anchors the elapsed baseline via the server-computed generation age', async () => { + const observedStarts: Array = []; + mockUseStreamStatus.mockReturnValue(INACTIVE_STATUS); + + const { rerender } = renderUseResumeOnLoad({ + messages: [buildUserMessage(CONVERSATION_ID)], + onSubmissionStart: (start) => observedStarts.push(start), + }); + await act(async () => { + await Promise.resolve(); + }); + expect(observedStarts[observedStarts.length - 1]).toBeNull(); + + mockUseActiveJobs.mockReturnValue({ + data: { activeJobIds: [CONVERSATION_ID] }, + dataUpdatedAt: 2, + }); + mockUseStreamStatus.mockReturnValue({ + ...ACTIVE_STATUS, + dataUpdatedAt: 1_000_000, + data: { ...ACTIVE_STATUS.data, elapsedMs: 30_000 }, + }); + rerender(); + await act(async () => { + await Promise.resolve(); + }); + + expect(observedStarts[observedStarts.length - 1]).toBe(1_000_000 - 30_000); + }); + + /** An older server without `elapsedMs` still beats counting from attach: + * the raw server start is adopted and the indicator clamps its skew. */ + it('falls back to the server-recorded generation start without elapsedMs', async () => { + const observedStarts: Array = []; + mockUseStreamStatus.mockReturnValue(INACTIVE_STATUS); + + const { rerender } = renderUseResumeOnLoad({ + messages: [buildUserMessage(CONVERSATION_ID)], + onSubmissionStart: (start) => observedStarts.push(start), + }); + await act(async () => { + await Promise.resolve(); + }); + + mockUseActiveJobs.mockReturnValue({ + data: { activeJobIds: [CONVERSATION_ID] }, + dataUpdatedAt: 2, + }); + mockUseStreamStatus.mockReturnValue(ACTIVE_STATUS); + rerender(); + await act(async () => { + await Promise.resolve(); + }); + + expect(observedStarts[observedStarts.length - 1]).toBe(4242); + }); + + /** A reattach to the run this session already anchored must keep the ask + * baseline — the atom deliberately outlives the submission it timed. */ + it('preserves a surviving elapsed baseline over the server-recorded start', async () => { + const observedStarts: Array = []; + mockUseStreamStatus.mockReturnValue(INACTIVE_STATUS); + + const { rerender } = renderUseResumeOnLoad({ + messages: [buildUserMessage(CONVERSATION_ID)], + submissionStart: 1111, + onSubmissionStart: (start) => observedStarts.push(start), + }); + await act(async () => { + await Promise.resolve(); + }); + + mockUseActiveJobs.mockReturnValue({ + data: { activeJobIds: [CONVERSATION_ID] }, + dataUpdatedAt: 2, + }); + mockUseStreamStatus.mockReturnValue(ACTIVE_STATUS); + rerender(); + await act(async () => { + await Promise.resolve(); + }); + + expect(observedStarts[observedStarts.length - 1]).toBe(1111); + }); + it('restores an externally started regeneration after history refreshes', async () => { const rootUser = buildUserMessage(CONVERSATION_ID, 'root-user'); const olderResponse = { diff --git a/client/src/hooks/SSE/useResumeOnLoad.ts b/client/src/hooks/SSE/useResumeOnLoad.ts index 6d6bad01d3..1727758b24 100644 --- a/client/src/hooks/SSE/useResumeOnLoad.ts +++ b/client/src/hooks/SSE/useResumeOnLoad.ts @@ -463,6 +463,7 @@ export default function useResumeOnLoad( const { data: streamStatus, + dataUpdatedAt: streamStatusUpdatedAt, isSuccess, isFetching, } = useStreamStatus(conversationId, shouldCheck); @@ -609,9 +610,23 @@ export default function useResumeOnLoad( const messages = getMessages() || []; /** Fill the elapsed baseline only when none survives: a reattach to the run * this session already anchored keeps its original start (the atom outlives - * the submission), while a run it never anchored — another client's, or any - * attach after the previous run's terminal clear — counts from attach. */ - setSubmissionStart((prev) => prev ?? Date.now()); + * the submission). A run it never anchored — after a reload, or started by + * another client — rebuilds the real baseline from the server-computed + * generation age, which keeps both clocks comparing only to themselves. + * The age is subtracted from the moment the status RESPONSE arrived + * (`dataUpdatedAt`), not from now: this effect trails the response behind + * `messagesLoaded`, and a slow history fetch would silently shrink the + * reading. Raw `createdAt` (an older server) still beats counting from + * attach; the indicator clamps whatever skew that carries. */ + setSubmissionStart((prev) => { + if (prev != null) { + return prev; + } + if (streamStatus.elapsedMs != null) { + return (streamStatusUpdatedAt || Date.now()) - streamStatus.elapsedMs; + } + return streamStatus.createdAt ?? Date.now(); + }); // Build submission from resume state if available if (streamStatus.resumeState) { @@ -679,6 +694,7 @@ export default function useResumeOnLoad( isSuccess, isFetching, streamStatus, + streamStatusUpdatedAt, getMessages, setSubmission, setSubmissionStart, diff --git a/client/src/store/families.ts b/client/src/store/families.ts index e3956516bc..40184b34a2 100644 --- a/client/src/store/families.ts +++ b/client/src/store/families.ts @@ -42,9 +42,10 @@ const submissionByIndex = atomFamily({ * Epoch ms baseline for the streaming elapsed indicator at this chat index. * Stamped when this session submits a generation (every path through `ask`), * cleared by the terminal handlers when that generation ends, and only FILLED - * — never overwritten — when resume-on-load attaches a run. The reading - * therefore survives mid-stream remounts (new-conversation id hydration, - * navigating away from a still-live run and back) without a later, + * — never overwritten — when resume-on-load attaches a run, preferring the + * server-recorded generation start so a reload reports real elapsed time. + * The reading therefore survives mid-stream remounts (new-conversation id + * hydration, navigating away from a still-live run and back) without a later, * externally-started generation inheriting a stale baseline. Known residual: * a run whose end this pane never observed (left mid-stream, finished * elsewhere) leaves its stamp for the next attach at this index to inherit. diff --git a/packages/api/src/stream/__tests__/elapsed.spec.ts b/packages/api/src/stream/__tests__/elapsed.spec.ts new file mode 100644 index 0000000000..19fbfd68ff --- /dev/null +++ b/packages/api/src/stream/__tests__/elapsed.spec.ts @@ -0,0 +1,13 @@ +import { getGenerationElapsedMs } from '../elapsed'; + +describe('getGenerationElapsedMs', () => { + it('reports the age of the job on this clock', () => { + const elapsed = getGenerationElapsedMs({ createdAt: Date.now() - 5_000 }); + expect(elapsed).toBeGreaterThanOrEqual(5_000); + expect(elapsed).toBeLessThan(6_000); + }); + + it('clamps a creator clock ahead of this replica to zero', () => { + expect(getGenerationElapsedMs({ createdAt: Date.now() + 60_000 })).toBe(0); + }); +}); diff --git a/packages/api/src/stream/elapsed.ts b/packages/api/src/stream/elapsed.ts new file mode 100644 index 0000000000..0582319382 --- /dev/null +++ b/packages/api/src/stream/elapsed.ts @@ -0,0 +1,9 @@ +/** + * Age of a generation computed entirely on this process's clock, for status + * responses whose clients rebuild a clock-local elapsed baseline instead of + * comparing timestamps across machines. Clamped: a job stamped by a replica + * whose clock ran ahead of this one must never report a negative age. + */ +export function getGenerationElapsedMs(job: { createdAt: number }): number { + return Math.max(0, Date.now() - job.createdAt); +} diff --git a/packages/api/src/stream/index.ts b/packages/api/src/stream/index.ts index 6ef140e0ae..61f788fc05 100644 --- a/packages/api/src/stream/index.ts +++ b/packages/api/src/stream/index.ts @@ -54,6 +54,7 @@ export type { RecoveredSteerPayload } from './SteerRecovery'; export { createStreamServices } from './createStreamServices'; export type { StreamServicesConfig, StreamServices } from './createStreamServices'; export { filterPersistableAbortContent, hasPersistableAbortContent } from './abortContent'; +export { getGenerationElapsedMs } from './elapsed'; // Implementations (for advanced use cases) export { InMemoryJobStore } from './implementations/InMemoryJobStore';