Stop the UI freeze: coalesce snapshot storms and bound chat rendering

Busy agents emitted a snapshot-changed broadcast per tool turn; every event
made the renderer reload the full snapshot while the main process re-read
whole audit tables synchronously, and ChatTimeline re-rendered every message
with per-row linear scans - quadratic work per keystroke of agent activity
until clicks stopped responding. Root fixes at each layer: all broadcasts
funnel through one 200ms-coalesced notifier (unit-tested), the renderer
reloads at most every 800ms with a trailing load, audit/access snapshot
reads use SQL limits instead of full-table scans, and the timeline renders
the latest 150 messages via O(n) lookup maps with a real show-earlier
control. cred.txt is now gitignored so the Composio key cannot leak.
This commit is contained in:
ZacharyZhang-NY committed 2026-07-06 22:40:32 -04:00
1 parent 538bf20057
commit f13ef48a3e
8 files changed
+158 -25

No files matched your search

+1
View File
@@ -6,3 +6,4 @@ playwright-report/
coverage/ coverage/
*.tsbuildinfo *.tsbuildinfo
design-source/ design-source/
cred.txt
+16
View File
@@ -0,0 +1,16 @@
/**
* Wraps a notification callback so bursts collapse into one trailing call.
* Agent run loops emit change events per tool turn; without coalescing every
* emission forces the renderer through a full snapshot reload, which starves
* the UI thread once the database grows.
*/
export function coalesce(fn: () => void, delayMs: number): () => void {
let timer: NodeJS.Timeout | null = null;
return () => {
if (timer) return;
timer = setTimeout(() => {
timer = null;
fn();
}, delayMs);
};
}
+13 -12
View File
@@ -27,6 +27,7 @@ import { KnowledgeService } from "../services/knowledge-service";
import { AutomationService } from "../services/automation-service"; import { AutomationService } from "../services/automation-service";
import { MeetingService } from "../services/meeting-service"; import { MeetingService } from "../services/meeting-service";
import { SprintService } from "../services/sprint-service"; import { SprintService } from "../services/sprint-service";
import { coalesce } from "./coalesce";
import { normalizeExternalBrowserUrl } from "./external-url"; import { normalizeExternalBrowserUrl } from "./external-url";
export interface IpcDeps { export interface IpcDeps {
@@ -43,13 +44,13 @@ export function registerIpcHandlers(deps: IpcDeps): void {
const providers = new ProviderService(deps.connection.db); const providers = new ProviderService(deps.connection.db);
const providerOAuth = new ProviderOAuthService(); const providerOAuth = new ProviderOAuthService();
const skills = new SkillService(deps.connection.db, deps.artifactRoot); const skills = new SkillService(deps.connection.db, deps.artifactRoot);
const notifyDataChanged = () => { const notifyDataChanged = coalesce(() => {
try { try {
if (!deps.mainWindow.isDestroyed() && !deps.mainWindow.webContents.isDestroyed()) { if (!deps.mainWindow.isDestroyed() && !deps.mainWindow.webContents.isDestroyed()) {
deps.mainWindow.webContents.send(IPC_CHANNELS.rendererEvent, { type: "snapshot-changed" }); deps.mainWindow.webContents.send(IPC_CHANNELS.rendererEvent, { type: "snapshot-changed" });
} }
} catch { /* renderer may have been disposed during agent run */ } } catch { /* renderer may have been disposed during agent run */ }
}; }, 200);
const taskExecution = new TaskExecutionService(deps.connection.db, providers, skills, notifyDataChanged); const taskExecution = new TaskExecutionService(deps.connection.db, providers, skills, notifyDataChanged);
const social = new SocialService(deps.connection.db); const social = new SocialService(deps.connection.db);
const files = new FileService(deps.connection.db, deps.artifactRoot); const files = new FileService(deps.connection.db, deps.artifactRoot);
@@ -74,7 +75,7 @@ export function registerIpcHandlers(deps: IpcDeps): void {
const mutate = <T>(operation: () => T): DesktopResult<AppSnapshot> => { const mutate = <T>(operation: () => T): DesktopResult<AppSnapshot> => {
try { try {
deps.connection.transaction(operation); deps.connection.transaction(operation);
deps.mainWindow.webContents.send(IPC_CHANNELS.rendererEvent, { type: "snapshot-changed" }); notifyDataChanged();
return ok(snapshot.load()); return ok(snapshot.load());
} catch (error) { } catch (error) {
return fail("operation_failed", error instanceof Error ? error.message : String(error)); return fail("operation_failed", error instanceof Error ? error.message : String(error));
@@ -95,7 +96,7 @@ export function registerIpcHandlers(deps: IpcDeps): void {
deps.connection.transaction(() => { deps.connection.transaction(() => {
summary = onboarding.bootstrap(input); summary = onboarding.bootstrap(input);
}); });
deps.mainWindow.webContents.send(IPC_CHANNELS.rendererEvent, { type: "snapshot-changed" }); notifyDataChanged();
return ok(summary); return ok(summary);
} catch (error) { } catch (error) {
return fail("onboarding_failed", error instanceof Error ? error.message : String(error)); return fail("onboarding_failed", error instanceof Error ? error.message : String(error));
@@ -114,10 +115,10 @@ export function registerIpcHandlers(deps: IpcDeps): void {
ipcMain.handle(IPC_CHANNELS.runTask, async (_event, input) => { ipcMain.handle(IPC_CHANNELS.runTask, async (_event, input) => {
try { try {
await taskExecution.runTask(input); await taskExecution.runTask(input);
deps.mainWindow.webContents.send(IPC_CHANNELS.rendererEvent, { type: "snapshot-changed" }); notifyDataChanged();
return ok(snapshot.load()); return ok(snapshot.load());
} catch (error) { } catch (error) {
deps.mainWindow.webContents.send(IPC_CHANNELS.rendererEvent, { type: "snapshot-changed" }); notifyDataChanged();
return fail("task_run_failed", error instanceof Error ? error.message : String(error)); return fail("task_run_failed", error instanceof Error ? error.message : String(error));
} }
}); });
@@ -151,7 +152,7 @@ export function registerIpcHandlers(deps: IpcDeps): void {
ipcMain.handle(IPC_CHANNELS.checkProviderHealth, async (_event, input) => { ipcMain.handle(IPC_CHANNELS.checkProviderHealth, async (_event, input) => {
try { try {
await providers.checkHealth(input.orgId, input.providerId); await providers.checkHealth(input.orgId, input.providerId);
deps.mainWindow.webContents.send(IPC_CHANNELS.rendererEvent, { type: "snapshot-changed" }); notifyDataChanged();
return ok(snapshot.load()); return ok(snapshot.load());
} catch (error) { } catch (error) {
return fail("provider_health_failed", error instanceof Error ? error.message : String(error)); return fail("provider_health_failed", error instanceof Error ? error.message : String(error));
@@ -196,7 +197,7 @@ export function registerIpcHandlers(deps: IpcDeps): void {
employeeId: input.employeeId, employeeId: input.employeeId,
}); });
emit({ phase: "active" }); emit({ phase: "active" });
deps.mainWindow.webContents.send(IPC_CHANNELS.rendererEvent, { type: "snapshot-changed" }); notifyDataChanged();
return ok({ secret: null }); return ok({ secret: null });
} }
emit({ phase: "active" }); emit({ phase: "active" });
@@ -222,7 +223,7 @@ export function registerIpcHandlers(deps: IpcDeps): void {
ipcMain.handle(IPC_CHANNELS.createProviderResponse, async (_event, input) => { ipcMain.handle(IPC_CHANNELS.createProviderResponse, async (_event, input) => {
try { try {
await providers.createResponse(input); await providers.createResponse(input);
deps.mainWindow.webContents.send(IPC_CHANNELS.rendererEvent, { type: "snapshot-changed" }); notifyDataChanged();
return ok(snapshot.load()); return ok(snapshot.load());
} catch (error) { } catch (error) {
return fail("provider_response_failed", error instanceof Error ? error.message : String(error)); return fail("provider_response_failed", error instanceof Error ? error.message : String(error));
@@ -233,7 +234,7 @@ export function registerIpcHandlers(deps: IpcDeps): void {
await providers.streamResponse(input, (streamEvent) => { await providers.streamResponse(input, (streamEvent) => {
deps.mainWindow.webContents.send(IPC_CHANNELS.rendererEvent, { type: "provider-response-stream", event: streamEvent }); deps.mainWindow.webContents.send(IPC_CHANNELS.rendererEvent, { type: "provider-response-stream", event: streamEvent });
}); });
deps.mainWindow.webContents.send(IPC_CHANNELS.rendererEvent, { type: "snapshot-changed" }); notifyDataChanged();
return ok(snapshot.load()); return ok(snapshot.load());
} catch (error) { } catch (error) {
return fail("provider_stream_failed", error instanceof Error ? error.message : String(error)); return fail("provider_stream_failed", error instanceof Error ? error.message : String(error));
@@ -242,7 +243,7 @@ export function registerIpcHandlers(deps: IpcDeps): void {
ipcMain.handle(IPC_CHANNELS.cancelProviderResponse, async (_event, input) => { ipcMain.handle(IPC_CHANNELS.cancelProviderResponse, async (_event, input) => {
try { try {
await providers.cancelResponse(input); await providers.cancelResponse(input);
deps.mainWindow.webContents.send(IPC_CHANNELS.rendererEvent, { type: "snapshot-changed" }); notifyDataChanged();
return ok(snapshot.load()); return ok(snapshot.load());
} catch (error) { } catch (error) {
return fail("provider_cancel_failed", error instanceof Error ? error.message : String(error)); return fail("provider_cancel_failed", error instanceof Error ? error.message : String(error));
@@ -269,7 +270,7 @@ export function registerIpcHandlers(deps: IpcDeps): void {
ipcMain.handle(IPC_CHANNELS.importFile, async (_event, input) => { ipcMain.handle(IPC_CHANNELS.importFile, async (_event, input) => {
try { try {
const fileId = await files.importFile(input); const fileId = await files.importFile(input);
if (fileId) deps.mainWindow.webContents.send(IPC_CHANNELS.rendererEvent, { type: "snapshot-changed" }); if (fileId) notifyDataChanged();
return ok(fileId); return ok(fileId);
} catch (error) { } catch (error) {
return fail("file_import_failed", error instanceof Error ? error.message : String(error)); return fail("file_import_failed", error instanceof Error ? error.message : String(error));
@@ -142,7 +142,7 @@ export class SnapshotService {
employeeIdentities, employeeIdentities,
serviceIdentities: list(this.db, "service_identities", "created_at desc"), serviceIdentities: list(this.db, "service_identities", "created_at desc"),
roleGrants: list(this.db, "role_grants", "created_at desc"), roleGrants: list(this.db, "role_grants", "created_at desc"),
accessDecisions: list<AppSnapshot["accessDecisions"][number]>(this.db, "access_decisions", "created_at desc").slice(0, 200), accessDecisions: recentList<AppSnapshot["accessDecisions"][number]>(this.db, "access_decisions", "created_at desc", 200),
orgDepartments: list(this.db, "org_departments", "name asc"), orgDepartments: list(this.db, "org_departments", "name asc"),
hiringPolicies: list(this.db, "hiring_policies", "department_name asc"), hiringPolicies: list(this.db, "hiring_policies", "department_name asc"),
approvalRoutingRules: list(this.db, "approval_routing_rules", "created_at asc"), approvalRoutingRules: list(this.db, "approval_routing_rules", "created_at asc"),
@@ -159,7 +159,7 @@ export class SnapshotService {
evalRuns: list(this.db, "eval_runs", "created_at desc"), evalRuns: list(this.db, "eval_runs", "created_at desc"),
prBotGates, prBotGates,
auditReplayRuns: list(this.db, "audit_replay_runs", "created_at desc"), auditReplayRuns: list(this.db, "audit_replay_runs", "created_at desc"),
auditEvents: list<AppSnapshot["auditEvents"][number]>(this.db, "audit_events", "created_at desc").slice(0, 200), auditEvents: recentList<AppSnapshot["auditEvents"][number]>(this.db, "audit_events", "created_at desc", 200),
documents: list(this.db, "documents", "updated_at desc"), documents: list(this.db, "documents", "updated_at desc"),
knowledgeEntries: list(this.db, "knowledge_entries", "updated_at desc"), knowledgeEntries: list(this.db, "knowledge_entries", "updated_at desc"),
automationRules: list<AppSnapshot["automationRules"][number]>(this.db, "automation_rules", "priority asc") automationRules: list<AppSnapshot["automationRules"][number]>(this.db, "automation_rules", "priority asc")
@@ -29,10 +29,32 @@ export function useSnapshot(): {
}, []); }, []);
useEffect(() => { useEffect(() => {
void Promise.resolve().then(load); let disposed = false;
return window.agentCompany.subscribe((event) => { let timer: ReturnType<typeof setTimeout> | null = null;
if (event.type === "snapshot-changed") void load(); let lastLoadAt = 0;
// Change events arrive in bursts while agents run; reload at most every
// 800ms with a trailing load so the UI converges without render storms.
const scheduleReload = () => {
if (disposed || timer) return;
const wait = Math.max(0, 800 - (Date.now() - lastLoadAt));
timer = setTimeout(() => {
timer = null;
lastLoadAt = Date.now();
void load();
}, wait);
};
void Promise.resolve().then(() => {
lastLoadAt = Date.now();
return load();
}); });
const unsubscribe = window.agentCompany.subscribe((event) => {
if (event.type === "snapshot-changed") scheduleReload();
});
return () => {
disposed = true;
if (timer) clearTimeout(timer);
unsubscribe();
};
}, [load]); }, [load]);
const actions = useMemo<SnapshotActions>(() => ({ run, reload: load }), [run, load]); const actions = useMemo<SnapshotActions>(() => ({ run, reload: load }), [run, load]);
@@ -72,6 +72,7 @@ export function ChatConversation({
<div style={{ flex: 1, overflow: "auto", padding: "20px 24px 8px" }}> <div style={{ flex: 1, overflow: "auto", padding: "20px 24px 8px" }}>
{space && ( {space && (
<ChatTimeline <ChatTimeline
key={space.id}
messagesInSpace={filtered} messagesInSpace={filtered}
threadsInSpace={threadsInSpace} threadsInSpace={threadsInSpace}
pendingApprovals={pendingApprovals} pendingApprovals={pendingApprovals}
@@ -1,4 +1,4 @@
import type { ReactNode } from "react"; import { useState, type ReactNode } from "react";
import { Button, Icon, Pill } from "../../shared/ui"; import { Button, Icon, Pill } from "../../shared/ui";
import type { import type {
ApprovalRecord, ApprovalRecord,
@@ -12,6 +12,22 @@ import type {
import { ChatMessageRow } from "./ChatMessageRow"; import { ChatMessageRow } from "./ChatMessageRow";
import { formatDay, type MessageContext } from "./chat-helpers"; import { formatDay, type MessageContext } from "./chat-helpers";
const WINDOW_SIZE = 150;
const EMPTY_ATTACHMENTS: ChatAttachmentRecord[] = [];
const EMPTY_MENTIONS: ChatMentionRecord[] = [];
const EMPTY_INVOCATIONS: ChatSlashInvocationRecord[] = [];
function groupBy<T>(items: T[], keyOf: (item: T) => string): Map<string, T[]> {
const groups = new Map<string, T[]>();
for (const item of items) {
const key = keyOf(item);
const group = groups.get(key);
if (group) group.push(item);
else groups.set(key, [item]);
}
return groups;
}
export function ChatTimeline({ export function ChatTimeline({
messagesInSpace, messagesInSpace,
threadsInSpace, threadsInSpace,
@@ -45,25 +61,52 @@ export function ChatTimeline({
onDeleteMessage: (messageId: string) => void; onDeleteMessage: (messageId: string) => void;
onOpenFile: (fileId: string) => void; onOpenFile: (fileId: string) => void;
}) { }) {
const elements: ReactNode[] = []; const [visibleCount, setVisibleCount] = useState(WINDOW_SIZE);
let lastDay = "";
const attachmentsByMessage = groupBy(attachments, (item) => item.messageId);
const mentionsByMessage = groupBy(mentions, (item) => item.messageId);
const invocationsByMessage = groupBy(slashInvocations, (item) => item.messageId);
const threadByRoot = new Map<string, ChatThreadRecord>();
for (const thread of threadsInSpace) {
if (thread.rootMessageId) threadByRoot.set(thread.rootMessageId, thread);
}
const replyCountByThread = new Map<string, number>();
for (const message of messagesInSpace) { for (const message of messagesInSpace) {
if (!message.threadId) continue;
replyCountByThread.set(message.threadId, (replyCountByThread.get(message.threadId) ?? 0) + 1);
}
const hiddenCount = Math.max(0, messagesInSpace.length - visibleCount);
const visible = hiddenCount > 0 ? messagesInSpace.slice(hiddenCount) : messagesInSpace;
const elements: ReactNode[] = [];
if (hiddenCount > 0) {
elements.push(
<div key="show-earlier" style={{ display: "flex", justifyContent: "center", marginBottom: 16 }}>
<Button size="small" onClick={() => setVisibleCount((count) => count + WINDOW_SIZE)}>
Show earlier messages
</Button>
</div>,
);
}
let lastDay = "";
for (const message of visible) {
const day = formatDay(message.createdAt); const day = formatDay(message.createdAt);
if (day !== lastDay) { if (day !== lastDay) {
lastDay = day; lastDay = day;
elements.push(<DayDivider key={`day-${day}`} label={day} />); elements.push(<DayDivider key={`day-${day}`} label={day} />);
} }
const thread = threadsInSpace.find((item) => item.rootMessageId === message.id) ?? null; const thread = threadByRoot.get(message.id) ?? null;
const replyCount = thread const replyCount = thread
? messagesInSpace.filter((m) => m.threadId === thread.id && m.id !== thread.rootMessageId).length ? Math.max(0, (replyCountByThread.get(thread.id) ?? 0) - (thread.rootMessageId && message.threadId === thread.id ? 1 : 0))
: 0; : 0;
elements.push( elements.push(
<ChatMessageRow <ChatMessageRow
key={message.id} key={message.id}
message={message} message={message}
attachments={attachments} attachments={attachmentsByMessage.get(message.id) ?? EMPTY_ATTACHMENTS}
mentions={mentions} mentions={mentionsByMessage.get(message.id) ?? EMPTY_MENTIONS}
slashInvocations={slashInvocations} slashInvocations={invocationsByMessage.get(message.id) ?? EMPTY_INVOCATIONS}
files={files} files={files}
replyCount={replyCount} replyCount={replyCount}
onOpenThread={thread ? () => onOpenThread(thread.id) : null} onOpenThread={thread ? () => onOpenThread(thread.id) : null}
@@ -0,0 +1,49 @@
import { afterEach, beforeEach, describe, expect, test, vi } from "vitest";
import { coalesce } from "../../src/main/app/coalesce";
describe("coalesce", () => {
beforeEach(() => {
vi.useFakeTimers();
});
afterEach(() => {
vi.useRealTimers();
});
test("collapses a burst into one trailing call", () => {
const calls: number[] = [];
const notify = coalesce(() => calls.push(Date.now()), 200);
notify();
notify();
notify();
expect(calls).toHaveLength(0);
vi.advanceTimersByTime(199);
expect(calls).toHaveLength(0);
vi.advanceTimersByTime(1);
expect(calls).toHaveLength(1);
});
test("fires again for events after the window closes", () => {
let count = 0;
const notify = coalesce(() => { count += 1; }, 200);
notify();
vi.advanceTimersByTime(200);
expect(count).toBe(1);
notify();
notify();
vi.advanceTimersByTime(200);
expect(count).toBe(2);
});
test("an event during the pending window is covered by the trailing call", () => {
let count = 0;
const notify = coalesce(() => { count += 1; }, 200);
notify();
vi.advanceTimersByTime(100);
notify();
vi.advanceTimersByTime(100);
expect(count).toBe(1);
vi.advanceTimersByTime(200);
expect(count).toBe(1);
});
});