Repository Public
1'use client';2/**3 * Live state of run timelines. The journal (`GET /runs/:id/steps`) is the source of truth and is4 * merged by `seq`; events only say "something changed" (their payloads are used for transient5 * things: streamed deltas, policy decisions, the steering echo). So a dropped socket, a replayed6 * event or an event received twice can never duplicate or lose a timeline item.7 */8import { create } from 'zustand';9import type { AosEvent, WaitDto } from '@aos/shared';10import { api, asList, errorMessage, qs } from '@/lib/api';11import { lastCompletedModelSeq, mergeSteps, ownerMessagesOf, syncCursor, type OwnerMessage, type PolicyDecision, type Step, type StreamState } from '@/lib/timeline';1213const PAGE = 200;14export interface RunTimeline {15 steps: Step[];16 status: 'loading' | 'ready' | 'error';17 error: string | null;18 stream: StreamState | null;19 decisions: PolicyDecision[];20 pending: OwnerMessage[];21 /** Bumped on every merge that changed something (drives auto-scroll). */22 rev: number;23}24interface TimelineState {25 runs: Record<string, RunTimeline>;26 /** Waits resolved during this session (the API lists pending ones only). */27 resolved: Record<string, WaitDto>;28 /** Latest thing each mission did, from live events: Home's activity line. */29 activity: Record<string, { text: string; ts: string }>;30}31const EMPTY: RunTimeline = { steps: [], status: 'loading', error: null, stream: null, decisions: [], pending: [], rev: 0 };32export const useTimeline = create<TimelineState>(() => ({ runs: {}, resolved: {}, activity: {} }));3334const patch = (runId: string, fn: (r: RunTimeline) => Partial<RunTimeline>) =>35 useTimeline.setState((s) => {36 const cur = s.runs[runId] ?? EMPTY;37 return { runs: { ...s.runs, [runId]: { ...cur, ...fn(cur) } } };38 });3940const inflight = new Map<string, Promise<void>>();41const dirty = new Set<string>();42const timers = new Map<string, ReturnType<typeof setTimeout>>();4344async function fetchFrom(runId: string, after: number | null): Promise<Step[]> {45 const out: Step[] = [];46 let cursor = after;47 for (let page = 0; page < 500; page++) {48 const list = asList<Step>(await api.get(`/runs/${runId}/steps${qs({ after: cursor, limit: PAGE })}`));49 out.push(...list);50 if (list.length < PAGE) break;51 cursor = list[list.length - 1]?.seq ?? cursor;52 }53 return out;54}5556function run(runId: string, full: boolean): Promise<void> {57 const existing = inflight.get(runId);58 if (existing) {59 dirty.add(runId); // something changed while a fetch was in flight: go again afterwards60 return existing;61 }62 const p = (async () => {63 try {64 const cur = useTimeline.getState().runs[runId];65 const incoming = await fetchFrom(runId, full || !cur ? null : syncCursor(cur.steps));66 patch(runId, (r) => {67 const before = lastCompletedModelSeq(r.steps);68 const steps = full ? mergeSteps([], incoming) : mergeSteps(r.steps, incoming);69 const modelDone = lastCompletedModelSeq(steps) > before;70 // the committed model step replaces what was streamed for it71 return { steps, status: 'ready', error: null, stream: modelDone ? null : r.stream, rev: r.rev + 1, pending: prunePending(r.pending, steps) };72 });73 } catch (e) {74 patch(runId, (r) => (r.status === 'ready' ? {} : { status: 'error', error: errorMessage(e) }));75 } finally {76 inflight.delete(runId);77 if (dirty.delete(runId)) void run(runId, false);78 }79 })();80 inflight.set(runId, p);81 return p;82}83/** A steering message stops being "queued" once an `owner_messages` step has journaled it: the agent has read it. */84function prunePending(pending: OwnerMessage[], steps: Step[]): OwnerMessage[] {85 if (pending.length === 0) return pending;86 const journaled = new Set<string>();87 for (const s of steps) if (s.kind === 'owner_messages') for (const m of ownerMessagesOf(s)) journaled.add(m.text);88 return pending.filter((m) => !journaled.has(m.text));89}9091export const timeline = {92 /** First load of a run (idempotent). */93 ensure(runId: string): void {94 if (useTimeline.getState().runs[runId]) return;95 patch(runId, () => ({}));96 void run(runId, true);97 },98 reload(runId: string): Promise<void> {99 patch(runId, (r) => (r.status === 'error' ? { status: 'loading', error: null } : {}));100 return run(runId, true);101 },102 /** Incremental sync, coalesced: many step events in a burst cost one request. */103 sync(runId: string): void {104 if (!useTimeline.getState().runs[runId] || timers.has(runId)) return;105 timers.set(106 runId,107 setTimeout(() => {108 timers.delete(runId);109 void run(runId, false);110 }, 40),111 );112 },113 /** Events may have been lost (first connection without a cursor, or a "gap" notice): reload what is on screen. */114 resyncAll(): void {115 for (const id of Object.keys(useTimeline.getState().runs)) void run(id, true);116 },117 /** Optimistic echo of a steering message, added BEFORE the request (the owner.message event then replaces it). */118 addPending(runId: string, text: string): string {119 const key = `local-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`;120 patch(runId, (r) => ({ pending: [...r.pending, { key, text, ts: new Date().toISOString(), pending: true }], rev: r.rev + 1 }));121 return key;122 },123 removePending(runId: string, key: string): void {124 patch(runId, (r) => ({ pending: r.pending.filter((m) => m.key !== key), rev: r.rev + 1 }));125 },126 rememberResolved(w: WaitDto): void {127 useTimeline.setState((s) => ({ resolved: { ...s.resolved, [w.id]: w } }));128 },129 onEvent(ev: AosEvent): void {130 const p = ev.payload;131 if (ev.kind === 'model.delta') {132 if (!ev.runId) return;133 // `{reset:true, turn}` opens every execution of a model step (first run, or again after a crash /134 // retry): whatever was streamed before for this run is void.135 if (p.reset === true) {136 if (useTimeline.getState().runs[ev.runId]) patch(ev.runId, (r) => (r.stream ? { stream: null, rev: r.rev + 1 } : {}));137 return;138 }139 const text = typeof p.text === 'string' ? p.text : '';140 if (!text) return;141 useTimeline.setState((s) => ({ activity: { ...s.activity, [ev.missionId]: { text: p.type === 'reasoning' ? 'Thinking' : 'Writing a reply', ts: ev.ts } } }));142 if (!useTimeline.getState().runs[ev.runId]) return;143 patch(ev.runId, (r) => {144 const cur = r.stream ?? { text: '', reasoning: '' };145 return { stream: p.type === 'reasoning' ? { ...cur, reasoning: cur.reasoning + text } : { ...cur, text: cur.text + text }, rev: r.rev + 1 };146 });147 return;148 }149 if (ev.kind === 'step.started') {150 const name = typeof p.name === 'string' ? p.name : null;151 const kind = typeof p.kind === 'string' ? p.kind : null;152 const text = kind === 'model_call' ? 'Thinking' : kind === 'tool_call' && name ? `Running ${name}` : kind === 'wait' ? 'Waiting for you' : null;153 if (text) useTimeline.setState((s) => ({ activity: { ...s.activity, [ev.missionId]: { text, ts: ev.ts } } }));154 }155 if (!ev.runId) return;156 const runId = ev.runId;157 if (ev.kind.startsWith('step.') || ev.kind === 'wait.created' || ev.kind === 'wait.resolved' || ev.kind === 'run.status') timeline.sync(runId);158 if (ev.kind === 'run.status' && typeof p.status === 'string' && p.status !== 'running') patch(runId, () => ({ stream: null }));159 if (ev.kind === 'policy.decision' && useTimeline.getState().runs[runId]) {160 const d: PolicyDecision = {161 id: ev.id,162 seq: ev.seq,163 ts: ev.ts,164 decision: String(p.decision ?? ''),165 tool: typeof p.tool === 'string' ? p.tool : null,166 reason: typeof p.reason === 'string' ? p.reason : typeof p.source === 'string' ? p.source : null,167 callId: typeof p.callId === 'string' ? p.callId : null,168 source: typeof p.source === 'string' ? p.source : null,169 };170 patch(runId, (r) => (r.decisions.some((x) => x.id === d.id) ? {} : { decisions: [...r.decisions, d], rev: r.rev + 1 }));171 }172 if (ev.kind === 'owner.message' && useTimeline.getState().runs[runId] && typeof p.text === 'string') {173 const text = p.text;174 // replace the optimistic local echo (same text) with the persisted event; never add twice175 patch(runId, (r) => {176 const key = `ev-${ev.id}`;177 if (r.pending.some((m) => m.key === key)) return {};178 const i = r.pending.findIndex((m) => m.key.startsWith('local-') && m.text === text);179 const next = [...r.pending];180 const msg: OwnerMessage = { key, text, ts: ev.ts, pending: true };181 if (i >= 0) next[i] = msg;182 else next.push(msg);183 return { pending: next, rev: r.rev + 1 };184 });185 }186 },187};188