diff --git a/docs/architecture.md b/docs/architecture.md index 100bd7967..44f84651a 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -207,6 +207,7 @@ initializing → idle ⇄ running - **AgentManager** is the source of truth for agent state and broadcasts updates to all subscribers - Timeline is append-only with epochs (each run starts a new epoch). Storage uses sequence numbers for client-side dedup; the default fetch page is 200 items +- Timeline row `timestamp` values are canonical daemon-owned timestamps. Providers may supply original replay timestamps, but clients must not guess timestamp trust or hide time UI based on local clock heuristics. - Events stream to all subscribed clients in real time - Agent state persists to `$PASEO_HOME/agents/{cwd-with-dashes}/{agent-id}.json` (timeline rows live alongside the record) diff --git a/packages/app/src/components/agent-stream-view.tsx b/packages/app/src/components/agent-stream-view.tsx index 7de6bf523..e6404780c 100644 --- a/packages/app/src/components/agent-stream-view.tsx +++ b/packages/app/src/components/agent-stream-view.tsx @@ -35,13 +35,14 @@ import { Check, ChevronDown, X } from "lucide-react-native"; import { usePanelStore } from "@/stores/panel-store"; import { AssistantMessage, + AssistantTurnFooter, SpeakMessage, UserMessage, ActivityLog, ToolCall, TodoListCard, CompactionMarker, - TurnCopyButton, + LiveElapsed, MessageOuterSpacingProvider, type InlinePathTarget, } from "./message"; @@ -89,6 +90,7 @@ import { WORKING_INDICATOR_CYCLE_MS, WORKING_INDICATOR_OFFSETS, } from "@/utils/working-indicator"; +import { findInFlightTurnStartedAt, findTurnHeaderForAssistantTurn } from "@/timeline/turn-time"; import { isWeb } from "@/constants/platform"; import { SPACING, type Theme } from "@/styles/theme"; @@ -274,6 +276,15 @@ const AgentStreamViewComponent = forwardRef + findInFlightTurnStartedAt({ + agentStatus: agent.status, + liveHead: baseRenderModel.segments.liveHead, + tail: streamItems, + }), + [agent.status, baseRenderModel.segments.liveHead, streamItems], + ); useImperativeHandle( ref, () => ({ @@ -549,6 +560,12 @@ const AgentStreamViewComponent = forwardRef ); + case "turn_header": + // Purely a data carrier - duration renders next to the assistant + // turn's copy button (TurnCopyButtonSlot), and the in-flight + // ticker renders next to the WorkingIndicator dots. + return null; + default: return null; } @@ -583,7 +600,7 @@ const AgentStreamViewComponent = forwardRef; + footer = ; } else if (isEndOfAssistantTurn) { footer = ( @@ -603,6 +620,7 @@ const AgentStreamViewComponent = forwardRef showAuxiliaryWorkingIndicator ? ( - + ) : null, - [showAuxiliaryWorkingIndicator], + [showAuxiliaryWorkingIndicator, inFlightTurnStartedAt], ); const renderModel = useMemo(() => { return { @@ -803,7 +821,13 @@ const AgentStreamViewComponent = forwardRef { @@ -867,14 +891,25 @@ function WorkingIndicator({ variant = "auxiliary" }: { variant?: "auxiliary" | " + {inFlightTurnStartedAt ? ( + + ) : null} ); } -function InlineWorkingIndicatorSlot() { +function InlineWorkingIndicatorSlot({ + inFlightTurnStartedAt, +}: { + inFlightTurnStartedAt: Date | null; +}) { return ( - + ); } @@ -900,7 +935,17 @@ function TurnCopyButtonSlot({ strategy, items, startIndex }: TurnCopyButtonSlotP }), [strategy, items, startIndex], ); - return ; + const header = useMemo( + () => findTurnHeaderForAssistantTurn({ strategy, items, startIndex }), + [strategy, items, startIndex], + ); + return ( + + ); } interface ToolCallSlotProps extends Omit< @@ -1251,8 +1296,15 @@ const stylesheet = StyleSheet.create((theme) => ({ }, inlineWorkingIndicatorFrame: { height: 18, + flexDirection: "row", alignItems: "center", - justifyContent: "center", + justifyContent: "flex-start", + gap: theme.spacing[2], + }, + workingElapsed: { + color: theme.colors.foregroundMuted, + fontSize: theme.fontSize.sm, + fontVariant: ["tabular-nums"], }, workingIndicatorBubble: { flexDirection: "row", diff --git a/packages/app/src/components/message.tsx b/packages/app/src/components/message.tsx index 95ca9e1b2..18da77ed4 100644 --- a/packages/app/src/components/message.tsx +++ b/packages/app/src/components/message.tsx @@ -74,6 +74,7 @@ import { type AssistantFileLinkSource } from "@/utils/assistant-file-link-resolv import { useAssistantFileLinkResolver } from "@/hooks/use-assistant-file-link-resolver"; import type { ToastApi } from "@/components/toast-host"; import { splitMarkdownBlocks } from "@/utils/split-markdown-blocks"; +import { formatDuration, formatMessageTimestamp } from "@/utils/time"; import { getAssistantImageLoadStateFromMetadata, getAssistantImageMetadata, @@ -391,16 +392,25 @@ const userMessageStylesheet = StyleSheet.create((theme) => ({ fontSize: theme.fontSize.sm, }, copyButton: { - alignSelf: "flex-end", padding: theme.spacing[1], + }, + trailingRow: { + alignSelf: "flex-end", + flexDirection: "row", + alignItems: "center", + gap: theme.spacing[2], marginTop: theme.spacing[2], }, - copyButtonHidden: { + trailingRowHidden: { opacity: 0, }, - copyButtonVisible: { + trailingRowVisible: { opacity: 1, }, + timestampText: { + color: theme.colors.foregroundMuted, + fontSize: theme.fontSize.xs, + }, })); function UserMessageAttachmentThumbnail({ image }: { image: UserMessageImageAttachment }) { @@ -433,22 +443,26 @@ export const UserMessage = memo(function UserMessage({ message, images = [], attachments = [], - timestamp: _timestamp, + timestamp, isFirstInGroup = true, isLastInGroup = true, disableOuterSpacing, }: UserMessageProps) { const isCompact = useIsCompactFormFactor(); - const [messageHovered, setMessageHovered] = useState(false); - const [copyButtonHovered, setCopyButtonHovered] = useState(false); + const [isHovered, setIsHovered] = useState(false); + const [isCopyHovered, setIsCopyHovered] = useState(false); const resolvedDisableOuterSpacing = useDisableOuterSpacing(disableOuterSpacing); const hasText = message.trim().length > 0; const hasImages = images.length > 0; const hasAttachments = attachments.length > 0; - const showCopyButton = hasText && (isCompact || messageHovered || copyButtonHovered); + const showTrailingRow = hasText && (isCompact || isNative || isHovered || isCopyHovered); + const formattedTimestamp = useMemo( + () => formatMessageTimestamp(new Date(timestamp)), + [timestamp], + ); - const handleHoverIn = useCallback(() => setMessageHovered(true), []); - const handleHoverOut = useCallback(() => setMessageHovered(false), []); + const handleHoverIn = useCallback(() => setIsHovered(true), []); + const handleHoverOut = useCallback(() => setIsHovered(false), []); const getMessageContent = useCallback(() => message, [message]); const containerStyle = useMemo( @@ -476,66 +490,220 @@ export const UserMessage = memo(function UserMessage({ ], [hasText], ); - const copyButtonStyle = useMemo( + const trailingRowStyle = useMemo( () => [ - userMessageStylesheet.copyButton, - showCopyButton - ? userMessageStylesheet.copyButtonVisible - : userMessageStylesheet.copyButtonHidden, + userMessageStylesheet.trailingRow, + showTrailingRow + ? userMessageStylesheet.trailingRowVisible + : userMessageStylesheet.trailingRowHidden, ], - [showCopyButton], + [showTrailingRow], ); return ( - - - {hasImages ? ( - - {images.map((image) => ( - - - - ))} - - ) : null} - {hasAttachments ? ( - - {attachments.map((attachment, index) => ( - - - {getUserMessageAttachmentLabel(attachment)} - - - ))} - - ) : null} - {hasText ? ( - - {message} - - ) : null} - + + + + {hasImages ? ( + + {images.map((image) => ( + + + + ))} + + ) : null} + {hasAttachments ? ( + + {attachments.map((attachment, index) => ( + + + {getUserMessageAttachmentLabel(attachment)} + + + ))} + + ) : null} + {hasText ? ( + + {message} + + ) : null} + + {hasText ? ( - + + {formattedTimestamp} + + ) : null} - + ); }); +interface AssistantTurnFooterProps { + getContent: () => string; + startedAt?: Date; + durationMs?: number; +} + +const assistantTurnFooterStylesheet = StyleSheet.create((theme) => ({ + container: { + flexDirection: "row", + alignItems: "center", + gap: theme.spacing[2], + paddingVertical: theme.spacing[1], + }, + copyButton: { + alignSelf: "center", + padding: theme.spacing[1], + paddingTop: theme.spacing[1], + marginTop: 0, + }, + labelWrapper: { + position: "relative", + }, + labelSizer: { + color: theme.colors.foregroundMuted, + fontSize: theme.fontSize.xs, + opacity: 0, + }, + labelOverlay: { + position: "absolute", + top: 0, + left: 0, + color: theme.colors.foregroundMuted, + fontSize: theme.fontSize.xs, + }, +})); + +const TIMESTAMP_REVEAL_MS = 3000; + +/** + * Footer rendered next to the copy button at the end of an assistant turn. + * Always shows the turn duration; swaps to the start timestamp on hover (web) + * or tap (native). The hidden sizer keeps the label width stable while the + * visible text swaps. + */ +export const AssistantTurnFooter = memo(function AssistantTurnFooter({ + getContent, + startedAt, + durationMs, +}: AssistantTurnFooterProps) { + const [hovered, setHovered] = useState(false); + const [pressedReveal, setPressedReveal] = useState(false); + const revealTimerRef = useRef | null>(null); + + useEffect(() => { + return () => { + if (revealTimerRef.current) { + clearTimeout(revealTimerRef.current); + revealTimerRef.current = null; + } + }; + }, []); + + const durationLabel = useMemo( + () => (durationMs !== undefined ? `Worked for ${formatDuration(durationMs)}` : ""), + [durationMs], + ); + const timestampLabel = useMemo( + () => (startedAt ? formatMessageTimestamp(startedAt) : ""), + [startedAt], + ); + + const canSwap = Boolean(timestampLabel); + const showTimestamp = canSwap && (isWeb ? hovered : pressedReveal); + + const handleHoverIn = useCallback(() => setHovered(true), []); + const handleHoverOut = useCallback(() => setHovered(false), []); + const handlePress = useCallback(() => { + if (isWeb || !canSwap) return; + if (revealTimerRef.current) { + clearTimeout(revealTimerRef.current); + } + setPressedReveal((prev) => !prev); + revealTimerRef.current = setTimeout(() => { + setPressedReveal(false); + revealTimerRef.current = null; + }, TIMESTAMP_REVEAL_MS); + }, [canSwap]); + + return ( + + + {durationLabel ? ( + + + {/* Sizer reserves space for whichever label is longer so the + container width is stable across hover transitions. */} + + {durationLabel.length >= timestampLabel.length ? durationLabel : timestampLabel} + + + {showTimestamp ? timestampLabel : durationLabel} + + + + ) : null} + + ); +}); + +interface LiveElapsedProps { + startedAt: Date; + style?: StyleProp; + testID?: string; +} + +/** + * Ticks every 100ms to render an elapsed duration. Isolated from parents so + * only this component re-renders on each tick. + */ +export const LiveElapsed = memo(function LiveElapsed({ + startedAt, + style, + testID, +}: LiveElapsedProps) { + const startedAtMs = startedAt.getTime(); + const [elapsedMs, setElapsedMs] = useState(() => Math.max(0, Date.now() - startedAtMs)); + + useEffect(() => { + setElapsedMs(Math.max(0, Date.now() - startedAtMs)); + const handle = setInterval(() => { + setElapsedMs(Math.max(0, Date.now() - startedAtMs)); + }, 100); + return () => clearInterval(handle); + }, [startedAtMs]); + + return ( + + {formatDuration(elapsedMs, { mode: "live" })} + + ); +}); + interface AssistantMessageProps { message: string; timestamp: number; diff --git a/packages/app/src/timeline/session-stream-reducers.test.ts b/packages/app/src/timeline/session-stream-reducers.test.ts index 556b41bca..f150b2a1b 100644 --- a/packages/app/src/timeline/session-stream-reducers.test.ts +++ b/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, diff --git a/packages/app/src/timeline/session-stream-reducers.ts b/packages/app/src/timeline/session-stream-reducers.ts index 4b932ebfc..630a11e24 100644 --- a/packages/app/src/timeline/session-stream-reducers.ts +++ b/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( (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 })); diff --git a/packages/app/src/timeline/turn-time.ts b/packages/app/src/timeline/turn-time.ts new file mode 100644 index 000000000..f27bcd00b --- /dev/null +++ b/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; +} diff --git a/packages/app/src/types/stream.test.ts b/packages/app/src/types/stream.test.ts index 395c743f4..42b5973f9 100644 --- a/packages/app/src/types/stream.test.ts +++ b/packages/app/src/types/stream.test.ts @@ -1,4 +1,5 @@ import assert from "node:assert/strict"; +import invariant from "tiny-invariant"; import { describe, it } from "vitest"; import { @@ -325,14 +326,14 @@ describe("stream reducer canonical tool calls", () => { }, ]); - assert.deepStrictEqual( - state.map((item) => (item.kind === "assistant_message" ? item.text : item.kind)), - ["First answer.", "Second answer."], - ); - assert.deepStrictEqual( - state.map((item) => (item.kind === "assistant_message" ? item.messageId : null)), - ["msg-first", "msg-second"], - ); + const messages = state.filter((item) => item.kind === "assistant_message"); + assert.strictEqual(messages.length, 2); + const first = messages[0]; + const second = messages[1]; + invariant(first?.kind === "assistant_message"); + invariant(second?.kind === "assistant_message"); + assert.deepStrictEqual([first.text, second.text], ["First answer.", "Second answer."]); + assert.deepStrictEqual([first.messageId, second.messageId], ["msg-first", "msg-second"]); }); it("merges adjacent assistant deltas when message ids match", () => { @@ -347,13 +348,13 @@ describe("stream reducer canonical tool calls", () => { }, ]); - assert.strictEqual(state.length, 1); - assert.strictEqual(state[0]?.kind, "assistant_message"); - if (state[0]?.kind === "assistant_message") { - assert.strictEqual(state[0].text, "Hello"); - assert.strictEqual(state[0].id, "msg-same"); - assert.strictEqual(state[0].messageId, "msg-same"); - } + const messages = state.filter((item) => item.kind === "assistant_message"); + assert.strictEqual(messages.length, 1); + const first = messages[0]; + invariant(first?.kind === "assistant_message"); + assert.strictEqual(first.text, "Hello"); + assert.strictEqual(first.id, "msg-same"); + assert.strictEqual(first.messageId, "msg-same"); }); it("preserves old assistant merge behavior when message ids are absent", () => { @@ -368,8 +369,11 @@ describe("stream reducer canonical tool calls", () => { }, ]); - assert.strictEqual(state.length, 1); - assert.strictEqual(state[0]?.kind === "assistant_message" ? state[0].text : null, "Hello"); + const messages = state.filter((item) => item.kind === "assistant_message"); + assert.strictEqual(messages.length, 1); + const first = messages[0]; + invariant(first?.kind === "assistant_message"); + assert.strictEqual(first.text, "Hello"); }); it("merges running and completed events by callId", () => { @@ -836,3 +840,113 @@ describe("stream reducer canonical tool calls", () => { ); }); }); + +describe("turn headers", () => { + it("inserts a live header on turn_started and closes it on turn_completed", () => { + const startedAt = new Date("2025-01-01T12:00:00Z"); + const completedAt = new Date("2025-01-01T12:00:05Z"); + + let state = reduceStreamUpdate([], { type: "turn_started", provider: "claude" }, startedAt); + state = reduceStreamUpdate( + state, + { type: "timeline", provider: "claude", item: { type: "assistant_message", text: "ok" } }, + new Date("2025-01-01T12:00:02Z"), + ); + state = reduceStreamUpdate(state, { type: "turn_completed", provider: "claude" }, completedAt); + + const header = state.find((item) => item.kind === "turn_header"); + invariant(header?.kind === "turn_header"); + assert.strictEqual(header.source, "live"); + assert.strictEqual(header.outcome, "completed"); + assert.strictEqual(header.durationMs, 5_000); + assert.strictEqual(header.startedAt.getTime(), startedAt.getTime()); + assert.strictEqual(header.completedAt?.getTime(), completedAt.getTime()); + }); + + it("synthesizes a derived header during hydration when turn events are missing", () => { + const state = hydrateStreamState([ + { + event: { + type: "timeline", + provider: "claude", + item: { type: "user_message", text: "hi" }, + }, + timestamp: new Date("2025-01-01T13:00:00Z"), + }, + { + event: assistantTimeline("Working on it.", "claude", "msg-1"), + timestamp: new Date("2025-01-01T13:00:01Z"), + }, + { + event: assistantTimeline("Done.", "claude", "msg-2"), + timestamp: new Date("2025-01-01T13:00:04Z"), + }, + ]); + + const headers = state.filter((item) => item.kind === "turn_header"); + assert.strictEqual(headers.length, 1); + const [header] = headers; + invariant(header?.kind === "turn_header"); + assert.strictEqual(header.source, "derived"); + assert.strictEqual(header.outcome, "completed"); + assert.strictEqual(header.durationMs, 3_000); + // Header should sit between the user message and the first assistant item. + const kinds = state.map((item) => item.kind); + assert.deepStrictEqual(kinds, [ + "user_message", + "turn_header", + "assistant_message", + "assistant_message", + ]); + }); + + it("does not synthesize a derived header when a live header is already present", () => { + const startedAt = new Date("2025-01-01T14:00:00Z"); + const state = hydrateStreamState([ + { + event: { + type: "timeline", + provider: "claude", + item: { type: "user_message", text: "hi" }, + }, + timestamp: new Date("2025-01-01T13:59:59Z"), + }, + { event: { type: "turn_started", provider: "claude" }, timestamp: startedAt }, + { + event: assistantTimeline("ok", "claude", "msg-1"), + timestamp: new Date("2025-01-01T14:00:02Z"), + }, + ]); + + const headers = state.filter((item) => item.kind === "turn_header"); + assert.strictEqual(headers.length, 1); + const header = headers[0]; + invariant(header?.kind === "turn_header"); + assert.strictEqual(header.source, "live"); + assert.strictEqual(header.completedAt, undefined); + }); + + it("skips synthesis for turns with no assistant output", () => { + const state = hydrateStreamState([ + { + event: { + type: "timeline", + provider: "claude", + item: { type: "user_message", text: "hi" }, + }, + timestamp: new Date("2025-01-01T15:00:00Z"), + }, + { + event: { + type: "timeline", + provider: "claude", + item: { type: "user_message", text: "still there?" }, + }, + timestamp: new Date("2025-01-01T15:01:00Z"), + }, + ]); + + const headers = state.filter((item) => item.kind === "turn_header"); + assert.strictEqual(headers.length, 0); + }); +}); diff --git a/packages/app/src/types/stream.ts b/packages/app/src/types/stream.ts index a6829eeb1..5c98db57e 100644 --- a/packages/app/src/types/stream.ts +++ b/packages/app/src/types/stream.ts @@ -3,6 +3,11 @@ import type { AgentAttachment, AgentStreamEventPayload } from "@server/shared/me import type { AttachmentMetadata } from "@/attachments/types"; import { extractTaskEntriesFromToolCall } from "../utils/tool-call-parsers"; import { splitMarkdownBlocks } from "@/utils/split-markdown-blocks"; +import { + appendLiveTurnHeader, + completeLastTurnHeader, + synthesizeMissingTurnHeaders, +} from "@/timeline/turn-time"; /** * Simple hash function for deterministic ID generation @@ -50,7 +55,21 @@ export type StreamItem = | ToolCallItem | TodoListItem | ActivityLogItem - | CompactionItem; + | CompactionItem + | TurnHeaderItem; + +export type TurnHeaderOutcome = "completed" | "failed" | "canceled"; + +export interface TurnHeaderItem { + kind: "turn_header"; + id: string; + timestamp: Date; + startedAt: Date; + completedAt?: Date; + durationMs?: number; + outcome?: TurnHeaderOutcome; + source: "live" | "derived"; +} export type UserMessageImageAttachment = AttachmentMetadata; @@ -704,10 +723,15 @@ export function reduceStreamUpdate( case "timeline": return reduceTimelineEvent(state, event, timestamp, source); case "thread_started": + return finalizeActiveThoughts(state); case "turn_started": + return appendLiveTurnHeader(finalizeActiveThoughts(state), timestamp); case "turn_completed": + return completeLastTurnHeader(finalizeActiveThoughts(state), timestamp, "completed"); case "turn_failed": + return completeLastTurnHeader(finalizeActiveThoughts(state), timestamp, "failed"); case "turn_canceled": + return completeLastTurnHeader(finalizeActiveThoughts(state), timestamp, "canceled"); case "permission_requested": case "permission_resolved": case "attention_required": @@ -721,14 +745,17 @@ export function reduceStreamUpdate( * Hydrate stream state from a batch of AgentManager stream events */ export function hydrateStreamState( - events: Array<{ event: AgentStreamEventPayload; timestamp: Date }>, + events: Array<{ + event: AgentStreamEventPayload; + timestamp: Date; + }>, options?: { source?: StreamUpdateSource }, ): StreamItem[] { const hydrated = events.reduce((state, { event, timestamp }) => { return reduceStreamUpdate(state, event, timestamp, options); }, []); - return finalizeActiveThoughts(hydrated); + return synthesizeMissingTurnHeaders(finalizeActiveThoughts(hydrated)); } /** @@ -749,6 +776,16 @@ const STREAM_COMPLETION_EVENTS = new Set([ "turn_canceled", ]); +function applyCompletionToTail( + tail: StreamItem[], + event: AgentStreamEventPayload, + timestamp: Date, + source: StreamUpdateSource, +): StreamItem[] { + const finalized = finalizeActiveThoughts(tail); + return reduceStreamUpdate(finalized, event, timestamp, { source }); +} + /** * Determine what kind of StreamItem an event would produce */ @@ -813,6 +850,28 @@ function getActiveAssistantHeadIndex(head: StreamItem[]): number { return -1; } +function getTailAssistantToResume(params: { + incomingKind: StreamItem["kind"] | null; + event: AgentStreamEventPayload; + nextHead: StreamItem[]; + tailAssistant: StreamItem | undefined; +}): AssistantMessageItem | null { + if (params.incomingKind !== "assistant_message" || params.nextHead.length !== 0) { + return null; + } + if (params.tailAssistant?.kind !== "assistant_message") { + return null; + } + const incomingMessageId = + params.event.type === "timeline" && params.event.item.type === "assistant_message" + ? params.event.item.messageId + : undefined; + if (incomingMessageId !== undefined && params.tailAssistant.messageId !== incomingMessageId) { + return null; + } + return params.tailAssistant; +} + function promoteCompletedAssistantBlocks(params: { tail: StreamItem[]; head: StreamItem[] }): { tail: StreamItem[]; head: StreamItem[]; @@ -986,8 +1045,7 @@ export function applyStreamEvent(params: { // Handle turn completion events - flush everything if (STREAM_COMPLETION_EVENTS.has(event.type)) { flushHead(); - // Also finalize any remaining thoughts in tail - const finalized = finalizeActiveThoughts(nextTail); + const finalized = applyCompletionToTail(nextTail, event, timestamp, source); if (finalized !== nextTail) { nextTail = finalized; changedTail = true; @@ -1002,21 +1060,17 @@ export function applyStreamEvent(params: { flushHead(); } - if (incomingKind === "assistant_message" && nextHead.length === 0) { - const tailAssistant = nextTail.at(-1); - const incomingMessageId = - event.type === "timeline" && event.item.type === "assistant_message" - ? event.item.messageId - : undefined; - const shouldContinueTailAssistant = - tailAssistant?.kind === "assistant_message" && - (incomingMessageId === undefined || tailAssistant.messageId === incomingMessageId); - if (shouldContinueTailAssistant) { - nextTail = nextTail.slice(0, -1); - nextHead = [tailAssistant]; - changedTail = true; - changedHead = true; - } + const tailAssistant = getTailAssistantToResume({ + incomingKind, + event, + nextHead, + tailAssistant: nextTail.at(-1), + }); + if (tailAssistant) { + nextTail = nextTail.slice(0, -1); + nextHead = [tailAssistant]; + changedTail = true; + changedHead = true; } // For streamable kinds, apply to head diff --git a/packages/app/src/utils/time.test.ts b/packages/app/src/utils/time.test.ts new file mode 100644 index 000000000..5a97f843d --- /dev/null +++ b/packages/app/src/utils/time.test.ts @@ -0,0 +1,63 @@ +import { describe, it, expect } from "vitest"; +import { formatDuration, formatMessageTimestamp } from "./time"; + +describe("formatDuration", () => { + it("renders static durations as integers", () => { + expect(formatDuration(5_600)).toBe("5s"); + expect(formatDuration(0)).toBe("0s"); + }); + + it("renders 10s-60s as whole seconds", () => { + expect(formatDuration(47_000)).toBe("47s"); + expect(formatDuration(10_400)).toBe("10s"); + }); + + it("renders minutes and remainder seconds", () => { + expect(formatDuration(132_000)).toBe("2m 12s"); + expect(formatDuration(120_000)).toBe("2m"); + }); + + it("renders hours and remainder minutes", () => { + expect(formatDuration(3_900_000)).toBe("1h 5m"); + expect(formatDuration(3_600_000)).toBe("1h"); + }); + + it("guards against negative and NaN", () => { + expect(formatDuration(-1)).toBe("0s"); + expect(formatDuration(Number.NaN)).toBe("0s"); + }); + + it("keeps one decimal for live durations", () => { + expect(formatDuration(5_600, { mode: "live" })).toBe("5.6s"); + expect(formatDuration(12_340, { mode: "live" })).toBe("12.3s"); + expect(formatDuration(75_230, { mode: "live" })).toBe("1m 15.2s"); + expect(formatDuration(0, { mode: "live" })).toBe("0.0s"); + }); +}); + +describe("formatMessageTimestamp", () => { + it("shows only time for same-day timestamps", () => { + const now = new Date(2026, 4, 14, 17, 30); + const date = new Date(2026, 4, 14, 12, 23); + const formatted = formatMessageTimestamp(date, now); + expect(formatted).toMatch(/12:23/); + expect(formatted).not.toMatch(/Thursday|Wednesday/); + }); + + it("includes weekday for timestamps within the last 6 days", () => { + // 2026-05-14 is a Thursday. 2026-05-11 is a Monday. + const now = new Date(2026, 4, 14, 17, 30); + const date = new Date(2026, 4, 11, 22, 12); + const formatted = formatMessageTimestamp(date, now); + expect(formatted).toMatch(/Monday/); + expect(formatted).toMatch(/10:12 PM|22:12/); + }); + + it("includes full date for older timestamps", () => { + const now = new Date(2026, 4, 14, 17, 30); + const date = new Date(2026, 3, 1, 9, 5); + const formatted = formatMessageTimestamp(date, now); + expect(formatted).toMatch(/Apr|April/); + expect(formatted).toMatch(/2026/); + }); +}); diff --git a/packages/app/src/utils/time.ts b/packages/app/src/utils/time.ts index 4b26e6f2f..3fbf16076 100644 --- a/packages/app/src/utils/time.ts +++ b/packages/app/src/utils/time.ts @@ -35,3 +35,87 @@ export function formatTimeAgo(date: Date): string { const day = date.getDate(); return `${month} ${day}`; } + +function isSameLocalDay(a: Date, b: Date): boolean { + return ( + a.getFullYear() === b.getFullYear() && + a.getMonth() === b.getMonth() && + a.getDate() === b.getDate() + ); +} + +/** + * Format a chat-message timestamp for hover-revealed UI. + * - Same day: "10:11 PM" + * - Within ~6 days: "Wednesday 10:11 PM" + * - Older: "14 May 2026, 10:11 PM" + */ +export function formatMessageTimestamp(date: Date, now: Date = new Date()): string { + const time = date.toLocaleTimeString(undefined, { + hour: "numeric", + minute: "2-digit", + }); + + if (isSameLocalDay(date, now)) { + return time; + } + + const diffMs = now.getTime() - date.getTime(); + const diffDays = Math.floor(diffMs / (1000 * 60 * 60 * 24)); + if (diffDays >= 0 && diffDays < 7) { + const weekday = date.toLocaleDateString(undefined, { weekday: "long" }); + return `${weekday} ${time}`; + } + + const dateLabel = date.toLocaleDateString(undefined, { + day: "numeric", + month: "short", + year: "numeric", + }); + return `${dateLabel}, ${time}`; +} + +type DurationFormatMode = "static" | "live"; + +/** + * Format a duration as a compact human-readable string. + * - Static: integer units only + * - Live: always keep one decimal in the smallest displayed unit + */ +export function formatDuration( + durationMs: number, + options?: { mode?: DurationFormatMode }, +): string { + const mode = options?.mode ?? "static"; + if (!Number.isFinite(durationMs) || durationMs < 0) { + return mode === "live" ? "0.0s" : "0s"; + } + const totalSeconds = durationMs / 1000; + + if (mode === "live") { + if (totalSeconds < 60) { + return `${totalSeconds.toFixed(1)}s`; + } + if (totalSeconds < 60 * 60) { + const minutes = Math.floor(totalSeconds / 60); + const seconds = totalSeconds - minutes * 60; + return `${minutes}m ${seconds.toFixed(1)}s`; + } + const hours = Math.floor(totalSeconds / (60 * 60)); + const minutes = (totalSeconds - hours * 60 * 60) / 60; + return `${hours}h ${minutes.toFixed(1)}m`; + } + + const wholeSeconds = Math.floor(totalSeconds); + if (wholeSeconds < 60) { + return `${wholeSeconds}s`; + } + const minutes = Math.floor(wholeSeconds / 60); + if (minutes < 60) { + const seconds = wholeSeconds % 60; + return seconds === 0 ? `${minutes}m` : `${minutes}m ${seconds}s`; + } + const hours = Math.floor(minutes / 60); + const remMinutes = minutes % 60; + return remMinutes === 0 ? `${hours}h` : `${hours}h ${remMinutes}m`; +} diff --git a/packages/server/src/server/agent/agent-manager.test.ts b/packages/server/src/server/agent/agent-manager.test.ts index b2425f584..412615fc0 100644 --- a/packages/server/src/server/agent/agent-manager.test.ts +++ b/packages/server/src/server/agent/agent-manager.test.ts @@ -5004,6 +5004,69 @@ test("hydrateTimeline keeps provider user_message items when no canonical user h expect(assistantMessages).toHaveLength(2); }); +test("hydrateTimeline preserves provider replay timestamps and marks missing ones untrusted", async () => { + const workdir = mkdtempSync(join(tmpdir(), "agent-manager-history-timestamps-")); + const storagePath = join(workdir, "agents"); + const storage = new AgentStorage(storagePath, logger); + + class TimestampedHistorySession extends TestAgentSession { + async *streamHistory(): AsyncGenerator { + yield { + type: "timeline", + provider: this.provider, + timestamp: "2026-05-01T10:00:00.000Z", + item: { type: "user_message", text: "hello", messageId: "msg_history_1" }, + }; + yield { + type: "timeline", + provider: this.provider, + item: { type: "assistant_message", text: "no original timestamp" }, + }; + } + } + + class TimestampedHistoryClient implements AgentClient { + readonly provider = "codex" as const; + readonly capabilities = TEST_CAPABILITIES; + + async isAvailable(): Promise { + return true; + } + + async createSession(config: AgentSessionConfig): Promise { + return new TimestampedHistorySession(config); + } + + async resumeSession(): Promise { + throw new Error("Not used in this test"); + } + } + + const manager = new AgentManager({ + clients: { + codex: new TimestampedHistoryClient(), + }, + registry: storage, + logger, + idFactory: () => "00000000-0000-4000-8000-000000000204", + }); + + const snapshot = await manager.createAgent({ + provider: "codex", + cwd: workdir, + }); + + await manager.hydrateTimelineFromProvider(snapshot.id); + const timeline = manager.fetchTimeline(snapshot.id, { direction: "tail", limit: 0 }).rows; + + expect(timeline).toHaveLength(2); + expect(timeline[0]).toMatchObject({ + timestamp: "2026-05-01T10:00:00.000Z", + item: { type: "user_message", text: "hello", messageId: "msg_history_1" }, + }); + expect(timeline[1]?.timestamp).toEqual(expect.any(String)); +}); + test("hydrateTimeline suppresses only matching canonical user_message messageId", async () => { const workdir = mkdtempSync(join(tmpdir(), "agent-manager-history-partial-dedup-")); const storagePath = join(workdir, "agents"); diff --git a/packages/server/src/server/agent/agent-manager.ts b/packages/server/src/server/agent/agent-manager.ts index 4b9c120cb..a8bf45bff 100644 --- a/packages/server/src/server/agent/agent-manager.ts +++ b/packages/server/src/server/agent/agent-manager.ts @@ -122,6 +122,7 @@ export type AgentManagerEvent = event: AgentStreamEvent; seq?: number; epoch?: string; + timestamp?: string; }; export type AgentSubscriber = (event: AgentManagerEvent) => void; @@ -1391,10 +1392,11 @@ export class AgentManager { this.dispatchStream(agent.id, event, { seq: row.seq, epoch: this.timelineStore.getEpoch(agent.id), + timestamp: row.timestamp, }); return; } - this.dispatchStream(agent.id, event); + this.dispatchStream(agent.id, event, { timestamp: new Date().toISOString() }); }; void (async () => { try { @@ -1436,6 +1438,7 @@ export class AgentManager { { seq: row.seq, epoch: this.timelineStore.getEpoch(agentId), + timestamp: row.timestamp, }, ); if (options?.emitState !== false) { @@ -1457,6 +1460,7 @@ export class AgentManager { { seq: row.seq, epoch: this.timelineStore.getEpoch(agentId), + timestamp: row.timestamp, }, ); await this.persistSnapshot(agent); @@ -1786,7 +1790,7 @@ export class AgentManager { const bufferedResolution = agent.bufferedPermissionResolutions.get(requestId); if (bufferedResolution) { agent.bufferedPermissionResolutions.delete(requestId); - this.dispatchStream(agent.id, bufferedResolution); + this.dispatchStream(agent.id, bufferedResolution, { timestamp: new Date().toISOString() }); } return result; @@ -1876,12 +1880,16 @@ export class AgentManager { // Clear any pending permissions that weren't cleaned up by handleStreamEvent. if (agent.pendingPermissions.size > 0) { for (const [requestId] of agent.pendingPermissions) { - this.dispatchStream(agent.id, { - type: "permission_resolved", - provider: agent.provider, - requestId, - resolution: { behavior: "deny", message: "Interrupted" }, - }); + this.dispatchStream( + agent.id, + { + type: "permission_resolved", + provider: agent.provider, + requestId, + resolution: { behavior: "deny", message: "Interrupted" }, + }, + { timestamp: new Date().toISOString() }, + ); } agent.pendingPermissions.clear(); this.touchUpdatedAt(agent); @@ -2587,7 +2595,11 @@ export class AgentManager { if (isDuplicateLegacyUserMessage(event.item, canonicalUserMessagesById)) { continue; } - this.recordTimeline(agent.id, event.item); + this.recordTimeline( + agent.id, + event.item, + event.timestamp ? { timestamp: event.timestamp } : undefined, + ); } } catch { // ignore history failures @@ -2663,7 +2675,7 @@ export class AgentManager { } if (!options?.fromHistory && flags.shouldDispatchEvent) { - this.dispatchStream(agent.id, event); + this.dispatchStream(agent.id, event, { timestamp: new Date().toISOString() }); } this.traceHandleStreamEventEnd(agent, event, eventTurnId, flags); @@ -2862,7 +2874,11 @@ export class AgentManager { } if (options?.fromHistory) { - this.recordTimeline(agent.id, event.item); + this.recordTimeline( + agent.id, + event.item, + event.timestamp ? { timestamp: event.timestamp } : undefined, + ); flags.shouldDispatchEvent = false; flags.shouldNotifyWaiters = false; return; @@ -3068,6 +3084,7 @@ export class AgentManager { this.dispatchStream(agentId, event, { seq: row.seq, epoch: this.timelineStore.getEpoch(agentId), + timestamp: row.timestamp, }); return event; } @@ -3108,6 +3125,7 @@ export class AgentManager { { seq: row.seq, epoch: this.timelineStore.getEpoch(agent.id), + timestamp: row.timestamp, }, ); } @@ -3128,8 +3146,12 @@ export class AgentManager { return parts.join("\n\n"); } - private recordTimeline(agentId: string, item: AgentTimelineItem): AgentTimelineRow { - const row = this.timelineStore.append(agentId, item); + private recordTimeline( + agentId: string, + item: AgentTimelineItem, + options?: { timestamp?: string }, + ): AgentTimelineRow { + const row = this.timelineStore.append(agentId, item, options); this.enqueueDurableTimelineAppend(agentId, row); return row; } @@ -3282,7 +3304,7 @@ export class AgentManager { private dispatchStream( agentId: string, event: AgentStreamEvent, - metadata?: { seq?: number; epoch?: string }, + metadata?: { seq?: number; epoch?: string; timestamp?: string }, ): void { const agent = this.agents.get(agentId); this.logger.trace( diff --git a/packages/server/src/server/agent/agent-sdk-types.ts b/packages/server/src/server/agent/agent-sdk-types.ts index e6ff7339a..d66eacae6 100644 --- a/packages/server/src/server/agent/agent-sdk-types.ts +++ b/packages/server/src/server/agent/agent-sdk-types.ts @@ -346,7 +346,13 @@ export type AgentStreamEvent = turnId?: string; } | { type: "turn_canceled"; provider: AgentProvider; reason: string; turnId?: string } - | { type: "timeline"; item: AgentTimelineItem; provider: AgentProvider; turnId?: string } + | { + type: "timeline"; + item: AgentTimelineItem; + provider: AgentProvider; + turnId?: string; + timestamp?: string; + } | { type: "permission_requested"; provider: AgentProvider; diff --git a/packages/server/src/server/agent/provider-history-timestamps.test.ts b/packages/server/src/server/agent/provider-history-timestamps.test.ts new file mode 100644 index 000000000..9564b1281 --- /dev/null +++ b/packages/server/src/server/agent/provider-history-timestamps.test.ts @@ -0,0 +1,23 @@ +import { describe, expect, test } from "vitest"; + +import { normalizeProviderReplayTimestamp } from "./provider-history-timestamps.js"; + +describe("normalizeProviderReplayTimestamp", () => { + test("preserves valid string timestamps after trimming", () => { + expect(normalizeProviderReplayTimestamp(" 2026-05-01T10:00:00.000Z ")).toBe( + "2026-05-01T10:00:00.000Z", + ); + }); + + test("converts numeric second and millisecond timestamps to ISO", () => { + expect(normalizeProviderReplayTimestamp(1_778_762_475)).toBe("2026-05-14T12:41:15.000Z"); + expect(normalizeProviderReplayTimestamp(1_778_762_475_873)).toBe("2026-05-14T12:41:15.873Z"); + }); + + test("returns null for missing or invalid timestamps", () => { + expect(normalizeProviderReplayTimestamp(undefined)).toBeNull(); + expect(normalizeProviderReplayTimestamp("not a timestamp")).toBeNull(); + expect(normalizeProviderReplayTimestamp(Number.NaN)).toBeNull(); + expect(normalizeProviderReplayTimestamp(Number.POSITIVE_INFINITY)).toBeNull(); + }); +}); diff --git a/packages/server/src/server/agent/provider-history-timestamps.ts b/packages/server/src/server/agent/provider-history-timestamps.ts new file mode 100644 index 000000000..8eb2306ea --- /dev/null +++ b/packages/server/src/server/agent/provider-history-timestamps.ts @@ -0,0 +1,17 @@ +export function normalizeProviderReplayTimestamp(value: unknown): string | null { + if (typeof value === "string") { + const timestamp = value.trim(); + if (!timestamp || Number.isNaN(Date.parse(timestamp))) { + return null; + } + return timestamp; + } + + if (typeof value !== "number" || !Number.isFinite(value)) { + return null; + } + + const timestampMs = value > 1_000_000_000_000 ? value : value * 1000; + const date = new Date(timestampMs); + return Number.isNaN(date.getTime()) ? null : date.toISOString(); +} diff --git a/packages/server/src/server/agent/providers/claude/agent.ts b/packages/server/src/server/agent/providers/claude/agent.ts index 1475721ad..0dda6fb3d 100644 --- a/packages/server/src/server/agent/providers/claude/agent.ts +++ b/packages/server/src/server/agent/providers/claude/agent.ts @@ -41,6 +41,7 @@ import { import { appendOrReplaceGrowingAssistantMessage, runProviderTurn } from "../provider-runner.js"; import { renderPromptAttachmentAsText } from "../../prompt-attachments.js"; import { claudeQuery, type ClaudeOptions, type ClaudeQueryFactory } from "./query.js"; +import { normalizeProviderReplayTimestamp } from "../../provider-history-timestamps.js"; import { getAgentStreamEventTurnId, @@ -185,6 +186,11 @@ interface AsyncMessageInput { iterable: AsyncIterable; } +interface PersistedTimelineEntry { + item: AgentTimelineItem; + timestamp?: string; +} + const CLAUDE_CAPABILITIES: AgentCapabilityFlags = { supportsStreaming: true, supportsSessionPersistence: true, @@ -1548,7 +1554,7 @@ class ClaudeAgentSession implements AgentSession { private readonly sidechainTracker = new ClaudeSidechainTracker({ getToolInput: (toolUseId) => this.toolUseCache.get(toolUseId)?.input ?? null, }); - private persistedHistory: AgentTimelineItem[] = []; + private persistedHistory: PersistedTimelineEntry[] = []; private historyPending = false; private turnState: TurnState = "idle"; private nextTurnOrdinal = 1; @@ -1756,8 +1762,13 @@ class ClaudeAgentSession implements AgentSession { const history = this.persistedHistory; this.persistedHistory = []; this.historyPending = false; - for (const item of history) { - yield { type: "timeline", item, provider: "claude" }; + for (const entry of history) { + yield { + type: "timeline", + item: entry.item, + provider: "claude", + timestamp: entry.timestamp, + }; } } @@ -2142,9 +2153,9 @@ class ClaudeAgentSession implements AgentSession { pushUnique(historyIds[idx]); } for (let idx = this.persistedHistory.length - 1; idx >= 0; idx -= 1) { - const item = this.persistedHistory[idx]; - if (item?.type === "user_message") { - pushUnique(item.messageId); + const entry = this.persistedHistory[idx]; + if (entry?.item.type === "user_message") { + pushUnique(entry.item.messageId); } } for (let idx = this.userMessageIds.length - 1; idx >= 0; idx -= 1) { @@ -3624,7 +3635,7 @@ class ClaudeAgentSession implements AgentSession { return; } - const timeline: AgentTimelineItem[] = []; + const timeline: PersistedTimelineEntry[] = []; for (const line of content.split(/\r?\n/)) { this.ingestPersistedHistoryLine(line, timeline); } @@ -3635,7 +3646,7 @@ class ClaudeAgentSession implements AgentSession { } } - private ingestPersistedHistoryLine(line: string, timeline: AgentTimelineItem[]): void { + private ingestPersistedHistoryLine(line: string, timeline: PersistedTimelineEntry[]): void { const trimmed = line.trim(); if (!trimmed) { return; @@ -3660,9 +3671,15 @@ class ClaudeAgentSession implements AgentSession { this.rememberUserMessageId(entry.uuid); } + const historyTimestamp = normalizeProviderReplayTimestamp(entry.timestamp); const items = this.convertHistoryEntry(entry); if (items.length > 0) { - timeline.push(...items); + timeline.push( + ...items.map((item) => ({ + item, + timestamp: historyTimestamp ?? undefined, + })), + ); } } diff --git a/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts b/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts index 834d93f5d..15a554767 100644 --- a/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts +++ b/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts @@ -1370,10 +1370,12 @@ describe("Codex app-server provider", () => { type: "agentMessage", id: "message-history", text: "History loaded.", + timestamp: "2026-05-01T10:00:00.000Z", }, { type: "contextCompaction", id: "compact-history", + createdAt: "2026-05-01T10:00:01.000Z", }, ], }, @@ -1397,6 +1399,7 @@ describe("Codex app-server provider", () => { { type: "timeline", provider: "codex", + timestamp: "2026-05-01T10:00:00.000Z", item: { type: "assistant_message", text: "History loaded.", @@ -1406,6 +1409,7 @@ describe("Codex app-server provider", () => { { type: "timeline", provider: "codex", + timestamp: "2026-05-01T10:00:01.000Z", item: { type: "compaction", status: "completed", diff --git a/packages/server/src/server/agent/providers/codex-app-server-agent.ts b/packages/server/src/server/agent/providers/codex-app-server-agent.ts index ecb4c7e0a..3afc2be0a 100644 --- a/packages/server/src/server/agent/providers/codex-app-server-agent.ts +++ b/packages/server/src/server/agent/providers/codex-app-server-agent.ts @@ -64,6 +64,7 @@ import { renderProviderImageOutputAsAssistantMarkdown, type ProviderImageOutput, } from "./provider-image-output.js"; +import { normalizeProviderReplayTimestamp } from "../provider-history-timestamps.js"; import { formatDiagnosticStatus, formatProviderDiagnostic, @@ -358,6 +359,11 @@ interface CodexConfiguredDefaults { thinkingOptionId?: string; } +interface PersistedTimelineEntry { + item: AgentTimelineItem; + timestamp?: string; +} + function mergeCodexConfiguredDefaults( primary: CodexConfiguredDefaults, fallback: CodexConfiguredDefaults, @@ -1444,6 +1450,18 @@ function firstStringField( return null; } +function readCodexHistoryTimestamp(item: unknown): string | null { + const record = toObjectRecord(item); + if (!record) { + return null; + } + return ( + normalizeProviderReplayTimestamp(record.timestamp) ?? + normalizeProviderReplayTimestamp(record.createdAt) ?? + normalizeProviderReplayTimestamp(record.created_at) + ); +} + function codexImageOutputFromResult(result: unknown): ProviderImageOutput | null { if (typeof result === "string") { const trimmed = result.trim(); @@ -1591,14 +1609,18 @@ async function loadCodexThreadHistoryTimeline(params: { threadId: string; cwd: string | null; requestThread: CodexThreadReadRequest; -}): Promise { +}): Promise { const response = await requestCodexThreadHistory(params.requestThread, params.threadId); - const timeline: AgentTimelineItem[] = []; + const timeline: PersistedTimelineEntry[] = []; for (const turn of response.thread.turns) { for (const item of turn.items) { const timelineItem = threadItemToTimeline(item, { cwd: params.cwd }); if (timelineItem) { - timeline.push(timelineItem); + const timestamp = readCodexHistoryTimestamp(item); + timeline.push({ + item: timelineItem, + timestamp: timestamp ?? undefined, + }); } } } @@ -2678,7 +2700,7 @@ class CodexAppServerAgentSession implements AgentSession { private serviceTier: "fast" | null = null; private planModeEnabled = false; private historyPending = false; - private persistedHistory: AgentTimelineItem[] = []; + private persistedHistory: PersistedTimelineEntry[] = []; private pendingPermissions = new Map(); private pendingPermissionHandlers = new Map< string, @@ -3334,8 +3356,13 @@ class CodexAppServerAgentSession implements AgentSession { const history = this.persistedHistory; this.persistedHistory = []; this.historyPending = false; - for (const item of history) { - yield { type: "timeline", provider: CODEX_PROVIDER, item }; + for (const entry of history) { + yield { + type: "timeline", + provider: CODEX_PROVIDER, + item: entry.item, + timestamp: entry.timestamp, + }; } } @@ -5151,7 +5178,7 @@ export class CodexAppServerAgentClient implements AgentClient { const threadId = typeof thread.id === "string" ? thread.id : ""; const cwd = typeof thread.cwd === "string" ? thread.cwd : process.cwd(); const title = typeof thread.preview === "string" ? thread.preview : null; - let timeline: AgentTimelineItem[] = []; + let timeline: PersistedTimelineEntry[] = []; try { timeline = await loadCodexThreadHistoryTimeline({ @@ -5186,7 +5213,7 @@ export class CodexAppServerAgentClient implements AgentClient { threadId, }, }, - timeline, + timeline: timeline.map((entry) => entry.item), }; }), ); diff --git a/packages/server/src/server/agent/providers/opencode-agent.test.ts b/packages/server/src/server/agent/providers/opencode-agent.test.ts index 6708274fc..edfd2b355 100644 --- a/packages/server/src/server/agent/providers/opencode-agent.test.ts +++ b/packages/server/src/server/agent/providers/opencode-agent.test.ts @@ -833,6 +833,144 @@ describe("OpenCode adapter startTurn error handling", () => { expect(fakeClient.session.delete).not.toHaveBeenCalled(); }); + test("streamHistory preserves OpenCode replay timestamps from message and part times", async () => { + const fakeClient = { + session: { + messages: vi.fn().mockResolvedValue({ + data: [ + { + info: { + id: "msg_user", + sessionID: "ses_unit_test", + role: "user", + time: { created: 1778762475873 }, + }, + parts: [ + { + id: "prt_user", + sessionID: "ses_unit_test", + messageID: "msg_user", + type: "text", + text: "Reply with exactly: probe ok", + }, + ], + }, + { + info: { + id: "msg_assistant", + sessionID: "ses_unit_test", + role: "assistant", + time: { created: 1778762475884, completed: 1778762489358 }, + }, + parts: [ + { + id: "prt_reasoning", + sessionID: "ses_unit_test", + messageID: "msg_assistant", + type: "reasoning", + text: "thinking", + time: { start: 1778762482953, end: 1778762483610 }, + }, + { + id: "prt_text", + sessionID: "ses_unit_test", + messageID: "msg_assistant", + type: "text", + text: "probe ok", + time: { start: 1778762483612, end: 1778762489351 }, + }, + ], + }, + ], + error: undefined, + }), + }, + } as never; + + const session = new __openCodeInternals.OpenCodeAgentSession( + { provider: "opencode", cwd: "/tmp/test" }, + fakeClient, + "ses_unit_test", + createTestLogger(), + "/tmp/opencode-storage", + ); + + const history: AgentStreamEvent[] = []; + for await (const event of session.streamHistory()) { + history.push(event); + } + + expect(history).toEqual([ + { + type: "timeline", + provider: "opencode", + timestamp: "2026-05-14T12:41:15.873Z", + item: { type: "user_message", text: "Reply with exactly: probe ok" }, + }, + { + type: "timeline", + provider: "opencode", + timestamp: "2026-05-14T12:41:22.953Z", + item: { type: "reasoning", text: "thinking" }, + }, + { + type: "timeline", + provider: "opencode", + timestamp: "2026-05-14T12:41:23.612Z", + item: { type: "assistant_message", text: "probe ok" }, + }, + ]); + }); + + test("streamHistory omits replay timestamps when OpenCode omits times", async () => { + const fakeClient = { + session: { + messages: vi.fn().mockResolvedValue({ + data: [ + { + info: { + id: "msg_assistant", + sessionID: "ses_unit_test", + role: "assistant", + }, + parts: [ + { + id: "prt_text", + sessionID: "ses_unit_test", + messageID: "msg_assistant", + type: "text", + text: "no clocks here", + }, + ], + }, + ], + error: undefined, + }), + }, + } as never; + + const session = new __openCodeInternals.OpenCodeAgentSession( + { provider: "opencode", cwd: "/tmp/test" }, + fakeClient, + "ses_unit_test", + createTestLogger(), + "/tmp/opencode-storage", + ); + + const history: AgentStreamEvent[] = []; + for await (const event of session.streamHistory()) { + history.push(event); + } + + expect(history).toEqual([ + { + type: "timeline", + provider: "opencode", + item: { type: "assistant_message", text: "no clocks here" }, + }, + ]); + }); + test("emits turn_failed when client.session.promptAsync throws synchronously", async () => { // Yield the server-connected event, then park forever. The adapter waits // for that first event before sending the prompt. diff --git a/packages/server/src/server/agent/providers/opencode-agent.ts b/packages/server/src/server/agent/providers/opencode-agent.ts index e5d1cc488..890c1d725 100644 --- a/packages/server/src/server/agent/providers/opencode-agent.ts +++ b/packages/server/src/server/agent/providers/opencode-agent.ts @@ -62,6 +62,7 @@ import { type OpenCodeRuntime, type OpenCodeServerAcquisition, } from "./opencode/runtime.js"; +import { normalizeProviderReplayTimestamp } from "../provider-history-timestamps.js"; const OPENCODE_CAPABILITIES: AgentCapabilityFlags = { supportsStreaming: true, @@ -899,6 +900,71 @@ function getOpenCodePartTimestamp(part: OpenCodeStoredPart): number { return part.time?.start ?? part.time?.end ?? 0; } +function resolveOpenCodeReplayTimestamp(params: { + message: { time?: { created?: number; completed?: number } | undefined }; + part?: unknown; +}): string | null { + const timedPart = params.part as + | { time?: { start?: number; end?: number } | undefined } + | undefined; + const partTimestamp = + timedPart?.time?.start ?? + timedPart?.time?.end ?? + params.message.time?.created ?? + params.message.time?.completed; + return normalizeProviderReplayTimestamp(partTimestamp); +} + +function buildOpenCodeReplayTimelineEvent(params: { + item: AgentTimelineItem; + message: { time?: { created?: number; completed?: number } | undefined }; + part?: unknown; +}): Extract { + const timestamp = resolveOpenCodeReplayTimestamp({ + message: params.message, + part: params.part, + }); + return { + type: "timeline", + provider: "opencode", + item: params.item, + ...(timestamp ? { timestamp } : {}), + }; +} + +function buildOpenCodeReplayPartTimelineEvent(params: { + part: OpenCodePart; + message: { structured?: unknown; time?: { created?: number; completed?: number } | undefined }; +}): Extract | null { + const { part, message } = params; + if (part.type === "text" && part.text) { + return buildOpenCodeReplayTimelineEvent({ + item: { type: "assistant_message", text: part.text }, + message, + part, + }); + } + if (part.type === "reasoning" && part.text) { + return buildOpenCodeReplayTimelineEvent({ + item: { type: "reasoning", text: part.text }, + message, + part, + }); + } + if (part.type !== "tool") { + return null; + } + const parsedToolPart = OpencodeToolPartToTimelineItemSchema.safeParse(part); + if (!parsedToolPart.success || !parsedToolPart.data) { + return null; + } + return buildOpenCodeReplayTimelineEvent({ + item: parsedToolPart.data, + message, + part, + }); +} + export const __openCodeInternals = { buildOpenCodePromptParts, buildOpenCodeModelContextWindowLookup, @@ -2842,53 +2908,30 @@ class OpenCodeAgentSession implements AgentSession { .join(""); if (text) { - yield { - type: "timeline", - provider: "opencode", + yield buildOpenCodeReplayTimelineEvent({ item: { type: "user_message", text }, - }; + message: info, + }); } } else { let emittedAssistantText = false; for (const part of parts) { if (part.type === "text" && part.text) { emittedAssistantText = true; - yield { - type: "timeline", - provider: "opencode", - item: { type: "assistant_message", text: part.text }, - }; - continue; } - if (part.type === "reasoning" && part.text) { - yield { - type: "timeline", - provider: "opencode", - item: { type: "reasoning", text: part.text }, - }; - continue; - } - if (part.type !== "tool") { - continue; - } - const parsedToolPart = OpencodeToolPartToTimelineItemSchema.safeParse(part); - if (parsedToolPart.success && parsedToolPart.data) { - yield { - type: "timeline", - provider: "opencode", - item: parsedToolPart.data, - }; + const event = buildOpenCodeReplayPartTimelineEvent({ part, message: info }); + if (event) { + yield event; } } if (!emittedAssistantText) { const text = stringifyStructuredAssistantMessage(info.structured); if (text) { - yield { - type: "timeline", - provider: "opencode", + yield buildOpenCodeReplayTimelineEvent({ item: { type: "assistant_message", text }, - }; + message: info, + }); } } } diff --git a/packages/server/src/server/agent/timeline-projection.ts b/packages/server/src/server/agent/timeline-projection.ts index 62477ac5b..c86198407 100644 --- a/packages/server/src/server/agent/timeline-projection.ts +++ b/packages/server/src/server/agent/timeline-projection.ts @@ -142,6 +142,7 @@ function collapseToolLifecycle(entries: readonly WorkingEntry[]): WorkingEntry[] output[existingIndex] = { ...existing, item: mergedItem, + timestamp: entry.timestamp, seqEnd: Math.max(existing.seqEnd, entry.seqEnd), sourceSeqRanges: mergedRanges, collapsed, @@ -181,6 +182,7 @@ function mergeReasoningChunks(entries: readonly WorkingEntry[]): WorkingEntry[] type: "reasoning", text: `${previousReasoning.text}${entryReasoning.text}`, }, + timestamp: entry.timestamp, seqEnd: entry.seqEnd, sourceSeqRanges: mergeSeqRanges(previous.sourceSeqRanges, entry.sourceSeqRanges), collapsed: Array.from(collapsedKinds), @@ -231,6 +233,7 @@ function mergeAssistantChunks(entries: readonly WorkingEntry[]): WorkingEntry[] text: `${previousAssistant.text}${entryAssistant.text}`, ...(previousAssistant.messageId ? { messageId: previousAssistant.messageId } : {}), }, + timestamp: entry.timestamp, seqEnd: entry.seqEnd, sourceSeqRanges: mergeSeqRanges(previous.sourceSeqRanges, entry.sourceSeqRanges), collapsed: Array.from(collapsedKinds), diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index a66480525..e7c7cf65a 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -96,6 +96,7 @@ import type { import { AgentManager } from "./agent/agent-manager.js"; import { ProviderSnapshotManager, resolveSnapshotCwd } from "./agent/provider-snapshot-manager.js"; import type { + AgentManagerEvent, AgentTimelineCursor, AgentTimelineFetchDirection, ManagedAgent, @@ -1277,20 +1278,8 @@ export class Session { // Reduce bandwidth/CPU on mobile: only forward high-frequency agent stream events // for the focused agent, with a short grace window while backgrounded. // History catch-up is handled via pull-based `fetch_agent_timeline_request`. - const activity = this.clientActivity; - if (activity?.deviceType === "mobile") { - if (!activity.focusedAgentId) { - return; - } - if (activity.focusedAgentId !== event.agentId) { - return; - } - if (!activity.appVisible) { - const hiddenForMs = Date.now() - activity.appVisibilityChangedAt.getTime(); - if (hiddenForMs >= this.MOBILE_BACKGROUND_STREAM_GRACE_MS) { - return; - } - } + if (this.shouldSkipAgentStreamForward(event.agentId)) { + return; } const serializedEvent = serializeAgentStreamEvent(event.event); @@ -1309,17 +1298,9 @@ export class Session { "agent.session.forward_stream", ); - const payload = { - agentId: event.agentId, - event: serializedEvent, - timestamp: new Date().toISOString(), - ...(typeof event.seq === "number" ? { seq: event.seq } : {}), - ...(typeof event.epoch === "string" ? { epoch: event.epoch } : {}), - } as const; - this.emit({ type: "agent_stream", - payload, + payload: this.buildAgentStreamPayload(event, serializedEvent), }); if (event.event.type === "permission_requested") { @@ -1347,6 +1328,34 @@ export class Session { ); } + private shouldSkipAgentStreamForward(agentId: string): boolean { + const activity = this.clientActivity; + if (activity?.deviceType !== "mobile") { + return false; + } + if (!activity.focusedAgentId || activity.focusedAgentId !== agentId) { + return true; + } + if (activity.appVisible) { + return false; + } + const hiddenForMs = Date.now() - activity.appVisibilityChangedAt.getTime(); + return hiddenForMs >= this.MOBILE_BACKGROUND_STREAM_GRACE_MS; + } + + private buildAgentStreamPayload( + event: Extract, + serializedEvent: Extract["payload"]["event"], + ): Extract["payload"] { + return { + agentId: event.agentId, + event: serializedEvent, + timestamp: event.timestamp ?? new Date().toISOString(), + ...(typeof event.seq === "number" ? { seq: event.seq } : {}), + ...(typeof event.epoch === "string" ? { epoch: event.epoch } : {}), + }; + } + private async buildAgentPayload(agent: ManagedAgent): Promise { const storedRecord = await this.agentStorage.get(agent.id); const title = storedRecord?.title ?? storedRecord?.config?.title ?? null;