479 lines
16 KiB
TypeScript
479 lines
16 KiB
TypeScript
import {
|
|
extractImages,
|
|
extractText,
|
|
extractTextSegments,
|
|
extractToolUse,
|
|
isGeneratingStatusNarration,
|
|
isInternalAssistantReplyText,
|
|
isInternalProcessNarration,
|
|
} from './message-utils';
|
|
import { isInternalMessage } from '@/stores/chat/helpers';
|
|
import type { RawMessage, ToolStatus } from '@/stores/chat';
|
|
|
|
export type TaskStepStatus = 'running' | 'completed' | 'error';
|
|
|
|
export interface TaskStep {
|
|
id: string;
|
|
label: string;
|
|
status: TaskStepStatus;
|
|
kind: 'thinking' | 'tool' | 'system' | 'message';
|
|
detail?: string;
|
|
depth: number;
|
|
parentId?: string;
|
|
/** Extracted URL for web_fetch tool, used to render a clickable link icon. */
|
|
url?: string;
|
|
}
|
|
|
|
/**
|
|
* Detects the index of the "final reply" assistant message in a run segment.
|
|
*
|
|
* The reply is the last assistant message that carries non-empty text
|
|
* content, regardless of whether it ALSO carries tool calls. (Mixed
|
|
* `text + toolCall` replies are rare but real — the model can emit a parting
|
|
* text block alongside a final tool call. Treating such a message as the
|
|
* reply avoids mis-protecting an earlier narration as the "answer" and
|
|
* leaking the actual last text into the fold.)
|
|
*
|
|
* When this returns a non-negative index, the caller should avoid folding
|
|
* that message's text into the graph (it is the answer the user sees in the
|
|
* chat stream). When the run is still active (streaming) the final reply is
|
|
* produced via `streamingMessage` instead, so callers pass
|
|
* `hasStreamingReply = true` to skip protection and let every assistant-with-
|
|
* text message in history be folded into the graph as narration.
|
|
*/
|
|
export function findReplyMessageIndex(messages: RawMessage[], hasStreamingReply: boolean): number {
|
|
if (hasStreamingReply) return -1;
|
|
for (let idx = messages.length - 1; idx >= 0; idx -= 1) {
|
|
const message = messages[idx];
|
|
if (!message || message.role !== 'assistant') continue;
|
|
const replyText = extractText(message).trim();
|
|
if (messageHasUserVisibleImage(message)) return idx;
|
|
if (replyText.length === 0 || isInternalAssistantReplyText(replyText)) continue;
|
|
if (isGeneratingStatusNarration(replyText)) continue;
|
|
return idx;
|
|
}
|
|
return -1;
|
|
}
|
|
|
|
function messageHasUserVisibleImage(message: RawMessage): boolean {
|
|
if ((message._attachedFiles ?? []).some((file) => file.mimeType.startsWith('image/'))) {
|
|
return true;
|
|
}
|
|
return extractImages(message).length > 0;
|
|
}
|
|
|
|
/**
|
|
* When true, assistant history in the run segment should be folded into the
|
|
* execution graph because the live answer is (or will be) shown via streaming.
|
|
* When false but the run is still open, a final reply already in history must
|
|
* stay visible in the chat stream (history poll can beat stream teardown).
|
|
*/
|
|
export function hasActiveStreamingReplyInRun(
|
|
isLatestOpenRun: boolean,
|
|
hasAnyStreamContent: boolean,
|
|
streamingReplyText: string | null,
|
|
): boolean {
|
|
return isLatestOpenRun && (hasAnyStreamContent || streamingReplyText != null);
|
|
}
|
|
|
|
/**
|
|
* Message indices that belong to an agent run segment (strictly after a run
|
|
* trigger user message up to the next real user message). Used to fold tool
|
|
* cards and process attachments into ExecutionGraphCard without depending on
|
|
* whether a graph card was successfully materialized (e.g. after history
|
|
* reload when the step cache is empty).
|
|
*/
|
|
export function buildRunSegmentMessageIndices(
|
|
messages: RawMessage[],
|
|
nextUserMessageIndexes: number[],
|
|
isRunTrigger: (message: RawMessage, index: number) => boolean,
|
|
): Set<number> {
|
|
const indices = new Set<number>();
|
|
messages.forEach((message, triggerIndex) => {
|
|
if (!isRunTrigger(message, triggerIndex)) return;
|
|
const nextUserIndex = nextUserMessageIndexes[triggerIndex];
|
|
const segmentEnd = nextUserIndex === -1 ? messages.length - 1 : nextUserIndex - 1;
|
|
for (let idx = triggerIndex + 1; idx <= segmentEnd; idx += 1) {
|
|
indices.add(idx);
|
|
}
|
|
});
|
|
|
|
// History pagination loads a suffix of the transcript. When the triggering
|
|
// user turn fell off the window, assistant tool steps remain at the top of
|
|
// `messages[]` without a preceding user row — fold them into the first run.
|
|
let firstTriggerIndex = -1;
|
|
for (let idx = 0; idx < messages.length; idx += 1) {
|
|
if (isRunTrigger(messages[idx], idx)) {
|
|
firstTriggerIndex = idx;
|
|
break;
|
|
}
|
|
}
|
|
if (firstTriggerIndex > 0) {
|
|
for (let idx = 0; idx < firstTriggerIndex; idx += 1) {
|
|
if (messages[idx]?.role === 'assistant') {
|
|
indices.add(idx);
|
|
}
|
|
}
|
|
}
|
|
|
|
return indices;
|
|
}
|
|
|
|
/**
|
|
* Messages strictly after the triggering user turn up to the next user.
|
|
* Use this for run lifecycle (final reply detection, reply index, open-run
|
|
* state) — never count paginated orphan assistants from a prior turn.
|
|
*/
|
|
export function getPostTriggerSegmentMessages(
|
|
messages: RawMessage[],
|
|
triggerIndex: number,
|
|
nextUserIndex: number,
|
|
): RawMessage[] {
|
|
const segmentEnd = nextUserIndex === -1 ? messages.length : nextUserIndex;
|
|
return messages.slice(triggerIndex + 1, segmentEnd);
|
|
}
|
|
|
|
/**
|
|
* Slice messages for a user-triggered run, including leading assistant orphans
|
|
* that belong to the same run but were separated by paginated history.
|
|
*/
|
|
export function getRunSegmentMessages(
|
|
messages: RawMessage[],
|
|
triggerIndex: number,
|
|
nextUserIndex: number,
|
|
isRunTrigger: (message: RawMessage, index: number) => boolean,
|
|
): RawMessage[] {
|
|
const segmentEnd = nextUserIndex === -1 ? messages.length : nextUserIndex;
|
|
const core = messages.slice(triggerIndex + 1, segmentEnd);
|
|
const hasEarlierUser = messages.some((message, index) => index < triggerIndex && isRunTrigger(message, index));
|
|
if (hasEarlierUser || triggerIndex === 0) return core;
|
|
const orphans = messages.slice(0, triggerIndex).filter((message) => message.role === 'assistant');
|
|
return [...orphans, ...core];
|
|
}
|
|
|
|
/**
|
|
* True when a run segment already contains a conclusive assistant reply: the
|
|
* last assistant message with user-visible text that appears after all tool
|
|
* calls (if any). Intermediate narration before tools does not count.
|
|
*/
|
|
export function segmentHasFinalReply(segmentMessages: RawMessage[]): boolean {
|
|
let lastToolUseOffset = -1;
|
|
for (let i = segmentMessages.length - 1; i >= 0; i -= 1) {
|
|
const message = segmentMessages[i];
|
|
if (message.role === 'assistant' && extractToolUse(message).length > 0) {
|
|
lastToolUseOffset = i;
|
|
break;
|
|
}
|
|
}
|
|
return segmentMessages.some((message, index) => {
|
|
if (index <= lastToolUseOffset) return false;
|
|
if (message.role !== 'assistant') return false;
|
|
if (messageHasUserVisibleImage(message)) return true;
|
|
const replyText = extractText(message).trim();
|
|
if (replyText.length === 0 || isInternalAssistantReplyText(replyText)) return false;
|
|
if (isGeneratingStatusNarration(replyText)) return false;
|
|
const content = message.content;
|
|
if (!Array.isArray(content)) return true;
|
|
return !(content as Array<{ type?: string }>).some(
|
|
(block) => block.type === 'tool_use' || block.type === 'toolCall',
|
|
);
|
|
});
|
|
}
|
|
|
|
interface DeriveTaskStepsInput {
|
|
messages: RawMessage[];
|
|
streamingMessage: unknown | null;
|
|
streamingTools: ToolStatus[];
|
|
omitLastStreamingMessageSegment?: boolean;
|
|
}
|
|
|
|
export interface SubagentCompletionInfo {
|
|
sessionKey: string;
|
|
sessionId: string;
|
|
agentId: string;
|
|
}
|
|
|
|
function normalizeText(text: string | null | undefined): string | undefined {
|
|
if (!text) return undefined;
|
|
const normalized = text.replace(/[ \t]+/g, ' ').trim();
|
|
if (!normalized) return undefined;
|
|
return normalized;
|
|
}
|
|
|
|
function makeToolId(prefix: string, name: string, index: number): string {
|
|
return `${prefix}:${name}:${index}`;
|
|
}
|
|
|
|
export function parseAgentIdFromSessionKey(sessionKey: string): string | null {
|
|
const parts = sessionKey.split(':');
|
|
if (parts.length < 2 || parts[0] !== 'agent') return null;
|
|
return parts[1] || null;
|
|
}
|
|
|
|
export function parseSubagentCompletionInfo(message: RawMessage): SubagentCompletionInfo | null {
|
|
const text = typeof message.content === 'string'
|
|
? message.content
|
|
: Array.isArray(message.content)
|
|
? message.content.map((block) => ('text' in block && typeof block.text === 'string' ? block.text : '')).join('\n')
|
|
: '';
|
|
if (!text.includes('[Internal task completion event]')) return null;
|
|
|
|
const sessionKeyMatch = text.match(/session_key:\s*(.+)/);
|
|
const sessionIdMatch = text.match(/session_id:\s*(.+)/);
|
|
const sessionKey = sessionKeyMatch?.[1]?.trim();
|
|
const sessionId = sessionIdMatch?.[1]?.trim();
|
|
if (!sessionKey || !sessionId) return null;
|
|
const agentId = parseAgentIdFromSessionKey(sessionKey);
|
|
if (!agentId) return null;
|
|
return { sessionKey, sessionId, agentId };
|
|
}
|
|
|
|
function isSpawnLikeStep(label: string): boolean {
|
|
return /(spawn|subagent|delegate|parallel)/i.test(label);
|
|
}
|
|
|
|
function tryParseJsonObject(detail: string | undefined): Record<string, unknown> | null {
|
|
if (!detail) return null;
|
|
try {
|
|
const parsed = JSON.parse(detail) as unknown;
|
|
return parsed && typeof parsed === 'object' ? parsed as Record<string, unknown> : null;
|
|
} catch {
|
|
return null;
|
|
}
|
|
}
|
|
|
|
function extractBranchAgent(step: TaskStep): string | null {
|
|
const parsed = tryParseJsonObject(step.detail);
|
|
const agentId = parsed?.agentId;
|
|
if (typeof agentId === 'string' && agentId.trim()) return agentId.trim();
|
|
|
|
const message = typeof parsed?.message === 'string' ? parsed.message : step.detail;
|
|
if (!message) return null;
|
|
const match = message.match(/\b(coder|reviewer|project-manager|manager|planner|researcher|worker|subagent)\b/i);
|
|
return match ? match[1] : null;
|
|
}
|
|
|
|
function attachTopology(steps: TaskStep[]): TaskStep[] {
|
|
const withTopology: TaskStep[] = [];
|
|
let activeBranchNodeId: string | null = null;
|
|
|
|
for (const step of steps) {
|
|
if (step.kind === 'system') {
|
|
activeBranchNodeId = null;
|
|
withTopology.push({ ...step, depth: 1, parentId: 'agent-run' });
|
|
continue;
|
|
}
|
|
|
|
if (/sessions_spawn/i.test(step.label)) {
|
|
const branchAgent = extractBranchAgent(step) || 'subagent';
|
|
const branchNodeId = `${step.id}:branch`;
|
|
withTopology.push({ ...step, depth: 1, parentId: 'agent-run' });
|
|
withTopology.push({
|
|
id: branchNodeId,
|
|
label: `${branchAgent} run`,
|
|
status: step.status,
|
|
kind: 'system',
|
|
detail: `Spawned branch for ${branchAgent}`,
|
|
depth: 2,
|
|
parentId: step.id,
|
|
});
|
|
activeBranchNodeId = branchNodeId;
|
|
continue;
|
|
}
|
|
|
|
if (/sessions_yield/i.test(step.label)) {
|
|
withTopology.push({
|
|
...step,
|
|
depth: activeBranchNodeId ? 3 : 1,
|
|
parentId: activeBranchNodeId ?? 'agent-run',
|
|
});
|
|
activeBranchNodeId = null;
|
|
continue;
|
|
}
|
|
|
|
if (step.kind === 'thinking' || step.kind === 'message') {
|
|
withTopology.push({
|
|
...step,
|
|
depth: activeBranchNodeId ? 3 : 1,
|
|
parentId: activeBranchNodeId ?? 'agent-run',
|
|
});
|
|
continue;
|
|
}
|
|
|
|
if (isSpawnLikeStep(step.label)) {
|
|
activeBranchNodeId = step.id;
|
|
withTopology.push({
|
|
...step,
|
|
depth: 1,
|
|
parentId: 'agent-run',
|
|
});
|
|
continue;
|
|
}
|
|
|
|
withTopology.push({
|
|
...step,
|
|
depth: activeBranchNodeId ? 3 : 1,
|
|
parentId: activeBranchNodeId ?? 'agent-run',
|
|
});
|
|
}
|
|
|
|
return withTopology;
|
|
}
|
|
|
|
function appendDetailSegments(
|
|
segments: string[],
|
|
options: {
|
|
idPrefix: string;
|
|
label: string;
|
|
kind: Extract<TaskStep['kind'], 'thinking' | 'message'>;
|
|
running: boolean;
|
|
upsertStep: (step: TaskStep) => void;
|
|
},
|
|
): void {
|
|
const normalizedSegments = segments
|
|
.map((segment) => normalizeText(segment))
|
|
.filter((segment): segment is string => !!segment)
|
|
.filter((segment) => !isInternalProcessNarration(segment));
|
|
|
|
normalizedSegments.forEach((detail, index) => {
|
|
options.upsertStep({
|
|
id: `${options.idPrefix}-${index}`,
|
|
label: options.label,
|
|
status: options.running && index === normalizedSegments.length - 1 ? 'running' : 'completed',
|
|
kind: options.kind,
|
|
detail,
|
|
depth: 1,
|
|
});
|
|
});
|
|
}
|
|
|
|
export function deriveTaskSteps({
|
|
messages,
|
|
streamingMessage,
|
|
streamingTools,
|
|
omitLastStreamingMessageSegment = false,
|
|
}: DeriveTaskStepsInput): TaskStep[] {
|
|
const steps: TaskStep[] = [];
|
|
const stepIndexById = new Map<string, number>();
|
|
|
|
const upsertStep = (step: TaskStep): void => {
|
|
const existingIndex = stepIndexById.get(step.id);
|
|
if (existingIndex == null) {
|
|
stepIndexById.set(step.id, steps.length);
|
|
steps.push(step);
|
|
return;
|
|
}
|
|
const existing = steps[existingIndex];
|
|
steps[existingIndex] = {
|
|
...existing,
|
|
...step,
|
|
detail: step.detail ?? existing.detail,
|
|
};
|
|
};
|
|
|
|
const streamMessage = streamingMessage && typeof streamingMessage === 'object'
|
|
? streamingMessage as RawMessage
|
|
: null;
|
|
|
|
// The final answer the user sees as a chat bubble. We avoid folding it into
|
|
// the graph to prevent duplication. When a run is still streaming, the
|
|
// reply lives in `streamingMessage`, so every pure-text assistant message in
|
|
// `messages` is treated as intermediate narration.
|
|
const replyIndex = findReplyMessageIndex(messages, streamMessage != null);
|
|
|
|
for (const [messageIndex, message] of messages.entries()) {
|
|
if (!message || message.role !== 'assistant') continue;
|
|
|
|
const toolUses = extractToolUse(message);
|
|
if (!isInternalMessage(message)) {
|
|
// Fold any intermediate assistant text into the graph as a narration
|
|
// step — including text that lives on a mixed `text + toolCall` message.
|
|
// The narration step is emitted BEFORE the tool steps so the graph
|
|
// preserves the original ordering (the assistant "thinks out loud" and
|
|
// then invokes the tool).
|
|
const narrationSegments = extractTextSegments(message);
|
|
const graphNarrationSegments = messageIndex === replyIndex
|
|
? narrationSegments.slice(0, -1)
|
|
: narrationSegments;
|
|
appendDetailSegments(graphNarrationSegments, {
|
|
idPrefix: `history-message-${message.id || messageIndex}`,
|
|
label: 'Message',
|
|
kind: 'message',
|
|
running: false,
|
|
upsertStep,
|
|
});
|
|
} else if (toolUses.length === 0) {
|
|
continue;
|
|
}
|
|
|
|
toolUses.forEach((tool, index) => {
|
|
const input = tool.input as Record<string, unknown>;
|
|
const url = tool.name === 'web_fetch' && typeof input?.url === 'string' ? input.url : undefined;
|
|
upsertStep({
|
|
id: tool.id || makeToolId(`history-tool-${message.id || messageIndex}`, tool.name, index),
|
|
label: tool.name,
|
|
status: 'completed',
|
|
kind: 'tool',
|
|
detail: normalizeText(JSON.stringify(tool.input, null, 2)),
|
|
depth: 1,
|
|
url,
|
|
});
|
|
});
|
|
}
|
|
|
|
if (streamMessage) {
|
|
// Stream-time narration should also appear in the execution graph so that
|
|
// intermediate process output stays in P1 instead of leaking into the
|
|
// assistant reply area.
|
|
const streamNarrationSegments = extractTextSegments(streamMessage);
|
|
const graphStreamNarrationSegments = omitLastStreamingMessageSegment
|
|
? streamNarrationSegments.slice(0, -1)
|
|
: streamNarrationSegments;
|
|
appendDetailSegments(graphStreamNarrationSegments, {
|
|
idPrefix: 'stream-message',
|
|
label: 'Message',
|
|
kind: 'message',
|
|
running: !omitLastStreamingMessageSegment,
|
|
upsertStep,
|
|
});
|
|
}
|
|
|
|
const activeToolIds = new Set<string>();
|
|
const activeToolNamesWithoutIds = new Set<string>();
|
|
streamingTools.forEach((tool, index) => {
|
|
const id = tool.toolCallId || tool.id || makeToolId('stream-status', tool.name, index);
|
|
activeToolIds.add(id);
|
|
if (!tool.toolCallId && !tool.id) {
|
|
activeToolNamesWithoutIds.add(tool.name);
|
|
}
|
|
upsertStep({
|
|
id,
|
|
label: tool.name,
|
|
status: tool.status,
|
|
kind: 'tool',
|
|
detail: normalizeText(tool.summary),
|
|
depth: 1,
|
|
});
|
|
});
|
|
|
|
if (streamMessage) {
|
|
extractToolUse(streamMessage).forEach((tool, index) => {
|
|
const id = tool.id || makeToolId('stream-tool', tool.name, index);
|
|
if (activeToolIds.has(id) || activeToolNamesWithoutIds.has(tool.name)) return;
|
|
const input = tool.input as Record<string, unknown>;
|
|
const url = tool.name === 'web_fetch' && typeof input?.url === 'string' ? input.url : undefined;
|
|
upsertStep({
|
|
id,
|
|
label: tool.name,
|
|
status: 'running',
|
|
kind: 'tool',
|
|
detail: normalizeText(JSON.stringify(tool.input, null, 2)),
|
|
depth: 1,
|
|
url,
|
|
});
|
|
});
|
|
}
|
|
|
|
return attachTopology(steps);
|
|
}
|