6900c30935
Reopening a chat whose agent run is still going showed a frozen snapshot from the moment it was opened. Add a passive-observer reconnect-poll path: when this tab did NOT start the run locally, poll POST /ai-chat/run every 2s while the run is pending/running and merge its incrementally-persisted assistant message into the thread, so new steps/tool-calls and the growing text appear live. Polling stops on terminal status (refetchInterval keyed on run.status, mirroring the reindex polling); a final messages invalidate shows the persisted end state. Observer-vs-streamer detection: ChatThread reports its local useChat streaming status up; the window only polls/merges while NOT locally streaming (the streamer's SSE owns the view — no double-render). Gated by settings.ai.autonomousRuns; the query is disabled when the feature is off so the flag-gated endpoint is never hit, and a failed fetch can't loop (retry:false -> refetchInterval(undefined)=false). Pure decisions (poll interval, observe gate, message merge) extracted to run-polling.ts and unit-tested; added query enable-gating and ChatThread observer-merge tests. Client-only change — the reconnect endpoint already returns the run plus the assistant message with its metadata.parts. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
72 lines
3.0 KiB
TypeScript
72 lines
3.0 KiB
TypeScript
import type { UIMessage } from "@ai-sdk/react";
|
|
import type { IAiChatRun } from "@/features/ai-chat/types/ai-chat.types.ts";
|
|
|
|
/**
|
|
* Reconnect-and-live-follow helpers (#184). When a chat is reopened while its
|
|
* agent run is STILL going, this tab is a PASSIVE OBSERVER: it did not start the
|
|
* run here (no local SSE stream), so it catches up by POLLING the reconnect
|
|
* endpoint (`POST /ai-chat/run`) and merging the run's incrementally-persisted
|
|
* assistant message into the rendered thread. These are the small pure decisions
|
|
* that machinery hangs off, extracted so they can be unit-tested in isolation
|
|
* (mirrors how reindex polling / editor-sync-state are tested).
|
|
*/
|
|
|
|
/** How often to re-poll the reconnect endpoint while a run is ACTIVE. */
|
|
export const RUN_POLL_INTERVAL_MS = 2000;
|
|
|
|
// 'pending' and 'running' are the two ACTIVE statuses; 'succeeded' | 'failed' |
|
|
// 'aborted' are TERMINAL (and any unknown future status is treated as terminal,
|
|
// so a stale/odd value never polls forever).
|
|
const ACTIVE_STATUSES = new Set(["pending", "running"]);
|
|
|
|
/** Whether a run is still going (worth polling / merging live updates from). */
|
|
export function isRunActive(run: IAiChatRun | null | undefined): boolean {
|
|
return !!run && ACTIVE_STATUSES.has(run.status);
|
|
}
|
|
|
|
/**
|
|
* The TanStack Query `refetchInterval` value for the run query: poll every
|
|
* {@link RUN_POLL_INTERVAL_MS} while the run is active, and `false` (stop) once
|
|
* it is terminal or there is no run. Polling is thus naturally bounded by the run
|
|
* reaching a terminal status — no separate timeout cap is needed.
|
|
*/
|
|
export function runPollInterval(
|
|
run: IAiChatRun | null | undefined,
|
|
): number | false {
|
|
return isRunActive(run) ? RUN_POLL_INTERVAL_MS : false;
|
|
}
|
|
|
|
/**
|
|
* Observer-vs-streamer decision. We render the polled run message (catch up +
|
|
* keep advancing) ONLY when this tab is a passive observer: there IS a run AND
|
|
* this tab is NOT the one locally streaming it (we reconnected, we didn't start
|
|
* it here). When this tab is the streamer, the live SSE stream owns the view, so
|
|
* we neither poll nor merge — avoiding a double-render fight. Terminal runs still
|
|
* merge (so the final persisted output is shown on reopen); the poll itself is
|
|
* stopped separately by {@link runPollInterval}.
|
|
*/
|
|
export function shouldObserveRun(
|
|
run: IAiChatRun | null | undefined,
|
|
localStreaming: boolean,
|
|
): boolean {
|
|
return !!run && !localStreaming;
|
|
}
|
|
|
|
/**
|
|
* Merge an observed assistant message into the rendered list: replace the message
|
|
* with the same id in place (the in-progress assistant row is already seeded from
|
|
* history, so per-step growth replaces it), or append it when absent. Returns a
|
|
* new array; the input is never mutated.
|
|
*/
|
|
export function mergeObservedMessage(
|
|
messages: UIMessage[],
|
|
observed: UIMessage | null | undefined,
|
|
): UIMessage[] {
|
|
if (!observed) return messages;
|
|
const idx = messages.findIndex((m) => m.id === observed.id);
|
|
if (idx === -1) return [...messages, observed];
|
|
const next = messages.slice();
|
|
next[idx] = observed;
|
|
return next;
|
|
}
|