From bc21d1a6f866e0d9bffc7a011ac1c1993365a92d Mon Sep 17 00:00:00 2001 From: Federico Jaramillo Martinez Date: Thu, 7 May 2026 13:42:51 +0200 Subject: [PATCH] Fix live session UI updates --- src/client/src/api.ts | 8 ++ src/client/src/appState.ts | 5 +- src/client/src/components/PiWebApp.ts | 4 +- src/client/src/components/SessionList.ts | 5 +- src/client/src/components/StatusBar.ts | 15 ++- src/client/src/components/shared.ts | 7 +- .../src/controllers/sessionController.ts | 18 ++- src/client/src/sessionSocket.ts | 119 +++++++++++++++--- src/server/sessions/piSessionService.ts | 59 ++++++++- vite.config.ts | 2 +- 10 files changed, 215 insertions(+), 27 deletions(-) diff --git a/src/client/src/api.ts b/src/client/src/api.ts index fa2bb37..aaa46d3 100644 --- a/src/client/src/api.ts +++ b/src/client/src/api.ts @@ -26,6 +26,14 @@ export interface SessionInfo { firstMessage: string; } +export interface SessionActivity { + sessionId: string; + phase: "active" | "idle" | "error"; + label: string; + detail?: string; + at: string; +} + export interface SessionStatus { sessionId: string; model?: { provider?: string; id?: string; name?: string; contextWindow?: number; reasoning?: unknown }; diff --git a/src/client/src/appState.ts b/src/client/src/appState.ts index fb7deb1..abfce25 100644 --- a/src/client/src/appState.ts +++ b/src/client/src/appState.ts @@ -1,4 +1,4 @@ -import type { CommandResult, Project, SessionInfo, SessionStatus, Workspace } from "./api"; +import type { CommandResult, Project, SessionActivity, SessionInfo, SessionStatus, Workspace } from "./api"; import type { ChatLine } from "./components/shared"; export interface AppState { @@ -10,7 +10,9 @@ export interface AppState { selectedWorkspace?: Workspace; selectedSession?: SessionInfo; status?: SessionStatus; + activity?: SessionActivity; sessionStatuses: Record; + sessionActivities: Record; commandDialog?: Extract; error: string; } @@ -22,6 +24,7 @@ export function initialAppState(): AppState { sessions: [], messages: [], sessionStatuses: {}, + sessionActivities: {}, error: "", }; } diff --git a/src/client/src/components/PiWebApp.ts b/src/client/src/components/PiWebApp.ts index fdf9ade..24c3684 100644 --- a/src/client/src/components/PiWebApp.ts +++ b/src/client/src/components/PiWebApp.ts @@ -100,12 +100,12 @@ export class PiWebApp extends LitElement { this.withChatScrollTransition(() => this.workspaces.selectProject(project))}> this.withChatScrollTransition(() => this.workspaces.selectWorkspace(workspace))}> - this.withChatScrollTransition(() => this.sessions.startSession())} .onSelect=${(session: SessionInfo) => this.withChatScrollTransition(() => this.sessions.selectSession(session))}> + this.withChatScrollTransition(() => this.sessions.startSession())} .onSelect=${(session: SessionInfo) => this.withChatScrollTransition(() => this.sessions.selectSession(session))}>
${state.error ? html`
${state.error}
` : null} ${state.selectedSession ? html` - + this.sessions.send(text)} .onStopSession=${() => this.sessions.stopSession()}> ${state.commandDialog ? html` this.sessions.respondToCommand(state.commandDialog!.requestId, value)} .onCancel=${() => this.sessions.cancelCommand()}>` : null} diff --git a/src/client/src/components/SessionList.ts b/src/client/src/components/SessionList.ts index 738237d..007f514 100644 --- a/src/client/src/components/SessionList.ts +++ b/src/client/src/components/SessionList.ts @@ -1,12 +1,13 @@ import { LitElement, html } from "lit"; import { customElement, property } from "lit/decorators.js"; -import type { SessionInfo, SessionStatus } from "../api"; +import type { SessionActivity, SessionInfo, SessionStatus } from "../api"; import { listStyles } from "./shared"; @customElement("session-list") export class SessionList extends LitElement { @property({ attribute: false }) sessions: SessionInfo[] = []; @property({ attribute: false }) statuses: Record = {}; + @property({ attribute: false }) activities: Record = {}; @property({ attribute: false }) selected?: SessionInfo; @property({ type: Boolean }) canStart = false; @property({ attribute: false }) onSelect?: (session: SessionInfo) => void; @@ -27,6 +28,8 @@ export class SessionList extends LitElement { private renderStatus(session: SessionInfo) { const status = this.statuses[session.id]; + const activity = this.activities[session.id]; + if (activity?.phase === "active") return `● ${activity.label} · `; if (!status) return ""; if (status.isStreaming) return "● streaming · "; if (status.isBashRunning) return "● bash · "; diff --git a/src/client/src/components/StatusBar.ts b/src/client/src/components/StatusBar.ts index 5aae132..c73fdb6 100644 --- a/src/client/src/components/StatusBar.ts +++ b/src/client/src/components/StatusBar.ts @@ -1,12 +1,13 @@ import { LitElement, html } from "lit"; import { customElement, property } from "lit/decorators.js"; -import type { SessionStatus, Workspace } from "../api"; +import type { SessionActivity, SessionStatus, Workspace } from "../api"; import { formatCost, formatTokenCount } from "../utils/format"; import { statusBarStyles } from "./shared"; @customElement("status-bar") export class StatusBar extends LitElement { @property({ attribute: false }) status?: SessionStatus; + @property({ attribute: false }) activity?: SessionActivity; @property({ attribute: false }) workspace?: Workspace; render() { @@ -14,7 +15,8 @@ export class StatusBar extends LitElement { if (!status) return html`
No session status yet
`; const model = status.model?.id ?? "no model"; const provider = status.model?.provider ? `${status.model.provider}/` : ""; - const state = status.isCompacting ? "compacting" : status.isBashRunning ? "bash" : status.isStreaming ? "running" : "idle"; + const state = status.isCompacting ? "compacting" : status.isBashRunning ? "bash" : status.isStreaming ? "running" : status.pendingMessageCount ? "queued" : "idle"; + const active = state !== "idle" || this.activity?.phase === "active"; const context = status.contextUsage; const contextText = context ? `${context.percent == null ? "?" : context.percent.toFixed(1)}%/${formatTokenCount(context.contextWindow)}` @@ -23,7 +25,7 @@ export class StatusBar extends LitElement { return html`
${this.workspace?.label ?? "workspace"} - ${state} + ${this.activityText(state)} ${provider}${model} thinking ${status.thinkingLevel ?? "off"} ↑${formatTokenCount(tokens.input)} @@ -35,5 +37,12 @@ export class StatusBar extends LitElement { `; } + private activityText(state: string): string { + const activity = this.activity; + if (!activity) return state; + if (state !== "idle" && activity.phase === "idle") return state; + return activity.detail ? `${activity.label}: ${activity.detail}` : activity.label; + } + static styles = statusBarStyles; } diff --git a/src/client/src/components/shared.ts b/src/client/src/components/shared.ts index fc91e67..f72002a 100644 --- a/src/client/src/components/shared.ts +++ b/src/client/src/components/shared.ts @@ -92,8 +92,13 @@ export const statusBarStyles = css` :host { display: block; color: #8b949e; font: 12px system-ui, sans-serif; } .bar { display: flex; gap: 12px; align-items: center; min-width: 0; padding: 7px 12px; border-bottom: 1px solid #30363d; background: #0d1117; white-space: nowrap; overflow: hidden; } span { overflow: hidden; text-overflow: ellipsis; } - span:first-child { flex: 1 1 auto; min-width: 80px; } + .bar > span:first-child { flex: 1 1 auto; min-width: 80px; } + .activity { display: inline-flex; align-items: center; gap: 6px; color: #8b949e; } + .activity.active { color: #3fb950; } + .dot { width: 7px; height: 7px; border-radius: 50%; background: currentColor; opacity: .45; flex: 0 0 auto; } + .activity.active .dot { animation: pulse 1s ease-in-out infinite; opacity: 1; } .muted { color: #6e7681; } + @keyframes pulse { 0%, 100% { transform: scale(.75); opacity: .55; } 50% { transform: scale(1.2); opacity: 1; } } `; export const autocompleteStyles = css` diff --git a/src/client/src/controllers/sessionController.ts b/src/client/src/controllers/sessionController.ts index bd58ce0..9d3c4ad 100644 --- a/src/client/src/controllers/sessionController.ts +++ b/src/client/src/controllers/sessionController.ts @@ -1,4 +1,4 @@ -import { api, type CommandResult, type SessionInfo, type SessionStatus } from "../api"; +import { api, type CommandResult, type SessionActivity, type SessionInfo, type SessionStatus } from "../api"; import { appendText, normalizeMessages, textMessage } from "../chatMessages"; import { GlobalSessionSocket, SessionSocket, type SessionUiEvent } from "../sessionSocket"; import type { GetState, SetState, UpdateUrl } from "./types"; @@ -10,7 +10,10 @@ export class SessionController { constructor(private readonly getState: GetState, private readonly setState: SetState, private readonly updateUrl: UpdateUrl) {} connectStatusUpdates() { - this.globalSocket.connect((event) => this.applyStatus(event.status)); + this.globalSocket.connect((event) => { + if (event.type === "status.update") this.applyStatus(event.status); + else this.applyActivity(event.activity); + }); } dispose() { @@ -20,7 +23,7 @@ export class SessionController { clearActiveSession() { this.socket.close(); - this.setState({ selectedSession: undefined, messages: [], status: undefined }); + this.setState({ selectedSession: undefined, messages: [], status: undefined, activity: undefined }); } async startSession() { @@ -115,6 +118,13 @@ export class SessionController { } } + private applyActivity(activity: SessionActivity) { + this.setState({ + sessionActivities: { ...this.getState().sessionActivities, [activity.sessionId]: activity }, + activity: this.getState().selectedSession?.id === activity.sessionId ? activity : this.getState().activity, + }); + } + private applyStatus(status: SessionStatus) { this.setState({ sessionStatuses: { ...this.getState().sessionStatuses, [status.sessionId]: status }, @@ -132,6 +142,8 @@ export class SessionController { this.setState({ messages: [...messages, textMessage("tool", `${event.isError ? "✖" : "✓"} ${event.toolName}`)] }); } else if (event.type === "status.update") { this.applyStatus(event.status); + } else if (event.type === "activity.update") { + this.applyActivity(event.activity); } else if (event.type === "session.error") { this.setState({ messages: [...messages, textMessage("system", event.message)] }); } diff --git a/src/client/src/sessionSocket.ts b/src/client/src/sessionSocket.ts index e40ba8d..e7eb348 100644 --- a/src/client/src/sessionSocket.ts +++ b/src/client/src/sessionSocket.ts @@ -1,21 +1,27 @@ -import { globalSessionEvents, sessionEvents, type SessionStatus } from "./api"; +import { globalSessionEvents, sessionEvents, type SessionActivity, type SessionStatus } from "./api"; export type SessionUiEvent = | { type: "assistant.delta"; text: string } | { type: "tool.start"; toolName: string } | { type: "tool.end"; toolName: string; isError: boolean } | { type: "status.update"; status: SessionStatus } + | { type: "activity.update"; activity: SessionActivity } | { type: "session.error"; message: string }; export class SessionSocket { private socket?: WebSocket; + private sessionId?: string; private onEvent?: (event: SessionUiEvent) => void; + private reconnectTimer?: number; + private reconnectDelay = 500; + private shouldReconnect = false; connect(sessionId: string, onEvent: (event: SessionUiEvent) => void): void { this.close(); + this.sessionId = sessionId; this.onEvent = onEvent; - this.socket = sessionEvents(sessionId); - this.socket.onmessage = (message) => this.handleMessage(message.data); + this.shouldReconnect = true; + this.open(); } setHandler(onEvent: (event: SessionUiEvent) => void): void { @@ -23,36 +29,121 @@ export class SessionSocket { } close(): void { - this.socket?.close(); + this.shouldReconnect = false; + window.clearTimeout(this.reconnectTimer); + closeSocketQuietly(this.socket); this.socket = undefined; + this.sessionId = undefined; this.onEvent = undefined; } - private handleMessage(data: string): void { - const event = JSON.parse(data); + private open(): void { + if (!this.sessionId || !this.shouldReconnect) return; + const socket = sessionEvents(this.sessionId); + this.socket = socket; + socket.onopen = () => { + this.reconnectDelay = 500; + }; + socket.onmessage = (message) => void this.handleMessage(message.data); + socket.onerror = () => socket.close(); + socket.onclose = () => { + if (this.socket === socket) this.socket = undefined; + this.scheduleReconnect(); + }; + } + + private scheduleReconnect(): void { + if (!this.shouldReconnect) return; + window.clearTimeout(this.reconnectTimer); + const delay = this.reconnectDelay; + this.reconnectDelay = Math.min(this.reconnectDelay * 1.6, 5000); + this.reconnectTimer = window.setTimeout(() => this.open(), delay); + } + + private async handleMessage(data: MessageEvent["data"]): Promise { + const event = await parseSocketEvent(data); if (isSessionUiEvent(event)) this.onEvent?.(event); } } export class GlobalSessionSocket { private socket?: WebSocket; + private onEvent?: (event: Extract) => void; + private reconnectTimer?: number; + private reconnectDelay = 500; + private shouldReconnect = false; - connect(onEvent: (event: Extract) => void): void { + connect(onEvent: (event: Extract) => void): void { this.close(); - this.socket = globalSessionEvents(); - this.socket.onmessage = (message) => { - const event = JSON.parse(message.data); - if (event?.type === "status.update") onEvent(event); - }; + this.onEvent = onEvent; + this.shouldReconnect = true; + this.open(); } close(): void { - this.socket?.close(); + this.shouldReconnect = false; + window.clearTimeout(this.reconnectTimer); + closeSocketQuietly(this.socket); this.socket = undefined; + this.onEvent = undefined; + } + + private open(): void { + if (!this.shouldReconnect) return; + const socket = globalSessionEvents(); + this.socket = socket; + socket.onopen = () => { + this.reconnectDelay = 500; + }; + socket.onmessage = (message) => void this.handleMessage(message.data); + socket.onerror = () => socket.close(); + socket.onclose = () => { + if (this.socket === socket) this.socket = undefined; + this.scheduleReconnect(); + }; + } + + private scheduleReconnect(): void { + if (!this.shouldReconnect) return; + window.clearTimeout(this.reconnectTimer); + const delay = this.reconnectDelay; + this.reconnectDelay = Math.min(this.reconnectDelay * 1.6, 5000); + this.reconnectTimer = window.setTimeout(() => this.open(), delay); + } + + private async handleMessage(data: MessageEvent["data"]): Promise { + const event = await parseSocketEvent(data); + if (isGlobalSessionEvent(event)) this.onEvent?.(event); } } function isSessionUiEvent(event: any): event is SessionUiEvent { - return ["assistant.delta", "tool.start", "tool.end", "status.update", "session.error"].includes(event?.type); + return ["assistant.delta", "tool.start", "tool.end", "status.update", "activity.update", "session.error"].includes(event?.type); } +function isGlobalSessionEvent(event: unknown): event is Extract { + return typeof event === "object" && event !== null && ("type" in event) && ((event as any).type === "status.update" || (event as any).type === "activity.update"); +} + +async function parseSocketEvent(data: MessageEvent["data"]): Promise { + try { + if (typeof data === "string") return JSON.parse(data); + if (data instanceof Blob) return JSON.parse(await data.text()); + if (data instanceof ArrayBuffer) return JSON.parse(new TextDecoder().decode(data)); + return undefined; + } catch { + return undefined; + } +} + +function closeSocketQuietly(socket: WebSocket | undefined): void { + if (!socket) return; + socket.onmessage = null; + socket.onerror = null; + socket.onclose = null; + if (socket.readyState === WebSocket.CONNECTING) { + socket.onopen = () => socket.close(); + return; + } + socket.close(); +} diff --git a/src/server/sessions/piSessionService.ts b/src/server/sessions/piSessionService.ts index 9298d45..c39c446 100644 --- a/src/server/sessions/piSessionService.ts +++ b/src/server/sessions/piSessionService.ts @@ -18,6 +18,8 @@ import type { ActiveSession } from "./sessionRuntimeStore.js"; export class PiSessionService { private readonly active = new Map(); + private readonly activities = new Map(); + private readonly heartbeat: NodeJS.Timeout; private readonly commandService: SessionCommandService; private readonly agentDir = getAgentDir(); private readonly authStorage = AuthStorage.create(); @@ -29,6 +31,7 @@ export class PiSessionService { }; constructor(private readonly events: SessionEventHub) { + this.heartbeat = setInterval(() => this.publishHeartbeats(), 2000); this.commandService = new SessionCommandService( (sessionId) => this.getActive(sessionId), (sessionId, text) => this.prompt(sessionId, text), @@ -90,8 +93,11 @@ export class PiSessionService { async prompt(sessionId: string, text: string): Promise { const session = await this.getOrOpen(sessionId); + this.publishActivity(session, "prompt accepted", "active"); void session.prompt(text).catch((error) => { - this.events.publish(sessionId, { type: "session.error", message: error instanceof Error ? error.message : String(error) }); + const message = error instanceof Error ? error.message : String(error); + this.publishActivity(session, "error", "error", message); + this.events.publish(sessionId, { type: "session.error", message }); }); } @@ -114,6 +120,7 @@ export class PiSessionService { active.unsubscribe(); void active.runtime.session.abort().finally(() => active.runtime.dispose()); this.active.delete(sessionId); + this.activities.delete(sessionId); } private async getOrOpen(sessionId: string): Promise { @@ -147,11 +154,61 @@ export class PiSessionService { const { session } = active.runtime; active.unsubscribe = session.subscribe((event) => { this.events.publish(session.sessionId, toClientEvent(event)); + this.publishActivityForEvent(session, event); this.publishStatus(session); }); this.active.set(session.sessionId, active); } + private publishHeartbeats(): void { + for (const active of this.active.values()) { + const { session } = active.runtime; + const activity = this.activities.get(session.sessionId); + const isActive = session.isStreaming || session.isBashRunning || session.isCompacting || session.pendingMessageCount > 0 || activity?.phase === "active"; + if (!isActive) continue; + this.publishStatus(session); + if (activity) this.publishActivity(session, activity.label, "active", activity.detail); + else this.publishActivity(session, this.activityLabelFromStatus(session), "active"); + } + } + + private activityLabelFromStatus(session: AgentSession): string { + if (session.isCompacting) return "compacting"; + if (session.isBashRunning) return "running bash"; + if (session.isStreaming) return "agent running"; + if (session.pendingMessageCount) return "queued"; + return "active"; + } + + private publishActivityForEvent(session: AgentSession, event: any): void { + if (event.type === "agent_start") return this.publishActivity(session, "agent running", "active"); + if (event.type === "agent_end") { + this.publishActivity(session, "idle", "idle"); + setTimeout(() => { + this.publishActivity(session, "idle", "idle"); + this.publishStatus(session); + }, 250); + return; + } + if (event.type === "turn_end") return this.publishActivity(session, "turn complete", "active"); + if (event.type === "message_start") return this.publishActivity(session, "message started", "active"); + if (event.type === "message_end") return this.publishActivity(session, "message complete", "idle"); + if (event.type === "message_update") return this.publishActivity(session, "receiving response", "active"); + if (event.type === "tool_execution_start") return this.publishActivity(session, "running tool", "active", event.toolName); + if (event.type === "tool_execution_end") return this.publishActivity(session, event.isError ? "tool failed" : "tool complete", event.isError ? "error" : "active", event.toolName); + if (event.type === "bash_execution_start") return this.publishActivity(session, "running bash", "active"); + if (event.type === "bash_execution_end") return this.publishActivity(session, "bash complete", "active"); + this.publishActivity(session, event.type.replaceAll("_", " "), "active"); + } + + private publishActivity(session: AgentSession, label: string, phase: "active" | "idle" | "error", detail?: string): void { + const at = new Date().toISOString(); + this.activities.set(session.sessionId, { phase, label, detail, at }); + const activity = { sessionId: session.sessionId, phase, label, detail, at }; + this.events.publish(session.sessionId, { type: "activity.update", activity }); + this.events.publishGlobal({ type: "activity.update", activity }); + } + private publishStatus(session: AgentSession): void { const status = this.statusFromSession(session); this.events.publish(session.sessionId, { type: "status.update", status }); diff --git a/vite.config.ts b/vite.config.ts index 5dc767f..31b2c01 100644 --- a/vite.config.ts +++ b/vite.config.ts @@ -10,7 +10,7 @@ export default defineConfig({ port: 5173, strictPort: true, proxy: { - "/api": "http://localhost:3000", + "/api": { target: "http://localhost:3000", ws: true }, }, }, });