mirror of
https://github.com/danny-avila/LibreChat.git
synced 2026-10-21 01:22:54 +00:00
⏳ fix: Anchor Resumed Elapsed Time at the Generation's Real Start (#15204)
Some checks failed
Publish `librechat-data-provider` to NPM / pack (push) Waiting to run
Publish `librechat-data-provider` to NPM / publish-npm (push) Blocked by required conditions
Docker Dev Images Build / build (Dockerfile, librechat-dev, node) (push) Waiting to run
Docker Dev Images Build / build (Dockerfile.multi, librechat-dev-api, api-build) (push) Waiting to run
Sync Helm Chart Tags / Ignore non-main push (push) Waiting to run
Sync Helm Chart Tags / Sync chart tags (push) Waiting to run
Publish `@librechat/client` to NPM / pack (push) Has been cancelled
Publish `@librechat/data-schemas` to NPM / pack (push) Has been cancelled
Publish `@librechat/client` to NPM / publish-npm (push) Has been cancelled
Publish `@librechat/data-schemas` to NPM / publish-npm (push) Has been cancelled
Some checks failed
Publish `librechat-data-provider` to NPM / pack (push) Waiting to run
Publish `librechat-data-provider` to NPM / publish-npm (push) Blocked by required conditions
Docker Dev Images Build / build (Dockerfile, librechat-dev, node) (push) Waiting to run
Docker Dev Images Build / build (Dockerfile.multi, librechat-dev-api, api-build) (push) Waiting to run
Sync Helm Chart Tags / Ignore non-main push (push) Waiting to run
Sync Helm Chart Tags / Sync chart tags (push) Waiting to run
Publish `@librechat/client` to NPM / pack (push) Has been cancelled
Publish `@librechat/data-schemas` to NPM / pack (push) Has been cancelled
Publish `@librechat/client` to NPM / publish-npm (push) Has been cancelled
Publish `@librechat/data-schemas` to NPM / publish-npm (push) Has been cancelled
* ⏳ fix: Anchor Resumed Elapsed Time at the Generation's Real Start A reload emptied the Recoil anchor, so the indicator fell back to its mount time and visibly reset to 0s over a run that had been generating for much longer. The stream status the resume path already reads carries the server-recorded generation start; the fill now prefers it, so a reattached run reports real elapsed time. The fill remains fill-only: a same-session reattach keeps its original ask baseline, and the indicator's existing clamp absorbs any client/server clock skew. * 🕰️ fix: Rebuild the Resumed Baseline From the Server-Computed Age Codex round 1: anchoring at the server's raw createdAt compares two clocks — a client behind the server froze the resumed reading at 0s for the skew, one ahead inflated it. The status route now also reports the generation's age computed on its own clock, and the client rebuilds a clock-local anchor as Date.now() minus that age, so each machine only ever compares to itself. Raw createdAt stays as the fallback for an older server mid-rolling-deploy. * 📥 fix: Compute Elapsed Age in TypeScript, Anchor It at Status Receipt Codex round 2: the elapsed computation moves into packages/api as getGenerationElapsedMs — the route now just delegates, keeping the response contract type-checked and the /api surface a thin wrapper — and the client subtracts the age from the moment the status response arrived (dataUpdatedAt) rather than from apply time, so a slow history fetch between receipt and apply can no longer shrink the reading. Declined with rationale: a shared clock source across replicas — the residual is inter-replica NTP drift, milliseconds against the minutes-scale client skew this PR eliminates, and the helper gives any future shared-clock upgrade a single home.
This commit is contained in:
parent
d9e6250d05
commit
6d499ba3ce
8 changed files with 155 additions and 6 deletions
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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. */
|
||||
|
|
|
|||
|
|
@ -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({
|
|||
<QueryClientProvider client={queryClient}>
|
||||
<RecoilRoot initializeState={initializeState}>
|
||||
<SubmissionProbe />
|
||||
<SubmissionStartProbe />
|
||||
<SiblingIndexProbe />
|
||||
<PendingSteersProbe />
|
||||
<QueuedMessagesProbe />
|
||||
|
|
@ -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<number | null> = [];
|
||||
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<number | null> = [];
|
||||
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<number | null> = [];
|
||||
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 = {
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -42,9 +42,10 @@ const submissionByIndex = atomFamily<TSubmission | null, string | number>({
|
|||
* 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.
|
||||
|
|
|
|||
13
packages/api/src/stream/__tests__/elapsed.spec.ts
Normal file
13
packages/api/src/stream/__tests__/elapsed.spec.ts
Normal file
|
|
@ -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);
|
||||
});
|
||||
});
|
||||
9
packages/api/src/stream/elapsed.ts
Normal file
9
packages/api/src/stream/elapsed.ts
Normal file
|
|
@ -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);
|
||||
}
|
||||
|
|
@ -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';
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue