From 9ba0174f6892ccd09348ac42efa63d3c2904f5af Mon Sep 17 00:00:00 2001 From: David Kaya Date: Sat, 21 Mar 2026 16:12:48 +0100 Subject: [PATCH] fix: stream session updates live Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- BACKEND_UI_CHANGES.md | 4 +- .../Services/CopilotWorkflowRunner.cs | 57 ++----- src/main/KopayaAppService.ts | 28 ++++ src/renderer/App.tsx | 2 + src/renderer/lib/sessionWorkspace.ts | 155 ++++++++++++++++++ tests/renderer/sessionWorkspace.test.ts | 130 +++++++++++++++ 6 files changed, 327 insertions(+), 49 deletions(-) create mode 100644 src/renderer/lib/sessionWorkspace.ts create mode 100644 tests/renderer/sessionWorkspace.test.ts diff --git a/BACKEND_UI_CHANGES.md b/BACKEND_UI_CHANGES.md index 51445f3..3658c6a 100644 --- a/BACKEND_UI_CHANGES.md +++ b/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 diff --git a/sidecar/src/Kopaya.AgentHost/Services/CopilotWorkflowRunner.cs b/sidecar/src/Kopaya.AgentHost/Services/CopilotWorkflowRunner.cs index 644a32c..353ac64 100644 --- a/sidecar/src/Kopaya.AgentHost/Services/CopilotWorkflowRunner.cs +++ b/sidecar/src/Kopaya.AgentHost/Services/CopilotWorkflowRunner.cs @@ -49,7 +49,6 @@ public sealed class CopilotWorkflowRunner : ITurnWorkflowRunner List completedMessages = []; AgentIdentity? activeAgent = null; HashSet startedAgents = new(StringComparer.OrdinalIgnoreCase); - HashSet 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 allMessages = outputEvent.As>() ?? []; List 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 messages, - ISet completedAgents, - Func 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, diff --git a/src/main/KopayaAppService.ts b/src/main/KopayaAppService.ts index cba9026..6a00cde 100644 --- a/src/main/KopayaAppService.ts +++ b/src/main/KopayaAppService.ts @@ -316,8 +316,34 @@ export class KopayaAppService extends EventEmitter { }); } + 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 { messageId: message.id, authorName: message.authorName, }); + + this.emitCompletedActivity(sessionId, pattern, message); } for (const message of session.messages) { diff --git a/src/renderer/App.tsx b/src/renderer/App.tsx index 31b46c3..1823fef 100644 --- a/src/renderer/App.tsx +++ b/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)); }); diff --git a/src/renderer/lib/sessionWorkspace.ts b/src/renderer/lib/sessionWorkspace.ts new file mode 100644 index 0000000..584a513 --- /dev/null +++ b/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, + }; +} diff --git a/tests/renderer/sessionWorkspace.test.ts b/tests/renderer/sessionWorkspace.test.ts new file mode 100644 index 0000000..ee7b97b --- /dev/null +++ b/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); + }); +});