Use daemon timestamps for turn timing (#1033)

packages/app/src/timeline/session-stream-reducers.test.ts

@@ -1,3 +1,4 @@
+import invariant from "tiny-invariant";
 import { describe, expect, it } from "vitest";
 import type { AgentStreamEventPayload } from "@server/shared/messages";
 import type { StreamItem } from "@/types/stream";
@@ -218,6 +219,36 @@ describe("processTimelineResponse", () => {
     expect(result.sideEffects.some((e) => e.type === "flush_pending_updates")).toBe(true);
   });
 
+  it("uses the timeline entry timestamp as canonical", () => {
+    const result = processTimelineResponse({
+      ...baseTimelineInput,
+      payload: {
+        ...baseTimelineInput.payload,
+        reset: true,
+        startCursor: { seq: 1 },
+        endCursor: { seq: 2 },
+        entries: [
+          {
+            ...makeTimelineEntry(1, "hello", "user_message"),
+            timestamp: new Date("2025-01-01T12:00:03Z").toISOString(),
+          },
+          {
+            ...makeTimelineEntry(2, "reply"),
+            timestamp: new Date("2025-01-01T12:00:04Z").toISOString(),
+          },
+        ],
+      },
+    });
+
+    const user = result.tail.find((item) => item.kind === "user_message");
+    const header = result.tail.find((item) => item.kind === "turn_header");
+
+    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");
+  });
+
   it("sets cursor to null when reset=true but no cursors in payload", () => {
     const result = processTimelineResponse({
       ...baseTimelineInput,

packages/app/src/timeline/session-stream-reducers.ts

@@ -395,7 +395,10 @@ function applyTimelineIncrementalPath(args: {
   if (acceptedUnits.length > 0) {
     if (payload.direction === "before") {
       const olderTail = hydrateStreamState(
-        acceptedUnits.map(({ event, timestamp }) => ({ event, timestamp })),
+        acceptedUnits.map(({ event, timestamp }) => ({
+          event,
+          timestamp,
+        })),
         { source: "canonical" },
       );
       nextTail = mergePrependedCanonicalTail(olderTail, currentTail);
@@ -414,7 +417,9 @@ function applyTimelineIncrementalPath(args: {
     } else {
       nextTail = acceptedUnits.reduce<StreamItem[]>(
         (state, { event, timestamp }) =>
-          reduceStreamUpdate(state, event, timestamp, { source: "canonical" }),
+          reduceStreamUpdate(state, event, timestamp, {
+            source: "canonical",
+          }),
         currentTail,
       );
     }
@@ -482,7 +487,7 @@ export function processTimelineResponse(
   }));
 
   const toHydratedEvents = (
-    units: typeof timelineUnits,
+    units: TimelineUnit[],
   ): Array<{ event: AgentStreamEventPayload; timestamp: Date }> =>
     units.map(({ event, timestamp }) => ({ event, timestamp }));
 

packages/app/src/timeline/turn-time.ts

@@ -0,0 +1,147 @@
+import type { StreamItem, TurnHeaderItem, TurnHeaderOutcome } 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 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 function synthesizeMissingTurnHeaders(state: StreamItem[]): StreamItem[] {
+  const result: StreamItem[] = [];
+  let segment: StreamItem[] = [];
+  let needsHeader = true;
+
+  function flushSegment() {
+    if (segment.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);
+    }
+    result.push(...segment);
+    segment = [];
+  }
+
+  for (const item of state) {
+    if (item.kind === "user_message") {
+      flushSegment();
+      needsHeader = true;
+      result.push(item);
+      continue;
+    }
+    if (item.kind === "turn_header") {
+      flushSegment();
+      needsHeader = false;
+      result.push(item);
+      continue;
+    }
+    segment.push(item);
+  }
+
+  flushSegment();
+  return result;
+}
+
+export function findInFlightTurnStartedAt(params: {
+  agentStatus: string;
+  liveHead: StreamItem[];
+  tail: StreamItem[];
+}): Date | null {
+  if (params.agentStatus !== "running") {
+    return null;
+  }
+  const headStartedAt = findLastOpenTurnHeader(params.liveHead)?.startedAt;
+  if (headStartedAt) {
+    return headStartedAt;
+  }
+  return findLastOpenTurnHeader(params.tail)?.startedAt ?? null;
+}
+
+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;
+    }
+  }
+  return null;
+}