feat(ui): add operator UI server, stores, and insights
This commit is contained in:
537
src/ui/session-insights.ts
Normal file
537
src/ui/session-insights.ts
Normal file
@@ -0,0 +1,537 @@
|
||||
import { readdir, readFile, stat } from "node:fs/promises";
|
||||
import { resolve } from "node:path";
|
||||
import type { PipelineEdge, PipelineNode, RouteCondition } from "../agents/manifest.js";
|
||||
import type { AgentManifest } from "../agents/manifest.js";
|
||||
import { isRecord, type JsonObject } from "../agents/types.js";
|
||||
import type { RuntimeEvent, RuntimeEventUsage } from "../telemetry/runtime-events.js";
|
||||
import { filterRuntimeEvents, readRuntimeEvents } from "./runtime-events-store.js";
|
||||
|
||||
export type SessionStatus = "success" | "failure" | "running" | "unknown";
|
||||
|
||||
export type SessionSummary = {
|
||||
sessionId: string;
|
||||
status: SessionStatus;
|
||||
startedAt?: string;
|
||||
endedAt?: string;
|
||||
nodeAttemptCount: number;
|
||||
distinctNodeCount: number;
|
||||
costUsd: number;
|
||||
durationMs: number;
|
||||
criticalEventCount: number;
|
||||
message?: string;
|
||||
aborted: boolean;
|
||||
};
|
||||
|
||||
export type NodeAttemptInsight = {
|
||||
timestamp: string;
|
||||
attempt: number;
|
||||
status: "success" | "validation_fail" | "failure";
|
||||
severity: RuntimeEvent["severity"];
|
||||
message: string;
|
||||
usage: RuntimeEventUsage;
|
||||
metadata: JsonObject;
|
||||
executionContext?: JsonObject;
|
||||
retrySpawned: boolean;
|
||||
securityViolation: boolean;
|
||||
subtasks: string[];
|
||||
};
|
||||
|
||||
export type NodeInsight = {
|
||||
nodeId: string;
|
||||
actorId: string;
|
||||
personaId: string;
|
||||
topology: string;
|
||||
fromNodeId?: string;
|
||||
attemptCount: number;
|
||||
lastStatus?: "success" | "validation_fail" | "failure";
|
||||
usage: RuntimeEventUsage;
|
||||
subtaskCount: number;
|
||||
securityViolationCount: number;
|
||||
attempts: NodeAttemptInsight[];
|
||||
domainEvents: RuntimeEvent[];
|
||||
sandboxPayload?: JsonObject;
|
||||
};
|
||||
|
||||
export type EdgeInsight = {
|
||||
from: string;
|
||||
to: string;
|
||||
trigger: string;
|
||||
conditionLabels: string[];
|
||||
visited: boolean;
|
||||
critical: boolean;
|
||||
};
|
||||
|
||||
export type SessionGraphInsight = {
|
||||
sessionId: string;
|
||||
status: SessionStatus;
|
||||
aborted: boolean;
|
||||
abortMessage?: string;
|
||||
nodes: NodeInsight[];
|
||||
edges: EdgeInsight[];
|
||||
runtimeEvents: RuntimeEvent[];
|
||||
criticalPathNodeIds: string[];
|
||||
};
|
||||
|
||||
type BuildSessionSummaryInput = {
|
||||
stateRoot: string;
|
||||
runtimeEventLogPath: string;
|
||||
};
|
||||
|
||||
type BuildSessionGraphInput = {
|
||||
stateRoot: string;
|
||||
runtimeEventLogPath: string;
|
||||
sessionId: string;
|
||||
manifest: AgentManifest;
|
||||
};
|
||||
|
||||
function toUsage(value: RuntimeEventUsage | undefined): RuntimeEventUsage {
|
||||
return {
|
||||
...(typeof value?.tokenInput === "number" ? { tokenInput: value.tokenInput } : {}),
|
||||
...(typeof value?.tokenOutput === "number" ? { tokenOutput: value.tokenOutput } : {}),
|
||||
...(typeof value?.tokenTotal === "number" ? { tokenTotal: value.tokenTotal } : {}),
|
||||
...(typeof value?.toolCalls === "number" ? { toolCalls: value.toolCalls } : {}),
|
||||
...(typeof value?.durationMs === "number" ? { durationMs: value.durationMs } : {}),
|
||||
...(typeof value?.costUsd === "number" ? { costUsd: value.costUsd } : {}),
|
||||
};
|
||||
}
|
||||
|
||||
function addUsage(target: RuntimeEventUsage, source: RuntimeEventUsage): RuntimeEventUsage {
|
||||
return {
|
||||
tokenInput: (target.tokenInput ?? 0) + (source.tokenInput ?? 0),
|
||||
tokenOutput: (target.tokenOutput ?? 0) + (source.tokenOutput ?? 0),
|
||||
tokenTotal: (target.tokenTotal ?? 0) + (source.tokenTotal ?? 0),
|
||||
toolCalls: (target.toolCalls ?? 0) + (source.toolCalls ?? 0),
|
||||
durationMs: (target.durationMs ?? 0) + (source.durationMs ?? 0),
|
||||
costUsd: (target.costUsd ?? 0) + (source.costUsd ?? 0),
|
||||
};
|
||||
}
|
||||
|
||||
function toConditionLabel(condition: RouteCondition): string {
|
||||
if (condition.type === "always") {
|
||||
return "always";
|
||||
}
|
||||
if (condition.type === "state_flag") {
|
||||
return `state_flag:${condition.key}=${String(condition.equals)}`;
|
||||
}
|
||||
if (condition.type === "history_has_event") {
|
||||
return `history_has_event:${condition.event}`;
|
||||
}
|
||||
return `file_exists:${condition.path}`;
|
||||
}
|
||||
|
||||
function toEdgeTrigger(edge: PipelineEdge): string {
|
||||
if (edge.event) {
|
||||
return `event:${edge.event}`;
|
||||
}
|
||||
return `on:${edge.on ?? "unknown"}`;
|
||||
}
|
||||
|
||||
function toStatusFromAttemptEvent(event: RuntimeEvent): "success" | "validation_fail" | "failure" {
|
||||
const metadata = event.metadata;
|
||||
if (isRecord(metadata)) {
|
||||
const status = metadata.status;
|
||||
if (status === "success" || status === "validation_fail" || status === "failure") {
|
||||
return status;
|
||||
}
|
||||
}
|
||||
|
||||
if (event.severity === "critical") {
|
||||
return "failure";
|
||||
}
|
||||
if (event.severity === "warning") {
|
||||
return "validation_fail";
|
||||
}
|
||||
return "success";
|
||||
}
|
||||
|
||||
function toSubtasks(metadata: JsonObject): string[] {
|
||||
const raw = metadata.subtasks;
|
||||
if (!Array.isArray(raw)) {
|
||||
return [];
|
||||
}
|
||||
|
||||
const subtasks: string[] = [];
|
||||
for (const item of raw) {
|
||||
if (typeof item !== "string") {
|
||||
continue;
|
||||
}
|
||||
const normalized = item.trim();
|
||||
if (!normalized) {
|
||||
continue;
|
||||
}
|
||||
subtasks.push(normalized);
|
||||
}
|
||||
|
||||
return subtasks;
|
||||
}
|
||||
|
||||
function readBoolean(record: JsonObject, key: string): boolean {
|
||||
return record[key] === true;
|
||||
}
|
||||
|
||||
function readExecutionContext(metadata: JsonObject): JsonObject | undefined {
|
||||
const raw = metadata.executionContext;
|
||||
return isRecord(raw) ? (raw as JsonObject) : undefined;
|
||||
}
|
||||
|
||||
function summarizeStatusForSession(events: readonly RuntimeEvent[]): {
|
||||
status: SessionStatus;
|
||||
startedAt?: string;
|
||||
endedAt?: string;
|
||||
message?: string;
|
||||
aborted: boolean;
|
||||
} {
|
||||
const started = events.find((event) => event.type === "session.started");
|
||||
const completed = [...events].reverse().find((event) => event.type === "session.completed");
|
||||
const failed = [...events].reverse().find((event) => event.type === "session.failed");
|
||||
|
||||
if (failed) {
|
||||
const message = failed.message;
|
||||
const lower = message.toLowerCase();
|
||||
return {
|
||||
status: "failure",
|
||||
startedAt: started?.timestamp,
|
||||
endedAt: failed.timestamp,
|
||||
message,
|
||||
aborted: lower.includes("abort"),
|
||||
};
|
||||
}
|
||||
|
||||
if (completed) {
|
||||
const metadata = isRecord(completed.metadata) ? completed.metadata : undefined;
|
||||
const completedStatus = metadata?.status;
|
||||
const status = completedStatus === "success" ? "success" : "failure";
|
||||
|
||||
return {
|
||||
status,
|
||||
startedAt: started?.timestamp,
|
||||
endedAt: completed.timestamp,
|
||||
message: completed.message,
|
||||
aborted: false,
|
||||
};
|
||||
}
|
||||
|
||||
if (started) {
|
||||
return {
|
||||
status: "running",
|
||||
startedAt: started.timestamp,
|
||||
message: started.message,
|
||||
aborted: false,
|
||||
};
|
||||
}
|
||||
|
||||
return {
|
||||
status: "unknown",
|
||||
aborted: false,
|
||||
};
|
||||
}
|
||||
|
||||
async function listSessionDirectories(stateRoot: string): Promise<string[]> {
|
||||
try {
|
||||
const entries = await readdir(resolve(stateRoot), { withFileTypes: true });
|
||||
return entries
|
||||
.filter((entry) => entry.isDirectory())
|
||||
.map((entry) => entry.name)
|
||||
.sort((left, right) => left.localeCompare(right));
|
||||
} catch (error) {
|
||||
if ((error as NodeJS.ErrnoException).code === "ENOENT") {
|
||||
return [];
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
export async function buildSessionSummaries(
|
||||
input: BuildSessionSummaryInput,
|
||||
): Promise<SessionSummary[]> {
|
||||
const [sessionDirectories, allEvents] = await Promise.all([
|
||||
listSessionDirectories(input.stateRoot),
|
||||
readRuntimeEvents(input.runtimeEventLogPath),
|
||||
]);
|
||||
|
||||
const sessionIds = new Set<string>(sessionDirectories);
|
||||
for (const event of allEvents) {
|
||||
if (event.sessionId) {
|
||||
sessionIds.add(event.sessionId);
|
||||
}
|
||||
}
|
||||
|
||||
const summaries: SessionSummary[] = [];
|
||||
|
||||
for (const sessionId of sessionIds) {
|
||||
const sessionEvents = filterRuntimeEvents(allEvents, {
|
||||
sessionId,
|
||||
});
|
||||
const attempts = sessionEvents.filter((event) => event.type === "node.attempt.completed");
|
||||
const distinctNodeIds = new Set(attempts.map((event) => event.nodeId).filter(Boolean));
|
||||
|
||||
const totalUsage = attempts.reduce<RuntimeEventUsage>(
|
||||
(aggregate, event) => addUsage(aggregate, toUsage(event.usage)),
|
||||
{},
|
||||
);
|
||||
|
||||
const criticalEventCount = sessionEvents.filter((event) => event.severity === "critical").length;
|
||||
const statusInfo = summarizeStatusForSession(sessionEvents);
|
||||
|
||||
summaries.push({
|
||||
sessionId,
|
||||
status: statusInfo.status,
|
||||
startedAt: statusInfo.startedAt,
|
||||
endedAt: statusInfo.endedAt,
|
||||
nodeAttemptCount: attempts.length,
|
||||
distinctNodeCount: distinctNodeIds.size,
|
||||
costUsd: totalUsage.costUsd ?? 0,
|
||||
durationMs: totalUsage.durationMs ?? 0,
|
||||
criticalEventCount,
|
||||
message: statusInfo.message,
|
||||
aborted: statusInfo.aborted,
|
||||
});
|
||||
}
|
||||
|
||||
summaries.sort((left, right) => {
|
||||
const leftTime = left.startedAt ?? left.endedAt ?? "";
|
||||
const rightTime = right.startedAt ?? right.endedAt ?? "";
|
||||
return rightTime.localeCompare(leftTime);
|
||||
});
|
||||
|
||||
return summaries;
|
||||
}
|
||||
|
||||
async function readHandoffParentByNode(
|
||||
stateRoot: string,
|
||||
sessionId: string,
|
||||
): Promise<Record<string, string | undefined>> {
|
||||
const handoffDirectory = resolve(stateRoot, sessionId, "handoffs");
|
||||
const output: Record<string, string | undefined> = {};
|
||||
|
||||
try {
|
||||
const entries = await readdir(handoffDirectory, { withFileTypes: true });
|
||||
for (const entry of entries) {
|
||||
if (!entry.isFile() || !entry.name.endsWith(".json")) {
|
||||
continue;
|
||||
}
|
||||
|
||||
const path = resolve(handoffDirectory, entry.name);
|
||||
const content = await readFile(path, "utf8");
|
||||
const parsed = JSON.parse(content) as unknown;
|
||||
|
||||
if (!isRecord(parsed)) {
|
||||
continue;
|
||||
}
|
||||
|
||||
const nodeId = parsed.nodeId;
|
||||
if (typeof nodeId !== "string" || !nodeId) {
|
||||
continue;
|
||||
}
|
||||
|
||||
const fromNodeId = parsed.fromNodeId;
|
||||
output[nodeId] = typeof fromNodeId === "string" && fromNodeId
|
||||
? fromNodeId
|
||||
: undefined;
|
||||
}
|
||||
} catch (error) {
|
||||
if ((error as NodeJS.ErrnoException).code === "ENOENT") {
|
||||
return output;
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
|
||||
return output;
|
||||
}
|
||||
|
||||
function buildCriticalPath(input: {
|
||||
manifest: AgentManifest;
|
||||
nodes: readonly NodeInsight[];
|
||||
fromNodeByNode: Record<string, string | undefined>;
|
||||
status: SessionStatus;
|
||||
}): string[] {
|
||||
const nodeById = new Map(input.nodes.map((node) => [node.nodeId, node]));
|
||||
const failed = [...input.nodes]
|
||||
.reverse()
|
||||
.find((node) => node.lastStatus === "failure" || node.lastStatus === "validation_fail");
|
||||
|
||||
const targetNodeId = failed?.nodeId ?? (() => {
|
||||
if (input.status !== "success") {
|
||||
return undefined;
|
||||
}
|
||||
|
||||
const executedNodeIds = new Set(input.nodes.filter((node) => node.attemptCount > 0).map((node) => node.nodeId));
|
||||
const terminalNode = [...executedNodeIds].find((nodeId) => {
|
||||
const outgoing = input.manifest.pipeline.edges.filter((edge) => edge.from === nodeId);
|
||||
return !outgoing.some((edge) => executedNodeIds.has(edge.to));
|
||||
});
|
||||
return terminalNode;
|
||||
})();
|
||||
|
||||
if (!targetNodeId || !nodeById.has(targetNodeId)) {
|
||||
return [];
|
||||
}
|
||||
|
||||
const path: string[] = [];
|
||||
const visited = new Set<string>();
|
||||
let current: string | undefined = targetNodeId;
|
||||
|
||||
while (current && !visited.has(current)) {
|
||||
visited.add(current);
|
||||
path.push(current);
|
||||
current = input.fromNodeByNode[current];
|
||||
}
|
||||
|
||||
return path.reverse();
|
||||
}
|
||||
|
||||
function toCriticalEdgeSet(pathNodeIds: readonly string[]): Set<string> {
|
||||
const output = new Set<string>();
|
||||
for (let index = 1; index < pathNodeIds.length; index += 1) {
|
||||
const from = pathNodeIds[index - 1];
|
||||
const to = pathNodeIds[index];
|
||||
if (from && to) {
|
||||
output.add(`${from}->${to}`);
|
||||
}
|
||||
}
|
||||
return output;
|
||||
}
|
||||
|
||||
function toAttemptInsight(event: RuntimeEvent): NodeAttemptInsight {
|
||||
const metadata = isRecord(event.metadata) ? (event.metadata as JsonObject) : {};
|
||||
|
||||
return {
|
||||
timestamp: event.timestamp,
|
||||
attempt: typeof event.attempt === "number" ? event.attempt : 1,
|
||||
status: toStatusFromAttemptEvent(event),
|
||||
severity: event.severity,
|
||||
message: event.message,
|
||||
usage: toUsage(event.usage),
|
||||
metadata,
|
||||
executionContext: readExecutionContext(metadata),
|
||||
retrySpawned: readBoolean(metadata, "retrySpawned"),
|
||||
securityViolation: readBoolean(metadata, "securityViolation"),
|
||||
subtasks: toSubtasks(metadata),
|
||||
};
|
||||
}
|
||||
|
||||
function inferTopology(node: PipelineNode): string {
|
||||
return node.topology?.kind ?? "sequential";
|
||||
}
|
||||
|
||||
export async function buildSessionGraphInsight(
|
||||
input: BuildSessionGraphInput,
|
||||
): Promise<SessionGraphInsight> {
|
||||
const [allEvents, handoffParentByNode] = await Promise.all([
|
||||
readRuntimeEvents(input.runtimeEventLogPath),
|
||||
readHandoffParentByNode(input.stateRoot, input.sessionId),
|
||||
]);
|
||||
|
||||
const sessionEvents = filterRuntimeEvents(allEvents, {
|
||||
sessionId: input.sessionId,
|
||||
});
|
||||
|
||||
const statusInfo = summarizeStatusForSession(sessionEvents);
|
||||
|
||||
const attemptsByNode = new Map<string, NodeAttemptInsight[]>();
|
||||
const domainEventsByNode = new Map<string, RuntimeEvent[]>();
|
||||
|
||||
for (const event of sessionEvents) {
|
||||
if (event.type === "node.attempt.completed" && event.nodeId) {
|
||||
const attempt = toAttemptInsight(event);
|
||||
const list = attemptsByNode.get(event.nodeId);
|
||||
if (list) {
|
||||
list.push(attempt);
|
||||
} else {
|
||||
attemptsByNode.set(event.nodeId, [attempt]);
|
||||
}
|
||||
continue;
|
||||
}
|
||||
|
||||
if (event.type.startsWith("domain.") && event.nodeId) {
|
||||
const list = domainEventsByNode.get(event.nodeId);
|
||||
if (list) {
|
||||
list.push(event);
|
||||
} else {
|
||||
domainEventsByNode.set(event.nodeId, [event]);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
const nodes: NodeInsight[] = input.manifest.pipeline.nodes.map((node) => {
|
||||
const attempts = [...(attemptsByNode.get(node.id) ?? [])].sort((left, right) => {
|
||||
if (left.attempt !== right.attempt) {
|
||||
return left.attempt - right.attempt;
|
||||
}
|
||||
return left.timestamp.localeCompare(right.timestamp);
|
||||
});
|
||||
|
||||
const usage = attempts.reduce<RuntimeEventUsage>((aggregate, attempt) => {
|
||||
return addUsage(aggregate, attempt.usage);
|
||||
}, {});
|
||||
|
||||
const last = attempts[attempts.length - 1];
|
||||
const sandboxPayload = [...attempts]
|
||||
.reverse()
|
||||
.map((attempt) => attempt.executionContext)
|
||||
.find((payload) => Boolean(payload));
|
||||
|
||||
const subtasks = attempts.flatMap((attempt) => attempt.subtasks);
|
||||
|
||||
return {
|
||||
nodeId: node.id,
|
||||
actorId: node.actorId,
|
||||
personaId: node.personaId,
|
||||
topology: inferTopology(node),
|
||||
fromNodeId: handoffParentByNode[node.id],
|
||||
attemptCount: attempts.length,
|
||||
...(last ? { lastStatus: last.status } : {}),
|
||||
usage,
|
||||
subtaskCount: subtasks.length,
|
||||
securityViolationCount: attempts.filter((attempt) => attempt.securityViolation).length,
|
||||
attempts,
|
||||
domainEvents: [...(domainEventsByNode.get(node.id) ?? [])],
|
||||
...(sandboxPayload ? { sandboxPayload } : {}),
|
||||
};
|
||||
});
|
||||
|
||||
const criticalPathNodeIds = buildCriticalPath({
|
||||
manifest: input.manifest,
|
||||
nodes,
|
||||
fromNodeByNode: handoffParentByNode,
|
||||
status: statusInfo.status,
|
||||
});
|
||||
const criticalEdgeSet = toCriticalEdgeSet(criticalPathNodeIds);
|
||||
|
||||
const edges: EdgeInsight[] = input.manifest.pipeline.edges.map((edge) => {
|
||||
const visited = handoffParentByNode[edge.to] === edge.from;
|
||||
|
||||
return {
|
||||
from: edge.from,
|
||||
to: edge.to,
|
||||
trigger: toEdgeTrigger(edge),
|
||||
conditionLabels: (edge.when ?? []).map(toConditionLabel),
|
||||
visited,
|
||||
critical: criticalEdgeSet.has(`${edge.from}->${edge.to}`),
|
||||
};
|
||||
});
|
||||
|
||||
return {
|
||||
sessionId: input.sessionId,
|
||||
status: statusInfo.status,
|
||||
aborted: statusInfo.aborted,
|
||||
...(statusInfo.message ? { abortMessage: statusInfo.message } : {}),
|
||||
nodes,
|
||||
edges,
|
||||
runtimeEvents: sessionEvents,
|
||||
criticalPathNodeIds,
|
||||
};
|
||||
}
|
||||
|
||||
export async function readSessionUpdatedAt(stateRoot: string, sessionId: string): Promise<string | undefined> {
|
||||
const sessionPath = resolve(stateRoot, sessionId);
|
||||
try {
|
||||
const metadata = await stat(sessionPath);
|
||||
return metadata.mtime.toISOString();
|
||||
} catch (error) {
|
||||
if ((error as NodeJS.ErrnoException).code === "ENOENT") {
|
||||
return undefined;
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user