mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-09-14 08:00:10 +00:00
fix(frontend): keep human input cards with their turn (#4892)
* fix(frontend): keep human input cards with their turn * fix(frontend): keep human input cards with the correct turn Multi-turn ordering in restoreLocalTurnMessageOrder could place an `ask_clarification` (needYourHelp) card on the wrong side of a newly submitted human message, and after an interrupt/stop it could move the current run's own already-executed steps above the human that started them. - Restore established messages that a live checkpoint tail wove after the new human (displacedBaselineMessages). - Treat cards/messages confirmed only by the REST history page as established-past-turn too (confirmedHistoryIdentities), not as in-flight pending steps (displacedHistoryMessages). - Never displace the CURRENT run's own steps after an interrupt/stop; they belong after the human even once canonical history confirms them (currentTurnRunIds, anchored by the pending human's run_id). Fixes #4889 * fix(frontend): preserve ordering across displaced messages * fix(frontend): close canonical history ordering gaps * fix(frontend): preserve current turn anchor after compaction * fix(frontend): anchor the local turn on the submitted human identity Follow-up to #4892. R2 is reachable through the full hook chain: when the checkpoint baseline covers only the latest turn, the server echo of the submitted human confirms the optimistic copy against the unthrottled SDK state while the ~80ms render snapshot cannot show it yet; the baseline-only anchor scan then promoted an older history-only human into the current turn's anchor and moved established history behind it. - Record a LocalTurnAnchor at dispatch: one client-generated human id is shared by the optimistic display copy and the submitted message, so the server X__user echo confirms the exact identity already on screen. - restoreLocalTurnMessageOrder repairs only when that identity is present in the display; a null anchor (hidden human-input reply, regenerate replay) or a not-yet-rendered identity keeps established history untouched. - Optimistic confirmation now observes the same coalesced render snapshot (identity match first, rendered human-count growth as fallback for runtime-re-keyed first turns) instead of the per-chunk array. - Edit replays adopt the prepare response's replacement identity; the render ledger excludes unconfirmed optimistic copies by identity now that the local input no longer uses an opt- prefix, so a failed send cannot pin a message the server never saw. - Anchor lifecycle matches the previous baseline: kept across finish/stop/error until canonical data takes over, replaced by the next local submit, cleared on send failure, thread switch, and replay gaps. * test(threads): type submit mock calls in local-turn-order dom tests * fix(frontend): bound local turn repair to pre-submit history * fix(frontend): preserve pre-submit bridge ordering --------- Co-authored-by: 肘子香香 <hyh112300@163.com> Co-authored-by: 霍英豪 <huoyinghao250707@credithc.com> Co-authored-by: wangzeren <1004695029@qq.com>
This commit is contained in:
Beverly621
GitHub
parent
6d5d7bb1d5
commit
0464502af1
File diff suppressed because one or more lines are too long
@ -34,6 +34,7 @@ import { taskEventToSubtaskUpdate } from "../tasks/lifecycle";
|
||||
import { messageToStep } from "../tasks/steps";
|
||||
import type { UploadedFileInfo } from "../uploads";
|
||||
import { promptInputFilePartToFile, uploadFiles } from "../uploads";
|
||||
import { uuid } from "../utils/uuid";
|
||||
|
||||
import {
|
||||
branchThreadFromTurn,
|
||||
@ -161,16 +162,24 @@ export function buildThreadSubmitMessages({
|
||||
additionalKwargs,
|
||||
additionalInputMessages = [],
|
||||
filesForSubmit = [],
|
||||
humanMessageId,
|
||||
}: {
|
||||
text: string;
|
||||
additionalKwargs?: Record<string, unknown>;
|
||||
additionalInputMessages?: Message[];
|
||||
filesForSubmit?: FileInMessage[];
|
||||
/**
|
||||
* Client-generated id for the visible human message. The optimistic display
|
||||
* copy and the actual submit share it, so the server echo (`<id>__user`)
|
||||
* confirms the exact message the user already sees.
|
||||
*/
|
||||
humanMessageId?: string;
|
||||
}): Message[] {
|
||||
return [
|
||||
...additionalInputMessages,
|
||||
{
|
||||
type: "human",
|
||||
...(humanMessageId ? { id: humanMessageId } : {}),
|
||||
content: [
|
||||
{
|
||||
type: "text",
|
||||
@ -190,6 +199,31 @@ export function buildThreadSubmitMessages({
|
||||
const EMPTY_MESSAGES: Message[] = [];
|
||||
const EMPTY_RUN_MESSAGES: RunMessage[] = [];
|
||||
const EMPTY_MESSAGE_IDENTITIES: readonly string[] = [];
|
||||
const EMPTY_MESSAGE_IDENTITIES_SET: ReadonlySet<string> = new Set<string>();
|
||||
/**
|
||||
* The turn this client submitted, recorded at dispatch time. The visible human
|
||||
* input gets one client-generated identity shared by the optimistic display
|
||||
* copy and the submitted message, so the turn anchor is a *known* identity
|
||||
* instead of a guess derived from the pre-submit baseline. `humanIdentity` is
|
||||
* the normalized identity (`message:<id>`) of that human, or null when the
|
||||
* turn has no visible human (hidden human-input reply, regenerate replay) —
|
||||
* such turns must never borrow an older visible human as their anchor.
|
||||
*/
|
||||
export type LocalTurnAnchor = {
|
||||
threadId: string;
|
||||
humanIdentity: string | null;
|
||||
baselineIdentities: ReadonlySet<string>;
|
||||
/** Canonical REST-history identities already loaded when this turn began. */
|
||||
preSubmitHistoryIdentities: ReadonlySet<string>;
|
||||
/** Transient-bridge identities already established before this turn began. */
|
||||
preSubmitBridgeIdentities: ReadonlySet<string>;
|
||||
/**
|
||||
* Highest authoritative feed position known before submit. Older pages that
|
||||
* arrive later may still be confirmed as pre-submit history through this
|
||||
* boundary; messages from later external turns may not.
|
||||
*/
|
||||
preSubmitMaxSeq?: number;
|
||||
};
|
||||
|
||||
function isNonEmptyString(value: string | undefined): value is string {
|
||||
return typeof value === "string" && value.length > 0;
|
||||
@ -207,6 +241,58 @@ const SUMMARIZATION_MIDDLEWARE_UPDATE_KEYS = new Set([
|
||||
"DeerFlowSummarizationMiddleware.before_model",
|
||||
]);
|
||||
|
||||
function maxMessageSeq(messages: Message[]): number | undefined {
|
||||
let maxSeq: number | undefined;
|
||||
for (const message of messages) {
|
||||
const seq = trustedMessageSeq(message);
|
||||
if (seq !== undefined && (maxSeq === undefined || seq > maxSeq)) {
|
||||
maxSeq = seq;
|
||||
}
|
||||
}
|
||||
return maxSeq;
|
||||
}
|
||||
|
||||
function getConfirmedPreSubmitHistoryIdentities(
|
||||
visibleHistory: Message[],
|
||||
localTurnAnchor: LocalTurnAnchor | null,
|
||||
): Set<string> {
|
||||
if (localTurnAnchor === null) {
|
||||
return new Set();
|
||||
}
|
||||
const confirmed = new Set([
|
||||
...localTurnAnchor.preSubmitHistoryIdentities,
|
||||
...localTurnAnchor.preSubmitBridgeIdentities,
|
||||
]);
|
||||
const maxSeq = localTurnAnchor.preSubmitMaxSeq;
|
||||
if (maxSeq === undefined) {
|
||||
return confirmed;
|
||||
}
|
||||
for (const message of visibleHistory) {
|
||||
const identity = messageIdentity(message);
|
||||
const seq = trustedMessageSeq(message);
|
||||
if (identity !== undefined && seq !== undefined && seq <= maxSeq) {
|
||||
confirmed.add(identity);
|
||||
}
|
||||
}
|
||||
return confirmed;
|
||||
}
|
||||
|
||||
function findMessageRunIdByIdentity(
|
||||
messages: Message[],
|
||||
identity: string,
|
||||
): string | undefined {
|
||||
for (const message of messages) {
|
||||
if (messageIdentity(message) !== identity) {
|
||||
continue;
|
||||
}
|
||||
const runId = getMessageRunId(message);
|
||||
if (runId) {
|
||||
return runId;
|
||||
}
|
||||
}
|
||||
return undefined;
|
||||
}
|
||||
|
||||
function dedupeRunMessagesByIdentity(messages: RunMessage[]): RunMessage[] {
|
||||
const lastIndexByIdentity = new Map<string, number>();
|
||||
messages.forEach((message, index) => {
|
||||
@ -456,31 +542,133 @@ export function reconcileThreadHistoryRows(
|
||||
// this module keep working unchanged.
|
||||
export { mergeMessages };
|
||||
|
||||
/**
|
||||
* Collect live run ids that were not part of the pre-submit checkpoint.
|
||||
* An empty result is safe because restoreLocalTurnMessageOrder independently
|
||||
* anchors the current turn from the pending human's run_id when interrupt/stop
|
||||
* has already flushed the live steps into canonical history.
|
||||
*/
|
||||
export function getCurrentTurnRunIds(
|
||||
messages: Message[],
|
||||
baselineMessageIdentities: ReadonlySet<string> | null,
|
||||
confirmedHistoryIdentities: ReadonlySet<string> = EMPTY_MESSAGE_IDENTITIES_SET,
|
||||
): Set<string> {
|
||||
const runIds = new Set<string>();
|
||||
if (baselineMessageIdentities === null) {
|
||||
return runIds;
|
||||
}
|
||||
|
||||
for (const message of messages) {
|
||||
if (
|
||||
(message.type !== "ai" && message.type !== "tool") ||
|
||||
isHiddenFromUIMessage(message)
|
||||
) {
|
||||
continue;
|
||||
}
|
||||
const identity = messageIdentity(message);
|
||||
const runId = getMessageRunId(message);
|
||||
if (
|
||||
runId &&
|
||||
(!identity ||
|
||||
(!baselineMessageIdentities.has(identity) &&
|
||||
!confirmedHistoryIdentities.has(identity)))
|
||||
) {
|
||||
runIds.add(runId);
|
||||
}
|
||||
}
|
||||
return runIds;
|
||||
}
|
||||
|
||||
/**
|
||||
* Keep messages from a locally submitted turn behind that turn's user input.
|
||||
* LangGraph `messages-tuple` events can publish the first AI/tool steps before
|
||||
* canonical history contains the user message. Those steps are not part of the
|
||||
* pre-submit baseline, so move only that visible pending segment behind the
|
||||
* first new human message without disturbing established history or hidden
|
||||
* checkpoint controls. The caller keeps the baseline after stream completion
|
||||
* because the SDK may retain its transient event order until the next submit.
|
||||
* latest new human message. Conversely, a baseline or history-confirmed message
|
||||
* from an established turn can be woven after that human before a live
|
||||
* checkpoint tail; move those established messages back before the input. The
|
||||
* caller keeps the baseline after stream completion because the SDK may retain
|
||||
* its transient event order until the next submit.
|
||||
*/
|
||||
export function restoreLocalTurnMessageOrder(
|
||||
messages: Message[],
|
||||
baselineMessageIdentities: ReadonlySet<string>,
|
||||
confirmedHistoryIdentities: ReadonlySet<string> = EMPTY_MESSAGE_IDENTITIES_SET,
|
||||
currentTurnRunIds: ReadonlySet<string> = EMPTY_MESSAGE_IDENTITIES_SET,
|
||||
anchorHumanIdentity?: string | null,
|
||||
canonicalHistoryIdentities: ReadonlySet<string> = confirmedHistoryIdentities,
|
||||
): Message[] {
|
||||
const pendingHumanIndex = messages.findIndex((message) => {
|
||||
const identity = messageIdentity(message);
|
||||
return (
|
||||
message.type === "human" &&
|
||||
!isHiddenFromUIMessage(message) &&
|
||||
identity !== undefined &&
|
||||
!baselineMessageIdentities.has(identity)
|
||||
);
|
||||
});
|
||||
if (pendingHumanIndex <= 0) {
|
||||
// When the caller recorded the exact human identity this turn submitted
|
||||
// (LocalTurnAnchor), only that message may anchor the repair. `null` means
|
||||
// the turn has no visible human at all (hidden human-input reply,
|
||||
// regenerate replay): no human may be borrowed from history. An identity
|
||||
// that has not reached the render snapshot yet means the display is a frame
|
||||
// behind — keep the established order instead of re-anchoring on an older
|
||||
// history-only human (absence from the checkpoint baseline is not proof
|
||||
// that a message belongs to this turn).
|
||||
if (anchorHumanIdentity === null) {
|
||||
return messages;
|
||||
}
|
||||
let pendingHumanIndex = -1;
|
||||
if (anchorHumanIdentity !== undefined) {
|
||||
for (let index = messages.length - 1; index >= 0; index -= 1) {
|
||||
const message = messages[index]!;
|
||||
if (
|
||||
message.type === "human" &&
|
||||
!isHiddenFromUIMessage(message) &&
|
||||
messageIdentity(message) === anchorHumanIdentity
|
||||
) {
|
||||
pendingHumanIndex = index;
|
||||
break;
|
||||
}
|
||||
}
|
||||
} else {
|
||||
// Compat path for callers without a local-turn anchor. Context compaction
|
||||
// can omit an older human from the checkpoint while the REST history page
|
||||
// still supplies it, so anchor on the LATEST visible human that is not in
|
||||
// the baseline; an old history-only turn then cannot claim the current
|
||||
// stream. A freshly submitted human may not have a server id yet, and
|
||||
// identity-less messages are necessarily absent from the baseline.
|
||||
for (let index = messages.length - 1; index >= 0; index -= 1) {
|
||||
const message = messages[index]!;
|
||||
const identity = messageIdentity(message);
|
||||
if (
|
||||
message.type === "human" &&
|
||||
!isHiddenFromUIMessage(message) &&
|
||||
(identity === undefined || !baselineMessageIdentities.has(identity))
|
||||
) {
|
||||
pendingHumanIndex = index;
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
if (pendingHumanIndex < 0) {
|
||||
return messages;
|
||||
}
|
||||
|
||||
// The fixed confirmed set decides which suffix messages may move back across
|
||||
// this turn's human. The wider canonical set has a different job in the
|
||||
// prefix: a REST-history message is not speculative current-turn output just
|
||||
// because the latest-page window advanced after submit.
|
||||
const isConfirmedHistoryMessage = (identity: string | undefined) =>
|
||||
identity !== undefined && confirmedHistoryIdentities.has(identity);
|
||||
const isCanonicalHistoryMessage = (identity: string | undefined) =>
|
||||
identity !== undefined && canonicalHistoryIdentities.has(identity);
|
||||
// Steps of the CURRENT run must never be treated as displaced history: after
|
||||
// an interrupt/stop the current turn's already-executed steps are persisted
|
||||
// into canonical history, but they still belong AFTER the new human input.
|
||||
// The pending human message itself carries the current run_id, so it is the
|
||||
// most reliable anchor even when the live checkpoint no longer holds the
|
||||
// current turn's steps (stop/interrupt can flush them to history).
|
||||
const effectiveCurrentTurnRunIds = new Set(currentTurnRunIds);
|
||||
const pendingHumanRunId = getMessageRunId(messages[pendingHumanIndex]!);
|
||||
if (pendingHumanRunId) {
|
||||
effectiveCurrentTurnRunIds.add(pendingHumanRunId);
|
||||
}
|
||||
const isCurrentTurnStep = (message: Message) => {
|
||||
const runId = getMessageRunId(message);
|
||||
return runId !== undefined && effectiveCurrentTurnRunIds.has(runId);
|
||||
};
|
||||
|
||||
const stablePrefix: Message[] = [];
|
||||
const earlyPendingSteps: Message[] = [];
|
||||
@ -490,22 +678,43 @@ export function restoreLocalTurnMessageOrder(
|
||||
(message.type === "ai" || message.type === "tool") &&
|
||||
!isHiddenFromUIMessage(message) &&
|
||||
identity !== undefined &&
|
||||
!baselineMessageIdentities.has(identity);
|
||||
!baselineMessageIdentities.has(identity) &&
|
||||
((!isCanonicalHistoryMessage(identity) &&
|
||||
!isConfirmedHistoryMessage(identity)) ||
|
||||
isCurrentTurnStep(message));
|
||||
if (isVisiblePendingStep) {
|
||||
earlyPendingSteps.push(message);
|
||||
} else {
|
||||
stablePrefix.push(message);
|
||||
}
|
||||
}
|
||||
if (earlyPendingSteps.length === 0) {
|
||||
const displacedMessages: Message[] = [];
|
||||
const stableSuffix: Message[] = [];
|
||||
for (const message of messages.slice(pendingHumanIndex + 1)) {
|
||||
const identity = messageIdentity(message);
|
||||
const wasPresentBeforeSubmit =
|
||||
identity !== undefined && baselineMessageIdentities.has(identity);
|
||||
const wasConfirmedInPreviousHistory =
|
||||
(message.type === "ai" || message.type === "tool") &&
|
||||
!isHiddenFromUIMessage(message) &&
|
||||
isConfirmedHistoryMessage(identity) &&
|
||||
!isCurrentTurnStep(message);
|
||||
if (wasPresentBeforeSubmit || wasConfirmedInPreviousHistory) {
|
||||
displacedMessages.push(message);
|
||||
} else {
|
||||
stableSuffix.push(message);
|
||||
}
|
||||
}
|
||||
if (earlyPendingSteps.length === 0 && displacedMessages.length === 0) {
|
||||
return messages;
|
||||
}
|
||||
|
||||
return [
|
||||
...stablePrefix,
|
||||
...displacedMessages,
|
||||
messages[pendingHumanIndex]!,
|
||||
...earlyPendingSteps,
|
||||
...messages.slice(pendingHumanIndex + 1),
|
||||
...stableSuffix,
|
||||
];
|
||||
}
|
||||
|
||||
@ -1695,7 +1904,7 @@ export function useThreadStream({
|
||||
transientHistoryThreadIdRef.current = null;
|
||||
summarizedRef.current = new Set<string>();
|
||||
pendingUsageBaselineMessageIdsRef.current = new Set();
|
||||
localTurnOrderBaselineIdentitiesRef.current = null;
|
||||
localTurnAnchorRef.current = null;
|
||||
tasksRef.current = {};
|
||||
setTasks({});
|
||||
invalidateStoppedThreadCaches(queryClient, threadIdRef.current, isMock);
|
||||
@ -1809,7 +2018,17 @@ export function useThreadStream({
|
||||
() => (threadId ? history : []),
|
||||
[history, threadId],
|
||||
);
|
||||
const humanMessageCount = persistedMessages.filter(
|
||||
// Render-facing coalesced snapshot. Optimistic-input confirmation and the
|
||||
// turn anchor observe THIS snapshot — the same frames the user sees — so a
|
||||
// human echo landing in the per-chunk SDK array one coalesce interval early
|
||||
// can no longer withdraw the local input before the snapshot shows it.
|
||||
// Refs, summarization capture, and token-usage tracking keep consuming the
|
||||
// per-chunk `persistedMessages` array above, unchanged.
|
||||
const renderMessages = useCoalescedStreamMessages(
|
||||
persistedMessages,
|
||||
thread.isLoading,
|
||||
);
|
||||
const humanMessageCount = renderMessages.filter(
|
||||
(m) => m.type === "human",
|
||||
).length;
|
||||
const latestMessageCountsRef = useRef({ humanMessageCount });
|
||||
@ -1820,7 +2039,7 @@ export function useThreadStream({
|
||||
// the settled frame. The next local submit replaces it and a thread switch or
|
||||
// replay gap clears it. An empty set is meaningful for a new thread and must
|
||||
// not be confused with a reconnect that has no local turn anchor.
|
||||
const localTurnOrderBaselineIdentitiesRef = useRef<Set<string> | null>(null);
|
||||
const localTurnAnchorRef = useRef<LocalTurnAnchor | null>(null);
|
||||
// Current-stream lifecycle bridge for messages removed from the checkpoint
|
||||
// tail before the canonical run-event page refetch observes the journal
|
||||
// flush. It is never appended into useThreadHistory's persisted pages.
|
||||
@ -1869,7 +2088,7 @@ export function useThreadStream({
|
||||
};
|
||||
summarizedRef.current = new Set<string>();
|
||||
pendingUsageBaselineMessageIdsRef.current = new Set();
|
||||
localTurnOrderBaselineIdentitiesRef.current = null;
|
||||
localTurnAnchorRef.current = null;
|
||||
pendingPreparedReplayRef.current = null;
|
||||
setPendingSupersededRunIds(new Set());
|
||||
setPendingSupersededMessageIds(new Set());
|
||||
@ -1919,9 +2138,11 @@ export function useThreadStream({
|
||||
|
||||
// Clear optimistic when server messages arrive.
|
||||
// For messages with a human optimistic message, wait until the server's
|
||||
// human message has arrived to avoid clearing before canonical history (or
|
||||
// replay-gap recovery) reports the input after individual messages-tuple
|
||||
// events for AI messages.
|
||||
// human message has arrived in the RENDER SNAPSHOT — identity match first,
|
||||
// human-count growth of the rendered frames as fallback for runtime-re-keyed
|
||||
// first turns — never in the unthrottled per-chunk array, which would
|
||||
// withdraw the local input one coalesce interval before the user can see
|
||||
// its confirmed copy.
|
||||
const optimisticMessageCount = optimisticMessages.length;
|
||||
const hasHumanOptimistic = optimisticMessages.some((m) => m.type === "human");
|
||||
useEffect(() => {
|
||||
@ -1938,12 +2159,12 @@ export function useThreadStream({
|
||||
useEffect(() => {
|
||||
if (
|
||||
optimisticMessageCount > 0 &&
|
||||
areOptimisticMessagesConfirmed(optimisticMessages, persistedMessages)
|
||||
areOptimisticMessagesConfirmed(optimisticMessages, renderMessages)
|
||||
) {
|
||||
setOptimisticMessages([]);
|
||||
setOptimisticThreadId(null);
|
||||
}
|
||||
}, [optimisticMessageCount, optimisticMessages, persistedMessages]);
|
||||
}, [optimisticMessageCount, optimisticMessages, renderMessages]);
|
||||
|
||||
const sendMessage = useCallback(
|
||||
async (
|
||||
@ -1971,9 +2192,33 @@ export function useThreadStream({
|
||||
.map(messageIdentity)
|
||||
.filter((id): id is string => Boolean(id)),
|
||||
);
|
||||
localTurnOrderBaselineIdentitiesRef.current = new Set(
|
||||
pendingUsageBaselineMessageIdsRef.current,
|
||||
);
|
||||
// One client-generated id for this turn's human input: the optimistic
|
||||
// display copy and the submitted message share it, so the render
|
||||
// snapshot confirms the exact identity it already shows instead of the
|
||||
// ordering repair guessing from the baseline (a compaction-trimmed
|
||||
// checkpoint must never promote an older history-only human into this
|
||||
// turn's anchor).
|
||||
const hideFromUI = options?.additionalKwargs?.hide_from_ui === true;
|
||||
const humanMessageId = `local-human-${uuid()}`;
|
||||
localTurnAnchorRef.current = {
|
||||
threadId,
|
||||
humanIdentity: hideFromUI ? null : `message:${humanMessageId}`,
|
||||
baselineIdentities: new Set(pendingUsageBaselineMessageIdsRef.current),
|
||||
preSubmitHistoryIdentities: new Set(
|
||||
visibleHistory.map(messageIdentity).filter(isNonEmptyString),
|
||||
),
|
||||
preSubmitBridgeIdentities: new Set(
|
||||
transientHistoryThreadIdRef.current === threadId
|
||||
? transientHistoryBridgeRef.current
|
||||
.map(messageIdentity)
|
||||
.filter(isNonEmptyString)
|
||||
: EMPTY_MESSAGE_IDENTITIES,
|
||||
),
|
||||
preSubmitMaxSeq: maxMessageSeq([
|
||||
...visibleHistory,
|
||||
...persistedMessages,
|
||||
]),
|
||||
};
|
||||
|
||||
// Build optimistic files list with uploading status
|
||||
const optimisticFiles: FileInMessage[] = (message.files ?? []).map(
|
||||
@ -1984,7 +2229,6 @@ export function useThreadStream({
|
||||
}),
|
||||
);
|
||||
|
||||
const hideFromUI = options?.additionalKwargs?.hide_from_ui === true;
|
||||
const optimisticAdditionalKwargs = {
|
||||
...options?.additionalKwargs,
|
||||
...(optimisticFiles.length > 0 ? { files: optimisticFiles } : {}),
|
||||
@ -1994,7 +2238,7 @@ export function useThreadStream({
|
||||
if (!hideFromUI) {
|
||||
newOptimistic.push({
|
||||
type: "human",
|
||||
id: `opt-human-${Date.now()}`,
|
||||
id: humanMessageId,
|
||||
content: text ? [{ type: "text", text }] : "",
|
||||
additional_kwargs: optimisticAdditionalKwargs,
|
||||
});
|
||||
@ -2101,6 +2345,7 @@ export function useThreadStream({
|
||||
additionalKwargs: options?.additionalKwargs,
|
||||
additionalInputMessages: options?.additionalInputMessages,
|
||||
filesForSubmit,
|
||||
humanMessageId,
|
||||
}),
|
||||
},
|
||||
{
|
||||
@ -2140,7 +2385,7 @@ export function useThreadStream({
|
||||
setOptimisticThreadId(null);
|
||||
setLiveMessagesThreadId(null);
|
||||
setIsUploading(false);
|
||||
localTurnOrderBaselineIdentitiesRef.current = null;
|
||||
localTurnAnchorRef.current = null;
|
||||
throw error;
|
||||
} finally {
|
||||
sendInFlightRef.current = false;
|
||||
@ -2153,6 +2398,7 @@ export function useThreadStream({
|
||||
queryClient,
|
||||
humanMessageCount,
|
||||
persistedMessages,
|
||||
visibleHistory,
|
||||
],
|
||||
);
|
||||
|
||||
@ -2178,9 +2424,27 @@ export function useThreadStream({
|
||||
.map(messageIdentity)
|
||||
.filter((id): id is string => Boolean(id)),
|
||||
);
|
||||
localTurnOrderBaselineIdentitiesRef.current = new Set(
|
||||
pendingUsageBaselineMessageIdsRef.current,
|
||||
);
|
||||
localTurnAnchorRef.current = {
|
||||
threadId,
|
||||
// Replay turns submit no new visible human; an edit replay adopts the
|
||||
// prepare response's replacement identity once it lands below.
|
||||
humanIdentity: null,
|
||||
baselineIdentities: new Set(pendingUsageBaselineMessageIdsRef.current),
|
||||
preSubmitHistoryIdentities: new Set(
|
||||
visibleHistory.map(messageIdentity).filter(isNonEmptyString),
|
||||
),
|
||||
preSubmitBridgeIdentities: new Set(
|
||||
transientHistoryThreadIdRef.current === threadId
|
||||
? transientHistoryBridgeRef.current
|
||||
.map(messageIdentity)
|
||||
.filter(isNonEmptyString)
|
||||
: EMPTY_MESSAGE_IDENTITIES,
|
||||
),
|
||||
preSubmitMaxSeq: maxMessageSeq([
|
||||
...visibleHistory,
|
||||
...persistedMessages,
|
||||
]),
|
||||
};
|
||||
setLiveMessagesThreadId(threadId);
|
||||
listeners.current.onSend?.(threadId);
|
||||
let preparedSupersededRunId: string | null = null;
|
||||
@ -2199,6 +2463,15 @@ export function useThreadStream({
|
||||
typeof prepared.replacement_human_message_id === "string"
|
||||
? prepared.replacement_human_message_id
|
||||
: undefined;
|
||||
const replayAnchor = localTurnAnchorRef.current;
|
||||
if (replayAnchor?.threadId === threadId && replacementHumanMessageId) {
|
||||
// The edit replay reuses the server-prepared replacement identity;
|
||||
// supersede semantics stay with the prepare response.
|
||||
localTurnAnchorRef.current = {
|
||||
...replayAnchor,
|
||||
humanIdentity: `message:${replacementHumanMessageId}`,
|
||||
};
|
||||
}
|
||||
const pendingReplay: PendingPreparedReplayMask = {
|
||||
kind: replacementHumanMessageId ? "edit" : "regenerate",
|
||||
targetRunId: prepared.target_run_id,
|
||||
@ -2267,7 +2540,7 @@ export function useThreadStream({
|
||||
setOptimisticMessages([]);
|
||||
setOptimisticThreadId(null);
|
||||
setLiveMessagesThreadId(null);
|
||||
localTurnOrderBaselineIdentitiesRef.current = null;
|
||||
localTurnAnchorRef.current = null;
|
||||
if (preparedSupersededRunId) {
|
||||
const supersededRunId = preparedSupersededRunId;
|
||||
pendingPreparedReplayRef.current = null;
|
||||
@ -2284,7 +2557,14 @@ export function useThreadStream({
|
||||
sendInFlightRef.current = false;
|
||||
}
|
||||
},
|
||||
[context, humanMessageCount, persistedMessages, queryClient, thread],
|
||||
[
|
||||
context,
|
||||
humanMessageCount,
|
||||
persistedMessages,
|
||||
queryClient,
|
||||
thread,
|
||||
visibleHistory,
|
||||
],
|
||||
);
|
||||
|
||||
const regenerateMessage = useCallback(
|
||||
@ -2369,14 +2649,6 @@ export function useThreadStream({
|
||||
messagesRef.current = persistedMessages;
|
||||
}
|
||||
|
||||
// Render-facing coalesced snapshot. Refs, counters and usage tracking keep
|
||||
// consuming the per-chunk array above so lifecycle semantics (optimistic
|
||||
// clearing, summarization capture, token-usage baselines) are unchanged.
|
||||
const renderMessages = useCoalescedStreamMessages(
|
||||
persistedMessages,
|
||||
thread.isLoading,
|
||||
);
|
||||
|
||||
const rawVisibleOptimisticMessages = getVisibleOptimisticMessages(
|
||||
optimisticThreadId === currentViewThreadId ? optimisticMessages : [],
|
||||
prevHumanMsgCountRef.current,
|
||||
@ -2433,10 +2705,58 @@ export function useThreadStream({
|
||||
renderMessages,
|
||||
visibleOptimisticMessages,
|
||||
);
|
||||
const localTurnOrderBaseline = localTurnOrderBaselineIdentitiesRef.current;
|
||||
return localTurnOrderBaseline === null
|
||||
const localTurnAnchor =
|
||||
localTurnAnchorRef.current?.threadId === threadId
|
||||
? localTurnAnchorRef.current
|
||||
: null;
|
||||
const canonicalHistoryIdentities = new Set(
|
||||
visibleHistory.map(messageIdentity).filter(isNonEmptyString),
|
||||
);
|
||||
// Only established history known to predate this local submit may be moved
|
||||
// across its human anchor. The fixed identity snapshots cover messages
|
||||
// already loaded from REST and pre-existing transient-bridge rescue; the
|
||||
// authoritative seq boundary also admits older pages that finish loading
|
||||
// after submit. Post-submit rescue and later external turns stay outside.
|
||||
const confirmedHistoryIdentities = getConfirmedPreSubmitHistoryIdentities(
|
||||
visibleHistory,
|
||||
localTurnAnchor,
|
||||
);
|
||||
// The current turn's run(s): visible ai/tool steps that appear in the live
|
||||
// checkpoint but are neither part of the pre-submit baseline nor already
|
||||
// canonical REST history. These are output from the in-flight submit and
|
||||
// must never be moved before their human.
|
||||
const currentTurnRunIds = getCurrentTurnRunIds(
|
||||
renderMessages,
|
||||
localTurnAnchor ? localTurnAnchor.baselineIdentities : null,
|
||||
canonicalHistoryIdentities,
|
||||
);
|
||||
if (localTurnAnchor?.humanIdentity) {
|
||||
// The surviving merged copy of the submitted human can be the run_id-less
|
||||
// optimistic one; recover the run from any rendered or canonical copy so
|
||||
// an interrupt-flushed current-run step is still recognised as ours.
|
||||
const anchorRunId =
|
||||
findMessageRunIdByIdentity(
|
||||
renderMessages,
|
||||
localTurnAnchor.humanIdentity,
|
||||
) ??
|
||||
findMessageRunIdByIdentity(
|
||||
effectiveHistory,
|
||||
localTurnAnchor.humanIdentity,
|
||||
);
|
||||
if (anchorRunId) {
|
||||
currentTurnRunIds.add(anchorRunId);
|
||||
}
|
||||
}
|
||||
return localTurnAnchor === null
|
||||
? restoreReconnectedTurnMessageOrder(merged)
|
||||
: restoreLocalTurnMessageOrder(merged, localTurnOrderBaseline);
|
||||
: restoreLocalTurnMessageOrder(
|
||||
merged,
|
||||
localTurnAnchor.baselineIdentities,
|
||||
confirmedHistoryIdentities,
|
||||
currentTurnRunIds,
|
||||
localTurnAnchor.humanIdentity,
|
||||
canonicalHistoryIdentities,
|
||||
);
|
||||
}, [
|
||||
previouslyRenderedOrder,
|
||||
renderMessages,
|
||||
@ -2446,10 +2766,27 @@ export function useThreadStream({
|
||||
visibleOptimisticMessages,
|
||||
]);
|
||||
useEffect(() => {
|
||||
const visibleMergedMessages = mergedMessages.filter(
|
||||
(message) =>
|
||||
!isHiddenFromUIMessage(message) && !message.id?.startsWith("opt-"),
|
||||
// The committed render ledger excludes hidden control copies and the
|
||||
// still-unconfirmed optimistic ones (keyed by identity, since the local
|
||||
// input now shares its id with the submit instead of an `opt-` prefix):
|
||||
// a failed send must never pin a message the server never saw.
|
||||
const pendingOptimisticIdentities = new Set(
|
||||
(optimisticThreadId === currentViewThreadId
|
||||
? optimisticMessages
|
||||
: EMPTY_MESSAGES
|
||||
)
|
||||
.map(messageIdentity)
|
||||
.filter(isNonEmptyString),
|
||||
);
|
||||
const visibleMergedMessages = mergedMessages.filter((message) => {
|
||||
if (isHiddenFromUIMessage(message)) {
|
||||
return false;
|
||||
}
|
||||
const identity = messageIdentity(message);
|
||||
return (
|
||||
identity === undefined || !pendingOptimisticIdentities.has(identity)
|
||||
);
|
||||
});
|
||||
const previousLedger =
|
||||
thread.isLoading &&
|
||||
renderedMessageSnapshotRef.current.threadId === threadId
|
||||
@ -2467,7 +2804,15 @@ export function useThreadStream({
|
||||
.map(messageIdentity)
|
||||
.filter(isNonEmptyString),
|
||||
};
|
||||
}, [mergedMessages, pendingSupersededMessageIds, thread.isLoading, threadId]);
|
||||
}, [
|
||||
mergedMessages,
|
||||
optimisticMessages,
|
||||
optimisticThreadId,
|
||||
currentViewThreadId,
|
||||
pendingSupersededMessageIds,
|
||||
thread.isLoading,
|
||||
threadId,
|
||||
]);
|
||||
const pendingUsageMessages = thread.isLoading
|
||||
? getMessagesAfterBaseline(
|
||||
persistedMessages,
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@ -10,6 +10,7 @@ import {
|
||||
computeSummarizationTransientMessages,
|
||||
countHumanMessagesExcludingSuperseded,
|
||||
flattenThreadHistoryPages,
|
||||
getCurrentTurnRunIds,
|
||||
getSummarizationMiddlewareMessages,
|
||||
getThreadHistoryNextPageParam,
|
||||
getVisibleOptimisticMessages,
|
||||
@ -1637,6 +1638,472 @@ test("local turn order keeps early streamed steps behind the user message", () =
|
||||
]);
|
||||
});
|
||||
|
||||
test("local turn order keeps an existing clarification card with its original turn", () => {
|
||||
const previousHuman = {
|
||||
id: "previous-human",
|
||||
type: "human",
|
||||
content: "Research today's market",
|
||||
} as Message;
|
||||
const previousAnswer = {
|
||||
id: "previous-answer",
|
||||
type: "ai",
|
||||
content: "Here is the completed report",
|
||||
} as Message;
|
||||
const clarificationCard = {
|
||||
id: "clarification-card",
|
||||
type: "tool",
|
||||
name: "ask_clarification",
|
||||
tool_call_id: "clarification-call",
|
||||
content: "Which market should I research?",
|
||||
} as Message;
|
||||
const currentHuman = {
|
||||
id: "current-human",
|
||||
type: "human",
|
||||
content: "Summarize the report",
|
||||
} as Message;
|
||||
const currentStep = {
|
||||
id: "current-step",
|
||||
type: "ai",
|
||||
content: "Reading the report",
|
||||
} as Message;
|
||||
const baselineIdentities = new Set([
|
||||
"message:previous-human",
|
||||
"message:previous-answer",
|
||||
"tool:clarification-call",
|
||||
]);
|
||||
|
||||
// A live checkpoint tail can be woven after the newly persisted human
|
||||
// message even though the card was already visible before submission.
|
||||
expect(
|
||||
restoreLocalTurnMessageOrder(
|
||||
[
|
||||
previousHuman,
|
||||
previousAnswer,
|
||||
currentHuman,
|
||||
clarificationCard,
|
||||
currentStep,
|
||||
],
|
||||
baselineIdentities,
|
||||
),
|
||||
).toEqual([
|
||||
previousHuman,
|
||||
previousAnswer,
|
||||
clarificationCard,
|
||||
currentHuman,
|
||||
currentStep,
|
||||
]);
|
||||
});
|
||||
|
||||
test("local turn order preserves relative order across displaced message sources", () => {
|
||||
const previousHuman = {
|
||||
id: "previous-human",
|
||||
type: "human",
|
||||
content: "Research today's market",
|
||||
} as Message;
|
||||
const currentHuman = {
|
||||
id: "current-human",
|
||||
type: "human",
|
||||
content: "Summarize the report",
|
||||
} as Message;
|
||||
const historyOnlyCard = {
|
||||
id: "history-only-card",
|
||||
type: "tool",
|
||||
name: "ask_clarification",
|
||||
tool_call_id: "history-only-call",
|
||||
content: "Which market should I research?",
|
||||
} as Message;
|
||||
const baselineAnswer = {
|
||||
id: "baseline-answer",
|
||||
type: "ai",
|
||||
content: "Here is the completed report",
|
||||
} as Message;
|
||||
const currentStep = {
|
||||
id: "current-step",
|
||||
type: "ai",
|
||||
content: "Reading the report",
|
||||
} as Message;
|
||||
|
||||
expect(
|
||||
restoreLocalTurnMessageOrder(
|
||||
[
|
||||
previousHuman,
|
||||
currentHuman,
|
||||
historyOnlyCard,
|
||||
baselineAnswer,
|
||||
currentStep,
|
||||
],
|
||||
new Set(["message:previous-human", "message:baseline-answer"]),
|
||||
new Set(["tool:history-only-call"]),
|
||||
),
|
||||
).toEqual([
|
||||
previousHuman,
|
||||
historyOnlyCard,
|
||||
baselineAnswer,
|
||||
currentHuman,
|
||||
currentStep,
|
||||
]);
|
||||
});
|
||||
|
||||
test("current turn run id derivation selects only visible unconfirmed steps", () => {
|
||||
const baselineStep = {
|
||||
id: "baseline-step",
|
||||
type: "ai",
|
||||
content: "Previous answer",
|
||||
run_id: "run-previous",
|
||||
} as Message;
|
||||
const currentStep = {
|
||||
id: "current-step",
|
||||
type: "ai",
|
||||
content: "Current progress",
|
||||
run_id: "run-current",
|
||||
} as Message;
|
||||
const runlessStep = {
|
||||
id: "runless-step",
|
||||
type: "tool",
|
||||
content: "Pending metadata",
|
||||
tool_call_id: "runless-call",
|
||||
} as Message;
|
||||
const hiddenStep = {
|
||||
id: "hidden-step",
|
||||
type: "ai",
|
||||
content: "Internal control",
|
||||
run_id: "run-hidden",
|
||||
additional_kwargs: { hide_from_ui: true },
|
||||
} as Message;
|
||||
const confirmedHistoryStep = {
|
||||
id: "confirmed-history-step",
|
||||
type: "tool",
|
||||
content: "Previously persisted result",
|
||||
tool_call_id: "confirmed-history-call",
|
||||
run_id: "run-history",
|
||||
} as Message;
|
||||
|
||||
expect(
|
||||
getCurrentTurnRunIds(
|
||||
[
|
||||
baselineStep,
|
||||
currentStep,
|
||||
runlessStep,
|
||||
hiddenStep,
|
||||
confirmedHistoryStep,
|
||||
],
|
||||
new Set(["message:baseline-step"]),
|
||||
new Set(["tool:confirmed-history-call"]),
|
||||
),
|
||||
).toEqual(new Set(["run-current"]));
|
||||
expect(
|
||||
getCurrentTurnRunIds([runlessStep], new Set(["message:baseline-step"])),
|
||||
).toEqual(new Set());
|
||||
expect(getCurrentTurnRunIds([currentStep], null)).toEqual(new Set());
|
||||
});
|
||||
|
||||
test("history-confirmed live cards are not derived as current-run steps", () => {
|
||||
const previousHuman = {
|
||||
id: "previous-human",
|
||||
type: "human",
|
||||
content: "Research today's market",
|
||||
} as Message;
|
||||
const currentHuman = {
|
||||
id: "current-human",
|
||||
type: "human",
|
||||
content: "Summarize the report",
|
||||
run_id: "run-current",
|
||||
} as Message;
|
||||
const oldHistoryCard = {
|
||||
id: "old-history-card",
|
||||
type: "tool",
|
||||
name: "ask_clarification",
|
||||
tool_call_id: "old-history-call",
|
||||
content: "Which market should I research?",
|
||||
run_id: "run-previous",
|
||||
} as Message;
|
||||
const baselineIdentities = new Set(["message:previous-human"]);
|
||||
const historyIdentities = new Set(["tool:old-history-call"]);
|
||||
const currentTurnRunIds = getCurrentTurnRunIds(
|
||||
[oldHistoryCard],
|
||||
baselineIdentities,
|
||||
historyIdentities,
|
||||
);
|
||||
|
||||
expect(currentTurnRunIds).toEqual(new Set());
|
||||
expect(
|
||||
restoreLocalTurnMessageOrder(
|
||||
[previousHuman, currentHuman, oldHistoryCard],
|
||||
baselineIdentities,
|
||||
historyIdentities,
|
||||
currentTurnRunIds,
|
||||
),
|
||||
).toEqual([previousHuman, oldHistoryCard, currentHuman]);
|
||||
});
|
||||
|
||||
test("local turn order repairs a displaced suffix when the human is first", () => {
|
||||
const currentHuman = {
|
||||
id: "current-human",
|
||||
type: "human",
|
||||
content: "Summarize the report",
|
||||
run_id: "run-current",
|
||||
} as Message;
|
||||
const oldHistoryCard = {
|
||||
id: "old-history-card",
|
||||
type: "tool",
|
||||
name: "ask_clarification",
|
||||
tool_call_id: "old-history-call",
|
||||
content: "Which market should I research?",
|
||||
run_id: "run-previous",
|
||||
} as Message;
|
||||
|
||||
expect(
|
||||
restoreLocalTurnMessageOrder(
|
||||
[currentHuman, oldHistoryCard],
|
||||
new Set(),
|
||||
new Set(["tool:old-history-call"]),
|
||||
),
|
||||
).toEqual([oldHistoryCard, currentHuman]);
|
||||
});
|
||||
|
||||
test("local turn order anchors on the latest non-baseline human after compaction", () => {
|
||||
// A summarized checkpoint may omit an older human that remains in canonical
|
||||
// history. The latest non-baseline human is the submitted turn even before
|
||||
// the server assigns it an id; choosing the first identifiable one would move
|
||||
// established intervening turns around the old human.
|
||||
const oldHistoryHuman = {
|
||||
id: "old-history-human",
|
||||
type: "human",
|
||||
content: "An earlier request omitted by the checkpoint",
|
||||
} as Message;
|
||||
const oldHistoryAnswer = {
|
||||
id: "old-history-answer",
|
||||
type: "ai",
|
||||
content: "An earlier answer",
|
||||
} as Message;
|
||||
const previousHuman = {
|
||||
id: "previous-human",
|
||||
type: "human",
|
||||
content: "The immediately previous request",
|
||||
} as Message;
|
||||
const previousAnswer = {
|
||||
id: "previous-answer",
|
||||
type: "ai",
|
||||
content: "The immediately previous answer",
|
||||
} as Message;
|
||||
const currentHuman = {
|
||||
type: "human",
|
||||
content: "The newly submitted request",
|
||||
} as Message;
|
||||
const currentStep = {
|
||||
id: "current-step",
|
||||
type: "ai",
|
||||
content: "Current streamed progress",
|
||||
} as Message;
|
||||
|
||||
expect(
|
||||
restoreLocalTurnMessageOrder(
|
||||
[
|
||||
oldHistoryHuman,
|
||||
oldHistoryAnswer,
|
||||
previousHuman,
|
||||
previousAnswer,
|
||||
currentHuman,
|
||||
currentStep,
|
||||
],
|
||||
new Set(["message:previous-human", "message:previous-answer"]),
|
||||
new Set(["message:old-history-answer"]),
|
||||
),
|
||||
).toEqual([
|
||||
oldHistoryHuman,
|
||||
oldHistoryAnswer,
|
||||
previousHuman,
|
||||
previousAnswer,
|
||||
currentHuman,
|
||||
currentStep,
|
||||
]);
|
||||
});
|
||||
|
||||
test("history-confirmed current steps still move behind their human", () => {
|
||||
const previousHuman = {
|
||||
id: "previous-human",
|
||||
type: "human",
|
||||
content: "Research the robot sector",
|
||||
} as Message;
|
||||
const currentStep = {
|
||||
id: "current-step",
|
||||
type: "ai",
|
||||
content: "Searching for catalysts",
|
||||
run_id: "run-current",
|
||||
} as Message;
|
||||
const currentHuman = {
|
||||
id: "current-human",
|
||||
type: "human",
|
||||
content: "Continue tracking the robot sector",
|
||||
run_id: "run-current",
|
||||
} as Message;
|
||||
|
||||
expect(
|
||||
restoreLocalTurnMessageOrder(
|
||||
[previousHuman, currentStep, currentHuman],
|
||||
new Set(["message:previous-human"]),
|
||||
new Set(["message:current-step"]),
|
||||
),
|
||||
).toEqual([previousHuman, currentHuman, currentStep]);
|
||||
});
|
||||
|
||||
test("local turn order keeps the current run's persisted steps after the new human", () => {
|
||||
// After an interrupt/stop, the current turn's already-executed steps are
|
||||
// flushed into canonical history. They belong AFTER the new human input even
|
||||
// though history now confirms them — treating them as displaced old message
|
||||
// would move the current run's own progress above the message that owns it.
|
||||
const previousHuman = {
|
||||
id: "previous-human",
|
||||
type: "human",
|
||||
content: "Research the robot sector",
|
||||
} as Message;
|
||||
const currentHuman = {
|
||||
id: "current-human",
|
||||
type: "human",
|
||||
content: "Continue tracking the robot sector",
|
||||
run_id: "run-current",
|
||||
} as Message;
|
||||
const currentAnsweredStep = {
|
||||
id: "current-step",
|
||||
type: "ai",
|
||||
content: "Searching for catalysts",
|
||||
run_id: "run-current",
|
||||
} as Message;
|
||||
const currentToolStep = {
|
||||
id: "current-tool",
|
||||
type: "tool",
|
||||
content: "results",
|
||||
tool_call_id: "call-current",
|
||||
run_id: "run-current",
|
||||
} as Message;
|
||||
const baselineIdentities = new Set(["message:previous-human"]);
|
||||
const historyIdentities = new Set([
|
||||
"message:previous-human",
|
||||
"message:current-human",
|
||||
"message:current-step",
|
||||
"tool:call-current",
|
||||
]);
|
||||
|
||||
// The human already at its correct position, followed by its own steps.
|
||||
expect(
|
||||
restoreLocalTurnMessageOrder(
|
||||
[previousHuman, currentHuman, currentAnsweredStep, currentToolStep],
|
||||
baselineIdentities,
|
||||
historyIdentities,
|
||||
new Set(["run-current"]),
|
||||
),
|
||||
).toEqual([
|
||||
previousHuman,
|
||||
currentHuman,
|
||||
currentAnsweredStep,
|
||||
currentToolStep,
|
||||
]);
|
||||
|
||||
// An empty explicit set exercises the pending-human run_id anchor and still
|
||||
// protects the persisted current-run steps.
|
||||
expect(
|
||||
restoreLocalTurnMessageOrder(
|
||||
[previousHuman, currentHuman, currentAnsweredStep, currentToolStep],
|
||||
baselineIdentities,
|
||||
historyIdentities,
|
||||
new Set(),
|
||||
),
|
||||
).toEqual([
|
||||
previousHuman,
|
||||
currentHuman,
|
||||
currentAnsweredStep,
|
||||
currentToolStep,
|
||||
]);
|
||||
|
||||
// Even when the current turn's human is missing its run_id, explicit
|
||||
// current-turn run ids keep its steps in place below the human.
|
||||
const humanWithoutRun = {
|
||||
id: "current-human",
|
||||
type: "human",
|
||||
content: "Continue tracking the robot sector",
|
||||
} as Message;
|
||||
expect(
|
||||
restoreLocalTurnMessageOrder(
|
||||
[previousHuman, humanWithoutRun, currentAnsweredStep, currentToolStep],
|
||||
baselineIdentities,
|
||||
historyIdentities,
|
||||
new Set(["run-current"]),
|
||||
),
|
||||
).toEqual([
|
||||
previousHuman,
|
||||
humanWithoutRun,
|
||||
currentAnsweredStep,
|
||||
currentToolStep,
|
||||
]);
|
||||
});
|
||||
|
||||
test("local turn order restores a clarification card that only canonical history confirmed", () => {
|
||||
// The card can reach the merged list through the REST history page without
|
||||
// ever entering the live checkpoint `messages` value, so it is not in the
|
||||
// pre-submit baseline. It must still be treated as an established-past-turn
|
||||
// message (not an in-flight pending step) and stay above the new human.
|
||||
const previousHuman = {
|
||||
id: "previous-human",
|
||||
type: "human",
|
||||
content: "Research today's market",
|
||||
} as Message;
|
||||
const previousAnswer = {
|
||||
id: "previous-answer",
|
||||
type: "ai",
|
||||
content: "Here is the completed report",
|
||||
} as Message;
|
||||
const clarificationCard = {
|
||||
id: "clarification-card",
|
||||
type: "tool",
|
||||
name: "ask_clarification",
|
||||
tool_call_id: "clarification-call",
|
||||
content: "Which market should I research?",
|
||||
} as Message;
|
||||
const currentHuman = {
|
||||
id: "current-human",
|
||||
type: "human",
|
||||
content: "Summarize the report",
|
||||
} as Message;
|
||||
const currentStep = {
|
||||
id: "current-step",
|
||||
type: "ai",
|
||||
content: "Reading the report",
|
||||
} as Message;
|
||||
// Baseline captured at submit time: the previous turn is there, but the
|
||||
// card has not reached the live checkpoint yet, so it is absent.
|
||||
const baselineIdentities = new Set([
|
||||
"message:previous-human",
|
||||
"message:previous-answer",
|
||||
]);
|
||||
// Canonical history has already committed the card identity.
|
||||
const historyIdentities = new Set([
|
||||
"message:previous-human",
|
||||
"message:previous-answer",
|
||||
"tool:clarification-call",
|
||||
]);
|
||||
|
||||
// The merged list has the card woven after the new human (from history).
|
||||
expect(
|
||||
restoreLocalTurnMessageOrder(
|
||||
[
|
||||
previousHuman,
|
||||
previousAnswer,
|
||||
currentHuman,
|
||||
clarificationCard,
|
||||
currentStep,
|
||||
],
|
||||
baselineIdentities,
|
||||
historyIdentities,
|
||||
),
|
||||
).toEqual([
|
||||
previousHuman,
|
||||
previousAnswer,
|
||||
clarificationCard,
|
||||
currentHuman,
|
||||
currentStep,
|
||||
]);
|
||||
});
|
||||
|
||||
test("reconnected turn order moves same-run steps back behind the user message", () => {
|
||||
// Reload mid-run: replayed `messages-tuple` steps reach the merged list
|
||||
// before the turn's human message (the retained replay buffer may have
|
||||
@ -2306,6 +2773,164 @@ test("a checkpoint message earlier than the loaded window is placed by its seq e
|
||||
]);
|
||||
});
|
||||
|
||||
test("local turn order anchors on the exact submitted human identity (X__user normalized)", () => {
|
||||
// The anchor recorded at submit time names one identity; the server's
|
||||
// visible copy arrives as `<id>__user` and normalizes onto it.
|
||||
const previousHuman = {
|
||||
id: "previous-human",
|
||||
type: "human",
|
||||
content: "Research the robot sector",
|
||||
} as Message;
|
||||
const previousAnswer = {
|
||||
id: "previous-answer",
|
||||
type: "ai",
|
||||
content: "Done",
|
||||
} as Message;
|
||||
const earlyStep = {
|
||||
id: "early-step",
|
||||
type: "ai",
|
||||
content: "Searching for catalysts",
|
||||
run_id: "run-new",
|
||||
} as Message;
|
||||
const serverHumanCopy = {
|
||||
id: "new-human__user",
|
||||
type: "human",
|
||||
content: "Continue tracking",
|
||||
run_id: "run-new",
|
||||
} as Message;
|
||||
|
||||
expect(
|
||||
restoreLocalTurnMessageOrder(
|
||||
[previousHuman, previousAnswer, earlyStep, serverHumanCopy],
|
||||
new Set(["message:previous-human", "message:previous-answer"]),
|
||||
new Set(["message:previous-human", "message:previous-answer"]),
|
||||
new Set(["run-new"]),
|
||||
"message:new-human",
|
||||
),
|
||||
).toEqual([previousHuman, previousAnswer, serverHumanCopy, earlyStep]);
|
||||
});
|
||||
|
||||
test("local turn order keeps established history while the anchored human is absent", () => {
|
||||
// R2: the checkpoint baseline covers only the latest turn while canonical
|
||||
// history holds an older one. Until the submitted human reaches the render
|
||||
// snapshot, no reordering may happen — a history-only human outside the
|
||||
// baseline is not proof of the current turn.
|
||||
const earlierAnswer = {
|
||||
id: "earlier-answer",
|
||||
type: "ai",
|
||||
content: "Earlier answer",
|
||||
} as Message;
|
||||
const oldHuman = {
|
||||
id: "old-human",
|
||||
type: "human",
|
||||
content: "An older request",
|
||||
} as Message;
|
||||
const recentHuman = {
|
||||
id: "recent-human",
|
||||
type: "human",
|
||||
content: "The recent request",
|
||||
} as Message;
|
||||
const recentAnswer = {
|
||||
id: "recent-answer",
|
||||
type: "ai",
|
||||
content: "The recent answer",
|
||||
} as Message;
|
||||
const newStep = {
|
||||
id: "new-step",
|
||||
type: "ai",
|
||||
content: "Working on the follow-up",
|
||||
run_id: "run-new",
|
||||
} as Message;
|
||||
const display = [earlierAnswer, oldHuman, recentHuman, recentAnswer, newStep];
|
||||
const baseline = new Set(["message:recent-human", "message:recent-answer"]);
|
||||
const confirmed = new Set([
|
||||
"message:earlier-answer",
|
||||
"message:old-human",
|
||||
"message:recent-human",
|
||||
"message:recent-answer",
|
||||
]);
|
||||
|
||||
expect(
|
||||
restoreLocalTurnMessageOrder(
|
||||
display,
|
||||
baseline,
|
||||
confirmed,
|
||||
new Set(["run-new"]),
|
||||
"message:new-human",
|
||||
),
|
||||
).toEqual(display);
|
||||
});
|
||||
|
||||
test("local turn order with a null anchor never borrows a history-only human", () => {
|
||||
// Hidden human-input replies and regenerate replays submit no visible
|
||||
// human, so no human identity may anchor the repair at all.
|
||||
const earlierAnswer = {
|
||||
id: "earlier-answer",
|
||||
type: "ai",
|
||||
content: "Earlier answer",
|
||||
} as Message;
|
||||
const oldHuman = {
|
||||
id: "old-human",
|
||||
type: "human",
|
||||
content: "An older request",
|
||||
} as Message;
|
||||
const recentHuman = {
|
||||
id: "recent-human",
|
||||
type: "human",
|
||||
content: "The recent request",
|
||||
} as Message;
|
||||
const replyStep = {
|
||||
id: "reply-step",
|
||||
type: "ai",
|
||||
content: "Applying the answer",
|
||||
run_id: "run-reply",
|
||||
} as Message;
|
||||
const display = [earlierAnswer, oldHuman, recentHuman, replyStep];
|
||||
|
||||
expect(
|
||||
restoreLocalTurnMessageOrder(
|
||||
display,
|
||||
new Set(["message:recent-human"]),
|
||||
new Set(["message:earlier-answer", "message:old-human"]),
|
||||
new Set(["run-reply"]),
|
||||
null,
|
||||
),
|
||||
).toEqual(display);
|
||||
});
|
||||
|
||||
test("local turn order repair is idempotent across repeated deliveries", () => {
|
||||
const previousHuman = {
|
||||
id: "previous-human",
|
||||
type: "human",
|
||||
content: "Research the robot sector",
|
||||
} as Message;
|
||||
const earlyStep = {
|
||||
id: "early-step",
|
||||
type: "ai",
|
||||
content: "Searching for catalysts",
|
||||
run_id: "run-new",
|
||||
} as Message;
|
||||
const submittedHuman = {
|
||||
id: "local-human-1",
|
||||
type: "human",
|
||||
content: "Continue tracking",
|
||||
} as Message;
|
||||
const displaced = [previousHuman, earlyStep, submittedHuman];
|
||||
const baseline = new Set(["message:previous-human"]);
|
||||
const args = [
|
||||
baseline,
|
||||
new Set(["message:previous-human"]),
|
||||
new Set(["run-new"]),
|
||||
"message:local-human-1",
|
||||
] as const;
|
||||
|
||||
const once = restoreLocalTurnMessageOrder(displaced, ...args);
|
||||
expect(once).toEqual([previousHuman, submittedHuman, earlyStep]);
|
||||
// Re-merging the same repaired snapshot converges to the identical order.
|
||||
expect(restoreLocalTurnMessageOrder(once, ...args)).toEqual(once);
|
||||
expect(restoreLocalTurnMessageOrder(displaced, ...args)).toEqual(once);
|
||||
});
|
||||
|
||||
test("mergeMessages preserves canonical seq when a live copy without seq replaces the content (R3)", () => {
|
||||
// Content replacement must not drop trusted ordering metadata: the live
|
||||
// checkpoint copy refreshes the text, but the thread-global position the
|
||||
|
||||
@ -96,3 +96,19 @@ test("keeps human input response metadata on the hidden user message", () => {
|
||||
},
|
||||
]);
|
||||
});
|
||||
|
||||
test("uses the caller-provided human message id for the visible user message", () => {
|
||||
const messages = buildThreadSubmitMessages({
|
||||
text: "hello",
|
||||
humanMessageId: "local-human-1",
|
||||
});
|
||||
|
||||
expect(messages).toEqual([
|
||||
{
|
||||
type: "human",
|
||||
id: "local-human-1",
|
||||
content: [{ type: "text", text: "hello" }],
|
||||
additional_kwargs: {},
|
||||
},
|
||||
]);
|
||||
});
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user