feat(observer): draw relayflow runs; carry message metadata on realtime events

CHANGELOG.md

@@ -16,7 +16,16 @@ 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` 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
 

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}`);

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:

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
 

packages/engine/src/engine/__tests__/wsTransform.test.ts

@@ -18,11 +18,23 @@ 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<WsEvent> & { type: string; data: Record<string, unknown> }): WsEvent {
   return { ...BASE, ...partial } as WsEvent;
 }
 
 describe('transformForClient - message.created', () => {
+  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<string, unknown>).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({

packages/engine/src/engine/wsTransform.ts

@@ -36,6 +36,9 @@ export function transformForClient(event: WsEvent): Record<string, unknown> {
           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,8 @@ export function transformForClient(event: WsEvent): Record<string, unknown> {
     }
   }
 }
+
+/** Narrow public metadata to a JSON object before projecting it to clients. */
+function isRecord(value: unknown): value is Record<string, unknown> {
+  return typeof value === 'object' && value !== null && !Array.isArray(value);
+}

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';
 
@@ -188,15 +190,23 @@ 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<HTMLDivElement>; onOpenThread?: (messageId: string) => void; mentionNames?: string[]; onOpenAgent?: (agentName: string | null) => void; }) {
   const { messages, loading, fetchMore } = useMessages(channel);
   const sorted = sortMessagesChronologically(messages);
   const feed = usePaginatedFeed(scrollRef, sorted, fetchMore);
 
+  const run = latestRelayflowRun(sorted);
+
   if (loading && sorted.length === 0) return <LoadingState label="Loading messages…" />;
   if (sorted.length === 0) return <EmptyState label="No messages yet" />;
 
-  return <FeedList sorted={sorted} feed={feed} onOpenThread={onOpenThread} mentionNames={mentionNames} onOpenAgent={onOpenAgent} />;
+  return (
+    <>
+      {run && <RunPanel run={run} />}
+      <FeedList sorted={sorted} feed={feed} onOpenThread={onOpenThread} mentionNames={mentionNames} onOpenAgent={onOpenAgent} />
+    </>
+  );
 }
 
 function toMessageWithMeta(m: DmMessage): MessageWithMeta {

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(<RunPanel run={{
+      runId: '01TEST',
+      flow: 'review',
+      status: 'running',
+      steps: [
+        { id: 'inspect', type: 'deterministic', dependsOn: [], state: 'running' },
+        { id: 'report', type: 'agent', dependsOn: ['inspect'], state: 'pending' },
+      ],
+    }} />);
+
+    expect(html).toContain('running · deterministic');
+    expect(html).toContain('pending · agent');
+  });
+});

packages/observer-dashboard/src/components/RunPanel.tsx

@@ -0,0 +1,86 @@
+'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<RelayflowStepState, { icon: typeof Circle; tone: string }> = {
+  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<string, string> = {
+  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 (
+    <div className="sticky top-0 z-10 border-b border-[var(--border-default)] bg-[var(--surface-card)] px-5 py-3">
+      <div className="flex flex-wrap items-center gap-2 text-sm">
+        <Workflow className="h-4 w-4 text-[var(--brand-primary)]" />
+        <span className="font-semibold text-[var(--foreground)]">{run.flow}</span>
+        <span className={cn('brand-pill text-[11px] font-medium uppercase tracking-[0.16em]', RUN_TONE[run.status])}>
+          {run.status}
+        </span>
+        {run.completionReason && run.completionReason !== 'success' && (
+          <span className="text-xs text-[var(--status-danger)]">{run.completionReason}</span>
+        )}
+        <span className="ml-auto text-xs text-[var(--text-muted)]">
+          {done}/{run.steps.length} steps · <span className="font-mono">{run.runId}</span>
+        </span>
+      </div>
+      {columns.length > 0 && (
+        <div className="mt-3 flex items-start gap-2 overflow-x-auto pb-1">
+          {columns.map((column, index) => (
+            <div key={index} className="flex items-center gap-2">
+              {index > 0 && <ChevronRight className="h-4 w-4 shrink-0 text-[var(--text-faint)]" />}
+              <div className="flex flex-col gap-1.5">
+                {column.map(step => <StepChip key={step.id} step={step} />)}
+              </div>
+            </div>
+          ))}
+        </div>
+      )}
+    </div>
+  );
+}
+
+/** 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,
+  ].filter(Boolean).join(' · ');
+  return (
+    <div
+      className={cn('flex min-w-[9rem] items-center gap-2 rounded-xl border px-2.5 py-1.5', tone)}
+      title={step.completionReason ? `completionReason: ${step.completionReason}` : undefined}
+    >
+      <Icon className={cn('h-3.5 w-3.5 shrink-0', step.state === 'running' && 'animate-spin')} />
+      <div className="min-w-0">
+        <div className="truncate text-xs font-semibold text-[var(--foreground)]">{step.id}</div>
+        <div className="truncate text-[11px] opacity-80">{detail}</div>
+      </div>
+    </div>
+  );
+}
+
+/** 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`;
+  return `${Math.floor(ms / 60_000)}m ${Math.round((ms % 60_000) / 1_000)}s`;
+}

packages/observer-dashboard/src/lib/relayflow-run.test.ts

@@ -0,0 +1,81 @@
+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<string, unknown>): 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;
+}
+
+/** Build a valid single-step relayflow v1 snapshot. */
+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('uses the later snowflake even when timestamps are skewed or replayed', () => {
+    const run = latestRelayflowRun([
+      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');
+  });
+
+  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']);
+  });
+
+  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', () => {
+  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('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'],
+    ]);
+  });
+});