mirror of
https://github.com/block/buzz.git
synced 2026-11-12 20:31:12 +00:00
fix(desktop): resolve O(n²) render cascade in activity sidebar (#555)
Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -5,18 +5,38 @@ import type { RelayEvent, ManagedAgent } from "@/shared/api/types";
|
||||
import { getIdentity } from "@/shared/api/tauri";
|
||||
import { decryptObserverEvent } from "@/shared/api/tauriObserver";
|
||||
import { normalizePubkey } from "@/shared/lib/pubkey";
|
||||
import type { ConnectionState, ObserverEvent } from "./ui/agentSessionTypes";
|
||||
import type {
|
||||
ConnectionState,
|
||||
ObserverEvent,
|
||||
TranscriptItem,
|
||||
} from "./ui/agentSessionTypes";
|
||||
import {
|
||||
type TranscriptState,
|
||||
buildTranscriptState,
|
||||
createEmptyTranscriptState,
|
||||
processTranscriptEvent,
|
||||
} from "./ui/agentSessionTranscript";
|
||||
|
||||
const MAX_OBSERVER_EVENTS = 800;
|
||||
|
||||
type ObserverSnapshot = {
|
||||
export type ObserverSnapshot = {
|
||||
connectionState: ConnectionState;
|
||||
errorMessage: string | null;
|
||||
events: ObserverEvent[];
|
||||
};
|
||||
|
||||
const IDLE_SNAPSHOT: ObserverSnapshot = {
|
||||
connectionState: "idle",
|
||||
errorMessage: null,
|
||||
events: [],
|
||||
};
|
||||
|
||||
const EMPTY_TRANSCRIPT: TranscriptItem[] = [];
|
||||
|
||||
const listeners = new Set<() => void>();
|
||||
const eventsByAgent = new Map<string, ObserverEvent[]>();
|
||||
const transcriptByAgent = new Map<string, TranscriptState>();
|
||||
const snapshotByAgent = new Map<string, ObserverSnapshot>();
|
||||
|
||||
// Normalized pubkeys of agents we are actively managing. Only events whose
|
||||
// "agent" tag matches an entry here will be decrypted (defense-in-depth).
|
||||
@@ -35,12 +55,18 @@ function notifyListeners() {
|
||||
}
|
||||
}
|
||||
|
||||
function invalidateSnapshot(key: string) {
|
||||
snapshotByAgent.delete(key);
|
||||
}
|
||||
|
||||
function setConnectionState(
|
||||
nextState: ConnectionState,
|
||||
nextErrorMessage: string | null = errorMessage,
|
||||
) {
|
||||
connectionState = nextState;
|
||||
errorMessage = nextErrorMessage;
|
||||
// Invalidate all cached snapshots since connectionState changed
|
||||
snapshotByAgent.clear();
|
||||
notifyListeners();
|
||||
}
|
||||
|
||||
@@ -60,13 +86,32 @@ function appendAgentEvent(agentPubkey: string, event: ObserverEvent) {
|
||||
return;
|
||||
}
|
||||
|
||||
const next = [...current, event].sort(compareObserverEvents);
|
||||
eventsByAgent.set(
|
||||
key,
|
||||
next.length > MAX_OBSERVER_EVENTS
|
||||
? next.slice(next.length - MAX_OBSERVER_EVENTS)
|
||||
: next,
|
||||
);
|
||||
const sorted = [...current, event].sort(compareObserverEvents);
|
||||
const trimmed = sorted.length > MAX_OBSERVER_EVENTS;
|
||||
const final = trimmed
|
||||
? sorted.slice(sorted.length - MAX_OBSERVER_EVENTS)
|
||||
: sorted;
|
||||
eventsByAgent.set(key, final);
|
||||
|
||||
// Determine whether the new event landed at the end of the sorted array.
|
||||
// If it did (common case), we can incrementally process just this event.
|
||||
// If not (out-of-order arrival) or if we trimmed, fall back to full rebuild.
|
||||
const eventAtEnd = sorted[sorted.length - 1] === event;
|
||||
|
||||
if (eventAtEnd && !trimmed) {
|
||||
// Fast path: incremental update
|
||||
const transcriptState =
|
||||
transcriptByAgent.get(key) ?? createEmptyTranscriptState();
|
||||
const updatedTranscript = processTranscriptEvent(transcriptState, event);
|
||||
transcriptByAgent.set(key, updatedTranscript);
|
||||
} else {
|
||||
// Slow path: full rebuild (out-of-order insertion or trim fired)
|
||||
transcriptByAgent.set(key, buildTranscriptState(final));
|
||||
}
|
||||
|
||||
// Invalidate cached snapshot for this agent
|
||||
invalidateSnapshot(key);
|
||||
|
||||
notifyListeners();
|
||||
}
|
||||
|
||||
@@ -188,13 +233,40 @@ export function subscribeAgentObserverStore(listener: () => void) {
|
||||
}
|
||||
|
||||
export function getAgentObserverSnapshot(
|
||||
agentPubkey: string,
|
||||
agentPubkey?: string | null,
|
||||
enabled?: boolean,
|
||||
): ObserverSnapshot {
|
||||
return {
|
||||
if (!enabled || !agentPubkey) {
|
||||
return IDLE_SNAPSHOT;
|
||||
}
|
||||
const key = normalizePubkey(agentPubkey);
|
||||
const cached = snapshotByAgent.get(key);
|
||||
if (
|
||||
cached &&
|
||||
cached.connectionState === connectionState &&
|
||||
cached.errorMessage === errorMessage
|
||||
) {
|
||||
return cached;
|
||||
}
|
||||
const snapshot: ObserverSnapshot = {
|
||||
connectionState,
|
||||
errorMessage,
|
||||
events: eventsByAgent.get(normalizePubkey(agentPubkey)) ?? [],
|
||||
events: eventsByAgent.get(key) ?? [],
|
||||
};
|
||||
snapshotByAgent.set(key, snapshot);
|
||||
return snapshot;
|
||||
}
|
||||
|
||||
export function getAgentTranscript(
|
||||
agentPubkey?: string | null,
|
||||
enabled?: boolean,
|
||||
): TranscriptItem[] {
|
||||
if (!enabled || !agentPubkey) {
|
||||
return EMPTY_TRANSCRIPT;
|
||||
}
|
||||
const key = normalizePubkey(agentPubkey);
|
||||
const state = transcriptByAgent.get(key);
|
||||
return state?.items ?? EMPTY_TRANSCRIPT;
|
||||
}
|
||||
|
||||
export function useManagedAgentObserverBridge(agents: readonly ManagedAgent[]) {
|
||||
@@ -229,6 +301,8 @@ export function resetAgentObserverStore() {
|
||||
startPromise = null;
|
||||
eventProcessingQueue = Promise.resolve();
|
||||
eventsByAgent.clear();
|
||||
transcriptByAgent.clear();
|
||||
snapshotByAgent.clear();
|
||||
knownAgentPubkeys.clear();
|
||||
connectionState = "idle";
|
||||
errorMessage = null;
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import * as React from "react";
|
||||
import { Bot, Brain, ChevronDown, Radio, TerminalSquare } from "lucide-react";
|
||||
|
||||
import { cn } from "@/shared/lib/cn";
|
||||
@@ -42,7 +43,7 @@ export function AgentSessionTranscriptList({
|
||||
);
|
||||
}
|
||||
|
||||
function TranscriptItemView({
|
||||
const TranscriptItemView = React.memo(function TranscriptItemView({
|
||||
agentName,
|
||||
item,
|
||||
}: {
|
||||
@@ -62,7 +63,7 @@ function TranscriptItemView({
|
||||
return <MetadataItem item={item} />;
|
||||
}
|
||||
return <LifecycleItem item={item} />;
|
||||
}
|
||||
});
|
||||
|
||||
function MessageItem({
|
||||
agentName,
|
||||
|
||||
@@ -19,9 +19,8 @@ import type {
|
||||
ObserverEvent,
|
||||
TranscriptItem,
|
||||
} from "./agentSessionTypes";
|
||||
import { buildTranscript } from "./agentSessionTranscript";
|
||||
import { shorten } from "./agentSessionUtils";
|
||||
import { useObserverEvents } from "./useObserverEvents";
|
||||
import { useObserverEvents, useAgentTranscript } from "./useObserverEvents";
|
||||
|
||||
type ManagedAgentSessionPanelProps = {
|
||||
agent: ManagedAgent;
|
||||
@@ -40,10 +39,23 @@ export function ManagedAgentSessionPanel({
|
||||
showHeader = true,
|
||||
showRaw = true,
|
||||
}: ManagedAgentSessionPanelProps) {
|
||||
const isRunning = agent.status === "running";
|
||||
const { connectionState, errorMessage, events } = useObserverEvents(
|
||||
agent.status === "running",
|
||||
isRunning,
|
||||
agent.pubkey,
|
||||
);
|
||||
const transcript = useAgentTranscript(isRunning, agent.pubkey);
|
||||
|
||||
// Filter transcript items by channelId (lightweight — items now carry channelId)
|
||||
const scopedTranscript = React.useMemo(
|
||||
() =>
|
||||
channelId
|
||||
? transcript.filter((item) => item.channelId === channelId)
|
||||
: transcript,
|
||||
[channelId, transcript],
|
||||
);
|
||||
|
||||
// Filter raw events by channelId for the RawEventRail
|
||||
const scopedEvents = React.useMemo(
|
||||
() =>
|
||||
channelId
|
||||
@@ -51,15 +63,14 @@ export function ManagedAgentSessionPanel({
|
||||
: events,
|
||||
[channelId, events],
|
||||
);
|
||||
const transcript = React.useMemo(
|
||||
() => buildTranscript(scopedEvents),
|
||||
[scopedEvents],
|
||||
);
|
||||
const latestSessionId = React.useMemo(
|
||||
() =>
|
||||
[...scopedEvents].reverse().find((event) => event.sessionId)?.sessionId,
|
||||
[scopedEvents],
|
||||
);
|
||||
|
||||
// Derive latestSessionId from channel-scoped events
|
||||
const latestSessionId = React.useMemo(() => {
|
||||
for (let i = scopedEvents.length - 1; i >= 0; i--) {
|
||||
if (scopedEvents[i].sessionId) return scopedEvents[i].sessionId;
|
||||
}
|
||||
return null;
|
||||
}, [scopedEvents]);
|
||||
|
||||
return (
|
||||
<section
|
||||
@@ -72,7 +83,7 @@ export function ManagedAgentSessionPanel({
|
||||
<SessionHeader
|
||||
connectionState={connectionState}
|
||||
eventCount={scopedEvents.length}
|
||||
hasObserver={agent.status === "running"}
|
||||
hasObserver={isRunning}
|
||||
latestSessionId={latestSessionId}
|
||||
/>
|
||||
) : null}
|
||||
@@ -83,9 +94,9 @@ export function ManagedAgentSessionPanel({
|
||||
emptyDescription={emptyDescription}
|
||||
errorMessage={errorMessage}
|
||||
events={scopedEvents}
|
||||
hasObserver={agent.status === "running"}
|
||||
hasObserver={isRunning}
|
||||
showRaw={showRaw}
|
||||
transcript={transcript}
|
||||
transcript={scopedTranscript}
|
||||
/>
|
||||
</section>
|
||||
);
|
||||
|
||||
@@ -24,194 +24,269 @@ import {
|
||||
|
||||
export { describeRawEvent } from "./agentSessionTranscriptHelpers";
|
||||
|
||||
export function buildTranscript(events: ObserverEvent[]): TranscriptItem[] {
|
||||
const items: TranscriptItem[] = [];
|
||||
const itemsById = new Map<string, TranscriptItem>();
|
||||
export type TranscriptState = {
|
||||
items: TranscriptItem[];
|
||||
itemsById: Map<string, TranscriptItem>;
|
||||
activeMessageKey: Map<string, string>;
|
||||
sealedKeys: Set<string>;
|
||||
continuationSeq: number;
|
||||
latestSessionId: string | null;
|
||||
};
|
||||
|
||||
// Maps a logical message ID (e.g. `assistant:msg1`) to the *actual* key
|
||||
// currently being appended to. When a non-message item interleaves, we
|
||||
// seal the current key so subsequent chunks create a fresh entry.
|
||||
const activeMessageKey = new Map<string, string>();
|
||||
const sealedKeys = new Set<string>();
|
||||
let continuationSeq = 0;
|
||||
export function createEmptyTranscriptState(): TranscriptState {
|
||||
return {
|
||||
items: [],
|
||||
itemsById: new Map(),
|
||||
activeMessageKey: new Map(),
|
||||
sealedKeys: new Set(),
|
||||
continuationSeq: 0,
|
||||
latestSessionId: null,
|
||||
};
|
||||
}
|
||||
|
||||
/** Seal every currently-open message so the next chunk starts a new entry. */
|
||||
function sealOpenMessages() {
|
||||
for (const [, currentKey] of activeMessageKey) {
|
||||
if (!sealedKeys.has(currentKey)) {
|
||||
sealedKeys.add(currentKey);
|
||||
/**
|
||||
* Mutable draft that collects changes during a single processTranscriptEvent
|
||||
* call. Replaces the previous pattern of nested closures capturing bare `let`
|
||||
* bindings — all mutation now targets this explicit object.
|
||||
*/
|
||||
type TranscriptDraft = {
|
||||
items: TranscriptItem[];
|
||||
itemsById: Map<string, TranscriptItem>;
|
||||
activeMessageKey: Map<string, string>;
|
||||
sealedKeys: Set<string>;
|
||||
continuationSeq: number;
|
||||
latestSessionId: string | null;
|
||||
changed: boolean;
|
||||
};
|
||||
|
||||
function draftFrom(state: TranscriptState): TranscriptDraft {
|
||||
return {
|
||||
items: state.items,
|
||||
itemsById: state.itemsById,
|
||||
activeMessageKey: state.activeMessageKey,
|
||||
sealedKeys: state.sealedKeys,
|
||||
continuationSeq: state.continuationSeq,
|
||||
latestSessionId: state.latestSessionId,
|
||||
changed: false,
|
||||
};
|
||||
}
|
||||
|
||||
/** Lazily copy items + itemsById on first mutation so callers get new refs. */
|
||||
function ensureMutable(d: TranscriptDraft) {
|
||||
if (!d.changed) {
|
||||
d.items = [...d.items];
|
||||
d.itemsById = new Map(d.itemsById);
|
||||
d.changed = true;
|
||||
}
|
||||
}
|
||||
|
||||
function replaceItem(d: TranscriptDraft, id: string, updated: TranscriptItem) {
|
||||
ensureMutable(d);
|
||||
const idx = d.items.findIndex((it) => it.id === id);
|
||||
if (idx !== -1) {
|
||||
d.items[idx] = updated;
|
||||
}
|
||||
d.itemsById.set(id, updated);
|
||||
}
|
||||
|
||||
function pushItem(d: TranscriptDraft, item: TranscriptItem) {
|
||||
ensureMutable(d);
|
||||
d.items.push(item);
|
||||
d.itemsById.set(item.id, item);
|
||||
}
|
||||
|
||||
function sealOpenMessages(d: TranscriptDraft) {
|
||||
let copied = false;
|
||||
for (const [, currentKey] of d.activeMessageKey) {
|
||||
if (!d.sealedKeys.has(currentKey)) {
|
||||
if (!copied) {
|
||||
d.sealedKeys = new Set(d.sealedKeys);
|
||||
copied = true;
|
||||
}
|
||||
d.sealedKeys.add(currentKey);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function upsertMessage(
|
||||
id: string,
|
||||
role: "assistant" | "user",
|
||||
title: string,
|
||||
text: string,
|
||||
timestamp: string,
|
||||
) {
|
||||
const currentKey = activeMessageKey.get(id);
|
||||
function upsertMessage(
|
||||
d: TranscriptDraft,
|
||||
id: string,
|
||||
role: "assistant" | "user",
|
||||
title: string,
|
||||
text: string,
|
||||
timestamp: string,
|
||||
channelId: string | null,
|
||||
) {
|
||||
const currentKey = d.activeMessageKey.get(id);
|
||||
|
||||
// If there is an active (non-sealed) key, append to it.
|
||||
if (currentKey && !sealedKeys.has(currentKey)) {
|
||||
const existing = itemsById.get(currentKey);
|
||||
if (existing?.type === "message") {
|
||||
existing.text += text;
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
// Otherwise create a new entry (either first time, or continuation).
|
||||
continuationSeq += 1;
|
||||
const newKey = currentKey ? `${id}:c${continuationSeq}` : id;
|
||||
const item: TranscriptItem = {
|
||||
id: newKey,
|
||||
type: "message",
|
||||
role,
|
||||
title,
|
||||
text,
|
||||
timestamp,
|
||||
};
|
||||
items.push(item);
|
||||
itemsById.set(newKey, item);
|
||||
activeMessageKey.set(id, newKey);
|
||||
}
|
||||
|
||||
function upsertTextItem(
|
||||
id: string,
|
||||
type: "thought" | "lifecycle",
|
||||
title: string,
|
||||
text: string,
|
||||
timestamp: string,
|
||||
) {
|
||||
const existing = itemsById.get(id);
|
||||
if (existing && existing.type === type) {
|
||||
existing.text += text;
|
||||
if (currentKey && !d.sealedKeys.has(currentKey)) {
|
||||
const existing = d.itemsById.get(currentKey);
|
||||
if (existing?.type === "message") {
|
||||
replaceItem(d, currentKey, {
|
||||
...existing,
|
||||
text: existing.text + text,
|
||||
channelId,
|
||||
});
|
||||
return;
|
||||
}
|
||||
sealOpenMessages();
|
||||
const item: TranscriptItem = { id, type, title, text, timestamp };
|
||||
items.push(item);
|
||||
itemsById.set(id, item);
|
||||
}
|
||||
|
||||
function upsertMetadata(
|
||||
id: string,
|
||||
title: string,
|
||||
sections: PromptSection[],
|
||||
timestamp: string,
|
||||
) {
|
||||
const existing = itemsById.get(id);
|
||||
if (existing?.type === "metadata") {
|
||||
existing.sections = sections;
|
||||
return;
|
||||
d.continuationSeq += 1;
|
||||
const newKey = currentKey ? `${id}:c${d.continuationSeq}` : id;
|
||||
pushItem(d, {
|
||||
id: newKey,
|
||||
type: "message",
|
||||
role,
|
||||
title,
|
||||
text,
|
||||
timestamp,
|
||||
channelId,
|
||||
});
|
||||
d.activeMessageKey = new Map(d.activeMessageKey);
|
||||
d.activeMessageKey.set(id, newKey);
|
||||
}
|
||||
|
||||
function upsertTextItem(
|
||||
d: TranscriptDraft,
|
||||
id: string,
|
||||
type: "thought" | "lifecycle",
|
||||
title: string,
|
||||
text: string,
|
||||
timestamp: string,
|
||||
channelId: string | null,
|
||||
) {
|
||||
const existing = d.itemsById.get(id);
|
||||
if (existing && existing.type === type) {
|
||||
replaceItem(d, id, { ...existing, text: existing.text + text, channelId });
|
||||
return;
|
||||
}
|
||||
sealOpenMessages(d);
|
||||
pushItem(d, { id, type, title, text, timestamp, channelId });
|
||||
}
|
||||
|
||||
function upsertMetadata(
|
||||
d: TranscriptDraft,
|
||||
id: string,
|
||||
title: string,
|
||||
sections: PromptSection[],
|
||||
timestamp: string,
|
||||
channelId: string | null,
|
||||
) {
|
||||
const existing = d.itemsById.get(id);
|
||||
if (existing?.type === "metadata") {
|
||||
replaceItem(d, id, { ...existing, sections, channelId });
|
||||
return;
|
||||
}
|
||||
sealOpenMessages(d);
|
||||
pushItem(d, { id, type: "metadata", title, sections, timestamp, channelId });
|
||||
}
|
||||
|
||||
function upsertTool(
|
||||
d: TranscriptDraft,
|
||||
id: string,
|
||||
title: string,
|
||||
toolName: string,
|
||||
sproutToolName: string | null,
|
||||
status: ToolStatus,
|
||||
args: Record<string, unknown>,
|
||||
result: string,
|
||||
isError: boolean,
|
||||
timestamp: string,
|
||||
channelId: string | null,
|
||||
) {
|
||||
const existing = d.itemsById.get(id);
|
||||
const canonicalSproutToolName =
|
||||
sproutToolName ?? findSproutToolName(toolName, true);
|
||||
if (existing?.type === "tool") {
|
||||
const updatedTitle = !isGenericToolTitle(title) ? title : existing.title;
|
||||
let updatedToolName = existing.toolName;
|
||||
let updatedSproutToolName = existing.sproutToolName;
|
||||
if (canonicalSproutToolName) {
|
||||
updatedSproutToolName = canonicalSproutToolName;
|
||||
updatedToolName = canonicalSproutToolName;
|
||||
} else if (!existing.sproutToolName && !isGenericToolTitle(toolName)) {
|
||||
updatedToolName = toolName;
|
||||
}
|
||||
sealOpenMessages();
|
||||
const item: TranscriptItem = {
|
||||
id,
|
||||
type: "metadata",
|
||||
title,
|
||||
sections,
|
||||
timestamp,
|
||||
};
|
||||
items.push(item);
|
||||
itemsById.set(id, item);
|
||||
}
|
||||
|
||||
function upsertTool(
|
||||
id: string,
|
||||
title: string,
|
||||
toolName: string,
|
||||
sproutToolName: string | null,
|
||||
status: ToolStatus,
|
||||
args: Record<string, unknown>,
|
||||
result: string,
|
||||
isError: boolean,
|
||||
timestamp: string,
|
||||
) {
|
||||
const existing = itemsById.get(id);
|
||||
const canonicalSproutToolName =
|
||||
sproutToolName ?? findSproutToolName(toolName, true);
|
||||
if (existing?.type === "tool") {
|
||||
if (!isGenericToolTitle(title)) {
|
||||
existing.title = title;
|
||||
}
|
||||
if (canonicalSproutToolName) {
|
||||
existing.sproutToolName = canonicalSproutToolName;
|
||||
existing.toolName = canonicalSproutToolName;
|
||||
} else if (!existing.sproutToolName && !isGenericToolTitle(toolName)) {
|
||||
existing.toolName = toolName;
|
||||
}
|
||||
existing.status = status;
|
||||
existing.args = Object.keys(args).length > 0 ? args : existing.args;
|
||||
if (result) existing.result = result;
|
||||
existing.isError = isError || existing.isError;
|
||||
if (
|
||||
replaceItem(d, id, {
|
||||
...existing,
|
||||
title: updatedTitle,
|
||||
toolName: updatedToolName,
|
||||
sproutToolName: updatedSproutToolName,
|
||||
status,
|
||||
args: Object.keys(args).length > 0 ? args : existing.args,
|
||||
result: result || existing.result,
|
||||
isError: isError || existing.isError,
|
||||
completedAt:
|
||||
(status === "completed" || status === "failed") &&
|
||||
existing.completedAt == null
|
||||
) {
|
||||
existing.completedAt = timestamp;
|
||||
}
|
||||
return;
|
||||
}
|
||||
sealOpenMessages();
|
||||
const item: TranscriptItem = {
|
||||
id,
|
||||
type: "tool",
|
||||
title,
|
||||
toolName: canonicalSproutToolName ?? toolName,
|
||||
sproutToolName: canonicalSproutToolName,
|
||||
status,
|
||||
args,
|
||||
result,
|
||||
isError,
|
||||
timestamp,
|
||||
startedAt: timestamp,
|
||||
completedAt: null,
|
||||
};
|
||||
items.push(item);
|
||||
itemsById.set(id, item);
|
||||
? timestamp
|
||||
: existing.completedAt,
|
||||
channelId,
|
||||
});
|
||||
return;
|
||||
}
|
||||
sealOpenMessages(d);
|
||||
pushItem(d, {
|
||||
id,
|
||||
type: "tool",
|
||||
title,
|
||||
toolName: canonicalSproutToolName ?? toolName,
|
||||
sproutToolName: canonicalSproutToolName,
|
||||
status,
|
||||
args,
|
||||
result,
|
||||
isError,
|
||||
timestamp,
|
||||
startedAt: timestamp,
|
||||
completedAt: null,
|
||||
channelId,
|
||||
});
|
||||
}
|
||||
|
||||
export function processTranscriptEvent(
|
||||
state: TranscriptState,
|
||||
event: ObserverEvent,
|
||||
): TranscriptState {
|
||||
const d = draftFrom(state);
|
||||
|
||||
if (event.sessionId && event.sessionId !== d.latestSessionId) {
|
||||
d.latestSessionId = event.sessionId;
|
||||
}
|
||||
|
||||
for (const event of events) {
|
||||
if (event.kind === "turn_started") {
|
||||
upsertTextItem(
|
||||
`turn:${event.turnId ?? event.seq}`,
|
||||
"lifecycle",
|
||||
"Turn started",
|
||||
describeTurnStarted(event.payload),
|
||||
event.timestamp,
|
||||
);
|
||||
continue;
|
||||
}
|
||||
|
||||
if (event.kind === "session_resolved") {
|
||||
upsertTextItem(
|
||||
`session:${event.turnId ?? event.seq}`,
|
||||
"lifecycle",
|
||||
"Session ready",
|
||||
describeSessionResolved(event.payload),
|
||||
event.timestamp,
|
||||
);
|
||||
continue;
|
||||
}
|
||||
|
||||
if (event.kind === "acp_parse_error") {
|
||||
upsertTextItem(
|
||||
`parse-error:${event.seq}`,
|
||||
"lifecycle",
|
||||
"Wire parse error",
|
||||
extractBlockText(event.payload),
|
||||
event.timestamp,
|
||||
);
|
||||
continue;
|
||||
}
|
||||
|
||||
if (event.kind !== "acp_read" && event.kind !== "acp_write") {
|
||||
continue;
|
||||
}
|
||||
const channelId = event.channelId ?? null;
|
||||
const ch = channelId ?? "global";
|
||||
|
||||
if (event.kind === "turn_started") {
|
||||
upsertTextItem(
|
||||
d,
|
||||
`turn:${ch}:${event.turnId ?? event.seq}`,
|
||||
"lifecycle",
|
||||
"Turn started",
|
||||
describeTurnStarted(event.payload),
|
||||
event.timestamp,
|
||||
channelId,
|
||||
);
|
||||
} else if (event.kind === "session_resolved") {
|
||||
upsertTextItem(
|
||||
d,
|
||||
`session:${ch}:${event.turnId ?? event.seq}`,
|
||||
"lifecycle",
|
||||
"Session ready",
|
||||
describeSessionResolved(event.payload),
|
||||
event.timestamp,
|
||||
channelId,
|
||||
);
|
||||
} else if (event.kind === "acp_parse_error") {
|
||||
upsertTextItem(
|
||||
d,
|
||||
`parse-error:${ch}:${event.seq}`,
|
||||
"lifecycle",
|
||||
"Wire parse error",
|
||||
extractBlockText(event.payload),
|
||||
event.timestamp,
|
||||
channelId,
|
||||
);
|
||||
} else if (event.kind === "acp_read" || event.kind === "acp_write") {
|
||||
const payload = asRecord(event.payload);
|
||||
const method = asString(payload.method);
|
||||
|
||||
@@ -221,115 +296,134 @@ export function buildTranscript(events: ObserverEvent[]): TranscriptItem[] {
|
||||
const parsedPrompt = parsePromptText(promptText);
|
||||
if (parsedPrompt.userText) {
|
||||
upsertMessage(
|
||||
`prompt:${event.turnId ?? event.seq}`,
|
||||
d,
|
||||
`prompt:${ch}:${event.turnId ?? event.seq}`,
|
||||
"user",
|
||||
parsedPrompt.userTitle,
|
||||
parsedPrompt.userText,
|
||||
event.timestamp,
|
||||
channelId,
|
||||
);
|
||||
}
|
||||
if (parsedPrompt.sections.length > 0) {
|
||||
upsertMetadata(
|
||||
`prompt-context:${event.turnId ?? event.seq}`,
|
||||
d,
|
||||
`prompt-context:${ch}:${event.turnId ?? event.seq}`,
|
||||
"Prompt context",
|
||||
parsedPrompt.sections,
|
||||
event.timestamp,
|
||||
channelId,
|
||||
);
|
||||
}
|
||||
}
|
||||
continue;
|
||||
}
|
||||
} else if (event.kind === "acp_read" && method === "session/update") {
|
||||
const params = asRecord(payload.params);
|
||||
const update = asRecord(params.update);
|
||||
const updateType = asString(update.sessionUpdate) ?? "unknown";
|
||||
const turnKey = event.turnId ?? event.sessionId ?? "unknown";
|
||||
const messageId = asString(update.messageId);
|
||||
|
||||
if (event.kind !== "acp_read" || method !== "session/update") {
|
||||
continue;
|
||||
}
|
||||
|
||||
const params = asRecord(payload.params);
|
||||
const update = asRecord(params.update);
|
||||
const updateType = asString(update.sessionUpdate) ?? "unknown";
|
||||
const turnKey = event.turnId ?? event.sessionId ?? "unknown";
|
||||
const messageId = asString(update.messageId);
|
||||
|
||||
if (updateType === "agent_message_chunk") {
|
||||
upsertMessage(
|
||||
`assistant:${messageId ?? turnKey}`,
|
||||
"assistant",
|
||||
"Assistant",
|
||||
extractContentText(update.content),
|
||||
event.timestamp,
|
||||
);
|
||||
continue;
|
||||
}
|
||||
|
||||
if (updateType === "user_message_chunk") {
|
||||
upsertMessage(
|
||||
`user:${messageId ?? turnKey}`,
|
||||
"user",
|
||||
"User",
|
||||
extractContentText(update.content),
|
||||
event.timestamp,
|
||||
);
|
||||
continue;
|
||||
}
|
||||
|
||||
if (updateType === "agent_thought_chunk") {
|
||||
upsertTextItem(
|
||||
`thinking:${messageId ?? turnKey}`,
|
||||
"thought",
|
||||
"Thinking",
|
||||
extractContentText(update.content),
|
||||
event.timestamp,
|
||||
);
|
||||
continue;
|
||||
}
|
||||
|
||||
if (updateType === "tool_call") {
|
||||
const toolId = asString(update.toolCallId) ?? `tool:${event.seq}`;
|
||||
const identity = extractToolIdentity(update);
|
||||
upsertTool(
|
||||
`tool:${toolId}`,
|
||||
identity.title,
|
||||
identity.toolName,
|
||||
identity.sproutToolName,
|
||||
normalizeToolStatus(asString(update.status) ?? "executing"),
|
||||
extractToolArgs(update),
|
||||
extractToolResult(update),
|
||||
false,
|
||||
event.timestamp,
|
||||
);
|
||||
continue;
|
||||
}
|
||||
|
||||
if (updateType === "tool_call_update") {
|
||||
const toolId = asString(update.toolCallId) ?? `tool:${event.seq}`;
|
||||
const status = normalizeToolStatus(
|
||||
asString(update.status) ?? "completed",
|
||||
);
|
||||
const identity = extractToolIdentity(update);
|
||||
upsertTool(
|
||||
`tool:${toolId}`,
|
||||
identity.title,
|
||||
identity.toolName,
|
||||
identity.sproutToolName,
|
||||
status,
|
||||
extractToolArgs(update),
|
||||
extractToolResult(update),
|
||||
status === "failed",
|
||||
event.timestamp,
|
||||
);
|
||||
continue;
|
||||
}
|
||||
|
||||
if (updateType === "plan") {
|
||||
upsertTextItem(
|
||||
`plan:${turnKey}`,
|
||||
"thought",
|
||||
"Plan",
|
||||
extractContentText(update.content) || JSON.stringify(update, null, 2),
|
||||
event.timestamp,
|
||||
);
|
||||
if (updateType === "agent_message_chunk") {
|
||||
upsertMessage(
|
||||
d,
|
||||
`assistant:${ch}:${messageId ?? turnKey}`,
|
||||
"assistant",
|
||||
"Assistant",
|
||||
extractContentText(update.content),
|
||||
event.timestamp,
|
||||
channelId,
|
||||
);
|
||||
} else if (updateType === "user_message_chunk") {
|
||||
upsertMessage(
|
||||
d,
|
||||
`user:${ch}:${messageId ?? turnKey}`,
|
||||
"user",
|
||||
"User",
|
||||
extractContentText(update.content),
|
||||
event.timestamp,
|
||||
channelId,
|
||||
);
|
||||
} else if (updateType === "agent_thought_chunk") {
|
||||
upsertTextItem(
|
||||
d,
|
||||
`thinking:${ch}:${messageId ?? turnKey}`,
|
||||
"thought",
|
||||
"Thinking",
|
||||
extractContentText(update.content),
|
||||
event.timestamp,
|
||||
channelId,
|
||||
);
|
||||
} else if (updateType === "tool_call") {
|
||||
const toolId = asString(update.toolCallId) ?? `tool:${event.seq}`;
|
||||
const identity = extractToolIdentity(update);
|
||||
upsertTool(
|
||||
d,
|
||||
`tool:${ch}:${toolId}`,
|
||||
identity.title,
|
||||
identity.toolName,
|
||||
identity.sproutToolName,
|
||||
normalizeToolStatus(asString(update.status) ?? "executing"),
|
||||
extractToolArgs(update),
|
||||
extractToolResult(update),
|
||||
false,
|
||||
event.timestamp,
|
||||
channelId,
|
||||
);
|
||||
} else if (updateType === "tool_call_update") {
|
||||
const toolId = asString(update.toolCallId) ?? `tool:${event.seq}`;
|
||||
const status = normalizeToolStatus(
|
||||
asString(update.status) ?? "completed",
|
||||
);
|
||||
const identity = extractToolIdentity(update);
|
||||
upsertTool(
|
||||
d,
|
||||
`tool:${ch}:${toolId}`,
|
||||
identity.title,
|
||||
identity.toolName,
|
||||
identity.sproutToolName,
|
||||
status,
|
||||
extractToolArgs(update),
|
||||
extractToolResult(update),
|
||||
status === "failed",
|
||||
event.timestamp,
|
||||
channelId,
|
||||
);
|
||||
} else if (updateType === "plan") {
|
||||
upsertTextItem(
|
||||
d,
|
||||
`plan:${ch}:${turnKey}`,
|
||||
"thought",
|
||||
"Plan",
|
||||
extractContentText(update.content) || JSON.stringify(update, null, 2),
|
||||
event.timestamp,
|
||||
channelId,
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return items;
|
||||
if (!d.changed && d.latestSessionId === state.latestSessionId) {
|
||||
return state;
|
||||
}
|
||||
|
||||
return {
|
||||
items: d.items,
|
||||
itemsById: d.itemsById,
|
||||
activeMessageKey: d.activeMessageKey,
|
||||
sealedKeys: d.sealedKeys,
|
||||
continuationSeq: d.continuationSeq,
|
||||
latestSessionId: d.latestSessionId,
|
||||
};
|
||||
}
|
||||
|
||||
export function buildTranscriptState(events: ObserverEvent[]): TranscriptState {
|
||||
let state = createEmptyTranscriptState();
|
||||
for (const event of events) {
|
||||
state = processTranscriptEvent(state, event);
|
||||
}
|
||||
return state;
|
||||
}
|
||||
|
||||
export function buildTranscript(events: ObserverEvent[]): TranscriptItem[] {
|
||||
return buildTranscriptState(events).items;
|
||||
}
|
||||
|
||||
@@ -28,6 +28,7 @@ export type TranscriptItem =
|
||||
title: string;
|
||||
text: string;
|
||||
timestamp: string;
|
||||
channelId?: string | null;
|
||||
}
|
||||
| {
|
||||
id: string;
|
||||
@@ -35,6 +36,7 @@ export type TranscriptItem =
|
||||
title: string;
|
||||
text: string;
|
||||
timestamp: string;
|
||||
channelId?: string | null;
|
||||
}
|
||||
| {
|
||||
id: string;
|
||||
@@ -42,6 +44,7 @@ export type TranscriptItem =
|
||||
title: string;
|
||||
text: string;
|
||||
timestamp: string;
|
||||
channelId?: string | null;
|
||||
}
|
||||
| {
|
||||
id: string;
|
||||
@@ -49,6 +52,7 @@ export type TranscriptItem =
|
||||
title: string;
|
||||
sections: PromptSection[];
|
||||
timestamp: string;
|
||||
channelId?: string | null;
|
||||
}
|
||||
| {
|
||||
id: string;
|
||||
@@ -63,6 +67,7 @@ export type TranscriptItem =
|
||||
timestamp: string;
|
||||
startedAt: string;
|
||||
completedAt: string | null;
|
||||
channelId?: string | null;
|
||||
};
|
||||
|
||||
export type PromptSection = {
|
||||
|
||||
@@ -3,48 +3,45 @@ import * as React from "react";
|
||||
import {
|
||||
ensureRelayObserverSubscription,
|
||||
getAgentObserverSnapshot,
|
||||
getAgentTranscript,
|
||||
subscribeAgentObserverStore,
|
||||
} from "@/features/agents/observerRelayStore";
|
||||
import type { ConnectionState, ObserverEvent } from "./agentSessionTypes";
|
||||
import type { TranscriptItem } from "./agentSessionTypes";
|
||||
|
||||
// Stable subscribe reference shared by all useSyncExternalStore hooks.
|
||||
// subscribeAgentObserverStore already has a fixed identity, so this thin
|
||||
// wrapper satisfies React's requirement without per-hook useCallback.
|
||||
const subscribeToStore = (onStoreChange: () => void) =>
|
||||
subscribeAgentObserverStore(onStoreChange);
|
||||
|
||||
export function useObserverEvents(
|
||||
enabled: boolean,
|
||||
agentPubkey?: string | null,
|
||||
) {
|
||||
const [events, setEvents] = React.useState<ObserverEvent[]>([]);
|
||||
const [connectionState, setConnectionState] =
|
||||
React.useState<ConnectionState>("idle");
|
||||
const [errorMessage, setErrorMessage] = React.useState<string | null>(null);
|
||||
const getSnapshot = React.useCallback(
|
||||
() => getAgentObserverSnapshot(agentPubkey, enabled),
|
||||
[agentPubkey, enabled],
|
||||
);
|
||||
|
||||
const snapshot = React.useSyncExternalStore(subscribeToStore, getSnapshot);
|
||||
|
||||
React.useEffect(() => {
|
||||
setEvents([]);
|
||||
setErrorMessage(null);
|
||||
|
||||
if (!enabled) {
|
||||
setConnectionState("idle");
|
||||
return;
|
||||
if (enabled && agentPubkey) {
|
||||
void ensureRelayObserverSubscription();
|
||||
}
|
||||
}, [enabled, agentPubkey]);
|
||||
|
||||
if (!agentPubkey) {
|
||||
setConnectionState("idle");
|
||||
return;
|
||||
}
|
||||
|
||||
const syncSnapshot = () => {
|
||||
const snapshot = getAgentObserverSnapshot(agentPubkey);
|
||||
setConnectionState(snapshot.connectionState);
|
||||
setErrorMessage(snapshot.errorMessage);
|
||||
setEvents(snapshot.events);
|
||||
};
|
||||
|
||||
syncSnapshot();
|
||||
const unsubscribe = subscribeAgentObserverStore(syncSnapshot);
|
||||
void ensureRelayObserverSubscription();
|
||||
|
||||
return () => {
|
||||
unsubscribe();
|
||||
};
|
||||
}, [agentPubkey, enabled]);
|
||||
|
||||
return { connectionState, errorMessage, events };
|
||||
return snapshot;
|
||||
}
|
||||
|
||||
export function useAgentTranscript(
|
||||
enabled: boolean,
|
||||
agentPubkey?: string | null,
|
||||
): TranscriptItem[] {
|
||||
const getSnapshot = React.useCallback(
|
||||
() => getAgentTranscript(agentPubkey, enabled),
|
||||
[agentPubkey, enabled],
|
||||
);
|
||||
|
||||
return React.useSyncExternalStore(subscribeToStore, getSnapshot);
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user