fix: stream session updates live

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

BACKEND_UI_CHANGES.md

@@ -83,13 +83,13 @@ The Electron main process maps this to a `SessionEventRecord` with `kind: 'agent
 
 ### Renderer consumption
 
-`App.tsx` now subscribes to `onSessionEvent` and tracks live activity per agent for the selected session. `ChatPane.tsx` uses that state to show a status row for each agent while the run is active.
+`App.tsx` now subscribes to `onSessionEvent`, applies message-delta / message-complete updates into renderer workspace state so assistant responses can stream live, and tracks live activity per agent for the selected session. `ChatPane.tsx` uses that state to show a status row for each agent while the run is active.
 
 - "Code Reviewer is thinking…"
 - "Code Reviewer is using read_file…"
 - "Handing off to Summarizer…"
 
-The activity panel in `ChatPane.tsx` is now wired to this data, showing every agent in the pattern with observed activity such as thinking, tool usage, handoff, or completed. If no event has been observed for an agent yet, the UI states that no status has been reported instead of inventing a synthetic state. The panel also keeps the last observed statuses visible after a run completes, and resets them when the next run begins.
+The activity panel in `ChatPane.tsx` is now wired to this data, showing every agent in the pattern with observed activity such as thinking, tool usage, handoff, or completed. If no event has been observed for an agent yet, the UI states that no status has been reported instead of inventing a synthetic state. The panel also keeps the last observed statuses visible after a run completes, and resets them when the next run begins. Completed activity is emitted when final messages are applied, so the status no longer jumps to `Completed` before the corresponding response becomes visible.
 
 ## Files involved
 

sidecar/src/Kopaya.AgentHost/Services/CopilotWorkflowRunner.cs

@@ -49,7 +49,6 @@ public sealed class CopilotWorkflowRunner : ITurnWorkflowRunner
         List<ChatMessageDto> completedMessages = [];
         AgentIdentity? activeAgent = null;
         HashSet<string> startedAgents = new(StringComparer.OrdinalIgnoreCase);
-        HashSet<string> completedAgents = new(StringComparer.OrdinalIgnoreCase);
 
         await using StreamingRun run = await InProcessExecution.RunStreamingAsync(workflow, inputMessages).ConfigureAwait(false);
         await run.TrySendMessageAsync(new TurnToken(emitEvents: true)).ConfigureAwait(false);
@@ -81,7 +80,7 @@ public sealed class CopilotWorkflowRunner : ITurnWorkflowRunner
                     await onActivity(activity).ConfigureAwait(false);
                 }
             }
-            else if (evt is AgentResponseUpdateEvent update && !string.IsNullOrEmpty(update.Update.Text))
+            else if (evt is AgentResponseUpdateEvent update)
             {
                 AgentIdentity? updateAgent = null;
                 string authorName = update.ExecutorId;
@@ -94,10 +93,6 @@ public sealed class CopilotWorkflowRunner : ITurnWorkflowRunner
                     authorName = resolvedUpdateAgent.AgentName;
                 }
 
-                string messageId = update.Update.MessageId ?? $"{command.RequestId}-delta-{fallbackMessageIndex++}";
-                StreamingSegment segment = GetOrCreateSegment(segments, messageId, authorName);
-                segment.Content.Append(update.Update.Text);
-
                 if (updateAgent.HasValue)
                 {
                     activeAgent = updateAgent.Value;
@@ -108,6 +103,15 @@ public sealed class CopilotWorkflowRunner : ITurnWorkflowRunner
                         onActivity).ConfigureAwait(false);
                 }
 
+                if (string.IsNullOrEmpty(update.Update.Text))
+                {
+                    continue;
+                }
+
+                string messageId = update.Update.MessageId ?? $"{command.RequestId}-delta-{fallbackMessageIndex++}";
+                StreamingSegment segment = GetOrCreateSegment(segments, messageId, authorName);
+                segment.Content.Append(update.Update.Text);
+
                 await onDelta(new TurnDeltaEventDto
                 {
                     Type = "turn-delta",
@@ -124,14 +128,6 @@ public sealed class CopilotWorkflowRunner : ITurnWorkflowRunner
                     completed.ExecutorId,
                     out AgentIdentity completedAgent))
             {
-                if (completedAgents.Add(completedAgent.AgentId))
-                {
-                    await onActivity(CreateActivityEvent(
-                        command,
-                        activityType: "completed",
-                        agent: completedAgent)).ConfigureAwait(false);
-                }
-
                 if (activeAgent.HasValue
                     && string.Equals(activeAgent.Value.AgentId, completedAgent.AgentId, StringComparison.Ordinal))
                 {
@@ -143,11 +139,6 @@ public sealed class CopilotWorkflowRunner : ITurnWorkflowRunner
                 List<ChatMessage> allMessages = outputEvent.As<List<ChatMessage>>() ?? [];
                 List<ChatMessage> newMessages = allMessages.Skip(inputMessages.Count).ToList();
                 completedMessages = ConvertOutputMessages(command, newMessages, segments);
-                await EmitCompletedActivitiesForMessages(
-                    command,
-                    completedMessages,
-                    completedAgents,
-                    onActivity).ConfigureAwait(false);
             }
         }
 
@@ -214,34 +205,6 @@ public sealed class CopilotWorkflowRunner : ITurnWorkflowRunner
             agent: agent)).ConfigureAwait(false);
     }
 
-    private static async Task EmitCompletedActivitiesForMessages(
-        RunTurnCommandDto command,
-        IReadOnlyList<ChatMessageDto> messages,
-        ISet<string> completedAgents,
-        Func<AgentActivityEventDto, Task> onActivity)
-    {
-        foreach (ChatMessageDto message in messages)
-        {
-            if (!AgentIdentityResolver.TryResolveKnownAgentIdentity(
-                command.Pattern,
-                message.AuthorName,
-                out AgentIdentity messageAgent))
-            {
-                continue;
-            }
-
-            if (!completedAgents.Add(messageAgent.AgentId))
-            {
-                continue;
-            }
-
-            await onActivity(CreateActivityEvent(
-                command,
-                activityType: "completed",
-                agent: messageAgent)).ConfigureAwait(false);
-        }
-    }
-
     private static bool TryGetHandoffTarget(
         PatternDefinitionDto pattern,
         RequestInfoEvent requestInfo,

src/main/KopayaAppService.ts

@@ -316,8 +316,34 @@ export class KopayaAppService extends EventEmitter<AppServiceEvents> {
     });
   }
 
+  private emitCompletedActivity(
+    sessionId: string,
+    pattern: PatternDefinition,
+    message: ChatMessageRecord,
+  ): void {
+    if (message.role !== 'assistant') {
+      return;
+    }
+
+    const agent = pattern.agents.find((candidate) =>
+      candidate.id === message.authorName || candidate.name === message.authorName);
+    if (!agent) {
+      return;
+    }
+
+    this.emitSessionEvent({
+      sessionId,
+      kind: 'agent-activity',
+      occurredAt: nowIso(),
+      activityType: 'completed',
+      agentId: agent.id,
+      agentName: agent.name,
+    });
+  }
+
   private finalizeTurn(workspace: WorkspaceState, sessionId: string, messages: ChatMessageRecord[]): void {
     const session = this.requireSession(workspace, sessionId);
+    const pattern = this.requirePattern(workspace, session.patternId);
     const incomingIds = new Set(messages.map((message) => message.id));
 
     for (const message of messages) {
@@ -337,6 +363,8 @@ export class KopayaAppService extends EventEmitter<AppServiceEvents> {
         messageId: message.id,
         authorName: message.authorName,
       });
+
+      this.emitCompletedActivity(sessionId, pattern, message);
     }
 
     for (const message of session.messages) {

src/renderer/App.tsx

@@ -10,6 +10,7 @@ import {
   pruneSessionActivities,
   type SessionActivityMap,
 } from '@renderer/lib/sessionActivity';
+import { applySessionEventWorkspace } from '@renderer/lib/sessionWorkspace';
 import { WelcomePane } from '@renderer/components/WelcomePane';
 import { getElectronApi } from '@renderer/lib/electronApi';
 import type { PatternDefinition } from '@shared/domain/pattern';
@@ -70,6 +71,7 @@ export default function App() {
     });
 
     const offSessionEvent = api.onSessionEvent((event) => {
+      setWorkspace((current) => applySessionEventWorkspace(current, event));
       setSessionActivities((current) => applySessionEventActivity(current, event));
     });
 

src/renderer/lib/sessionWorkspace.ts

@@ -0,0 +1,155 @@
+import type { SessionEventRecord } from '@shared/domain/event';
+import type { ChatMessageRecord, SessionRecord } from '@shared/domain/session';
+import type { WorkspaceState } from '@shared/domain/workspace';
+
+export function applySessionEventWorkspace(
+  current: WorkspaceState | undefined,
+  event: SessionEventRecord,
+): WorkspaceState | undefined {
+  if (!current) {
+    return current;
+  }
+
+  const sessionIndex = current.sessions.findIndex((session) => session.id === event.sessionId);
+  if (sessionIndex < 0) {
+    return current;
+  }
+
+  const session = current.sessions[sessionIndex];
+  const nextSession = applySessionEvent(session, event);
+  if (nextSession === session) {
+    return current;
+  }
+
+  const nextSessions = current.sessions.slice();
+  nextSessions[sessionIndex] = nextSession;
+  return {
+    ...current,
+    sessions: nextSessions,
+  };
+}
+
+function applySessionEvent(session: SessionRecord, event: SessionEventRecord): SessionRecord {
+  switch (event.kind) {
+    case 'status':
+      return applyStatusEvent(session, event);
+    case 'error':
+      return applyErrorEvent(session, event);
+    case 'message-delta':
+      return applyMessageDeltaEvent(session, event);
+    case 'message-complete':
+      return applyMessageCompleteEvent(session, event);
+    default:
+      return session;
+  }
+}
+
+function applyStatusEvent(session: SessionRecord, event: SessionEventRecord): SessionRecord {
+  if (!event.status) {
+    return session;
+  }
+
+  if (session.status === event.status && (event.status === 'error' || !session.lastError)) {
+    return session;
+  }
+
+  return {
+    ...session,
+    status: event.status,
+    lastError: event.status === 'error' ? session.lastError : undefined,
+    updatedAt: event.occurredAt,
+  };
+}
+
+function applyErrorEvent(session: SessionRecord, event: SessionEventRecord): SessionRecord {
+  const error = event.error?.trim();
+  if (session.status === 'error' && session.lastError === error) {
+    return session;
+  }
+
+  return {
+    ...session,
+    status: 'error',
+    lastError: error,
+    updatedAt: event.occurredAt,
+  };
+}
+
+function applyMessageDeltaEvent(session: SessionRecord, event: SessionEventRecord): SessionRecord {
+  if (!event.messageId || event.contentDelta === undefined) {
+    return session;
+  }
+
+  const messageIndex = session.messages.findIndex((message) => message.id === event.messageId);
+  if (messageIndex >= 0) {
+    const existing = session.messages[messageIndex];
+    const nextMessage: ChatMessageRecord = {
+      ...existing,
+      authorName: event.authorName ?? existing.authorName,
+      content: `${existing.content}${event.contentDelta}`,
+      pending: true,
+    };
+
+    if (
+      nextMessage.authorName === existing.authorName
+      && nextMessage.content === existing.content
+      && existing.pending
+    ) {
+      return session;
+    }
+
+    const nextMessages = session.messages.slice();
+    nextMessages[messageIndex] = nextMessage;
+    return {
+      ...session,
+      messages: nextMessages,
+      updatedAt: event.occurredAt,
+    };
+  }
+
+  return {
+    ...session,
+    messages: [
+      ...session.messages,
+      {
+        id: event.messageId,
+        role: 'assistant',
+        authorName: event.authorName ?? 'assistant',
+        content: event.contentDelta,
+        createdAt: event.occurredAt,
+        pending: true,
+      },
+    ],
+    updatedAt: event.occurredAt,
+  };
+}
+
+function applyMessageCompleteEvent(session: SessionRecord, event: SessionEventRecord): SessionRecord {
+  if (!event.messageId) {
+    return session;
+  }
+
+  const messageIndex = session.messages.findIndex((message) => message.id === event.messageId);
+  if (messageIndex < 0) {
+    return session;
+  }
+
+  const existing = session.messages[messageIndex];
+  const nextMessage: ChatMessageRecord = {
+    ...existing,
+    authorName: event.authorName ?? existing.authorName,
+    pending: false,
+  };
+
+  if (nextMessage.authorName === existing.authorName && !existing.pending) {
+    return session;
+  }
+
+  const nextMessages = session.messages.slice();
+  nextMessages[messageIndex] = nextMessage;
+  return {
+    ...session,
+    messages: nextMessages,
+    updatedAt: event.occurredAt,
+  };
+}

tests/renderer/sessionWorkspace.test.ts

@@ -0,0 +1,130 @@
+import { describe, expect, test } from 'bun:test';
+
+import { applySessionEventWorkspace } from '@renderer/lib/sessionWorkspace';
+import type { SessionEventRecord } from '@shared/domain/event';
+import type { WorkspaceState } from '@shared/domain/workspace';
+
+describe('session workspace helpers', () => {
+  function createWorkspace(): WorkspaceState {
+    return {
+      projects: [],
+      patterns: [],
+      sessions: [
+        {
+          id: 'session-1',
+          projectId: 'project-1',
+          patternId: 'pattern-1',
+          title: 'Session',
+          createdAt: '2026-03-23T00:00:00.000Z',
+          updatedAt: '2026-03-23T00:00:00.000Z',
+          status: 'idle',
+          messages: [],
+        },
+      ],
+      selectedSessionId: 'session-1',
+      lastUpdatedAt: '2026-03-23T00:00:00.000Z',
+    };
+  }
+
+  test('applies message deltas by creating and appending assistant messages', () => {
+    const created = applySessionEventWorkspace(createWorkspace(), {
+      sessionId: 'session-1',
+      kind: 'message-delta',
+      occurredAt: '2026-03-23T00:00:01.000Z',
+      messageId: 'assistant-1',
+      authorName: 'Architect',
+      contentDelta: 'Hello',
+    } satisfies SessionEventRecord);
+
+    expect(created?.sessions[0].messages).toEqual([
+      {
+        id: 'assistant-1',
+        role: 'assistant',
+        authorName: 'Architect',
+        content: 'Hello',
+        createdAt: '2026-03-23T00:00:01.000Z',
+        pending: true,
+      },
+    ]);
+
+    const appended = applySessionEventWorkspace(created, {
+      sessionId: 'session-1',
+      kind: 'message-delta',
+      occurredAt: '2026-03-23T00:00:02.000Z',
+      messageId: 'assistant-1',
+      authorName: 'Architect',
+      contentDelta: ' world',
+    } satisfies SessionEventRecord);
+
+    expect(appended?.sessions[0].messages[0]).toMatchObject({
+      authorName: 'Architect',
+      content: 'Hello world',
+      pending: true,
+    });
+  });
+
+  test('marks streamed messages complete when the session event arrives', () => {
+    const workspace = applySessionEventWorkspace(createWorkspace(), {
+      sessionId: 'session-1',
+      kind: 'message-delta',
+      occurredAt: '2026-03-23T00:00:01.000Z',
+      messageId: 'assistant-1',
+      authorName: 'Implementer',
+      contentDelta: 'Done',
+    } satisfies SessionEventRecord);
+
+    const completed = applySessionEventWorkspace(workspace, {
+      sessionId: 'session-1',
+      kind: 'message-complete',
+      occurredAt: '2026-03-23T00:00:02.000Z',
+      messageId: 'assistant-1',
+      authorName: 'Implementer',
+    } satisfies SessionEventRecord);
+
+    expect(completed?.sessions[0].messages[0]).toMatchObject({
+      authorName: 'Implementer',
+      content: 'Done',
+      pending: false,
+    });
+  });
+
+  test('updates session status and error state from session events', () => {
+    const running = applySessionEventWorkspace(createWorkspace(), {
+      sessionId: 'session-1',
+      kind: 'status',
+      occurredAt: '2026-03-23T00:00:01.000Z',
+      status: 'running',
+    } satisfies SessionEventRecord);
+
+    expect(running?.sessions[0]).toMatchObject({
+      status: 'running',
+      lastError: undefined,
+    });
+
+    const failed = applySessionEventWorkspace(running, {
+      sessionId: 'session-1',
+      kind: 'error',
+      occurredAt: '2026-03-23T00:00:02.000Z',
+      error: 'Boom',
+    } satisfies SessionEventRecord);
+
+    expect(failed?.sessions[0]).toMatchObject({
+      status: 'error',
+      lastError: 'Boom',
+    });
+  });
+
+  test('ignores events for unknown sessions', () => {
+    const workspace = createWorkspace();
+    expect(
+      applySessionEventWorkspace(workspace, {
+        sessionId: 'missing',
+        kind: 'message-delta',
+        occurredAt: '2026-03-23T00:00:01.000Z',
+        messageId: 'assistant-1',
+        authorName: 'Architect',
+        contentDelta: 'Hello',
+      } satisfies SessionEventRecord),
+    ).toBe(workspace);
+  });
+});