Fix agent turn timing persistence
This commit is contained in:
Mohamed Boudra
@@ -1,4 +1,3 @@
|
||||
import invariant from "tiny-invariant";
|
||||
import { describe, expect, it } from "vitest";
|
||||
import type { AgentStreamEventPayload } from "@server/shared/messages";
|
||||
import type { StreamItem } from "@/types/stream";
|
||||
@@ -241,12 +240,10 @@ describe("processTimelineResponse", () => {
|
||||
});
|
||||
|
||||
const user = result.tail.find((item) => item.kind === "user_message");
|
||||
const header = result.tail.find((item) => item.kind === "turn_header");
|
||||
const assistant = result.tail.find((item) => item.kind === "assistant_message");
|
||||
|
||||
expect(user?.timestamp.toISOString()).toBe("2025-01-01T12:00:03.000Z");
|
||||
invariant(header?.kind === "turn_header");
|
||||
expect(header.startedAt.toISOString()).toBe("2025-01-01T12:00:04.000Z");
|
||||
expect(header.completedAt?.toISOString()).toBe("2025-01-01T12:00:04.000Z");
|
||||
expect(assistant?.timestamp.toISOString()).toBe("2025-01-01T12:00:04.000Z");
|
||||
});
|
||||
|
||||
it("sets cursor to null when reset=true but no cursors in payload", () => {
|
||||
|
||||
@@ -0,0 +1,87 @@
|
||||
import assert from "node:assert/strict";
|
||||
import { describe, it } from "vitest";
|
||||
import { deriveStreamTurnTiming } from "./turn-time";
|
||||
import type { StreamItem } from "@/types/stream";
|
||||
|
||||
function user(id: string, timestamp: Date): StreamItem {
|
||||
return {
|
||||
kind: "user_message",
|
||||
id,
|
||||
text: id,
|
||||
timestamp,
|
||||
};
|
||||
}
|
||||
|
||||
function assistant(id: string, timestamp: Date): StreamItem {
|
||||
return {
|
||||
kind: "assistant_message",
|
||||
id,
|
||||
text: id,
|
||||
timestamp,
|
||||
};
|
||||
}
|
||||
|
||||
describe("deriveStreamTurnTiming", () => {
|
||||
it("uses the last user message as the running turn start", () => {
|
||||
const firstUserAt = new Date("2026-05-15T00:00:00.000Z");
|
||||
const secondUserAt = new Date("2026-05-15T00:01:00.000Z");
|
||||
|
||||
const timing = deriveStreamTurnTiming({
|
||||
agentStatus: "running",
|
||||
tail: [
|
||||
user("u1", firstUserAt),
|
||||
assistant("a1", new Date("2026-05-15T00:00:05.000Z")),
|
||||
user("u2", secondUserAt),
|
||||
],
|
||||
head: [assistant("a2", new Date("2026-05-15T00:01:04.000Z"))],
|
||||
});
|
||||
|
||||
assert.equal(timing.runningStartedAt, secondUserAt);
|
||||
assert.equal(timing.byAssistantId.has("a2"), false);
|
||||
});
|
||||
|
||||
it("derives completed turn timing from user and assistant item timestamps", () => {
|
||||
const userAt = new Date("2026-05-15T00:00:00.000Z");
|
||||
const assistantAt = new Date("2026-05-15T00:00:07.000Z");
|
||||
|
||||
const timing = deriveStreamTurnTiming({
|
||||
agentStatus: "idle",
|
||||
tail: [
|
||||
user("u1", userAt),
|
||||
assistant("a1", assistantAt),
|
||||
user("u2", new Date("2026-05-15T00:01:00.000Z")),
|
||||
],
|
||||
head: [],
|
||||
});
|
||||
|
||||
assert.deepEqual(timing.byAssistantId.get("a1"), {
|
||||
startedAt: userAt,
|
||||
completedAt: assistantAt,
|
||||
durationMs: 7000,
|
||||
});
|
||||
});
|
||||
|
||||
it("maps multiple assistant chunks in one turn to the same timing", () => {
|
||||
const userAt = new Date("2026-05-15T00:00:00.000Z");
|
||||
const firstAssistantAt = new Date("2026-05-15T00:00:03.000Z");
|
||||
const lastAssistantAt = new Date("2026-05-15T00:00:07.000Z");
|
||||
|
||||
const timing = deriveStreamTurnTiming({
|
||||
agentStatus: "idle",
|
||||
tail: [
|
||||
user("u1", userAt),
|
||||
assistant("a1", firstAssistantAt),
|
||||
assistant("a2", lastAssistantAt),
|
||||
],
|
||||
head: [],
|
||||
});
|
||||
|
||||
const expected = {
|
||||
startedAt: userAt,
|
||||
completedAt: lastAssistantAt,
|
||||
durationMs: 7000,
|
||||
};
|
||||
assert.deepEqual(timing.byAssistantId.get("a1"), expected);
|
||||
assert.deepEqual(timing.byAssistantId.get("a2"), expected);
|
||||
});
|
||||
});
|
||||
@@ -1,146 +1,83 @@
|
||||
import type { StreamItem, TurnHeaderItem, TurnHeaderOutcome } from "@/types/stream";
|
||||
import type { StreamItem } from "@/types/stream";
|
||||
|
||||
export function appendLiveTurnHeader(state: StreamItem[], startedAt: Date): StreamItem[] {
|
||||
const lastHeader = findLastTurnHeader(state);
|
||||
if (lastHeader && lastHeader.completedAt === undefined) {
|
||||
return state;
|
||||
}
|
||||
const header: TurnHeaderItem = {
|
||||
kind: "turn_header",
|
||||
id: `turn_${startedAt.getTime()}_${state.length.toString(36)}`,
|
||||
timestamp: startedAt,
|
||||
startedAt,
|
||||
source: "live",
|
||||
};
|
||||
return [...state, header];
|
||||
export interface TurnTiming {
|
||||
startedAt: Date;
|
||||
completedAt: Date;
|
||||
durationMs: number;
|
||||
}
|
||||
|
||||
export function completeLastTurnHeader(
|
||||
state: StreamItem[],
|
||||
completedAt: Date,
|
||||
outcome: TurnHeaderOutcome,
|
||||
): StreamItem[] {
|
||||
for (let i = state.length - 1; i >= 0; i -= 1) {
|
||||
const item = state[i];
|
||||
if (item?.kind !== "turn_header") {
|
||||
continue;
|
||||
}
|
||||
if (item.completedAt !== undefined) {
|
||||
return state;
|
||||
}
|
||||
const updated: TurnHeaderItem = {
|
||||
...item,
|
||||
completedAt,
|
||||
durationMs: Math.max(0, completedAt.getTime() - item.startedAt.getTime()),
|
||||
outcome,
|
||||
};
|
||||
const next = [...state];
|
||||
next[i] = updated;
|
||||
return next;
|
||||
}
|
||||
return state;
|
||||
export interface StreamTurnTiming {
|
||||
byAssistantId: Map<string, TurnTiming>;
|
||||
runningStartedAt: Date | null;
|
||||
}
|
||||
|
||||
export function synthesizeMissingTurnHeaders(state: StreamItem[]): StreamItem[] {
|
||||
const result: StreamItem[] = [];
|
||||
let segment: StreamItem[] = [];
|
||||
let needsHeader = true;
|
||||
export function deriveStreamTurnTiming(params: {
|
||||
agentStatus: string;
|
||||
tail: StreamItem[];
|
||||
head: StreamItem[];
|
||||
}): StreamTurnTiming {
|
||||
const byAssistantId = new Map<string, TurnTiming>();
|
||||
let currentUserAt: Date | null = null;
|
||||
let currentLastItemAt: Date | null = null;
|
||||
let currentAssistantIds: string[] = [];
|
||||
|
||||
function flushSegment() {
|
||||
if (segment.length === 0) {
|
||||
const flushCompletedTurn = () => {
|
||||
if (!currentUserAt || !currentLastItemAt || currentAssistantIds.length === 0) {
|
||||
return;
|
||||
}
|
||||
if (needsHeader) {
|
||||
const first = segment[0];
|
||||
const last = segment[segment.length - 1];
|
||||
const startedAt = first.timestamp;
|
||||
const completedAt = last.timestamp;
|
||||
const header: TurnHeaderItem = {
|
||||
kind: "turn_header",
|
||||
id: `turn_derived_${startedAt.getTime()}_${result.length.toString(36)}`,
|
||||
timestamp: startedAt,
|
||||
startedAt,
|
||||
completedAt,
|
||||
durationMs: Math.max(0, completedAt.getTime() - startedAt.getTime()),
|
||||
outcome: "completed",
|
||||
source: "derived",
|
||||
};
|
||||
result.push(header);
|
||||
const timing: TurnTiming = {
|
||||
startedAt: currentUserAt,
|
||||
completedAt: currentLastItemAt,
|
||||
durationMs: Math.max(0, currentLastItemAt.getTime() - currentUserAt.getTime()),
|
||||
};
|
||||
for (const id of currentAssistantIds) {
|
||||
byAssistantId.set(id, timing);
|
||||
}
|
||||
result.push(...segment);
|
||||
segment = [];
|
||||
}
|
||||
};
|
||||
|
||||
for (const item of state) {
|
||||
const visitItem = (item: StreamItem) => {
|
||||
if (item.kind === "user_message") {
|
||||
flushSegment();
|
||||
needsHeader = true;
|
||||
result.push(item);
|
||||
continue;
|
||||
flushCompletedTurn();
|
||||
currentUserAt = item.timestamp;
|
||||
currentLastItemAt = null;
|
||||
currentAssistantIds = [];
|
||||
return;
|
||||
}
|
||||
if (item.kind === "turn_header") {
|
||||
flushSegment();
|
||||
needsHeader = false;
|
||||
result.push(item);
|
||||
continue;
|
||||
if (!currentUserAt) {
|
||||
return;
|
||||
}
|
||||
segment.push(item);
|
||||
currentLastItemAt = item.timestamp;
|
||||
if (item.kind === "assistant_message") {
|
||||
currentAssistantIds.push(item.id);
|
||||
}
|
||||
};
|
||||
|
||||
for (const item of params.tail) {
|
||||
visitItem(item);
|
||||
}
|
||||
for (const item of params.head) {
|
||||
visitItem(item);
|
||||
}
|
||||
|
||||
flushSegment();
|
||||
return result;
|
||||
}
|
||||
|
||||
export function findInFlightTurnStartedAt(params: {
|
||||
agentStatus: string;
|
||||
liveHead: StreamItem[];
|
||||
tail: StreamItem[];
|
||||
}): Date | null {
|
||||
const runningStartedAt =
|
||||
params.agentStatus === "running"
|
||||
? (findLastUserMessageTimestamp(params.head) ?? currentUserAt)
|
||||
: null;
|
||||
if (params.agentStatus !== "running") {
|
||||
return null;
|
||||
flushCompletedTurn();
|
||||
}
|
||||
const headStartedAt = findLastOpenTurnHeader(params.liveHead)?.startedAt;
|
||||
if (headStartedAt) {
|
||||
return headStartedAt;
|
||||
}
|
||||
return findLastOpenTurnHeader(params.tail)?.startedAt ?? null;
|
||||
|
||||
return {
|
||||
byAssistantId,
|
||||
runningStartedAt,
|
||||
};
|
||||
}
|
||||
|
||||
export interface TurnHeaderNeighborResolver {
|
||||
getNeighborIndex(index: number, relation: "above" | "below"): number;
|
||||
}
|
||||
|
||||
export function findTurnHeaderForAssistantTurn(params: {
|
||||
strategy: TurnHeaderNeighborResolver;
|
||||
items: StreamItem[];
|
||||
startIndex: number;
|
||||
}): TurnHeaderItem | null {
|
||||
let index = params.startIndex;
|
||||
while (index >= 0 && index < params.items.length) {
|
||||
const item = params.items[index];
|
||||
if (!item) return null;
|
||||
if (item.kind === "turn_header") return item;
|
||||
if (item.kind === "user_message") return null;
|
||||
index = params.strategy.getNeighborIndex(index, "above");
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
function findLastTurnHeader(state: StreamItem[]): TurnHeaderItem | null {
|
||||
for (let i = state.length - 1; i >= 0; i -= 1) {
|
||||
const item = state[i];
|
||||
if (item?.kind === "turn_header") {
|
||||
return item;
|
||||
}
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
function findLastOpenTurnHeader(state: StreamItem[]): TurnHeaderItem | null {
|
||||
for (let i = state.length - 1; i >= 0; i -= 1) {
|
||||
const item = state[i];
|
||||
if (item?.kind === "turn_header" && item.completedAt === undefined) {
|
||||
return item;
|
||||
function findLastUserMessageTimestamp(items: StreamItem[]): Date | null {
|
||||
for (let i = items.length - 1; i >= 0; i -= 1) {
|
||||
const item = items[i];
|
||||
if (item?.kind === "user_message") {
|
||||
return item.timestamp;
|
||||
}
|
||||
}
|
||||
return null;
|
||||
|
||||
Reference in New Issue
Block a user