Render all transcript message kinds live

.github/workflows/pr.yml

@@ -29,8 +29,15 @@ jobs:
       - name: Unit tests
         run: npm run test:unit
 
-      - name: E2E tests
-        run: npx playwright test --reporter=html,list
+      - name: Build
+        run: npm run build
+
+      - name: Functional realtime E2E suite
+        run: >-
+          npx playwright test --project=desktop --reporter=html,list
+          tests/e2e/custom-messages.spec.ts
+          tests/e2e/interleaving.spec.ts
+          tests/e2e/retry-errors.spec.ts
 
       - name: Upload Playwright report
         if: always()

server/mock.ts

@@ -1,5 +1,6 @@
 import { join } from "node:path";
 import type { PiWebSession, PiWebSessionInfo } from "./types.js";
+import { simplifyMessage } from "./session/projection.js";
 
 interface MockSessionOptions {
   piCwd: string;
@@ -217,12 +218,19 @@ export function createMockHarness(options: MockSessionOptions) {
 
     function broadcastPiEvent(event: Record<string, unknown>, activityAt?: string | false) {
       const lastActivityAt = activityAt === false ? runtimeLastActivityAt : markRuntimeActivity(activityAt || new Date().toISOString());
+      const committedMessage = event.type === "message_end" ? simplifyMessage(event.message) : undefined;
       broadcast({
         type: "pi_event",
         sessionId: mockSession.sessionId,
         sessionFile: mockSession.sessionFile,
         event: lastActivityAt ? { ...event, lastActivityAt } : event,
       });
+      if (committedMessage) broadcast({
+        type: "committed_message",
+        sessionId: mockSession.sessionId,
+        sessionFile: mockSession.sessionFile,
+        message: committedMessage,
+      });
     }
 
     async function runMockCompaction(customInstructions?: string, slow = false) {
@@ -488,6 +496,7 @@ export function createMockHarness(options: MockSessionOptions) {
         const withoutAgentEnd = /missing agent end|no agent end/i.test(message);
         const withStaleRuntimeAfterEnd = /stale runtime after end/i.test(message);
         const withPendingToolRefresh = /pending tool refresh/i.test(message) || withProgressDemo;
+        const withLiveMessageKinds = /live message kinds/i.test(message);
         const withTools = !withShowcase && !withEditTool && !withMalformedEditTool && !withInterruptedTool && (/tool|interleav/i.test(message) || withProgressDemo || withLateToolTimestamp);
         mockSession.isStreaming = true;
         if (withQuietRuntime) {
@@ -497,7 +506,26 @@ export function createMockHarness(options: MockSessionOptions) {
         }
         broadcastRuntimeChanged();
         broadcastPiEvent({ type: "agent_start", startedAt: runtimeStartedAt }, runtimeLastActivityAt || runtimeStartedAt);
-        if (withQuietRuntime) {
+        if (withLiveMessageKinds) {
+          // Let the browser apply agent_start before exercising interleaved
+          // committed messages; this keeps the scenario deterministic on CI.
+          if (!(await waitForMockRun(150))) return;
+          if (!(await waitForMockRun(500))) return;
+          const timestamp = new Date().toISOString();
+          const visibleCustom = { role: "custom", customType: "probe", content: "hello from an extension", details: { source: "mock-extension" }, display: true, timestamp };
+          appendMockMessage(visibleCustom);
+          broadcastPiEvent({ type: "message_end", message: visibleCustom });
+          if (!(await waitForMockRun(500))) return;
+          broadcastPiEvent({ type: "message_update", assistantMessageEvent: { type: "text_delta", delta: "streamed prefix" } });
+          const hiddenCustom = { role: "custom", customType: "probe-hidden", content: "hidden extension message", details: { source: "mock-extension" }, display: false, timestamp };
+          appendMockMessage(hiddenCustom);
+          broadcastPiEvent({ type: "message_end", message: hiddenCustom });
+          const unknownMessage = { role: "futureKind", content: "future message content", timestamp };
+          appendMockMessage(unknownMessage);
+          broadcastPiEvent({ type: "message_end", message: unknownMessage });
+          broadcastPiEvent({ type: "message_update", assistantMessageEvent: { type: "text_delta", delta: "streamed suffix" } });
+        }
+        if (withQuietRuntime || withLiveMessageKinds) {
           if (!(await waitForMockRun(60_000))) return;
         } else if (slow && !(await waitForMockRun(/queue demo/i.test(message) ? 2_500 : 750))) return;
         if (withProviderError) {

server/session/dto.ts

@@ -36,20 +36,25 @@ export interface BaseSessionStateDto {
   stats: SessionStatsDto;
 }
 
-/** Serializable message projection consumed by the browser message list. */
-export interface MessageDto {
+/** Serializable, role-discriminated projection consumed by every transcript path. */
+type MessageDtoBase = {
   entryId?: string;
-  role?: string;
   text?: string;
-  toolCalls?: Array<{ id?: string; toolName: string; args: JsonValue; startedAt?: string }>;
-  toolCallId?: string;
-  toolName?: string;
-  toolArgs?: JsonValue;
-  isError?: boolean;
   timestamp?: string;
   raw?: JsonValue;
-  [key: string]: JsonValue | undefined;
-}
+};
+
+export type MessageDto = MessageDtoBase & (
+  | { role: "user"; isError?: boolean }
+  | { role: "assistant"; toolCalls?: Array<{ id?: string; toolName: string; args: JsonValue; startedAt?: string }>; isError: boolean }
+  | { role: "system"; isError?: boolean }
+  | { role: "toolResult"; toolCallId?: string; toolName?: string; toolArgs?: JsonValue; isError: boolean }
+  | { role: "bashExecution"; command?: JsonValue; output?: JsonValue; exitCode?: JsonValue; cancelled: boolean; truncated: boolean; fullOutputPath?: JsonValue; excludeFromContext: boolean }
+  | { role: "compactionSummary"; isError?: boolean }
+  | { role: "branchSummary"; isError?: boolean }
+  | { role: "unknown"; originalRole: string; isError?: boolean }
+  | { role: "custom"; customType: string; details?: JsonValue; display: true }
+);
 
 export interface TreeNodeDto {
   id: string;
@@ -111,6 +116,7 @@ export interface DeleteSessionResultDto {
 export type SessionServiceEvent =
   | { type: "pi"; sessionId: string; sessionFile: string; event: JsonValue; clientMessageId?: string; sourceClientId?: string }
   | { type: "state"; state: BaseSessionStateDto; includeThinkingLevels?: boolean }
+  | { type: "committed"; sessionId: string; sessionFile: string; message: MessageDto }
   | { type: "stats"; sessionId: string; sessionFile: string; stats: SessionStatsDto }
   | { type: "models"; sessionId: string; models: ModelDto[] }
   | { type: "error"; sessionId?: string; sessionFile?: string; error: string; clientMessageId?: string }

server/session/hostEvents.ts

@@ -57,7 +57,7 @@ export function decorateHostMessages(messages: MessageDto[], sessionFile: string
       : [];
     return {
       ...message,
-      ...(message.toolCalls ? {
+      ...(message.role === "assistant" && message.toolCalls ? {
         toolCalls: message.toolCalls.map((call, index) => {
           const startedAt = decoratedToolCalls[index]?.startedAt;
           return startedAt && !call.startedAt ? { ...call, startedAt } : call;
@@ -96,6 +96,14 @@ export function createHostSessionEventHandler(deps: HostEventDependencies) {
         });
         return;
       }
+      case "committed":
+        deps.broadcast({
+          type: "committed_message",
+          sessionId: serviceEvent.sessionId,
+          sessionFile: serviceEvent.sessionFile,
+          message: decorateHostMessages([serviceEvent.message], serviceEvent.sessionFile, deps.sessionActivity)[0],
+        });
+        return;
       case "state": {
         const target = deps.sessionForId(serviceEvent.state.sessionId);
         if (target) deps.broadcast({ type: "state_changed", ...decorate(serviceEvent.state, target, Boolean(serviceEvent.includeThinkingLevels)) });

server/session/projection.ts

@@ -1,8 +1,10 @@
 import type { PiWebSession } from "../types.js";
-import type { BaseSessionStateDto, ConversationTreeDto, ModelDto, SessionStatsDto, SlashCommandDto } from "./dto.js";
+import { jsonRoundTrip, type BaseSessionStateDto, type ConversationTreeDto, type MessageDto, type ModelDto, type SessionStatsDto, type SlashCommandDto } from "./dto.js";
 
 export type ContentDecorator = (content: unknown) => unknown;
 
+const warnedUnknownMessageRoles = new Set<string>();
+
 export function textFromContent(content: unknown): string {
   if (typeof content === "string") return content;
   if (!Array.isArray(content)) return "";
@@ -76,14 +78,14 @@ export function messageEntryRefs(targetSession: PiWebSession): Array<{ entryId?:
 export function simplifyMessage(
   message: unknown,
   options: { toolCallArgs?: Map<string, Record<string, unknown>>; decorateContent?: ContentDecorator; entryId?: string } = {},
-) {
-  if (!message || typeof message !== "object") return message;
+): MessageDto | undefined {
+  if (!message || typeof message !== "object") return undefined;
   const m = message as Record<string, unknown>;
   const content = options.decorateContent ? options.decorateContent(m.content) : m.content;
   const entry = options.entryId ? { entryId: options.entryId } : {};
   const toolCallArgs = options.toolCallArgs;
   if (m.role === "bashExecution") {
-    return {
+    return jsonRoundTrip({
       ...entry,
       role: "bashExecution",
       command: m.command,
@@ -95,11 +97,11 @@ export function simplifyMessage(
       excludeFromContext: Boolean(m.excludeFromContext),
       timestamp: m.timestamp,
       raw: m,
-    };
+    }) as MessageDto;
   }
   if (m.role === "toolResult") {
     const args = toolCallArgs?.get(m.toolCallId as string);
-    return {
+    return jsonRoundTrip({
       ...entry,
       role: "toolResult",
       toolCallId: m.toolCallId,
@@ -109,7 +111,35 @@ export function simplifyMessage(
       text: textFromContent(m.content),
       timestamp: m.timestamp,
       raw: m,
-    };
+    }) as MessageDto;
+  }
+  if (m.role === "custom") {
+    if (m.display === false) return undefined;
+    return jsonRoundTrip({
+      ...entry,
+      role: "custom",
+      customType: typeof m.customType === "string" ? m.customType : "",
+      text: textFromContent(content),
+      details: m.details,
+      display: true,
+      timestamp: m.timestamp,
+      raw: content === m.content ? m : { ...m, content },
+    }) as MessageDto;
+  }
+  if (!["user", "assistant", "system", "compactionSummary", "branchSummary"].includes(String(m.role))) {
+    const originalRole = typeof m.role === "string" && m.role ? m.role : "unknown";
+    if (!warnedUnknownMessageRoles.has(originalRole)) {
+      warnedUnknownMessageRoles.add(originalRole);
+      console.warn(`Projecting unknown transcript message role: ${originalRole}`);
+    }
+    return jsonRoundTrip({
+      ...entry,
+      role: "unknown",
+      originalRole,
+      text: textFromContent(content),
+      timestamp: m.timestamp,
+      raw: content === m.content ? m : { ...m, content },
+    }) as MessageDto;
   }
   const text = textFromContent(content);
   const errorText = m.role === "assistant" && m.errorMessage ? assistantErrorPreview(m) : "";
@@ -123,15 +153,41 @@ export function simplifyMessage(
       startedAt: part.startedAt,
     }))
     : undefined;
-  return {
+  return jsonRoundTrip({
     ...entry,
     role: m.role,
     text: displayText,
     toolCalls,
     isError: Boolean(m.errorMessage || m.stopReason === "error" || stopReasonText),
     timestamp: m.timestamp,
     raw: content === m.content ? m : { ...m, content },
-  };
+  }) as MessageDto;
+}
+
+function messageProjectionContext(targetSession: PiWebSession) {
+  const toolCallArgs = new Map<string, Record<string, unknown>>();
+  for (const message of targetSession.messages as any[]) {
+    if (message?.role !== "assistant" || !Array.isArray(message.content)) continue;
+    for (const part of message.content) {
+      if (part?.type === "toolCall" && part.id) toolCallArgs.set(part.id, part.arguments || {});
+    }
+  }
+  return { toolCallArgs, refs: messageEntryRefs(targetSession) };
+}
+
+export function projectMessages(targetSession: PiWebSession): MessageDto[] {
+  const { toolCallArgs, refs } = messageProjectionContext(targetSession);
+  return targetSession.messages.flatMap((message, index) => {
+    const projected = simplifyMessage(message, { toolCallArgs, entryId: refs[index]?.entryId });
+    return projected ? [projected] : [];
+  });
+}
+
+export function projectCommittedMessage(targetSession: PiWebSession, committed: unknown): MessageDto | undefined {
+  const index = targetSession.messages.lastIndexOf(committed as never);
+  if (index < 0) return undefined;
+  const { toolCallArgs, refs } = messageProjectionContext(targetSession);
+  return simplifyMessage(targetSession.messages[index], { toolCallArgs, entryId: refs[index]?.entryId });
 }
 
 export function truncatePreview(value: string, max = 220) {
@@ -141,7 +197,14 @@ export function truncatePreview(value: string, max = 220) {
 
 export function entryMessage(entry: any) {
   if (entry?.type === "message") return entry.message;
-  if (entry?.type === "custom_message") return { role: "custom", content: entry.content, timestamp: entry.timestamp };
+  if (entry?.type === "custom_message") return {
+    role: "custom",
+    customType: entry.customType,
+    content: entry.content,
+    details: entry.details,
+    display: entry.display,
+    timestamp: entry.timestamp,
+  };
   return undefined;
 }
 

server/session/service.ts

@@ -33,10 +33,10 @@ import {
   isAssistantAbortedMessage,
   isAssistantFailureMessage,
   isIncompleteToolResultMessage,
-  messageEntryRefs,
+  projectCommittedMessage,
+  projectMessages,
   projectSessionState,
   sessionStats,
-  simplifyMessage,
   simplifyModel,
 } from "./projection.js";
 
@@ -266,19 +266,7 @@ export class LocalSessionService implements SessionService {
   }
 
   async messages(sessionId: string): Promise<MessageDto[]> {
-    const value = await this.require(sessionId);
-    const toolCallArgs = new Map<string, Record<string, unknown>>();
-    for (const message of value.messages as any[]) {
-      if (message?.role !== "assistant" || !Array.isArray(message.content)) continue;
-      for (const part of message.content) {
-        if (part?.type === "toolCall" && part.id) toolCallArgs.set(part.id, part.arguments || {});
-      }
-    }
-    const refs = messageEntryRefs(value);
-    return jsonSafe(value.messages.map((message, index) => simplifyMessage(message, {
-      toolCallArgs,
-      entryId: refs[index]?.entryId,
-    }) as MessageDto));
+    return jsonSafe(projectMessages(await this.require(sessionId)));
   }
 
   async commands(sessionId: string) {
@@ -663,6 +651,16 @@ export class LocalSessionService implements SessionService {
       event: event as JsonValue,
       ...(correlation ? { clientMessageId: correlation.clientMessageId, sourceClientId: correlation.sourceClientId } : {}),
     });
+    if (e?.type === "message_end") {
+      const committed = e.message;
+      // agent-core inserts this object before notifying listeners; the agent
+      // relay persists its entry after listeners return, while idle custom
+      // messages persist before emitting. Defer so both paths expose entry metadata.
+      queueMicrotask(() => {
+        const message = projectCommittedMessage(value, committed);
+        if (message) this.emit({ type: "committed", sessionId, sessionFile: value.sessionFile, message });
+      });
+    }
     if (e?.type === "session_info_changed") this.emit({ type: "state", state: this.projectState(value) });
     if (e?.type === "message_end" || e?.type === "agent_end" || e?.type === "compaction_end") {
       this.emit({ type: "stats", sessionId, sessionFile, stats: sessionStats(value) });