Revert the timeline optimistic and pagination rework (#2596)
* Add npm release track invariant and beta dist-tag procedure * Revert "Keep older chat history and image previews stable (#2490)" The timeline optimistic/pagination rework needs more work than main releases can wait for. Reverted together with #2484; both are reapplied on integration/timeline-optimistic-rework. * Revert "Stop completed turns from appearing stuck (#2484)" See the #2490 revert. #2490 built on this submission-authority change, so the two revert and reapply as a pair.
This commit is contained in:
Mohamed Boudra
File diff suppressed because it is too large
Load Diff
@@ -1,15 +1,15 @@
|
||||
import type { AgentStreamEventPayload } from "@getpaseo/protocol/messages";
|
||||
import { selectAgentTimelineState, useSessionStore } from "@/stores/session-store";
|
||||
import type { AssistantMessageItem, StreamItem } from "@/types/stream";
|
||||
import type { AgentLifecycleStatus } from "@getpaseo/protocol/agent-lifecycle";
|
||||
import type { Agent } from "@/stores/session-store";
|
||||
import { useSessionStore } from "@/stores/session-store";
|
||||
import type { AssistantMessageItem, StreamItem, UserMessageItem } from "@/types/stream";
|
||||
import {
|
||||
applyStreamEvent,
|
||||
flushHeadToTail,
|
||||
hydrateStreamState,
|
||||
isAgentToolCallItem,
|
||||
mergeAgentToolCallItem,
|
||||
replaceWithCanonicalStream,
|
||||
reduceStreamUpdate,
|
||||
upsertUserMessageAcrossStream,
|
||||
} from "@/types/stream";
|
||||
|
||||
const AGENT_STREAM_REDUCER_FLUSH_DELAY_MS = 16 * 3;
|
||||
@@ -88,8 +88,6 @@ export interface ProcessTimelineResponseInput {
|
||||
isInitializing: boolean;
|
||||
hasActiveInitDeferred: boolean;
|
||||
initRequestDirection: InitRequestDirection;
|
||||
sendingClientMessageIds: readonly string[];
|
||||
hasAuthoritativeBaseline: boolean;
|
||||
}
|
||||
|
||||
export interface ProcessTimelineResponseOutput {
|
||||
@@ -101,7 +99,6 @@ export interface ProcessTimelineResponseOutput {
|
||||
clearInitializing: boolean;
|
||||
error: string | null;
|
||||
sideEffects: TimelineReducerSideEffect[];
|
||||
acknowledgedClientMessageIds: string[];
|
||||
}
|
||||
|
||||
interface TimelineUnit {
|
||||
@@ -118,7 +115,6 @@ interface TimelinePathResult {
|
||||
cursor: TimelineCursor | null | undefined;
|
||||
cursorChanged: boolean;
|
||||
sideEffects: TimelineReducerSideEffect[];
|
||||
acknowledgedClientMessageIds: string[];
|
||||
}
|
||||
|
||||
function classifySessionTimelineSeq({
|
||||
@@ -205,39 +201,76 @@ function shouldResolveTimelineInit({
|
||||
return responseDirection === initRequestDirection;
|
||||
}
|
||||
|
||||
function deriveOptimisticLifecycleStatus(
|
||||
currentStatus: AgentLifecycleStatus,
|
||||
event: AgentStreamEventPayload,
|
||||
): AgentLifecycleStatus | null {
|
||||
if (currentStatus !== "running") {
|
||||
return null;
|
||||
}
|
||||
switch (event.type) {
|
||||
case "turn_completed":
|
||||
return "idle";
|
||||
case "turn_failed":
|
||||
return "error";
|
||||
case "turn_canceled":
|
||||
// A canceled turn can be either a final user cancel or an interrupt before
|
||||
// a replacement turn starts. The daemon snapshot is authoritative here.
|
||||
return null;
|
||||
default:
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
function preserveReplacePathAssistantHead(params: {
|
||||
tail: StreamItem[];
|
||||
currentHead: StreamItem[];
|
||||
}): {
|
||||
tail: StreamItem[];
|
||||
head: StreamItem[];
|
||||
} {
|
||||
const { tail, currentHead } = params;
|
||||
const liveAssistant = currentHead.findLast(
|
||||
(item): item is Extract<StreamItem, { kind: "assistant_message" }> =>
|
||||
item.kind === "assistant_message",
|
||||
);
|
||||
if (!liveAssistant) {
|
||||
return { tail, head: [] };
|
||||
}
|
||||
const tailAssistant = tail.at(-1);
|
||||
if (!tailAssistant || tailAssistant.kind !== "assistant_message") {
|
||||
return { tail, head: currentHead };
|
||||
}
|
||||
if (!liveAssistant.text.startsWith(tailAssistant.text)) {
|
||||
return { tail, head: [] };
|
||||
}
|
||||
return {
|
||||
tail: tail.slice(0, -1),
|
||||
head: [{ ...liveAssistant, text: tailAssistant.text }],
|
||||
};
|
||||
}
|
||||
|
||||
function applyTimelineReplacePath(args: {
|
||||
timelineUnits: TimelineUnit[];
|
||||
payload: ProcessTimelineResponseInput["payload"];
|
||||
bootstrapPolicy: ReturnType<typeof deriveBootstrapTailTimelinePolicy>;
|
||||
currentTail: StreamItem[];
|
||||
currentHead: StreamItem[];
|
||||
sendingClientMessageIds: readonly string[];
|
||||
preserveLiveHead: boolean;
|
||||
toHydratedEvents: (
|
||||
units: TimelineUnit[],
|
||||
) => Array<{ event: AgentStreamEventPayload; timestamp: Date }>;
|
||||
}): TimelinePathResult {
|
||||
const {
|
||||
timelineUnits,
|
||||
payload,
|
||||
bootstrapPolicy,
|
||||
currentTail,
|
||||
currentHead,
|
||||
sendingClientMessageIds,
|
||||
preserveLiveHead,
|
||||
toHydratedEvents,
|
||||
} = args;
|
||||
const { timelineUnits, payload, bootstrapPolicy, currentTail, currentHead, toHydratedEvents } =
|
||||
args;
|
||||
const hydratedTail = hydrateStreamState(toHydratedEvents(timelineUnits), { source: "canonical" });
|
||||
const { tail, head, acknowledgedClientMessageIds } = replaceWithCanonicalStream({
|
||||
canonical: hydratedTail,
|
||||
const reconciledTail = reconcileLocalUserPresentationAfterReplace({
|
||||
canonicalTail: hydratedTail,
|
||||
previousTail: currentTail,
|
||||
previousHead: currentHead,
|
||||
sendingClientMessageIds,
|
||||
preserveLiveHead,
|
||||
canonicalCoverage: {
|
||||
epoch: payload.epoch,
|
||||
endSeq: payload.endCursor?.seq ?? null,
|
||||
},
|
||||
});
|
||||
const { tail, head } = preserveReplacePathAssistantHead({
|
||||
tail: reconciledTail,
|
||||
currentHead,
|
||||
});
|
||||
const cursor: TimelineCursor | null =
|
||||
payload.startCursor && payload.endCursor
|
||||
@@ -251,15 +284,136 @@ function applyTimelineReplacePath(args: {
|
||||
if (bootstrapPolicy.catchUpCursor) {
|
||||
sideEffects.push({ type: "catch_up", cursor: bootstrapPolicy.catchUpCursor });
|
||||
}
|
||||
return { tail, head, cursor, cursorChanged: true, sideEffects };
|
||||
}
|
||||
|
||||
function collectLocallyPresentedUserMessages(items: StreamItem[]): Array<{
|
||||
ordinal: number;
|
||||
item: UserMessageItem;
|
||||
}> {
|
||||
const localUsers: Array<{ ordinal: number; item: UserMessageItem }> = [];
|
||||
let ordinal = 0;
|
||||
for (const item of items) {
|
||||
if (item.kind !== "user_message") {
|
||||
continue;
|
||||
}
|
||||
if (item.optimistic || item.images?.length || item.attachments?.length) {
|
||||
localUsers.push({ ordinal, item });
|
||||
}
|
||||
ordinal += 1;
|
||||
}
|
||||
return localUsers;
|
||||
}
|
||||
|
||||
function mergeCanonicalUserWithLocalPresentation(
|
||||
canonical: UserMessageItem,
|
||||
local: UserMessageItem,
|
||||
): UserMessageItem {
|
||||
return {
|
||||
tail,
|
||||
head,
|
||||
cursor,
|
||||
cursorChanged: true,
|
||||
sideEffects,
|
||||
acknowledgedClientMessageIds,
|
||||
kind: "user_message",
|
||||
id: canonical.id,
|
||||
...(canonical.clientMessageId ? { clientMessageId: canonical.clientMessageId } : {}),
|
||||
text: local.text,
|
||||
timestamp: local.timestamp,
|
||||
...(local.images && local.images.length > 0 ? { images: local.images } : {}),
|
||||
...(local.attachments && local.attachments.length > 0
|
||||
? { attachments: local.attachments }
|
||||
: {}),
|
||||
};
|
||||
}
|
||||
|
||||
interface CanonicalUserMessageIdentity {
|
||||
messageId?: string;
|
||||
clientMessageId?: string;
|
||||
text: string;
|
||||
}
|
||||
|
||||
function matchesLocalUserMessageIdentity(
|
||||
canonical: CanonicalUserMessageIdentity,
|
||||
optimistic: UserMessageItem,
|
||||
): boolean {
|
||||
if (canonical.clientMessageId !== undefined) {
|
||||
return canonical.clientMessageId === optimistic.id;
|
||||
}
|
||||
if (canonical.messageId === optimistic.id) {
|
||||
return true;
|
||||
}
|
||||
// COMPAT(userMessageClientId): added in v0.2.0, remove after 2027-01-20 once
|
||||
// the supported daemon floor emits clientMessageId on submitted user messages.
|
||||
return canonical.text.length > 0 && canonical.text === optimistic.text;
|
||||
}
|
||||
|
||||
function reconcileLocalUserPresentationAfterReplace(params: {
|
||||
canonicalTail: StreamItem[];
|
||||
previousTail: StreamItem[];
|
||||
previousHead: StreamItem[];
|
||||
}): StreamItem[] {
|
||||
const localUsers = collectLocallyPresentedUserMessages([
|
||||
...params.previousTail,
|
||||
...params.previousHead,
|
||||
]);
|
||||
if (localUsers.length === 0) {
|
||||
return params.canonicalTail;
|
||||
}
|
||||
|
||||
const canonicalUserIndexes: number[] = [];
|
||||
params.canonicalTail.forEach((item, index) => {
|
||||
if (item.kind === "user_message") {
|
||||
canonicalUserIndexes.push(index);
|
||||
}
|
||||
});
|
||||
|
||||
const nextTail = [...params.canonicalTail];
|
||||
const claimedCanonicalIndexes = new Set<number>();
|
||||
const unmatched: UserMessageItem[] = [];
|
||||
|
||||
for (const local of localUsers) {
|
||||
const exactIndex = canonicalUserIndexes.find((index) => {
|
||||
if (claimedCanonicalIndexes.has(index)) return false;
|
||||
const canonical = params.canonicalTail[index];
|
||||
return (
|
||||
canonical?.kind === "user_message" &&
|
||||
matchesLocalUserMessageIdentity(
|
||||
{
|
||||
messageId: canonical.id,
|
||||
clientMessageId: canonical.clientMessageId,
|
||||
text: canonical.text,
|
||||
},
|
||||
local.item,
|
||||
)
|
||||
);
|
||||
});
|
||||
const ordinalIndex = canonicalUserIndexes[local.ordinal];
|
||||
const ordinalItem = ordinalIndex === undefined ? undefined : params.canonicalTail[ordinalIndex];
|
||||
const canonicalIndex =
|
||||
exactIndex ??
|
||||
(ordinalIndex !== undefined &&
|
||||
!claimedCanonicalIndexes.has(ordinalIndex) &&
|
||||
ordinalItem?.kind === "user_message" &&
|
||||
ordinalItem.clientMessageId === undefined
|
||||
? ordinalIndex
|
||||
: undefined);
|
||||
const canonicalItem = canonicalIndex === undefined ? undefined : nextTail[canonicalIndex];
|
||||
if (canonicalIndex === undefined || !canonicalItem || canonicalItem.kind !== "user_message") {
|
||||
if (local.item.optimistic) {
|
||||
unmatched.push(local.item);
|
||||
}
|
||||
continue;
|
||||
}
|
||||
nextTail[canonicalIndex] = mergeCanonicalUserWithLocalPresentation(canonicalItem, local.item);
|
||||
claimedCanonicalIndexes.add(canonicalIndex);
|
||||
}
|
||||
|
||||
for (const item of unmatched) {
|
||||
const insertionIndex = nextTail.findIndex(
|
||||
(canonical) => canonical.timestamp.getTime() > item.timestamp.getTime(),
|
||||
);
|
||||
nextTail.splice(insertionIndex < 0 ? nextTail.length : insertionIndex, 0, item);
|
||||
}
|
||||
|
||||
return nextTail;
|
||||
}
|
||||
|
||||
interface IncrementalAcceptResult {
|
||||
acceptedUnits: TimelineUnit[];
|
||||
cursor: TimelineCursor | undefined;
|
||||
@@ -361,34 +515,6 @@ function mergePrependedCanonicalTail(olderTail: StreamItem[], currentTail: Strea
|
||||
return olderTail;
|
||||
}
|
||||
|
||||
const remainingOlder: StreamItem[] = [];
|
||||
let reconciledCurrent = currentTail;
|
||||
for (const item of olderTail) {
|
||||
if (item.kind !== "user_message") {
|
||||
remainingOlder.push(item);
|
||||
continue;
|
||||
}
|
||||
const result = upsertUserMessageAcrossStream({
|
||||
tail: reconciledCurrent,
|
||||
head: [],
|
||||
message: item,
|
||||
insert: "prepend-tail",
|
||||
presentation: "existing",
|
||||
});
|
||||
if (result.location?.matched) {
|
||||
remainingOlder.push(result.location.message);
|
||||
reconciledCurrent = [
|
||||
...result.tail.slice(0, result.location.index),
|
||||
...result.tail.slice(result.location.index + 1),
|
||||
];
|
||||
} else {
|
||||
remainingOlder.push(item);
|
||||
}
|
||||
}
|
||||
olderTail = remainingOlder;
|
||||
currentTail = reconciledCurrent;
|
||||
if (olderTail.length === 0) return currentTail;
|
||||
|
||||
const olderLast = olderTail.at(-1);
|
||||
const currentFirst = currentTail[0];
|
||||
|
||||
@@ -410,15 +536,16 @@ function mergePrependedCanonicalTail(olderTail: StreamItem[], currentTail: Strea
|
||||
return [...olderTail, ...currentTail];
|
||||
}
|
||||
|
||||
const mergedAssistant: AssistantMessageItem = {
|
||||
...currentFirst,
|
||||
text: `${olderLast.text}${currentFirst.text}`,
|
||||
};
|
||||
if (mergedAssistant.messageId === undefined && olderLast.messageId !== undefined) {
|
||||
mergedAssistant.messageId = olderLast.messageId;
|
||||
}
|
||||
|
||||
return [...olderTail.slice(0, -1), mergedAssistant, ...currentTail.slice(1)];
|
||||
return [
|
||||
...olderTail.slice(0, -1),
|
||||
{
|
||||
...olderLast,
|
||||
text: `${olderLast.text}${currentFirst.text}`,
|
||||
timestamp: currentFirst.timestamp,
|
||||
...(currentFirst.timelineCursor ? { timelineCursor: currentFirst.timelineCursor } : {}),
|
||||
},
|
||||
...currentTail.slice(1),
|
||||
];
|
||||
}
|
||||
|
||||
function replaceLiveAssistantWithProjectedText(params: {
|
||||
@@ -618,24 +745,9 @@ function applyCanonicalForwardUnit(params: {
|
||||
head: StreamItem[];
|
||||
unit: TimelineUnit;
|
||||
epoch: string;
|
||||
}): { tail: StreamItem[]; head: StreamItem[]; acknowledgedClientMessageIds: string[] } {
|
||||
}): { tail: StreamItem[]; head: StreamItem[] } {
|
||||
const { event, timestamp, seqEnd } = params.unit;
|
||||
const timelineCursor = { epoch: params.epoch, seq: seqEnd };
|
||||
if (event.type === "timeline" && event.item.type === "user_message") {
|
||||
const applied = applyStreamEvent({
|
||||
tail: params.tail,
|
||||
head: params.head,
|
||||
event,
|
||||
timestamp,
|
||||
source: "canonical",
|
||||
timelineCursor,
|
||||
});
|
||||
return {
|
||||
tail: applied.tail,
|
||||
head: applied.head,
|
||||
acknowledgedClientMessageIds: applied.acknowledgedClientMessageIds ?? [],
|
||||
};
|
||||
}
|
||||
if (params.head.length === 0) {
|
||||
return {
|
||||
tail: reduceStreamUpdate(params.tail, event, timestamp, {
|
||||
@@ -643,7 +755,6 @@ function applyCanonicalForwardUnit(params: {
|
||||
timelineCursor,
|
||||
}),
|
||||
head: params.head,
|
||||
acknowledgedClientMessageIds: [],
|
||||
};
|
||||
}
|
||||
const replacedHead = replaceLiveAssistantWithProjectedText({
|
||||
@@ -652,9 +763,7 @@ function applyCanonicalForwardUnit(params: {
|
||||
timestamp,
|
||||
timelineCursor,
|
||||
});
|
||||
if (replacedHead) {
|
||||
return { tail: params.tail, head: replacedHead, acknowledgedClientMessageIds: [] };
|
||||
}
|
||||
if (replacedHead) return { tail: params.tail, head: replacedHead };
|
||||
|
||||
const activeAssistant = params.head.findLast(
|
||||
(item): item is Extract<StreamItem, { kind: "assistant_message" }> =>
|
||||
@@ -672,7 +781,6 @@ function applyCanonicalForwardUnit(params: {
|
||||
source: "canonical",
|
||||
timelineCursor,
|
||||
}),
|
||||
acknowledgedClientMessageIds: [],
|
||||
};
|
||||
}
|
||||
|
||||
@@ -684,11 +792,7 @@ function applyCanonicalForwardUnit(params: {
|
||||
source: "canonical",
|
||||
timelineCursor,
|
||||
});
|
||||
return {
|
||||
tail: applied.tail,
|
||||
head: applied.head,
|
||||
acknowledgedClientMessageIds: applied.acknowledgedClientMessageIds ?? [],
|
||||
};
|
||||
return { tail: applied.tail, head: applied.head };
|
||||
}
|
||||
|
||||
function applyAcceptedForwardTimelineUnits(params: {
|
||||
@@ -697,7 +801,7 @@ function applyAcceptedForwardTimelineUnits(params: {
|
||||
currentTail: StreamItem[];
|
||||
currentHead: StreamItem[];
|
||||
currentEndSeq: number | undefined;
|
||||
}): { tail: StreamItem[]; head: StreamItem[]; acknowledgedClientMessageIds: string[] } {
|
||||
}): { tail: StreamItem[]; head: StreamItem[] } {
|
||||
const reconciled = reconcileOverlappingProjectedStreamItems({
|
||||
tail: params.currentTail,
|
||||
head: params.currentHead,
|
||||
@@ -707,19 +811,15 @@ function applyAcceptedForwardTimelineUnits(params: {
|
||||
});
|
||||
let tail = reconciled.tail;
|
||||
let head = reconciled.head;
|
||||
const acknowledgedClientMessageIds = new Set<string>();
|
||||
|
||||
for (const unit of params.units) {
|
||||
if (reconciled.reconciledUnits.has(unit)) continue;
|
||||
const applied = applyCanonicalForwardUnit({ tail, head, unit, epoch: params.epoch });
|
||||
tail = applied.tail;
|
||||
head = applied.head;
|
||||
for (const clientMessageId of applied.acknowledgedClientMessageIds) {
|
||||
acknowledgedClientMessageIds.add(clientMessageId);
|
||||
}
|
||||
}
|
||||
|
||||
return { tail, head, acknowledgedClientMessageIds: [...acknowledgedClientMessageIds] };
|
||||
return { tail, head };
|
||||
}
|
||||
|
||||
function applyTimelineIncrementalPath(args: {
|
||||
@@ -735,17 +835,9 @@ function applyTimelineIncrementalPath(args: {
|
||||
let nextCursor: TimelineCursor | null | undefined = currentCursor;
|
||||
let cursorChanged = false;
|
||||
const sideEffects: TimelineReducerSideEffect[] = [];
|
||||
let acknowledgedClientMessageIds: string[] = [];
|
||||
|
||||
if (timelineUnits.length === 0) {
|
||||
return {
|
||||
tail: nextTail,
|
||||
head: nextHead,
|
||||
cursor: nextCursor,
|
||||
cursorChanged,
|
||||
sideEffects,
|
||||
acknowledgedClientMessageIds,
|
||||
};
|
||||
return { tail: nextTail, head: nextHead, cursor: nextCursor, cursorChanged, sideEffects };
|
||||
}
|
||||
|
||||
const { acceptedUnits, cursor, gapCursor } =
|
||||
@@ -782,7 +874,6 @@ function applyTimelineIncrementalPath(args: {
|
||||
});
|
||||
nextTail = applied.tail;
|
||||
nextHead = applied.head;
|
||||
acknowledgedClientMessageIds = applied.acknowledgedClientMessageIds;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -801,14 +892,7 @@ function applyTimelineIncrementalPath(args: {
|
||||
sideEffects.push({ type: "catch_up", cursor: gapCursor });
|
||||
}
|
||||
|
||||
return {
|
||||
tail: nextTail,
|
||||
head: nextHead,
|
||||
cursor: nextCursor,
|
||||
cursorChanged,
|
||||
sideEffects,
|
||||
acknowledgedClientMessageIds,
|
||||
};
|
||||
return { tail: nextTail, head: nextHead, cursor: nextCursor, cursorChanged, sideEffects };
|
||||
}
|
||||
|
||||
export function processTimelineResponse(
|
||||
@@ -822,8 +906,6 @@ export function processTimelineResponse(
|
||||
isInitializing,
|
||||
hasActiveInitDeferred,
|
||||
initRequestDirection,
|
||||
sendingClientMessageIds,
|
||||
hasAuthoritativeBaseline,
|
||||
} = input;
|
||||
|
||||
// ------------------------------------------------------------------
|
||||
@@ -839,7 +921,6 @@ export function processTimelineResponse(
|
||||
clearInitializing: isInitializing,
|
||||
error: payload.error,
|
||||
sideEffects: [],
|
||||
acknowledgedClientMessageIds: [],
|
||||
};
|
||||
}
|
||||
|
||||
@@ -886,6 +967,7 @@ export function processTimelineResponse(
|
||||
hasActiveInitDeferred,
|
||||
});
|
||||
const replace = bootstrapPolicy.replace;
|
||||
|
||||
const sideEffects: TimelineReducerSideEffect[] = [];
|
||||
const timelineResult = replace
|
||||
? applyTimelineReplacePath({
|
||||
@@ -894,9 +976,6 @@ export function processTimelineResponse(
|
||||
bootstrapPolicy,
|
||||
currentTail,
|
||||
currentHead,
|
||||
sendingClientMessageIds,
|
||||
preserveLiveHead:
|
||||
currentCursor?.epoch === payload.epoch || (!payload.reset && !hasAuthoritativeBaseline),
|
||||
toHydratedEvents,
|
||||
})
|
||||
: applyTimelineIncrementalPath({
|
||||
@@ -945,7 +1024,6 @@ export function processTimelineResponse(
|
||||
clearInitializing,
|
||||
error: null,
|
||||
sideEffects,
|
||||
acknowledgedClientMessageIds: timelineResult.acknowledgedClientMessageIds,
|
||||
};
|
||||
}
|
||||
|
||||
@@ -960,10 +1038,20 @@ export interface ProcessAgentStreamEventInput {
|
||||
currentTail: StreamItem[];
|
||||
currentHead: StreamItem[];
|
||||
currentCursor: TimelineCursor | undefined;
|
||||
hasAuthoritativeBaseline?: boolean;
|
||||
currentAgent: {
|
||||
status: AgentLifecycleStatus;
|
||||
updatedAt: Date;
|
||||
lastActivityAt: Date;
|
||||
} | null;
|
||||
timestamp: Date;
|
||||
}
|
||||
|
||||
export interface AgentPatch {
|
||||
status: AgentLifecycleStatus;
|
||||
updatedAt: Date;
|
||||
lastActivityAt: Date;
|
||||
}
|
||||
|
||||
export interface ProcessAgentStreamEventOutput {
|
||||
tail: StreamItem[];
|
||||
head: StreamItem[];
|
||||
@@ -971,7 +1059,8 @@ export interface ProcessAgentStreamEventOutput {
|
||||
changedHead: boolean;
|
||||
cursor: TimelineCursor | null;
|
||||
cursorChanged: boolean;
|
||||
acknowledgedClientMessageIds: string[];
|
||||
agent: AgentPatch | null;
|
||||
agentChanged: boolean;
|
||||
sideEffects: AgentStreamReducerSideEffect[];
|
||||
}
|
||||
|
||||
@@ -990,12 +1079,18 @@ interface TimelineSequencingGateResult {
|
||||
sideEffects: AgentStreamReducerSideEffect[];
|
||||
}
|
||||
|
||||
export interface AgentStreamReducerAgentSnapshot {
|
||||
status: AgentLifecycleStatus;
|
||||
updatedAt: Date;
|
||||
lastActivityAt: Date;
|
||||
}
|
||||
|
||||
export interface ProcessAgentStreamEventsInput {
|
||||
events: AgentStreamReducerEvent[];
|
||||
currentTail: StreamItem[];
|
||||
currentHead: StreamItem[];
|
||||
currentCursor: TimelineCursor | undefined;
|
||||
hasAuthoritativeBaseline?: boolean;
|
||||
currentAgent: AgentStreamReducerAgentSnapshot | null;
|
||||
}
|
||||
|
||||
export type AgentStreamReducerSnapshot = Omit<ProcessAgentStreamEventsInput, "events">;
|
||||
@@ -1019,14 +1114,27 @@ export interface CreateAgentStreamReducerQueueInput {
|
||||
cancelFlush: (id: number) => void;
|
||||
}
|
||||
|
||||
function applyAgentPatch(
|
||||
currentAgent: AgentStreamReducerAgentSnapshot | null,
|
||||
patch: AgentPatch | null,
|
||||
): AgentStreamReducerAgentSnapshot | null {
|
||||
if (!currentAgent || !patch) {
|
||||
return currentAgent;
|
||||
}
|
||||
return {
|
||||
status: patch.status,
|
||||
updatedAt: patch.updatedAt,
|
||||
lastActivityAt: patch.lastActivityAt,
|
||||
};
|
||||
}
|
||||
|
||||
function processTimelineSequencingGate(input: {
|
||||
event: AgentStreamEventPayload;
|
||||
seq: number | undefined;
|
||||
epoch: string | undefined;
|
||||
currentCursor: TimelineCursor | undefined;
|
||||
hasAuthoritativeBaseline: boolean;
|
||||
}): TimelineSequencingGateResult {
|
||||
const { event, seq, epoch, currentCursor, hasAuthoritativeBaseline } = input;
|
||||
const { event, seq, epoch, currentCursor } = input;
|
||||
const base: TimelineSequencingGateResult = {
|
||||
shouldApplyStreamEvent: true,
|
||||
nextTimelineCursor: null,
|
||||
@@ -1037,9 +1145,6 @@ function processTimelineSequencingGate(input: {
|
||||
if (event.type !== "timeline" || typeof seq !== "number" || typeof epoch !== "string") {
|
||||
return base;
|
||||
}
|
||||
if (!hasAuthoritativeBaseline) {
|
||||
return base;
|
||||
}
|
||||
|
||||
const decision = classifySessionTimelineSeq({
|
||||
cursor: currentCursor ? { epoch: currentCursor.epoch, endSeq: currentCursor.endSeq } : null,
|
||||
@@ -1096,24 +1201,10 @@ function processTimelineSequencingGate(input: {
|
||||
export function processAgentStreamEvent(
|
||||
input: ProcessAgentStreamEventInput,
|
||||
): ProcessAgentStreamEventOutput {
|
||||
const {
|
||||
event,
|
||||
seq,
|
||||
epoch,
|
||||
currentTail,
|
||||
currentHead,
|
||||
currentCursor,
|
||||
timestamp,
|
||||
hasAuthoritativeBaseline = true,
|
||||
} = input;
|
||||
const { event, seq, epoch, currentTail, currentHead, currentCursor, currentAgent, timestamp } =
|
||||
input;
|
||||
|
||||
const sequencing = processTimelineSequencingGate({
|
||||
event,
|
||||
seq,
|
||||
epoch,
|
||||
currentCursor,
|
||||
hasAuthoritativeBaseline,
|
||||
});
|
||||
const sequencing = processTimelineSequencingGate({ event, seq, epoch, currentCursor });
|
||||
const timelineCursor =
|
||||
event.type === "timeline" && seq !== undefined && epoch !== undefined
|
||||
? { epoch, seq }
|
||||
@@ -1122,52 +1213,49 @@ export function processAgentStreamEvent(
|
||||
// ------------------------------------------------------------------
|
||||
// Apply stream event to tail/head
|
||||
// ------------------------------------------------------------------
|
||||
let streamResult: ReturnType<typeof applyStreamEvent>;
|
||||
if (!sequencing.shouldApplyStreamEvent) {
|
||||
streamResult = {
|
||||
tail: currentTail,
|
||||
head: currentHead,
|
||||
changedTail: false,
|
||||
changedHead: false,
|
||||
};
|
||||
} else if (!hasAuthoritativeBaseline) {
|
||||
if (event.type === "timeline" && event.item.type === "user_message") {
|
||||
streamResult = applyStreamEvent({
|
||||
const { tail, head, changedTail, changedHead } = sequencing.shouldApplyStreamEvent
|
||||
? applyStreamEvent({
|
||||
tail: sequencing.resetLiveTimeline ? [] : currentTail,
|
||||
head: sequencing.resetLiveTimeline ? [] : currentHead,
|
||||
event,
|
||||
timestamp,
|
||||
source: "live",
|
||||
timelineCursor,
|
||||
})
|
||||
: {
|
||||
tail: currentTail,
|
||||
head: currentHead,
|
||||
event,
|
||||
timestamp,
|
||||
source: "live",
|
||||
timelineCursor,
|
||||
unmatchedUserMessageInsert: "head",
|
||||
});
|
||||
} else {
|
||||
const overlay = applyStreamEvent({
|
||||
tail: currentHead,
|
||||
head: [],
|
||||
event,
|
||||
timestamp,
|
||||
source: "live",
|
||||
timelineCursor,
|
||||
});
|
||||
streamResult = {
|
||||
tail: currentTail,
|
||||
head: [...overlay.tail, ...overlay.head],
|
||||
changedTail: false,
|
||||
changedHead: overlay.changedTail || overlay.changedHead,
|
||||
changedHead: false,
|
||||
};
|
||||
|
||||
// ------------------------------------------------------------------
|
||||
// Optimistic lifecycle status
|
||||
// ------------------------------------------------------------------
|
||||
let agentPatch: AgentPatch | null = null;
|
||||
let agentChanged = false;
|
||||
|
||||
if (
|
||||
currentAgent &&
|
||||
(event.type === "turn_completed" ||
|
||||
event.type === "turn_canceled" ||
|
||||
event.type === "turn_failed")
|
||||
) {
|
||||
const optimisticStatus = deriveOptimisticLifecycleStatus(currentAgent.status, event);
|
||||
if (optimisticStatus) {
|
||||
const nextUpdatedAtMs = Math.max(currentAgent.updatedAt.getTime(), timestamp.getTime());
|
||||
const nextLastActivityAtMs = Math.max(
|
||||
currentAgent.lastActivityAt.getTime(),
|
||||
timestamp.getTime(),
|
||||
);
|
||||
agentPatch = {
|
||||
status: optimisticStatus,
|
||||
updatedAt: new Date(nextUpdatedAtMs),
|
||||
lastActivityAt: new Date(nextLastActivityAtMs),
|
||||
};
|
||||
agentChanged = true;
|
||||
}
|
||||
} else {
|
||||
streamResult = applyStreamEvent({
|
||||
tail: sequencing.resetLiveTimeline ? [] : currentTail,
|
||||
head: sequencing.resetLiveTimeline ? [] : currentHead,
|
||||
event,
|
||||
timestamp,
|
||||
source: "live",
|
||||
timelineCursor,
|
||||
});
|
||||
}
|
||||
const { tail, head, changedTail, changedHead } = streamResult;
|
||||
|
||||
return {
|
||||
tail,
|
||||
@@ -1176,7 +1264,8 @@ export function processAgentStreamEvent(
|
||||
changedHead,
|
||||
cursor: sequencing.nextTimelineCursor,
|
||||
cursorChanged: sequencing.cursorChanged,
|
||||
acknowledgedClientMessageIds: streamResult.acknowledgedClientMessageIds ?? [],
|
||||
agent: agentPatch,
|
||||
agentChanged,
|
||||
sideEffects: sequencing.sideEffects,
|
||||
};
|
||||
}
|
||||
@@ -1187,10 +1276,12 @@ export function processAgentStreamEvents(
|
||||
let tail = input.currentTail;
|
||||
let head = input.currentHead;
|
||||
let cursor = input.currentCursor;
|
||||
let agent = input.currentAgent;
|
||||
let changedTail = false;
|
||||
let changedHead = false;
|
||||
let cursorChanged = false;
|
||||
const acknowledgedClientMessageIds = new Set<string>();
|
||||
let agentPatch: AgentPatch | null = null;
|
||||
let agentChanged = false;
|
||||
const sideEffects: AgentStreamReducerSideEffect[] = [];
|
||||
|
||||
for (const reducerEvent of input.events) {
|
||||
@@ -1201,7 +1292,7 @@ export function processAgentStreamEvents(
|
||||
currentTail: tail,
|
||||
currentHead: head,
|
||||
currentCursor: cursor,
|
||||
hasAuthoritativeBaseline: input.hasAuthoritativeBaseline,
|
||||
currentAgent: agent,
|
||||
timestamp: reducerEvent.timestamp,
|
||||
});
|
||||
|
||||
@@ -1210,14 +1301,17 @@ export function processAgentStreamEvents(
|
||||
changedTail = changedTail || result.changedTail;
|
||||
changedHead = changedHead || result.changedHead;
|
||||
sideEffects.push(...result.sideEffects);
|
||||
for (const clientMessageId of result.acknowledgedClientMessageIds) {
|
||||
acknowledgedClientMessageIds.add(clientMessageId);
|
||||
}
|
||||
|
||||
if (result.cursorChanged) {
|
||||
cursor = result.cursor ?? undefined;
|
||||
cursorChanged = true;
|
||||
}
|
||||
|
||||
if (result.agentChanged) {
|
||||
agentPatch = result.agent;
|
||||
agentChanged = true;
|
||||
agent = applyAgentPatch(agent, result.agent);
|
||||
}
|
||||
}
|
||||
|
||||
return {
|
||||
@@ -1227,7 +1321,8 @@ export function processAgentStreamEvents(
|
||||
changedHead,
|
||||
cursor: cursor ?? null,
|
||||
cursorChanged,
|
||||
acknowledgedClientMessageIds: [...acknowledgedClientMessageIds],
|
||||
agent: agentPatch,
|
||||
agentChanged,
|
||||
sideEffects,
|
||||
};
|
||||
}
|
||||
@@ -1310,7 +1405,6 @@ export function createAgentStreamReducerQueue(
|
||||
interface StreamStatePatch {
|
||||
tail?: StreamItem[];
|
||||
head?: StreamItem[];
|
||||
acknowledgedClientMessageIds?: readonly string[];
|
||||
}
|
||||
|
||||
export interface CreateSessionAgentStreamReducerQueueInput {
|
||||
@@ -1320,6 +1414,7 @@ export interface CreateSessionAgentStreamReducerQueueInput {
|
||||
serverId: string,
|
||||
state: (prev: Map<string, TimelineCursor>) => Map<string, TimelineCursor>,
|
||||
) => void;
|
||||
setAgents: (serverId: string, state: (prev: Map<string, Agent>) => Map<string, Agent>) => void;
|
||||
recoverTimelineGap: (agentId: string, cursor: { epoch: string; endSeq: number }) => void;
|
||||
}
|
||||
|
||||
@@ -1334,31 +1429,31 @@ function cancelAgentStreamReducerFlush(id: number) {
|
||||
export function createSessionAgentStreamReducerQueue(
|
||||
input: CreateSessionAgentStreamReducerQueueInput,
|
||||
): AgentStreamReducerQueue {
|
||||
const { serverId, setAgentStreamState, setAgentTimelineCursor, recoverTimelineGap } = input;
|
||||
const { serverId, setAgentStreamState, setAgentTimelineCursor, setAgents, recoverTimelineGap } =
|
||||
input;
|
||||
|
||||
return createAgentStreamReducerQueue({
|
||||
getSnapshot: (agentId) => {
|
||||
const session = useSessionStore.getState().sessions[serverId];
|
||||
const timeline = selectAgentTimelineState(session, agentId);
|
||||
const currentAgentEntry = session?.agents.get(agentId);
|
||||
return {
|
||||
currentTail: timeline.status === "cold" ? [] : timeline.items,
|
||||
currentTail: session?.agentStreamTail.get(agentId) ?? [],
|
||||
currentHead: session?.agentStreamHead.get(agentId) ?? [],
|
||||
currentCursor: timeline.status === "synced" ? (timeline.range ?? undefined) : undefined,
|
||||
hasAuthoritativeBaseline: timeline.status === "synced",
|
||||
currentCursor: session?.agentTimelineCursor.get(agentId),
|
||||
currentAgent: currentAgentEntry
|
||||
? {
|
||||
status: currentAgentEntry.status,
|
||||
updatedAt: currentAgentEntry.updatedAt,
|
||||
lastActivityAt: currentAgentEntry.lastActivityAt,
|
||||
}
|
||||
: null,
|
||||
};
|
||||
},
|
||||
commit: (agentId, result, events) => {
|
||||
if (
|
||||
result.changedTail ||
|
||||
result.changedHead ||
|
||||
result.acknowledgedClientMessageIds.length > 0
|
||||
) {
|
||||
if (result.changedTail || result.changedHead) {
|
||||
setAgentStreamState(serverId, agentId, {
|
||||
...(result.changedTail ? { tail: result.tail } : {}),
|
||||
...(result.changedHead ? { head: result.head } : {}),
|
||||
...(result.acknowledgedClientMessageIds.length > 0
|
||||
? { acknowledgedClientMessageIds: result.acknowledgedClientMessageIds }
|
||||
: {}),
|
||||
});
|
||||
}
|
||||
|
||||
@@ -1391,6 +1486,24 @@ export function createSessionAgentStreamReducerQueue(
|
||||
return next;
|
||||
});
|
||||
}
|
||||
|
||||
if (result.agentChanged && result.agent) {
|
||||
const nextAgent = result.agent;
|
||||
setAgents(serverId, (prev) => {
|
||||
const current = prev.get(agentId);
|
||||
if (!current) {
|
||||
return prev;
|
||||
}
|
||||
const next = new Map(prev);
|
||||
next.set(agentId, {
|
||||
...current,
|
||||
status: nextAgent.status,
|
||||
updatedAt: nextAgent.updatedAt,
|
||||
lastActivityAt: nextAgent.lastActivityAt,
|
||||
});
|
||||
return next;
|
||||
});
|
||||
}
|
||||
},
|
||||
handleSideEffects: (agentId, sideEffects) => {
|
||||
for (const effect of sideEffects) {
|
||||
|
||||
@@ -22,16 +22,34 @@ function assistant(id: string, timestamp: Date): StreamItem {
|
||||
}
|
||||
|
||||
describe("deriveStreamTurnTiming", () => {
|
||||
it("starts elapsed time from the submitted prompt", () => {
|
||||
const submittedAt = new Date("2026-05-15T00:00:00.000Z");
|
||||
it("reserves a running footer for an optimistic prompt before the host starts the turn", () => {
|
||||
const optimisticPrompt = {
|
||||
...user("optimistic", new Date("2026-05-15T00:00:00.000Z")),
|
||||
optimistic: true as const,
|
||||
};
|
||||
|
||||
const timing = deriveStreamTurnTiming({
|
||||
agentStatus: "idle",
|
||||
tail: [],
|
||||
head: [optimisticPrompt],
|
||||
});
|
||||
|
||||
assert.equal(timing.isActive, true);
|
||||
});
|
||||
|
||||
it("does not start elapsed time from an optimistic prompt", () => {
|
||||
const optimisticPrompt = {
|
||||
...user("optimistic", new Date("2026-05-15T00:00:00.000Z")),
|
||||
optimistic: true as const,
|
||||
};
|
||||
|
||||
const timing = deriveStreamTurnTiming({
|
||||
agentStatus: "running",
|
||||
tail: [],
|
||||
head: [user("submitted", submittedAt)],
|
||||
head: [optimisticPrompt],
|
||||
});
|
||||
|
||||
assert.equal(timing.runningStartedAt, submittedAt);
|
||||
assert.equal(timing.runningStartedAt, null);
|
||||
});
|
||||
|
||||
it("uses the last user message as the running turn start", () => {
|
||||
|
||||
@@ -9,6 +9,7 @@ export interface TurnTiming {
|
||||
export interface StreamTurnTiming {
|
||||
byAssistantId: Map<string, TurnTiming>;
|
||||
runningStartedAt: Date | null;
|
||||
isActive: boolean;
|
||||
}
|
||||
|
||||
export function deriveStreamTurnTiming(params: {
|
||||
@@ -18,6 +19,8 @@ export function deriveStreamTurnTiming(params: {
|
||||
}): StreamTurnTiming {
|
||||
const byAssistantId = new Map<string, TurnTiming>();
|
||||
let currentUserAt: Date | null = null;
|
||||
let currentAuthoritativeUserAt: Date | null = null;
|
||||
let currentUserIsOptimistic = false;
|
||||
let currentLastItemAt: Date | null = null;
|
||||
let currentAssistantIds: string[] = [];
|
||||
|
||||
@@ -39,6 +42,8 @@ export function deriveStreamTurnTiming(params: {
|
||||
if (item.kind === "user_message") {
|
||||
flushCompletedTurn();
|
||||
currentUserAt = item.timestamp;
|
||||
currentAuthoritativeUserAt = item.optimistic ? null : item.timestamp;
|
||||
currentUserIsOptimistic = item.optimistic === true;
|
||||
currentLastItemAt = null;
|
||||
currentAssistantIds = [];
|
||||
return;
|
||||
@@ -60,7 +65,7 @@ export function deriveStreamTurnTiming(params: {
|
||||
}
|
||||
|
||||
const isRunning = params.agentStatus === "running";
|
||||
const runningStartedAt = isRunning ? currentUserAt : null;
|
||||
const runningStartedAt = isRunning ? currentAuthoritativeUserAt : null;
|
||||
if (params.agentStatus !== "running") {
|
||||
flushCompletedTurn();
|
||||
}
|
||||
@@ -68,5 +73,6 @@ export function deriveStreamTurnTiming(params: {
|
||||
return {
|
||||
byAssistantId,
|
||||
runningStartedAt,
|
||||
isActive: isRunning || currentUserIsOptimistic,
|
||||
};
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user