From 22a19c89ed088a7619461d60e5d059ac418b907a Mon Sep 17 00:00:00 2001 From: Test Date: Wed, 23 Sep 2026 13:02:28 -0700 Subject: [PATCH 1/4] feat(observer): draw relayflow runs; carry message metadata on realtime events The observer opened a relayflow run's link onto an arbitrary channel with no notion of the flow. Relayflows now projects each run into its wf- channel with a run snapshot in metadata.relayflow on every message; the dashboard renders that snapshot as the run's step graph above the feed. The realtime message.created event dropped metadata (REST returned it), and the React reducer dropped it again, so the graph froze at the first load. Both now carry it through. Co-Authored-By: Claude Opus 5.5 (1M context) --- CHANGELOG.md | 10 +- packages/engine/CHANGELOG.md | 6 +- .../src/engine/__tests__/wsTransform.test.ts | 11 ++ packages/engine/src/engine/wsTransform.ts | 7 ++ .../src/components/ChatFeed.tsx | 11 +- .../src/components/RunPanel.tsx | 83 ++++++++++++++ .../src/lib/relayflow-run.test.ts | 63 +++++++++++ .../src/lib/relayflow-run.ts | 104 ++++++++++++++++++ packages/react/src/__tests__/reducer.test.ts | 13 +++ packages/react/src/reducer.ts | 1 + 10 files changed, 306 insertions(+), 3 deletions(-) create mode 100644 packages/observer-dashboard/src/components/RunPanel.tsx create mode 100644 packages/observer-dashboard/src/lib/relayflow-run.test.ts create mode 100644 packages/observer-dashboard/src/lib/relayflow-run.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index da546697..1eb8e623 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,7 +16,15 @@ This project follows [Semantic Versioning](https://semver.org/spec/v2.0.0.html). Packages without a separate changelog are covered by the cross-package notes below. -## [Unreleased] +## [Unreleased - Minor] + +### Added + +- The observer dashboard draws a relayflow run: a channel whose messages carry `metadata.relayflow` run snapshots shows the run's step graph, with each step's state and timing, pinned above the feed. + +### Fixed + +- Realtime `message.created` events carry the message's `metadata`, so structured messages render live instead of only after a refetch. ## [8.11.7] - 2026-09-22 diff --git a/packages/engine/CHANGELOG.md b/packages/engine/CHANGELOG.md index a8445bdd..08705aea 100644 --- a/packages/engine/CHANGELOG.md +++ b/packages/engine/CHANGELOG.md @@ -7,7 +7,11 @@ See the [root changelog](../../CHANGELOG.md) for cross-package release highlight The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project follows [Semantic Versioning](https://semver.org/spec/v2.0.0.html). -## [Unreleased] +## [Unreleased - Patch] + +### Fixed + +- The realtime `message.created` event includes the message `metadata` that `GET /v1/channels/:name/messages` already returns. ## [8.11.7] - 2026-09-22 diff --git a/packages/engine/src/engine/__tests__/wsTransform.test.ts b/packages/engine/src/engine/__tests__/wsTransform.test.ts index c323780a..b9809db5 100644 --- a/packages/engine/src/engine/__tests__/wsTransform.test.ts +++ b/packages/engine/src/engine/__tests__/wsTransform.test.ts @@ -23,6 +23,17 @@ function ev(partial: Partial & { type: string; data: Record { + it('carries message metadata, and omits it when absent or not an object', () => { + const base = { id: 'msg_2', channel_name: 'wf-run', agent_id: 'a', from_name: 'flow', text: 't' }; + const metadata = { relayflow: { version: 1, run: { steps: [{ id: 'greet' }] } } }; + const withMeta = transformForClient(ev({ type: 'message.created', data: { ...base, metadata } })); + expect((withMeta.message as Record).metadata).toEqual(metadata); + for (const value of [undefined, null, 'x', [1]]) { + const out = transformForClient(ev({ type: 'message.created', data: { ...base, metadata: value } })); + expect(out.message).not.toHaveProperty('metadata'); + } + }); + it('renames from_name -> agent_name, derives stable id, keeps channel_id, prefers data.created_at', () => { const out = transformForClient( ev({ diff --git a/packages/engine/src/engine/wsTransform.ts b/packages/engine/src/engine/wsTransform.ts index efa7cf4f..f201d4fa 100644 --- a/packages/engine/src/engine/wsTransform.ts +++ b/packages/engine/src/engine/wsTransform.ts @@ -36,6 +36,9 @@ export function transformForClient(event: WsEvent): Record { text: d.text as string, attachments: (d.attachments as unknown[]) ?? [], injection_mode: d.injection_mode as 'wait' | 'steer' | undefined, + // The same document `GET .../messages` returns; a live feed that + // drops it cannot render structured messages until a refetch. + ...(isRecord(d.metadata) ? { metadata: d.metadata } : {}), }, }; @@ -261,3 +264,7 @@ export function transformForClient(event: WsEvent): Record { } } } + +function isRecord(value: unknown): value is Record { + return typeof value === 'object' && value !== null && !Array.isArray(value); +} diff --git a/packages/observer-dashboard/src/components/ChatFeed.tsx b/packages/observer-dashboard/src/components/ChatFeed.tsx index 3ad286b0..5960c880 100644 --- a/packages/observer-dashboard/src/components/ChatFeed.tsx +++ b/packages/observer-dashboard/src/components/ChatFeed.tsx @@ -4,6 +4,8 @@ import { useCallback, useEffect, useLayoutEffect, useRef, useState, type RefObje import { Hash, MessageSquare, UserRound } from 'lucide-react'; import { useMessages, useRelay, sortMessagesChronologically } from '@relaycast/react'; import { MessageCard } from './MessageCard'; +import { RunPanel } from './RunPanel'; +import { latestRelayflowRun } from '../lib/relayflow-run'; import { cn } from '../lib/utils'; import type { DmMessage, MessageWithMeta } from '@relaycast/sdk'; @@ -193,10 +195,17 @@ function ChannelMessages({ channel, scrollRef, onOpenThread, mentionNames, onOpe const sorted = sortMessagesChronologically(messages); const feed = usePaginatedFeed(scrollRef, sorted, fetchMore); + const run = latestRelayflowRun(sorted); + if (loading && sorted.length === 0) return ; if (sorted.length === 0) return ; - return ; + return ( + <> + {run && } + + + ); } function toMessageWithMeta(m: DmMessage): MessageWithMeta { diff --git a/packages/observer-dashboard/src/components/RunPanel.tsx b/packages/observer-dashboard/src/components/RunPanel.tsx new file mode 100644 index 00000000..c3a26996 --- /dev/null +++ b/packages/observer-dashboard/src/components/RunPanel.tsx @@ -0,0 +1,83 @@ +'use client'; + +import { CheckCircle2, ChevronRight, Circle, Loader2, PauseCircle, Workflow, XCircle } from 'lucide-react'; +import { cn } from '../lib/utils'; +import { stepColumns, type RelayflowRun, type RelayflowStep, type RelayflowStepState } from '../lib/relayflow-run'; + +const STATE_STYLE: Record = { + pending: { icon: Circle, tone: 'border-[var(--border-default)] text-[var(--text-faint)]' }, + running: { icon: Loader2, tone: 'border-[var(--brand-primary)] text-[var(--brand-primary-strong)] bg-[var(--brand-primary-faint)]' }, + completed: { icon: CheckCircle2, tone: 'border-[var(--status-success)] text-[var(--status-success)] bg-[var(--status-success-soft)]' }, + failed: { icon: XCircle, tone: 'border-[var(--status-danger)] text-[var(--status-danger)] bg-[var(--status-danger-soft)]' }, + parked: { icon: PauseCircle, tone: 'border-[var(--status-warning)] text-[var(--status-warning)] bg-[var(--status-warning-soft)]' }, +}; + +const RUN_TONE: Record = { + running: 'text-[var(--brand-primary-strong)]', + completed: 'text-[var(--status-success)]', + failed: 'text-[var(--status-danger)]', + canceled: 'text-[var(--status-danger)]', + parked: 'text-[var(--status-warning)]', +}; + +/** The run's step graph, drawn from the newest snapshot its channel carries. */ +export function RunPanel({ run }: { run: RelayflowRun }) { + const done = run.steps.filter(step => step.state === 'completed').length; + const columns = stepColumns(run.steps); + return ( +
+
+ + {run.flow} + + {run.status} + + {run.completionReason && run.completionReason !== 'success' && ( + {run.completionReason} + )} + + {done}/{run.steps.length} steps · {run.runId} + +
+ {columns.length > 0 && ( +
+ {columns.map((column, index) => ( +
+ {index > 0 && } +
+ {column.map(step => )} +
+
+ ))} +
+ )} +
+ ); +} + +function StepChip({ step }: { step: RelayflowStep }) { + const { icon: Icon, tone } = STATE_STYLE[step.state]; + const detail = [ + step.type, + step.elapsedMs === undefined ? undefined : formatElapsed(step.elapsedMs), + step.attempt !== undefined && step.attempt > 1 ? `attempt ${step.attempt}` : undefined, + ].filter(Boolean).join(' · '); + return ( +
+ +
+
{step.id}
+
{detail}
+
+
+ ); +} + +function formatElapsed(ms: number): string { + if (ms < 1_000) return `${ms}ms`; + if (ms < 60_000) return `${(ms / 1_000).toFixed(1)}s`; + return `${Math.floor(ms / 60_000)}m ${Math.round((ms % 60_000) / 1_000)}s`; +} diff --git a/packages/observer-dashboard/src/lib/relayflow-run.test.ts b/packages/observer-dashboard/src/lib/relayflow-run.test.ts new file mode 100644 index 00000000..d5c0b0c1 --- /dev/null +++ b/packages/observer-dashboard/src/lib/relayflow-run.test.ts @@ -0,0 +1,63 @@ +import { describe, expect, it } from 'vitest'; +import type { MessageWithMeta } from '@relaycast/sdk'; +import { latestRelayflowRun, stepColumns, type RelayflowStep } from './relayflow-run'; + +function message(id: string, createdAt: string, metadata?: Record): MessageWithMeta { + return { + id, channelId: 'c', agentName: 'flow', agentId: 'a', text: 't', blocks: null, + ...(metadata ? { metadata } : {}), + hasAttachments: false, threadId: null, attachments: [], createdAt, replyCount: 0, reactions: [], readByCount: 0, + } as MessageWithMeta; +} + +function snapshot(status: string, state: string) { + return { relayflow: { version: 1, event: 'x', run: { + runId: 'R1', flow: 'hello', status, + steps: [{ id: 'greet', type: 'deterministic', dependsOn: [], state }], + } } }; +} + +describe('latestRelayflowRun', () => { + it('returns the newest snapshot regardless of message order', () => { + const run = latestRelayflowRun([ + message('3', '2026-09-23T00:00:02.000Z', snapshot('completed', 'completed')), + message('1', '2026-09-23T00:00:00.000Z', snapshot('running', 'pending')), + message('2', '2026-09-23T00:00:01.000Z', snapshot('running', 'running')), + ]); + expect(run?.status).toBe('completed'); + expect(run?.steps[0]?.state).toBe('completed'); + }); + + it('breaks a timestamp tie by the later snowflake id', () => { + const at = '2026-09-23T00:00:00.000Z'; + const run = latestRelayflowRun([ + message('228601462569775105', at, snapshot('completed', 'completed')), + message('228601462569775104', at, snapshot('running', 'running')), + ]); + expect(run?.status).toBe('completed'); + }); + + it('ignores ordinary messages and malformed or unknown-version metadata', () => { + expect(latestRelayflowRun([message('1', '2026-09-23T00:00:00.000Z')])).toBeNull(); + expect(latestRelayflowRun([message('1', '2026-09-23T00:00:00.000Z', { relayflow: { version: 2, run: {} } })])).toBeNull(); + expect(latestRelayflowRun([message('1', '2026-09-23T00:00:00.000Z', { relayflow: { version: 1, run: { runId: 1 } } })])).toBeNull(); + const run = latestRelayflowRun([message('1', '2026-09-23T00:00:00.000Z', { relayflow: { version: 1, run: { + runId: 'R', flow: 'f', status: 'running', steps: [{ id: 'ok', state: 'running' }, { id: 'bad', state: 'exploded' }, 7], + } } })]); + expect(run?.steps.map(step => step.id)).toEqual(['ok']); + }); +}); + +describe('stepColumns', () => { + const step = (id: string, dependsOn: string[] = []): RelayflowStep => ({ id, type: 'deterministic', dependsOn, state: 'pending' }); + + it('groups steps by dependency depth in declaration order', () => { + const columns = stepColumns([step('a'), step('b'), step('c', ['a']), step('d', ['b', 'c'])]); + expect(columns.map(column => column.map(s => s.id))).toEqual([['a', 'b'], ['c'], ['d']]); + }); + + it('keeps steps with unknown dependencies or cycles visible', () => { + const columns = stepColumns([step('x', ['missing']), step('y', ['z']), step('z', ['y'])]); + expect(columns.flat().map(s => s.id).sort()).toEqual(['x', 'y', 'z']); + }); +}); diff --git a/packages/observer-dashboard/src/lib/relayflow-run.ts b/packages/observer-dashboard/src/lib/relayflow-run.ts new file mode 100644 index 00000000..bd3dc58d --- /dev/null +++ b/packages/observer-dashboard/src/lib/relayflow-run.ts @@ -0,0 +1,104 @@ +/** + * A relayflow run projects itself into its `wf-` channel: every message + * carries the whole run snapshot under `metadata.relayflow` (version 1). The + * newest valid snapshot is the run's current state. Metadata is caller-authored, + * so it is parsed field by field and anything malformed is ignored. + */ + +import type { MessageWithMeta } from '@relaycast/sdk'; + +export type RelayflowStepState = 'pending' | 'running' | 'completed' | 'failed' | 'parked'; + +export interface RelayflowStep { + id: string; + type: string; + dependsOn: string[]; + state: RelayflowStepState; + attempt?: number; + elapsedMs?: number; + completionReason?: string; +} + +export interface RelayflowRun { + runId: string; + flow: string; + status: string; + completionReason?: string; + steps: RelayflowStep[]; +} + +const STATES = new Set(['pending', 'running', 'completed', 'failed', 'parked']); + +/** The newest run snapshot among `messages` (any order), or null. */ +export function latestRelayflowRun(messages: readonly MessageWithMeta[]): RelayflowRun | null { + let latest: { at: number; id: string; run: RelayflowRun } | null = null; + for (const message of messages) { + const run = parseRun((message.metadata as Record | undefined)?.relayflow); + if (run === null) continue; + const at = Date.parse(message.createdAt) || 0; + // Ids are snowflakes: on a timestamp tie the later id is the later message. + if (latest === null || at > latest.at || (at === latest.at && compareIds(message.id, latest.id) > 0)) { + latest = { at, id: message.id, run }; + } + } + return latest?.run ?? null; +} + +function compareIds(a: string, b: string): number { + return a.length === b.length ? a.localeCompare(b) : a.length - b.length; +} + +function parseRun(value: unknown): RelayflowRun | null { + if (!isRecord(value) || value.version !== 1 || !isRecord(value.run)) return null; + const run = value.run; + if (typeof run.runId !== 'string' || typeof run.flow !== 'string' || typeof run.status !== 'string' + || !Array.isArray(run.steps)) return null; + return { + runId: run.runId, + flow: run.flow, + status: run.status, + ...(typeof run.completionReason === 'string' ? { completionReason: run.completionReason } : {}), + steps: run.steps.flatMap(parseStep), + }; +} + +function parseStep(value: unknown): RelayflowStep[] { + if (!isRecord(value) || typeof value.id !== 'string' || typeof value.state !== 'string' + || !STATES.has(value.state)) return []; + return [{ + id: value.id, + type: typeof value.type === 'string' ? value.type : 'step', + dependsOn: Array.isArray(value.dependsOn) ? value.dependsOn.filter((id): id is string => typeof id === 'string') : [], + state: value.state as RelayflowStepState, + ...(typeof value.attempt === 'number' ? { attempt: value.attempt } : {}), + ...(typeof value.elapsedMs === 'number' ? { elapsedMs: value.elapsedMs } : {}), + ...(typeof value.completionReason === 'string' ? { completionReason: value.completionReason } : {}), + }]; +} + +/** + * Group steps into columns by dependency depth, preserving declaration order + * within a column. A dependency that names no known step, or a cycle, counts + * as depth 0 rather than hiding the step. + */ +export function stepColumns(steps: readonly RelayflowStep[]): RelayflowStep[][] { + const byId = new Map(steps.map(step => [step.id, step])); + const depth = new Map(); + const visit = (step: RelayflowStep, seen: Set): number => { + const known = depth.get(step.id); + if (known !== undefined) return known; + if (seen.has(step.id)) return 0; + seen.add(step.id); + const parents = step.dependsOn.map(id => byId.get(id)).filter((parent): parent is RelayflowStep => parent !== undefined); + const value = parents.length === 0 ? 0 : 1 + Math.max(...parents.map(parent => visit(parent, seen))); + depth.set(step.id, value); + return value; + }; + const columns: RelayflowStep[][] = []; + for (const step of steps) (columns[visit(step, new Set())] ??= []).push(step); + return columns.filter(column => column !== undefined); +} + +function isRecord(value: unknown): value is Record { + return typeof value === 'object' && value !== null && !Array.isArray(value); +} diff --git a/packages/react/src/__tests__/reducer.test.ts b/packages/react/src/__tests__/reducer.test.ts index fda6455d..1c5be49e 100644 --- a/packages/react/src/__tests__/reducer.test.ts +++ b/packages/react/src/__tests__/reducer.test.ts @@ -72,6 +72,19 @@ describe('handleServerEvent', () => { expect(state.channelMessages['general'].messages[0].text).toBe('hello'); }); + it('keeps the live message metadata', () => { + const store = createStore(); + const metadata = { relayflow: { version: 1 } }; + + handleServerEvent(store, { + type: 'message.created', + channel: 'general', + message: { id: 'msg1', agentName: 'Alice', text: 'hello', attachments: [], metadata }, + }); + + expect(store.getState().channelMessages['general'].messages[0].metadata).toEqual(metadata); + }); + it('deduplicates by id', () => { const store = createStore(); diff --git a/packages/react/src/reducer.ts b/packages/react/src/reducer.ts index 856d62ab..bc7901f0 100644 --- a/packages/react/src/reducer.ts +++ b/packages/react/src/reducer.ts @@ -74,6 +74,7 @@ function handleMessageCreated(store: RelayStore, event: MessageCreatedEvent): vo hasAttachments: (event.message.attachments?.length ?? 0) > 0, threadId: null, attachments: event.message.attachments ?? [], + ...(event.message.metadata ? { metadata: event.message.metadata } : {}), createdAt: new Date().toISOString(), replyCount: 0, reactions: [], From a1de55e1eb0ab3bc3d4d620a75e2a5291b86cacb Mon Sep 17 00:00:00 2001 From: Miya Date: Thu, 24 Sep 2026 08:19:38 +0200 Subject: [PATCH 2/4] fix(observer): make relayflow live state monotonic --- CHANGELOG.md | 3 +- README.md | 4 +- openapi.yaml | 2 +- .../src/engine/__tests__/wsTransform.test.ts | 1 + packages/engine/src/engine/wsTransform.ts | 1 + .../src/components/ChatFeed.tsx | 1 + .../src/components/RunPanel.test.tsx | 21 +++++++ .../src/components/RunPanel.tsx | 4 ++ .../src/lib/relayflow-run.test.ts | 32 ++++++++--- .../src/lib/relayflow-run.ts | 57 +++++++++++++++---- packages/react/src/__tests__/reducer.test.ts | 8 ++- packages/react/src/reducer.ts | 5 +- packages/sdk-typescript/CHANGELOG.md | 4 ++ .../src/__tests__/agent-ws.test.ts | 9 ++- packages/sdk-typescript/src/ws.ts | 3 + packages/types/CHANGELOG.md | 4 ++ packages/types/src/events.ts | 1 + 17 files changed, 135 insertions(+), 25 deletions(-) create mode 100644 packages/observer-dashboard/src/components/RunPanel.test.tsx diff --git a/CHANGELOG.md b/CHANGELOG.md index 1eb8e623..a83638c1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -24,7 +24,8 @@ Packages without a separate changelog are covered by the cross-package notes bel ### Fixed -- Realtime `message.created` events carry the message's `metadata`, so structured messages render live instead of only after a refetch. +- Realtime `message.created` events carry the message's `metadata` on workspace and direct-node transports and preserve the server creation time, so structured messages render live without replayed events appearing newer than persisted messages. +- Relayflow observer panels order snapshots by their server-assigned snowflake IDs, render cyclic dependencies without inventing an execution sequence, and expose each step state as text instead of color alone. ## [8.11.7] - 2026-09-22 diff --git a/README.md b/README.md index d3fb92ec..8dd6cbf4 100644 --- a/README.md +++ b/README.md @@ -401,7 +401,9 @@ await me.send('#general', 'Hello from Relaycast'); const observer = new RelayCast({ apiKey: 'ot_live_...' }); observer.connect(); observer.on.messageCreated((event) => { - console.log(`[workspace] ${event.channel}: ${event.message.text}`); + // createdAt is the persisted server time; structured message data is + // delivered as message.metadata on both observer and direct-node streams. + console.log(`[workspace] ${event.createdAt} ${event.channel}: ${event.message.text}`, event.message.metadata); }); observer.on.actionCompleted((event) => { console.log(`[workspace] ${event.actionName} ${event.status}`); diff --git a/openapi.yaml b/openapi.yaml index 907df993..e94e91f5 100644 --- a/openapi.yaml +++ b/openapi.yaml @@ -6218,7 +6218,7 @@ paths: /ws: get: summary: Workspace observer WebSocket stream - description: Upgrade to the workspace observer realtime stream with an `ot_live_*` token that has `stream:read`. Workspace, agent, and node tokens cannot open this stream; agent SDK realtime uses `/v1/node/ws` with a direct node token. The token is supplied via the `token` query parameter; this endpoint does not read an Authorization header. Query-param tokens can appear in access logs. `file.uploaded` is emitted at upload completion before channel or DM attachment exists; channel and DM visibility are enforced on file REST reads and message attachment reads. + description: Upgrade to the workspace observer realtime stream with an `ot_live_*` token that has `stream:read`. Workspace, agent, and node tokens cannot open this stream; agent SDK realtime uses `/v1/node/ws` with a direct node token. The token is supplied via the `token` query parameter; this endpoint does not read an Authorization header. Query-param tokens can appear in access logs. `message.created` carries the persisted `created_at` plus public structured message `metadata`; direct-node delivery projects the same fields. `file.uploaded` is emitted at upload completion before channel or DM attachment exists; channel and DM visibility are enforced on file REST reads and message attachment reads. tags: - System security: diff --git a/packages/engine/src/engine/__tests__/wsTransform.test.ts b/packages/engine/src/engine/__tests__/wsTransform.test.ts index b9809db5..ccfcc0f5 100644 --- a/packages/engine/src/engine/__tests__/wsTransform.test.ts +++ b/packages/engine/src/engine/__tests__/wsTransform.test.ts @@ -18,6 +18,7 @@ const BASE = { timestamp: '2026-07-06T00:00:00.000Z', } as const; +/** Build a complete internal WebSocket event from focused test fields. */ function ev(partial: Partial & { type: string; data: Record }): WsEvent { return { ...BASE, ...partial } as WsEvent; } diff --git a/packages/engine/src/engine/wsTransform.ts b/packages/engine/src/engine/wsTransform.ts index f201d4fa..0c70c8a5 100644 --- a/packages/engine/src/engine/wsTransform.ts +++ b/packages/engine/src/engine/wsTransform.ts @@ -265,6 +265,7 @@ export function transformForClient(event: WsEvent): Record { } } +/** Narrow public metadata to a JSON object before projecting it to clients. */ function isRecord(value: unknown): value is Record { return typeof value === 'object' && value !== null && !Array.isArray(value); } diff --git a/packages/observer-dashboard/src/components/ChatFeed.tsx b/packages/observer-dashboard/src/components/ChatFeed.tsx index 5960c880..1318ef08 100644 --- a/packages/observer-dashboard/src/components/ChatFeed.tsx +++ b/packages/observer-dashboard/src/components/ChatFeed.tsx @@ -190,6 +190,7 @@ function FeedList({ ); } +/** Render one channel's live feed and its newest valid relayflow snapshot. */ function ChannelMessages({ channel, scrollRef, onOpenThread, mentionNames, onOpenAgent }: { channel: string; scrollRef: RefObject; onOpenThread?: (messageId: string) => void; mentionNames?: string[]; onOpenAgent?: (agentName: string | null) => void; }) { const { messages, loading, fetchMore } = useMessages(channel); const sorted = sortMessagesChronologically(messages); diff --git a/packages/observer-dashboard/src/components/RunPanel.test.tsx b/packages/observer-dashboard/src/components/RunPanel.test.tsx new file mode 100644 index 00000000..84e1719b --- /dev/null +++ b/packages/observer-dashboard/src/components/RunPanel.test.tsx @@ -0,0 +1,21 @@ +import React from 'react'; +import { renderToStaticMarkup } from 'react-dom/server'; +import { describe, expect, it } from 'vitest'; +import { RunPanel } from './RunPanel'; + +describe('RunPanel', () => { + it('renders each step state as text instead of relying on icon color', () => { + const html = renderToStaticMarkup(); + + expect(html).toContain('running · deterministic'); + expect(html).toContain('pending · agent'); + }); +}); diff --git a/packages/observer-dashboard/src/components/RunPanel.tsx b/packages/observer-dashboard/src/components/RunPanel.tsx index c3a26996..375c4f67 100644 --- a/packages/observer-dashboard/src/components/RunPanel.tsx +++ b/packages/observer-dashboard/src/components/RunPanel.tsx @@ -1,5 +1,6 @@ 'use client'; +import React from 'react'; import { CheckCircle2, ChevronRight, Circle, Loader2, PauseCircle, Workflow, XCircle } from 'lucide-react'; import { cn } from '../lib/utils'; import { stepColumns, type RelayflowRun, type RelayflowStep, type RelayflowStepState } from '../lib/relayflow-run'; @@ -55,9 +56,11 @@ export function RunPanel({ run }: { run: RelayflowRun }) { ); } +/** Render one step with a visible, screen-reader-readable state label. */ function StepChip({ step }: { step: RelayflowStep }) { const { icon: Icon, tone } = STATE_STYLE[step.state]; const detail = [ + step.state, step.type, step.elapsedMs === undefined ? undefined : formatElapsed(step.elapsedMs), step.attempt !== undefined && step.attempt > 1 ? `attempt ${step.attempt}` : undefined, @@ -76,6 +79,7 @@ function StepChip({ step }: { step: RelayflowStep }) { ); } +/** Format a millisecond duration compactly for the run panel. */ function formatElapsed(ms: number): string { if (ms < 1_000) return `${ms}ms`; if (ms < 60_000) return `${(ms / 1_000).toFixed(1)}s`; diff --git a/packages/observer-dashboard/src/lib/relayflow-run.test.ts b/packages/observer-dashboard/src/lib/relayflow-run.test.ts index d5c0b0c1..ace7550c 100644 --- a/packages/observer-dashboard/src/lib/relayflow-run.test.ts +++ b/packages/observer-dashboard/src/lib/relayflow-run.test.ts @@ -2,6 +2,7 @@ import { describe, expect, it } from 'vitest'; import type { MessageWithMeta } from '@relaycast/sdk'; import { latestRelayflowRun, stepColumns, type RelayflowStep } from './relayflow-run'; +/** Build a minimal persisted or realtime message for projection selection. */ function message(id: string, createdAt: string, metadata?: Record): MessageWithMeta { return { id, channelId: 'c', agentName: 'flow', agentId: 'a', text: 't', blocks: null, @@ -10,6 +11,7 @@ function message(id: string, createdAt: string, metadata?: Record { expect(run?.steps[0]?.state).toBe('completed'); }); - it('breaks a timestamp tie by the later snowflake id', () => { - const at = '2026-09-23T00:00:00.000Z'; + it('uses the later snowflake even when timestamps are skewed or replayed', () => { const run = latestRelayflowRun([ - message('228601462569775105', at, snapshot('completed', 'completed')), - message('228601462569775104', at, snapshot('running', 'running')), + message('228601462569775105', '2026-09-23T00:00:00.000Z', snapshot('completed', 'completed')), + message('228601462569775104', '2026-09-23T00:05:00.000Z', snapshot('running', 'running')), ]); expect(run?.status).toBe('completed'); }); @@ -46,6 +47,15 @@ describe('latestRelayflowRun', () => { } } })]); expect(run?.steps.map(step => step.id)).toEqual(['ok']); }); + + it('selects the newest valid snapshot across malformed messages', () => { + const run = latestRelayflowRun([ + message('228601462569775104', '2026-09-23T00:00:00.000Z', snapshot('running', 'running')), + message('228601462569775105', '2026-09-23T00:00:01.000Z', { relayflow: { version: 1, run: { runId: 7 } } }), + message('228601462569775106', '2026-09-23T00:00:02.000Z', snapshot('completed', 'completed')), + ]); + expect(run?.status).toBe('completed'); + }); }); describe('stepColumns', () => { @@ -56,8 +66,16 @@ describe('stepColumns', () => { expect(columns.map(column => column.map(s => s.id))).toEqual([['a', 'b'], ['c'], ['d']]); }); - it('keeps steps with unknown dependencies or cycles visible', () => { - const columns = stepColumns([step('x', ['missing']), step('y', ['z']), step('z', ['y'])]); - expect(columns.flat().map(s => s.id).sort()).toEqual(['x', 'y', 'z']); + it('puts every cycle member at depth zero and its dependents after it', () => { + const columns = stepColumns([ + step('x', ['missing']), + step('y', ['z']), + step('z', ['y']), + step('after-cycle', ['z']), + ]); + expect(columns.map(column => column.map(s => s.id))).toEqual([ + ['x', 'y', 'z'], + ['after-cycle'], + ]); }); }); diff --git a/packages/observer-dashboard/src/lib/relayflow-run.ts b/packages/observer-dashboard/src/lib/relayflow-run.ts index bd3dc58d..49a612b4 100644 --- a/packages/observer-dashboard/src/lib/relayflow-run.ts +++ b/packages/observer-dashboard/src/lib/relayflow-run.ts @@ -29,25 +29,29 @@ export interface RelayflowRun { const STATES = new Set(['pending', 'running', 'completed', 'failed', 'parked']); -/** The newest run snapshot among `messages` (any order), or null. */ +/** + * Return the newest valid run snapshot among `messages`, regardless of array + * or delivery order. Relaycast message IDs are sortable snowflakes, so their + * server-assigned order is authoritative even under replay or clock skew. + */ export function latestRelayflowRun(messages: readonly MessageWithMeta[]): RelayflowRun | null { - let latest: { at: number; id: string; run: RelayflowRun } | null = null; + let latest: { id: string; run: RelayflowRun } | null = null; for (const message of messages) { const run = parseRun((message.metadata as Record | undefined)?.relayflow); if (run === null) continue; - const at = Date.parse(message.createdAt) || 0; - // Ids are snowflakes: on a timestamp tie the later id is the later message. - if (latest === null || at > latest.at || (at === latest.at && compareIds(message.id, latest.id) > 0)) { - latest = { at, id: message.id, run }; + if (latest === null || compareIds(message.id, latest.id) > 0) { + latest = { id: message.id, run }; } } return latest?.run ?? null; } +/** Compare decimal snowflake IDs without losing precision to Number. */ function compareIds(a: string, b: string): number { return a.length === b.length ? a.localeCompare(b) : a.length - b.length; } +/** Parse the closed relayflow v1 run projection, ignoring invalid snapshots. */ function parseRun(value: unknown): RelayflowRun | null { if (!isRecord(value) || value.version !== 1 || !isRecord(value.run)) return null; const run = value.run; @@ -62,6 +66,7 @@ function parseRun(value: unknown): RelayflowRun | null { }; } +/** Parse one relayflow v1 step, returning no value for a malformed step. */ function parseStep(value: unknown): RelayflowStep[] { if (!isRecord(value) || typeof value.id !== 'string' || typeof value.state !== 'string' || !STATES.has(value.state)) return []; @@ -83,22 +88,52 @@ function parseStep(value: unknown): RelayflowStep[] { */ export function stepColumns(steps: readonly RelayflowStep[]): RelayflowStep[][] { const byId = new Map(steps.map(step => [step.id, step])); + const cyclic = findCyclicSteps(byId); const depth = new Map(); - const visit = (step: RelayflowStep, seen: Set): number => { + const visit = (step: RelayflowStep): number => { + if (cyclic.has(step.id)) return 0; const known = depth.get(step.id); if (known !== undefined) return known; - if (seen.has(step.id)) return 0; - seen.add(step.id); const parents = step.dependsOn.map(id => byId.get(id)).filter((parent): parent is RelayflowStep => parent !== undefined); - const value = parents.length === 0 ? 0 : 1 + Math.max(...parents.map(parent => visit(parent, seen))); + const value = parents.length === 0 ? 0 : 1 + Math.max(...parents.map(visit)); depth.set(step.id, value); return value; }; const columns: RelayflowStep[][] = []; - for (const step of steps) (columns[visit(step, new Set())] ??= []).push(step); + for (const step of steps) (columns[visit(step)] ??= []).push(step); return columns.filter(column => column !== undefined); } +/** Find every step participating in a dependency cycle. */ +function findCyclicSteps(byId: ReadonlyMap): Set { + const visited = new Set(); + const visiting = new Set(); + const path: string[] = []; + const cyclic = new Set(); + + const visit = (step: RelayflowStep): void => { + if (visited.has(step.id)) return; + if (visiting.has(step.id)) { + const start = path.lastIndexOf(step.id); + for (const id of path.slice(start)) cyclic.add(id); + return; + } + visiting.add(step.id); + path.push(step.id); + for (const dependencyId of step.dependsOn) { + const dependency = byId.get(dependencyId); + if (dependency) visit(dependency); + } + path.pop(); + visiting.delete(step.id); + visited.add(step.id); + }; + + for (const step of byId.values()) visit(step); + return cyclic; +} + +/** Narrow unknown metadata containers to plain records. */ function isRecord(value: unknown): value is Record { return typeof value === 'object' && value !== null && !Array.isArray(value); } diff --git a/packages/react/src/__tests__/reducer.test.ts b/packages/react/src/__tests__/reducer.test.ts index 1c5be49e..3ef7809a 100644 --- a/packages/react/src/__tests__/reducer.test.ts +++ b/packages/react/src/__tests__/reducer.test.ts @@ -72,17 +72,21 @@ describe('handleServerEvent', () => { expect(state.channelMessages['general'].messages[0].text).toBe('hello'); }); - it('keeps the live message metadata', () => { + it('keeps live metadata and the server creation time', () => { const store = createStore(); const metadata = { relayflow: { version: 1 } }; handleServerEvent(store, { type: 'message.created', + createdAt: '2026-09-24T05:00:00.000Z', channel: 'general', message: { id: 'msg1', agentName: 'Alice', text: 'hello', attachments: [], metadata }, }); - expect(store.getState().channelMessages['general'].messages[0].metadata).toEqual(metadata); + expect(store.getState().channelMessages['general'].messages[0]).toMatchObject({ + createdAt: '2026-09-24T05:00:00.000Z', + metadata, + }); }); it('deduplicates by id', () => { diff --git a/packages/react/src/reducer.ts b/packages/react/src/reducer.ts index bc7901f0..71a135ec 100644 --- a/packages/react/src/reducer.ts +++ b/packages/react/src/reducer.ts @@ -61,6 +61,7 @@ export function handleServerEvent(store: RelayStore, event: WsClientEvent): void } } +/** Append one validated realtime channel message without duplicating replays. */ function handleMessageCreated(store: RelayStore, event: MessageCreatedEvent): void { store.updateChannelMessages(event.channel, (prev) => { if (prev.messages.some((m) => m.id === event.message.id)) return prev; @@ -75,7 +76,9 @@ function handleMessageCreated(store: RelayStore, event: MessageCreatedEvent): vo threadId: null, attachments: event.message.attachments ?? [], ...(event.message.metadata ? { metadata: event.message.metadata } : {}), - createdAt: new Date().toISOString(), + // New servers provide their persisted timestamp. The fallback keeps the + // reducer compatible with older/self-hosted servers during an upgrade. + createdAt: event.createdAt ?? new Date().toISOString(), replyCount: 0, reactions: [], readByCount: 0, diff --git a/packages/sdk-typescript/CHANGELOG.md b/packages/sdk-typescript/CHANGELOG.md index e68e1d11..36371c67 100644 --- a/packages/sdk-typescript/CHANGELOG.md +++ b/packages/sdk-typescript/CHANGELOG.md @@ -9,6 +9,10 @@ and this project follows [Semantic Versioning](https://semver.org/spec/v2.0.0.ht ## [Unreleased] +### Fixed + +- Direct-node `message.created` events preserve public message metadata and the server creation timestamp, matching workspace WebSocket delivery. + ## [8.11.1] - 2026-09-17 ### Fixed diff --git a/packages/sdk-typescript/src/__tests__/agent-ws.test.ts b/packages/sdk-typescript/src/__tests__/agent-ws.test.ts index e40de74c..cbe7b260 100644 --- a/packages/sdk-typescript/src/__tests__/agent-ws.test.ts +++ b/packages/sdk-typescript/src/__tests__/agent-ws.test.ts @@ -224,18 +224,25 @@ describe('AgentClient WebSocket integration', () => { type: 'message.created', data: { id: 'm_1', + created_at: '2026-09-24T05:00:00.000Z', channel_name: 'general', agent_id: 'bot_1', from_name: 'Bot', text: 'hi', attachments: [], + metadata: { relayflow: { version: 1 } }, }, }, }); expect(handler).toHaveBeenCalledTimes(1); expect(handler).toHaveBeenCalledWith( - expect.objectContaining({ type: 'message.created', channel: 'general' }), + expect.objectContaining({ + type: 'message.created', + createdAt: '2026-09-24T05:00:00.000Z', + channel: 'general', + message: expect.objectContaining({ metadata: { relayflow: { version: 1 } } }), + }), ); expect(ws.send).toHaveBeenCalledWith(JSON.stringify({ v: 1, diff --git a/packages/sdk-typescript/src/ws.ts b/packages/sdk-typescript/src/ws.ts index 14f5d723..ea191752 100644 --- a/packages/sdk-typescript/src/ws.ts +++ b/packages/sdk-typescript/src/ws.ts @@ -495,12 +495,14 @@ export class WsClient { } } + /** Normalize a direct-node delivery into the public workspace event shape. */ private transformServerLikeEvent(eventType: string, data: Record): Record { switch (eventType) { case 'message.created': return { id: stableRelaycastEventId(String(data.id ?? '')), type: 'message.created', + ...(typeof data.created_at === 'string' ? { created_at: data.created_at } : {}), channel: data.channel_name, message: { id: data.id, @@ -509,6 +511,7 @@ export class WsClient { text: data.text, attachments: Array.isArray(data.attachments) ? data.attachments : [], ...(typeof data.injection_mode === 'string' ? { injection_mode: data.injection_mode } : {}), + ...(isRecord(data.metadata) ? { metadata: data.metadata } : {}), }, }; case 'thread.reply': diff --git a/packages/types/CHANGELOG.md b/packages/types/CHANGELOG.md index 7ade05de..987ffc67 100644 --- a/packages/types/CHANGELOG.md +++ b/packages/types/CHANGELOG.md @@ -9,6 +9,10 @@ and this project follows [Semantic Versioning](https://semver.org/spec/v2.0.0.ht ## [Unreleased] +### Added + +- `MessageCreatedEvent` accepts the server-provided `created_at` timestamp used to keep replayed realtime messages in persisted creation order. + ## [8.11.0] - 2026-09-17 - Add task action capability mode, correlated execution metadata, action.accept, and fenced final/interim result validation. diff --git a/packages/types/src/events.ts b/packages/types/src/events.ts index b532db46..cf526b59 100644 --- a/packages/types/src/events.ts +++ b/packages/types/src/events.ts @@ -48,6 +48,7 @@ export type ChannelMessagePayload = z.infer; export const MessageCreatedEventSchema = z.object({ id: z.string().uuid(), type: z.literal('message.created'), + created_at: z.string().optional(), channel: z.string(), message: ChannelMessagePayloadSchema, }); From e5c709c5da8ee3b9076e88e00abb9af404d1b330 Mon Sep 17 00:00:00 2001 From: Miya Date: Thu, 24 Sep 2026 08:35:41 +0200 Subject: [PATCH 3/4] docs: mark observer event changes as patch releases --- packages/sdk-typescript/CHANGELOG.md | 2 +- packages/types/CHANGELOG.md | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/packages/sdk-typescript/CHANGELOG.md b/packages/sdk-typescript/CHANGELOG.md index 36371c67..88377266 100644 --- a/packages/sdk-typescript/CHANGELOG.md +++ b/packages/sdk-typescript/CHANGELOG.md @@ -7,7 +7,7 @@ See the [root changelog](../../CHANGELOG.md) for cross-package release highlight The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project follows [Semantic Versioning](https://semver.org/spec/v2.0.0.html). -## [Unreleased] +## [Unreleased - Patch] ### Fixed diff --git a/packages/types/CHANGELOG.md b/packages/types/CHANGELOG.md index 987ffc67..d69bcb8c 100644 --- a/packages/types/CHANGELOG.md +++ b/packages/types/CHANGELOG.md @@ -7,7 +7,7 @@ See the [root changelog](../../CHANGELOG.md) for cross-package release highlight The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project follows [Semantic Versioning](https://semver.org/spec/v2.0.0.html). -## [Unreleased] +## [Unreleased - Patch] ### Added From f4a1c56d51c159d37a860053b47190ee57f90ed3 Mon Sep 17 00:00:00 2001 From: Relayflow Lead Date: Thu, 24 Sep 2026 07:20:49 -0700 Subject: [PATCH 4/4] test(observer-dashboard): compile JSX with the automatic runtime under vitest RunPanel imported React only so its test could render under vitest's classic JSX transform; Next compiles with the automatic runtime. Match Next in the test config and drop the unused import. Co-Authored-By: Claude Opus 5.5 (1M context) --- packages/observer-dashboard/src/components/RunPanel.tsx | 1 - packages/observer-dashboard/vitest.config.ts | 3 +++ 2 files changed, 3 insertions(+), 1 deletion(-) diff --git a/packages/observer-dashboard/src/components/RunPanel.tsx b/packages/observer-dashboard/src/components/RunPanel.tsx index 375c4f67..e8f5e124 100644 --- a/packages/observer-dashboard/src/components/RunPanel.tsx +++ b/packages/observer-dashboard/src/components/RunPanel.tsx @@ -1,6 +1,5 @@ 'use client'; -import React from 'react'; import { CheckCircle2, ChevronRight, Circle, Loader2, PauseCircle, Workflow, XCircle } from 'lucide-react'; import { cn } from '../lib/utils'; import { stepColumns, type RelayflowRun, type RelayflowStep, type RelayflowStepState } from '../lib/relayflow-run'; diff --git a/packages/observer-dashboard/vitest.config.ts b/packages/observer-dashboard/vitest.config.ts index 7382f40e..6e2aad86 100644 --- a/packages/observer-dashboard/vitest.config.ts +++ b/packages/observer-dashboard/vitest.config.ts @@ -1,6 +1,9 @@ import { defineConfig } from 'vitest/config'; export default defineConfig({ + // Next compiles JSX with the automatic runtime; tests must match, or a + // component that (correctly) never imports React fails only under vitest. + esbuild: { jsx: 'automatic' }, test: { globals: true, },