From cf386e1aaabb96628e8894b3f02e0d53caf15245 Mon Sep 17 00:00:00 2001 From: Josh Rzemien Date: Mon, 23 Feb 2026 18:49:53 -0500 Subject: [PATCH] feat(ui): add operator UI server, stores, and insights --- .env.example | 4 + README.md | 40 +++ docs/runtime-events.md | 7 + package.json | 1 + src/agents/lifecycle-observer.ts | 47 ++- src/agents/pipeline.ts | 55 ++- src/ui/config-store.ts | 183 ++++++++++ src/ui/env-store.ts | 130 +++++++ src/ui/http-utils.ts | 90 +++++ src/ui/manifest-store.ts | 116 ++++++ src/ui/provider-executor.ts | 595 +++++++++++++++++++++++++++++++ src/ui/run-service.ts | 475 ++++++++++++++++++++++++ src/ui/runtime-events-store.ts | 94 +++++ src/ui/server.ts | 587 ++++++++++++++++++++++++++++++ src/ui/session-insights.ts | 537 ++++++++++++++++++++++++++++ tests/env-store.test.ts | 35 ++ tests/provider-executor.test.ts | 66 ++++ tests/session-insights.test.ts | 207 +++++++++++ 18 files changed, 3252 insertions(+), 17 deletions(-) create mode 100644 src/ui/config-store.ts create mode 100644 src/ui/env-store.ts create mode 100644 src/ui/http-utils.ts create mode 100644 src/ui/manifest-store.ts create mode 100644 src/ui/provider-executor.ts create mode 100644 src/ui/run-service.ts create mode 100644 src/ui/runtime-events-store.ts create mode 100644 src/ui/server.ts create mode 100644 src/ui/session-insights.ts create mode 100644 tests/env-store.test.ts create mode 100644 tests/provider-executor.test.ts create mode 100644 tests/session-insights.test.ts diff --git a/.env.example b/.env.example index e8eb20f..3b5f346 100644 --- a/.env.example +++ b/.env.example @@ -57,6 +57,10 @@ AGENT_RUNTIME_DISCORD_WEBHOOK_URL= AGENT_RUNTIME_DISCORD_MIN_SEVERITY=critical AGENT_RUNTIME_DISCORD_ALWAYS_NOTIFY_TYPES=session.started,session.completed,session.failed +# Local operator UI +AGENT_UI_HOST=127.0.0.1 +AGENT_UI_PORT=4317 + # Runtime-injected (do not set manually): # AGENT_REPO_ROOT, AGENT_WORKTREE_PATH, AGENT_WORKTREE_BASE_REF, # AGENT_PORT_RANGE_START, AGENT_PORT_RANGE_END, AGENT_PORT_PRIMARY, AGENT_DISCOVERY_FILE diff --git a/README.md b/README.md index 14f703d..2ae0f70 100644 --- a/README.md +++ b/README.md @@ -42,6 +42,7 @@ TypeScript runtime for deterministic multi-agent execution with: - `src/mcp`: MCP config types/conversion/handlers - `src/security`: shell AST parsing, rules engine, secure executor, and audit sinks - `src/telemetry`: runtime event schema, sink fan-out, file sink, and Discord webhook sink +- `src/ui`: local operator UI server, API routes, run-control service, and graph/event aggregation - `src/examples`: provider entrypoints (`codex.ts`, `claude.ts`) - `src/config.ts`: centralized env parsing/validation/defaulting - `tests`: manager, manifest, pipeline/orchestration, state, provisioning, MCP @@ -68,6 +69,35 @@ npm run dev -- codex "List potential improvements." npm run dev -- claude "List potential improvements." ``` +## Operator UI + +Start the local UI server: + +```bash +npm run ui +``` + +Then open: + +- `http://127.0.0.1:4317` (default) + +The UI provides: + +- graph visualizer with topology/retry rendering, edge trigger labels, node economics (duration/cost/tokens), and critical-path highlighting +- node inspector with attempt metadata and injected `ResolvedExecutionContext` sandbox payload +- live runtime event feed from `AGENT_RUNTIME_EVENT_LOG_PATH` with severity coloring (including security mirror events) +- run trigger + kill switch backed by `SchemaDrivenExecutionEngine.runSession(...)` + - run mode selector: `provider` (real Codex/Claude execution) or `mock` (deterministic dry-run executor) + - provider selector: `codex` or `claude` +- run history from `AGENT_STATE_ROOT` +- forms for runtime Discord webhook settings, security policy, and manager/resource limits +- manifest editor/validator/saver for schema `"1"` manifests + +Provider mode notes: + +- `provider=codex` uses existing OpenAI/Codex auth settings (`OPENAI_AUTH_MODE`, `CODEX_API_KEY`, `OPENAI_API_KEY`). +- `provider=claude` uses Claude auth resolution (`CLAUDE_CODE_OAUTH_TOKEN` preferred, otherwise `ANTHROPIC_API_KEY`, or existing Claude Code login state). + ## Manifest Semantics `AgentManifest` (schema `"1"`) validates: @@ -121,6 +151,11 @@ Each runtime event is written as one NDJSON object with: - `message` - optional `usage` (`tokenInput`, `tokenOutput`, `tokenTotal`, `toolCalls`, `durationMs`, `costUsd`) - optional structured `metadata` + - `node.attempt.completed` metadata includes: + - `executionContext` (resolved sandbox payload injected into executor) + - `topologyKind` + - `retrySpawned` + - optional `fromNodeId`, `subtasks`, `securityViolation` ### Runtime Event Setup @@ -257,6 +292,11 @@ jq -c 'select(.severity=="critical")' .ai_ops/events/runtime-events.ndjson - `AGENT_RUNTIME_DISCORD_MIN_SEVERITY` (`info`, `warning`, or `critical`) - `AGENT_RUNTIME_DISCORD_ALWAYS_NOTIFY_TYPES` (CSV event types such as `session.started,session.completed,session.failed`) +### Operator UI + +- `AGENT_UI_HOST` (default `127.0.0.1`) +- `AGENT_UI_PORT` (default `4317`) + ### Runtime-Injected (Do Not Configure In `.env`) - `AGENT_REPO_ROOT` diff --git a/docs/runtime-events.md b/docs/runtime-events.md index a7f5443..38af8d2 100644 --- a/docs/runtime-events.md +++ b/docs/runtime-events.md @@ -28,6 +28,13 @@ Core emitted event types: - `session.failed` - `security.` (mirrored from security audit engine) +`node.attempt.completed` metadata includes orchestration-debug fields used by the operator UI: + +- `status`, optional `failureKind`, optional `failureCode` +- `executionContext` (`phase`, `modelConstraint`, `allowedTools`, security constraints) +- `topologyKind`, `retrySpawned`, optional `fromNodeId` +- optional `subtasks`, `securityViolation` + ## Sinks - File sink (`AGENT_RUNTIME_EVENT_LOG_PATH`) diff --git a/package.json b/package.json index 7720af4..7bfd069 100644 --- a/package.json +++ b/package.json @@ -10,6 +10,7 @@ "test": "node --import tsx/esm --test tests/**/*.test.ts", "verify": "npm run check && npm run check:tests && npm run test && npm run build", "dev": "node --import tsx/esm src/index.ts", + "ui": "node --import tsx/esm src/ui/server.ts", "codex": "node --import tsx/esm src/examples/codex.ts", "claude": "node --import tsx/esm src/examples/claude.ts", "start": "node dist/index.js" diff --git a/src/agents/lifecycle-observer.ts b/src/agents/lifecycle-observer.ts index 697abc0..40b86af 100644 --- a/src/agents/lifecycle-observer.ts +++ b/src/agents/lifecycle-observer.ts @@ -3,7 +3,7 @@ import { type DomainEvent, type DomainEventType, } from "./domain-events.js"; -import type { PipelineNode } from "./manifest.js"; +import type { NodeTopologyKind, PipelineNode } from "./manifest.js"; import { type ProjectContextPatch, type FileSystemProjectContextStore } from "./project-context.js"; import { PersonaRegistry } from "./persona-registry.js"; import { @@ -23,6 +23,10 @@ export type PipelineNodeAttemptObservedEvent = { attempt: number; result: ActorExecutionResult; domainEvents: DomainEvent[]; + executionContext: JsonObject; + fromNodeId?: string; + retrySpawned: boolean; + topologyKind: NodeTopologyKind; }; function toBehaviorEvent(status: ActorResultStatus): "onTaskComplete" | "onValidationFail" | undefined { @@ -149,6 +153,41 @@ function extractUsageMetrics(result: ActorExecutionResult): RuntimeEventUsage | return undefined; } +function extractSubtasks(result: ActorExecutionResult): string[] { + const candidates = [ + result.payload?.subtasks, + result.stateMetadata?.subtasks, + ]; + + for (const candidate of candidates) { + if (!Array.isArray(candidate)) { + continue; + } + + const subtasks: string[] = []; + for (const item of candidate) { + if (typeof item !== "string") { + continue; + } + const normalized = item.trim(); + if (!normalized) { + continue; + } + subtasks.push(normalized); + } + + if (subtasks.length > 0) { + return subtasks; + } + } + + return []; +} + +function hasSecurityViolation(result: ActorExecutionResult): boolean { + return isRecord(result.payload) && result.payload.security_violation === true; +} + export interface PipelineLifecycleObserver { onNodeAttempt(event: PipelineNodeAttemptObservedEvent): Promise; } @@ -213,6 +252,12 @@ export class PersistenceLifecycleObserver implements PipelineLifecycleObserver { status: event.result.status, ...(event.result.failureKind ? { failureKind: event.result.failureKind } : {}), ...(event.result.failureCode ? { failureCode: event.result.failureCode } : {}), + executionContext: event.executionContext, + topologyKind: event.topologyKind, + retrySpawned: event.retrySpawned, + ...(event.fromNodeId ? { fromNodeId: event.fromNodeId } : {}), + subtasks: extractSubtasks(event.result), + securityViolation: hasSecurityViolation(event.result), }, }); diff --git a/src/agents/pipeline.ts b/src/agents/pipeline.ts index 4e742bb..0315ba8 100644 --- a/src/agents/pipeline.ts +++ b/src/agents/pipeline.ts @@ -14,7 +14,13 @@ import { PersistenceLifecycleObserver, type PipelineLifecycleObserver, } from "./lifecycle-observer.js"; -import type { AgentManifest, PipelineEdge, PipelineNode, RouteCondition } from "./manifest.js"; +import type { + AgentManifest, + NodeTopologyKind, + PipelineEdge, + PipelineNode, + RouteCondition, +} from "./manifest.js"; import type { AgentManager, RecursiveChildIntent } from "./manager.js"; import type { CodexConfigObject, @@ -183,6 +189,10 @@ type NodeAttemptResult = { payloadForNext: JsonObject; domainEvents: DomainEvent[]; hardFailure: boolean; + executionContext: ResolvedExecutionContext; + fromNodeId?: string; + retrySpawned: boolean; + topologyKind: NodeTopologyKind; }; type NodeExecutionOutcome = { @@ -854,6 +864,12 @@ export class PipelineExecutor { attempt, }, }); + const toolClearance = this.personaRegistry.getToolClearance(node.personaId); + const executionContext = this.resolveExecutionContext({ + node, + toolClearance, + prompt, + }); const result = await this.invokeActorExecutor({ sessionId, @@ -861,6 +877,7 @@ export class PipelineExecutor { prompt, context, signal: recursiveSignal, + executionContext, executor, }); @@ -871,6 +888,12 @@ export class PipelineExecutor { status: result.status, customEvents: result.events, }); + const topologyKind: NodeTopologyKind = node.topology?.kind ?? "sequential"; + const payloadForNext = result.payload ?? context.handoff.payload; + const shouldRetry = + result.status === "validation_fail" && + this.shouldRetryValidation(node) && + attempt <= maxRetriesForNode; await this.lifecycleObserver.onNodeAttempt({ sessionId, @@ -878,6 +901,10 @@ export class PipelineExecutor { attempt, result, domainEvents, + executionContext, + fromNodeId: context.handoff.fromNodeId, + retrySpawned: shouldRetry, + topologyKind, }); const emittedEventTypes = domainEvents.map((event) => event.type); @@ -893,12 +920,6 @@ export class PipelineExecutor { const hardFailure = this.failurePolicy.isHardFailure(result); hardFailureAttempts.push(hardFailure); - const payloadForNext = result.payload ?? context.handoff.payload; - const shouldRetry = - result.status === "validation_fail" && - this.shouldRetryValidation(node) && - attempt <= maxRetriesForNode; - if (!shouldRetry) { return { type: "complete" as const, @@ -909,6 +930,10 @@ export class PipelineExecutor { payloadForNext, domainEvents, hardFailure, + executionContext, + fromNodeId: context.handoff.fromNodeId, + retrySpawned: false, + topologyKind, }, }; } @@ -935,7 +960,10 @@ export class PipelineExecutor { if (!first) { throw new Error(`Retry aggregation for node "${node.id}" did not receive child output.`); } - return first.output; + return { + ...first.output, + retrySpawned: true, + }; }, }; }, @@ -965,16 +993,11 @@ export class PipelineExecutor { prompt: string; context: NodeExecutionContext; signal: AbortSignal; + executionContext: ResolvedExecutionContext; executor: ActorExecutor; }): Promise { try { throwIfAborted(input.signal); - const toolClearance = this.personaRegistry.getToolClearance(input.node.personaId); - const executionContext = this.resolveExecutionContext({ - node: input.node, - toolClearance, - prompt: input.prompt, - }); return await input.executor({ sessionId: input.sessionId, @@ -982,8 +1005,8 @@ export class PipelineExecutor { prompt: input.prompt, context: input.context, signal: input.signal, - executionContext, - mcp: this.buildActorMcpContext(executionContext, input.prompt), + executionContext: input.executionContext, + mcp: this.buildActorMcpContext(input.executionContext, input.prompt), security: this.securityContext, }); } catch (error) { diff --git a/src/ui/config-store.ts b/src/ui/config-store.ts new file mode 100644 index 0000000..109af78 --- /dev/null +++ b/src/ui/config-store.ts @@ -0,0 +1,183 @@ +import { resolve } from "node:path"; +import { loadConfig, type AppConfig } from "../config.js"; +import { parseEnvFile, writeEnvFileUpdates } from "./env-store.js"; + +export type RuntimeNotificationSettings = { + webhookUrl: string; + minSeverity: "info" | "warning" | "critical"; + alwaysNotifyTypes: string[]; +}; + +export type SecurityPolicySettings = { + violationMode: "hard_abort" | "validation_fail"; + allowedBinaries: string[]; + commandTimeoutMs: number; + inheritedEnv: string[]; + scrubbedEnv: string[]; +}; + +export type LimitSettings = { + maxConcurrent: number; + maxSession: number; + maxRecursiveDepth: number; + topologyMaxDepth: number; + topologyMaxRetries: number; + relationshipMaxChildren: number; + portBase: number; + portBlockSize: number; + portBlockCount: number; + portPrimaryOffset: number; +}; + +export type UiConfigSnapshot = { + envFilePath: string; + runtimeEvents: RuntimeNotificationSettings; + security: SecurityPolicySettings; + limits: LimitSettings; + paths: { + stateRoot: string; + projectContextPath: string; + runtimeEventLogPath: string; + securityAuditLogPath: string; + }; +}; + +function toCsv(values: readonly string[]): string { + return values.join(","); +} + +function sanitizeCsv(values: readonly string[]): string[] { + const output: string[] = []; + const seen = new Set(); + + for (const value of values) { + const normalized = value.trim(); + if (!normalized || seen.has(normalized)) { + continue; + } + seen.add(normalized); + output.push(normalized); + } + + return output; +} + +function toRuntimeEvents(config: Readonly): RuntimeNotificationSettings { + return { + webhookUrl: config.runtimeEvents.discordWebhookUrl ?? "", + minSeverity: config.runtimeEvents.discordMinSeverity, + alwaysNotifyTypes: [...config.runtimeEvents.discordAlwaysNotifyTypes], + }; +} + +function toSecurity(config: Readonly): SecurityPolicySettings { + return { + violationMode: config.security.violationHandling, + allowedBinaries: [...config.security.shellAllowedBinaries], + commandTimeoutMs: config.security.commandTimeoutMs, + inheritedEnv: [...config.security.inheritedEnvVars], + scrubbedEnv: [...config.security.scrubbedEnvVars], + }; +} + +function toLimits(config: Readonly): LimitSettings { + return { + maxConcurrent: config.agentManager.maxConcurrentAgents, + maxSession: config.agentManager.maxSessionAgents, + maxRecursiveDepth: config.agentManager.maxRecursiveDepth, + topologyMaxDepth: config.orchestration.maxDepth, + topologyMaxRetries: config.orchestration.maxRetries, + relationshipMaxChildren: config.orchestration.maxChildren, + portBase: config.provisioning.portRange.basePort, + portBlockSize: config.provisioning.portRange.blockSize, + portBlockCount: config.provisioning.portRange.blockCount, + portPrimaryOffset: config.provisioning.portRange.primaryPortOffset, + }; +} + +function toSnapshot(config: Readonly, envFilePath: string): UiConfigSnapshot { + return { + envFilePath, + runtimeEvents: toRuntimeEvents(config), + security: toSecurity(config), + limits: toLimits(config), + paths: { + stateRoot: config.orchestration.stateRoot, + projectContextPath: config.orchestration.projectContextPath, + runtimeEventLogPath: config.runtimeEvents.logPath, + securityAuditLogPath: config.security.auditLogPath, + }, + }; +} + +function mergeEnv(fileValues: Record): NodeJS.ProcessEnv { + return { + ...process.env, + ...fileValues, + }; +} + +export class UiConfigStore { + private readonly envFilePath: string; + + constructor(input: { workspaceRoot: string; envFilePath?: string }) { + this.envFilePath = resolve(input.workspaceRoot, input.envFilePath ?? ".env"); + } + + getEnvFilePath(): string { + return this.envFilePath; + } + + async readSnapshot(): Promise { + const parsed = await parseEnvFile(this.envFilePath); + const config = loadConfig(mergeEnv(parsed.values)); + return toSnapshot(config, this.envFilePath); + } + + async updateRuntimeEvents(input: RuntimeNotificationSettings): Promise { + const alwaysNotifyTypes = sanitizeCsv(input.alwaysNotifyTypes); + + const updates: Record = { + AGENT_RUNTIME_DISCORD_WEBHOOK_URL: input.webhookUrl.trim(), + AGENT_RUNTIME_DISCORD_MIN_SEVERITY: input.minSeverity, + AGENT_RUNTIME_DISCORD_ALWAYS_NOTIFY_TYPES: toCsv(alwaysNotifyTypes), + }; + + const parsed = await writeEnvFileUpdates(this.envFilePath, updates); + const config = loadConfig(mergeEnv(parsed.values)); + return toSnapshot(config, this.envFilePath); + } + + async updateSecurityPolicy(input: SecurityPolicySettings): Promise { + const updates: Record = { + AGENT_SECURITY_VIOLATION_MODE: input.violationMode, + AGENT_SECURITY_ALLOWED_BINARIES: toCsv(sanitizeCsv(input.allowedBinaries)), + AGENT_SECURITY_COMMAND_TIMEOUT_MS: String(input.commandTimeoutMs), + AGENT_SECURITY_ENV_INHERIT: toCsv(sanitizeCsv(input.inheritedEnv)), + AGENT_SECURITY_ENV_SCRUB: toCsv(sanitizeCsv(input.scrubbedEnv)), + }; + + const parsed = await writeEnvFileUpdates(this.envFilePath, updates); + const config = loadConfig(mergeEnv(parsed.values)); + return toSnapshot(config, this.envFilePath); + } + + async updateLimits(input: LimitSettings): Promise { + const updates: Record = { + AGENT_MAX_CONCURRENT: String(input.maxConcurrent), + AGENT_MAX_SESSION: String(input.maxSession), + AGENT_MAX_RECURSIVE_DEPTH: String(input.maxRecursiveDepth), + AGENT_TOPOLOGY_MAX_DEPTH: String(input.topologyMaxDepth), + AGENT_TOPOLOGY_MAX_RETRIES: String(input.topologyMaxRetries), + AGENT_RELATIONSHIP_MAX_CHILDREN: String(input.relationshipMaxChildren), + AGENT_PORT_BASE: String(input.portBase), + AGENT_PORT_BLOCK_SIZE: String(input.portBlockSize), + AGENT_PORT_BLOCK_COUNT: String(input.portBlockCount), + AGENT_PORT_PRIMARY_OFFSET: String(input.portPrimaryOffset), + }; + + const parsed = await writeEnvFileUpdates(this.envFilePath, updates); + const config = loadConfig(mergeEnv(parsed.values)); + return toSnapshot(config, this.envFilePath); + } +} diff --git a/src/ui/env-store.ts b/src/ui/env-store.ts new file mode 100644 index 0000000..c0a2c51 --- /dev/null +++ b/src/ui/env-store.ts @@ -0,0 +1,130 @@ +import { mkdir, readFile, writeFile } from "node:fs/promises"; +import { dirname, resolve } from "node:path"; + +type AssignmentLine = { + kind: "assignment"; + key: string; + value: string; + leading: string; +}; + +type RawLine = { + kind: "raw"; + raw: string; +}; + +type EnvLine = AssignmentLine | RawLine; + +export type ParsedEnvFile = { + filePath: string; + lines: EnvLine[]; + values: Record; +}; + +const ASSIGNMENT_PATTERN = /^(\s*)([A-Za-z_][A-Za-z0-9_]*)\s*=(.*)$/; + +function unquoteValue(raw: string): string { + const trimmed = raw.trim(); + if (trimmed.length >= 2) { + const first = trimmed[0]; + const last = trimmed[trimmed.length - 1]; + if ((first === '"' && last === '"') || (first === "'" && last === "'")) { + return trimmed.slice(1, -1); + } + } + return trimmed; +} + +function formatValue(value: string): string { + if (value.length === 0) { + return ""; + } + + if (/\s|#|"|'/.test(value)) { + const escaped = value + .replace(/\\/g, "\\\\") + .replace(/"/g, '\\"'); + return `"${escaped}"`; + } + + return value; +} + +export async function parseEnvFile(filePath: string): Promise { + const resolvedPath = resolve(filePath); + let content = ""; + + try { + content = await readFile(resolvedPath, "utf8"); + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== "ENOENT") { + throw error; + } + } + + const lines = content.length > 0 ? content.split(/\r?\n/) : []; + const parsedLines: EnvLine[] = []; + const values: Record = {}; + + for (const line of lines) { + const match = line.match(ASSIGNMENT_PATTERN); + if (!match) { + parsedLines.push({ kind: "raw", raw: line }); + continue; + } + + const leading = match[1] ?? ""; + const key = match[2] ?? ""; + const rawValue = match[3] ?? ""; + const value = unquoteValue(rawValue); + + parsedLines.push({ + kind: "assignment", + key, + value, + leading, + }); + values[key] = value; + } + + return { + filePath: resolvedPath, + lines: parsedLines, + values, + }; +} + +export async function writeEnvFileUpdates( + filePath: string, + updates: Record, +): Promise { + const parsed = await parseEnvFile(filePath); + const keysToApply = new Set(Object.keys(updates)); + + const renderedLines = parsed.lines.map((line) => { + if (line.kind !== "assignment") { + return line.raw; + } + + if (!keysToApply.has(line.key)) { + return `${line.leading}${line.key}=${formatValue(line.value)}`; + } + + keysToApply.delete(line.key); + const nextValue = updates[line.key] ?? ""; + return `${line.leading}${line.key}=${formatValue(nextValue)}`; + }); + + for (const key of keysToApply) { + const nextValue = updates[key] ?? ""; + renderedLines.push(`${key}=${formatValue(nextValue)}`); + } + + const output = `${renderedLines.join("\n").replace(/\n+$/u, "")}\n`; + const resolvedPath = resolve(filePath); + + await mkdir(dirname(resolvedPath), { recursive: true }); + await writeFile(resolvedPath, output, "utf8"); + + return parseEnvFile(resolvedPath); +} diff --git a/src/ui/http-utils.ts b/src/ui/http-utils.ts new file mode 100644 index 0000000..7d09d75 --- /dev/null +++ b/src/ui/http-utils.ts @@ -0,0 +1,90 @@ +import { createReadStream } from "node:fs"; +import { stat } from "node:fs/promises"; +import { extname, resolve } from "node:path"; +import type { IncomingMessage, ServerResponse } from "node:http"; + +const CONTENT_TYPES: Record = { + ".html": "text/html; charset=utf-8", + ".js": "text/javascript; charset=utf-8", + ".css": "text/css; charset=utf-8", + ".json": "application/json; charset=utf-8", + ".svg": "image/svg+xml", +}; + +export function sendJson(response: ServerResponse, statusCode: number, body: unknown): void { + const payload = JSON.stringify(body); + response.statusCode = statusCode; + response.setHeader("Content-Type", "application/json; charset=utf-8"); + response.end(payload); +} + +export function sendText(response: ServerResponse, statusCode: number, body: string): void { + response.statusCode = statusCode; + response.setHeader("Content-Type", "text/plain; charset=utf-8"); + response.end(body); +} + +export async function parseJsonBody(request: IncomingMessage): Promise { + const chunks: Buffer[] = []; + + await new Promise((resolveBody, rejectBody) => { + request.on("data", (chunk: Buffer) => { + chunks.push(chunk); + }); + request.on("end", () => resolveBody()); + request.on("error", rejectBody); + }); + + const body = Buffer.concat(chunks).toString("utf8").trim(); + if (!body) { + throw new Error("Request body is required."); + } + + return JSON.parse(body) as T; +} + +export function methodNotAllowed(response: ServerResponse): void { + sendJson(response, 405, { + ok: false, + error: "Method not allowed.", + }); +} + +export function notFound(response: ServerResponse): void { + sendJson(response, 404, { + ok: false, + error: "Not found.", + }); +} + +export async function serveStaticFile(input: { + response: ServerResponse; + filePath: string; +}): Promise { + try { + const absolutePath = resolve(input.filePath); + const fileStats = await stat(absolutePath); + if (!fileStats.isFile()) { + return false; + } + + const extension = extname(absolutePath).toLowerCase(); + const contentType = CONTENT_TYPES[extension] ?? "application/octet-stream"; + input.response.statusCode = 200; + input.response.setHeader("Content-Type", contentType); + + await new Promise((resolveStream, rejectStream) => { + const stream = createReadStream(absolutePath); + stream.on("error", rejectStream); + stream.on("end", () => resolveStream()); + stream.pipe(input.response); + }); + + return true; + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") { + return false; + } + throw error; + } +} diff --git a/src/ui/manifest-store.ts b/src/ui/manifest-store.ts new file mode 100644 index 0000000..b42b192 --- /dev/null +++ b/src/ui/manifest-store.ts @@ -0,0 +1,116 @@ +import { mkdir, readdir, readFile, writeFile } from "node:fs/promises"; +import { dirname, extname, isAbsolute, relative, resolve, sep } from "node:path"; +import { parseAgentManifest, type AgentManifest } from "../agents/manifest.js"; + +export type ManifestRecord = { + path: string; + manifest: AgentManifest; + source: unknown; +}; + +export type ManifestListing = { + paths: string[]; +}; + +async function walkJsonFiles(root: string): Promise { + const output: string[] = []; + + const entries = await readdir(root, { withFileTypes: true }); + for (const entry of entries) { + const fullPath = resolve(root, entry.name); + if (entry.isDirectory()) { + output.push(...(await walkJsonFiles(fullPath))); + continue; + } + + if (entry.isFile() && extname(entry.name).toLowerCase() === ".json") { + output.push(fullPath); + } + } + + return output; +} + +function assertWorkspacePath(workspaceRoot: string, inputPath: string): string { + const resolved = isAbsolute(inputPath) + ? resolve(inputPath) + : resolve(workspaceRoot, inputPath); + const rel = relative(workspaceRoot, resolved); + + if (rel === ".." || rel.startsWith(`..${sep}`)) { + throw new Error("Path is outside workspace root."); + } + + return resolved; +} + +function toRelativePath(workspaceRoot: string, absolutePath: string): string { + return relative(workspaceRoot, absolutePath) || "."; +} + +export class ManifestStore { + private readonly workspaceRoot: string; + private readonly manifestDirectory: string; + + constructor(input: { workspaceRoot: string; manifestDirectory?: string }) { + this.workspaceRoot = resolve(input.workspaceRoot); + this.manifestDirectory = assertWorkspacePath( + this.workspaceRoot, + input.manifestDirectory ?? ".ai_ops/manifests", + ); + } + + getManifestDirectory(): string { + return this.manifestDirectory; + } + + async list(): Promise { + try { + const files = await walkJsonFiles(this.manifestDirectory); + const relPaths = files + .map((filePath) => toRelativePath(this.workspaceRoot, filePath)) + .sort((left, right) => left.localeCompare(right)); + + return { + paths: relPaths, + }; + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") { + return { + paths: [], + }; + } + throw error; + } + } + + async read(pathInput: string): Promise { + const absolutePath = assertWorkspacePath(this.workspaceRoot, pathInput); + const sourceText = await readFile(absolutePath, "utf8"); + const source = JSON.parse(sourceText) as unknown; + const manifest = parseAgentManifest(source); + + return { + path: toRelativePath(this.workspaceRoot, absolutePath), + manifest, + source, + }; + } + + async validate(source: unknown): Promise { + return parseAgentManifest(source); + } + + async save(pathInput: string, source: unknown): Promise { + const manifest = parseAgentManifest(source); + const absolutePath = assertWorkspacePath(this.workspaceRoot, pathInput); + await mkdir(dirname(absolutePath), { recursive: true }); + await writeFile(absolutePath, `${JSON.stringify(manifest, null, 2)}\n`, "utf8"); + + return { + path: toRelativePath(this.workspaceRoot, absolutePath), + manifest, + source: manifest, + }; + } +} diff --git a/src/ui/provider-executor.ts b/src/ui/provider-executor.ts new file mode 100644 index 0000000..a97a185 --- /dev/null +++ b/src/ui/provider-executor.ts @@ -0,0 +1,595 @@ +import { Codex } from "@openai/codex-sdk"; +import { query, type Options, type SDKMessage } from "@anthropic-ai/claude-agent-sdk"; +import { + buildClaudeAuthEnv, + resolveAnthropicToken, + resolveOpenAiApiKey, + type AppConfig, +} from "../config.js"; +import { isDomainEventType, type DomainEventEmission } from "../agents/domain-events.js"; +import type { ActorExecutionInput, ActorExecutionResult, ActorExecutor } from "../agents/pipeline.js"; +import { isRecord, type JsonObject, type JsonValue } from "../agents/types.js"; +import { createSessionContext, type SessionContext } from "../examples/session-context.js"; + +export type RunProvider = "codex" | "claude"; + +export type ProviderRunRuntime = { + provider: RunProvider; + config: Readonly; + sessionContext: SessionContext; + close: () => Promise; +}; + +type ProviderUsage = { + tokenInput?: number; + tokenOutput?: number; + tokenTotal?: number; + durationMs?: number; + costUsd?: number; +}; + +const ACTOR_RESPONSE_SCHEMA = { + type: "object", + additionalProperties: true, + properties: { + status: { + type: "string", + enum: ["success", "validation_fail", "failure"], + }, + payload: { + type: "object", + }, + stateFlags: { + type: "object", + additionalProperties: { + type: "boolean", + }, + }, + stateMetadata: { + type: "object", + }, + events: { + type: "array", + items: { + type: "object", + additionalProperties: true, + }, + }, + failureKind: { + type: "string", + enum: ["soft", "hard"], + }, + failureCode: { + type: "string", + }, + }, + required: ["status"], +}; + +const CLAUDE_OUTPUT_FORMAT = { + type: "json_schema", + name: "actor_execution_result", + schema: ACTOR_RESPONSE_SCHEMA, +} as const; + +function toErrorMessage(error: unknown): string { + if (error instanceof Error) { + return error.message; + } + return String(error); +} + +function toJsonValue(value: unknown): JsonValue { + return JSON.parse(JSON.stringify(value)) as JsonValue; +} + +function toJsonObject(value: unknown): JsonObject | undefined { + if (!isRecord(value)) { + return undefined; + } + + try { + const cloned = toJsonValue(value); + if (!isRecord(cloned)) { + return undefined; + } + return cloned as JsonObject; + } catch { + return undefined; + } +} + +function toBooleanRecord(value: unknown): Record | undefined { + if (!isRecord(value)) { + return undefined; + } + + const output: Record = {}; + for (const [key, candidate] of Object.entries(value)) { + if (typeof candidate === "boolean") { + output[key] = candidate; + } + } + + return Object.keys(output).length > 0 ? output : undefined; +} + +function toEventEmissions(value: unknown): DomainEventEmission[] | undefined { + if (!Array.isArray(value)) { + return undefined; + } + + const output: DomainEventEmission[] = []; + for (const item of value) { + if (!isRecord(item)) { + continue; + } + + const type = item.type; + if (typeof type !== "string" || !isDomainEventType(type)) { + continue; + } + + const payload = toJsonObject(item.payload); + output.push({ + type, + ...(payload ? { payload } : {}), + }); + } + + return output.length > 0 ? output : undefined; +} + +function extractJsonFromFencedBlock(text: string): unknown { + const matches = text.match(/```(?:json)?\s*([\s\S]*?)```/i); + if (!matches || !matches[1]) { + return undefined; + } + + try { + return JSON.parse(matches[1]); + } catch { + return undefined; + } +} + +function extractFirstBalancedJsonObject(text: string): unknown { + const start = text.indexOf("{"); + if (start < 0) { + return undefined; + } + + let depth = 0; + let inString = false; + let escaped = false; + + for (let index = start; index < text.length; index += 1) { + const character = text[index]; + if (!character) { + continue; + } + + if (inString) { + if (escaped) { + escaped = false; + } else if (character === "\\") { + escaped = true; + } else if (character === '"') { + inString = false; + } + continue; + } + + if (character === '"') { + inString = true; + continue; + } + + if (character === "{") { + depth += 1; + continue; + } + + if (character === "}") { + depth -= 1; + if (depth === 0) { + const candidate = text.slice(start, index + 1); + try { + return JSON.parse(candidate); + } catch { + return undefined; + } + } + } + } + + return undefined; +} + +function tryParseResponseObject(rawText: string, structuredOutput?: unknown): unknown { + if (structuredOutput !== undefined) { + return structuredOutput; + } + + const trimmed = rawText.trim(); + if (!trimmed) { + return undefined; + } + + try { + return JSON.parse(trimmed); + } catch { + // fall through + } + + const fenced = extractJsonFromFencedBlock(trimmed); + if (fenced !== undefined) { + return fenced; + } + + return extractFirstBalancedJsonObject(trimmed); +} + +function ensureUsageMetadata(input: { + result: ActorExecutionResult; + providerUsage: ProviderUsage; +}): ActorExecutionResult { + const stateMetadata = toJsonObject(input.result.stateMetadata) ?? {}; + const existingUsage = toJsonObject(stateMetadata.usage) ?? {}; + + const usage: JsonObject = { + ...existingUsage, + ...(typeof input.providerUsage.tokenInput === "number" + ? { tokenInput: input.providerUsage.tokenInput } + : {}), + ...(typeof input.providerUsage.tokenOutput === "number" + ? { tokenOutput: input.providerUsage.tokenOutput } + : {}), + ...(typeof input.providerUsage.tokenTotal === "number" + ? { tokenTotal: input.providerUsage.tokenTotal } + : {}), + ...(typeof input.providerUsage.durationMs === "number" + ? { durationMs: input.providerUsage.durationMs } + : {}), + ...(typeof input.providerUsage.costUsd === "number" + ? { costUsd: input.providerUsage.costUsd } + : {}), + }; + + return { + ...input.result, + stateMetadata: { + ...stateMetadata, + usage, + }, + }; +} + +export function parseActorExecutionResultFromModelOutput(input: { + rawText: string; + structuredOutput?: unknown; +}): ActorExecutionResult { + const parsed = tryParseResponseObject(input.rawText, input.structuredOutput); + if (!isRecord(parsed)) { + return { + status: "success", + payload: { + assistantResponse: input.rawText.trim(), + }, + }; + } + + const status = parsed.status; + if (status !== "success" && status !== "validation_fail" && status !== "failure") { + return { + status: "success", + payload: { + assistantResponse: input.rawText.trim(), + }, + }; + } + + const payload = toJsonObject(parsed.payload) ?? { + assistantResponse: input.rawText.trim(), + }; + const stateMetadata = toJsonObject(parsed.stateMetadata); + const stateFlags = toBooleanRecord(parsed.stateFlags); + const events = toEventEmissions(parsed.events); + const failureKind = parsed.failureKind === "soft" || parsed.failureKind === "hard" + ? parsed.failureKind + : undefined; + const failureCode = typeof parsed.failureCode === "string" + ? parsed.failureCode + : undefined; + + return { + status, + payload, + ...(stateFlags ? { stateFlags } : {}), + ...(stateMetadata ? { stateMetadata } : {}), + ...(events ? { events } : {}), + ...(failureKind ? { failureKind } : {}), + ...(failureCode ? { failureCode } : {}), + }; +} + +function buildActorPrompt(input: ActorExecutionInput): string { + const recentHistory = input.context.state.history.slice(-15); + + return [ + "You are executing one orchestration node in a schema-driven DAG runtime.", + "Return ONLY JSON with this object shape:", + JSON.stringify( + { + status: "success | validation_fail | failure", + payload: {}, + stateFlags: { + optional_boolean_flag: true, + }, + stateMetadata: { + optional_metadata: "value", + }, + events: [ + { + type: "requirements_defined | tasks_planned | code_committed | task_blocked | validation_passed | validation_failed | branch_merged", + payload: { + summary: "optional", + details: {}, + errorCode: "optional", + artifactPointer: "optional", + }, + }, + ], + failureKind: "soft | hard", + failureCode: "optional", + }, + null, + 2, + ), + "Do not include markdown or extra explanation outside JSON.", + `Node Prompt:\n${input.prompt}`, + `Execution Context:\n${JSON.stringify(input.executionContext, null, 2)}`, + `Current Handoff Payload:\n${JSON.stringify(input.context.handoff.payload, null, 2)}`, + `Session Flags:\n${JSON.stringify(input.context.state.flags, null, 2)}`, + `Recent Domain History:\n${JSON.stringify(recentHistory, null, 2)}`, + ].join("\n\n"); +} + +async function runCodexActor(input: { + runtime: ProviderRunRuntime; + actorInput: ActorExecutionInput; +}): Promise { + const { runtime, actorInput } = input; + const prompt = buildActorPrompt(actorInput); + const startedAt = Date.now(); + const apiKey = resolveOpenAiApiKey(runtime.config.provider); + + const codex = new Codex({ + ...(apiKey ? { apiKey } : {}), + ...(runtime.config.provider.openAiBaseUrl + ? { baseUrl: runtime.config.provider.openAiBaseUrl } + : {}), + ...(actorInput.mcp.resolvedConfig.codexConfig + ? { config: actorInput.mcp.resolvedConfig.codexConfig } + : {}), + env: runtime.sessionContext.runtimeInjection.env, + }); + + const thread = codex.startThread({ + workingDirectory: runtime.sessionContext.runtimeInjection.workingDirectory, + skipGitRepoCheck: runtime.config.provider.codexSkipGitCheck, + }); + + const turn = await runtime.sessionContext.runInSession(() => + thread.run(prompt, { + signal: actorInput.signal, + outputSchema: ACTOR_RESPONSE_SCHEMA, + }), + ); + + const usage: ProviderUsage = { + ...(turn.usage + ? { + tokenInput: turn.usage.input_tokens + turn.usage.cached_input_tokens, + tokenOutput: turn.usage.output_tokens, + tokenTotal: turn.usage.input_tokens + turn.usage.cached_input_tokens + turn.usage.output_tokens, + } + : {}), + durationMs: Date.now() - startedAt, + }; + + const parsed = parseActorExecutionResultFromModelOutput({ + rawText: turn.finalResponse, + }); + + return ensureUsageMetadata({ + result: parsed, + providerUsage: usage, + }); +} + +type ClaudeTurnResult = { + text: string; + structuredOutput?: unknown; + usage: ProviderUsage; +}; + +function buildClaudeOptions(input: { + runtime: ProviderRunRuntime; + actorInput: ActorExecutionInput; +}): Options { + const { runtime, actorInput } = input; + + const authOptionOverrides = runtime.config.provider.anthropicOauthToken + ? { authToken: runtime.config.provider.anthropicOauthToken } + : (() => { + const token = resolveAnthropicToken(runtime.config.provider); + return token ? { apiKey: token } : {}; + })(); + + const runtimeEnv = { + ...runtime.sessionContext.runtimeInjection.env, + ...buildClaudeAuthEnv(runtime.config.provider), + }; + + return { + maxTurns: 1, + ...(runtime.config.provider.claudeModel + ? { model: runtime.config.provider.claudeModel } + : {}), + ...(runtime.config.provider.claudeCodePath + ? { pathToClaudeCodeExecutable: runtime.config.provider.claudeCodePath } + : {}), + ...authOptionOverrides, + ...(actorInput.mcp.resolvedConfig.claudeMcpServers + ? { mcpServers: actorInput.mcp.resolvedConfig.claudeMcpServers as Options["mcpServers"] } + : {}), + canUseTool: actorInput.mcp.createClaudeCanUseTool(), + cwd: runtime.sessionContext.runtimeInjection.workingDirectory, + env: runtimeEnv, + outputFormat: CLAUDE_OUTPUT_FORMAT, + }; +} + +async function runClaudeTurn(input: { + runtime: ProviderRunRuntime; + actorInput: ActorExecutionInput; + prompt: string; +}): Promise { + const options = buildClaudeOptions({ + runtime: input.runtime, + actorInput: input.actorInput, + }); + + const startedAt = Date.now(); + const stream = query({ + prompt: input.prompt, + options, + }); + + let resultText = ""; + let structuredOutput: unknown; + let usage: ProviderUsage = {}; + + const onAbort = (): void => { + stream.close(); + }; + + input.actorInput.signal.addEventListener("abort", onAbort, { once: true }); + + try { + for await (const message of stream as AsyncIterable) { + if (message.type !== "result") { + continue; + } + + if (message.subtype !== "success") { + const detail = message.errors.join("; "); + throw new Error( + `Claude query failed (${message.subtype})${detail ? `: ${detail}` : ""}`, + ); + } + + resultText = message.result.trim(); + structuredOutput = message.structured_output; + usage = { + tokenInput: message.usage.input_tokens, + tokenOutput: message.usage.output_tokens, + tokenTotal: message.usage.input_tokens + message.usage.output_tokens, + durationMs: message.duration_ms, + costUsd: message.total_cost_usd, + }; + } + } finally { + input.actorInput.signal.removeEventListener("abort", onAbort); + stream.close(); + } + + if (!resultText && structuredOutput !== undefined) { + resultText = JSON.stringify(structuredOutput); + } + + if (!resultText) { + throw new Error("Claude run completed without a final result."); + } + + return { + text: resultText, + structuredOutput, + usage: { + ...usage, + durationMs: usage.durationMs ?? Date.now() - startedAt, + }, + }; +} + +async function runClaudeActor(input: { + runtime: ProviderRunRuntime; + actorInput: ActorExecutionInput; +}): Promise { + const prompt = buildActorPrompt(input.actorInput); + const turn = await input.runtime.sessionContext.runInSession(() => + runClaudeTurn({ + runtime: input.runtime, + actorInput: input.actorInput, + prompt, + }), + ); + + const parsed = parseActorExecutionResultFromModelOutput({ + rawText: turn.text, + structuredOutput: turn.structuredOutput, + }); + + return ensureUsageMetadata({ + result: parsed, + providerUsage: turn.usage, + }); +} + +export async function createProviderRunRuntime(input: { + provider: RunProvider; + initialPrompt: string; + config: Readonly; +}): Promise { + const sessionContext = await createSessionContext(input.provider, { + prompt: input.initialPrompt, + config: input.config, + }); + + return { + provider: input.provider, + config: input.config, + sessionContext, + close: async () => { + await sessionContext.close(); + }, + }; +} + +export function createProviderActorExecutor(runtime: ProviderRunRuntime): ActorExecutor { + return async (actorInput) => { + try { + if (runtime.provider === "codex") { + return await runCodexActor({ + runtime, + actorInput, + }); + } + + return await runClaudeActor({ + runtime, + actorInput, + }); + } catch (error) { + return { + status: "failure", + payload: { + error: toErrorMessage(error), + }, + failureKind: "hard", + failureCode: `provider_${runtime.provider}_execution_error`, + }; + } + }; +} diff --git a/src/ui/run-service.ts b/src/ui/run-service.ts new file mode 100644 index 0000000..ac0168f --- /dev/null +++ b/src/ui/run-service.ts @@ -0,0 +1,475 @@ +import { randomUUID } from "node:crypto"; +import { mkdir, readFile, writeFile } from "node:fs/promises"; +import { resolve } from "node:path"; +import { SchemaDrivenExecutionEngine } from "../agents/orchestration.js"; +import { parseAgentManifest, type AgentManifest } from "../agents/manifest.js"; +import type { ActorExecutionResult, ActorExecutor } from "../agents/pipeline.js"; +import { loadConfig, type AppConfig } from "../config.js"; +import { parseEnvFile } from "./env-store.js"; +import { + createProviderActorExecutor, + createProviderRunRuntime, + type RunProvider, +} from "./provider-executor.js"; + +const RUN_META_FILE_NAME = "ui-run-meta.json"; + +export type RunStatus = "running" | "success" | "failure" | "cancelled"; +export type RunExecutionMode = "mock" | "provider"; + +export type StartRunInput = { + prompt: string; + manifest: unknown; + sessionId?: string; + manifestPath?: string; + topologyHint?: string; + initialFlags?: Record; + runtimeContextOverrides?: Record; + simulateValidationNodeIds?: string[]; + executionMode?: RunExecutionMode; + provider?: RunProvider; +}; + +export type RunRecord = { + runId: string; + sessionId: string; + status: RunStatus; + startedAt: string; + endedAt?: string; + manifestPath?: string; + topologyHint?: string; + executionMode: RunExecutionMode; + provider: RunProvider; + error?: string; +}; + +type ActiveRun = { + controller: AbortController; + record: RunRecord; + promise: Promise; +}; + +function toSessionId(): string { + return `ui-session-${Date.now().toString(36)}-${randomUUID().slice(0, 8)}`; +} + +function dedupeStrings(values: readonly string[]): string[] { + const output: string[] = []; + const seen = new Set(); + for (const value of values) { + const normalized = value.trim(); + if (!normalized || seen.has(normalized)) { + continue; + } + seen.add(normalized); + output.push(normalized); + } + return output; +} + +async function waitWithSignal(ms: number, signal: AbortSignal): Promise { + if (signal.aborted) { + throw signal.reason instanceof Error + ? signal.reason + : new Error(String(signal.reason ?? "Run aborted.")); + } + + await new Promise((resolveWait, rejectWait) => { + const timeout = setTimeout(() => { + signal.removeEventListener("abort", onAbort); + resolveWait(); + }, ms); + + function onAbort(): void { + clearTimeout(timeout); + signal.removeEventListener("abort", onAbort); + rejectWait( + signal.reason instanceof Error + ? signal.reason + : new Error(String(signal.reason ?? "Run aborted.")), + ); + } + + signal.addEventListener("abort", onAbort, { once: true }); + }); +} + +function estimateUsage(prompt: string, toolCount: number): { + tokenInput: number; + tokenOutput: number; + durationMs: number; + costUsd: number; +} { + const tokenInput = Math.max(1, Math.ceil(prompt.length / 4)); + const tokenOutput = Math.max(16, Math.ceil(tokenInput * 0.7)); + const durationMs = 220 + (tokenInput % 11) * 35 + toolCount * 20; + const tokenTotal = tokenInput + tokenOutput; + const costUsd = Number((tokenTotal * 0.000002).toFixed(6)); + + return { + tokenInput, + tokenOutput, + durationMs, + costUsd, + }; +} + +function extractSubtasks(prompt: string): string[] { + const sentences = prompt + .split(/[.!?\n]/) + .map((part) => part.trim()) + .filter((part) => part.length > 0) + .slice(0, 3); + + if (sentences.length > 0) { + return sentences; + } + + const words = prompt.split(/\s+/).filter((word) => word.length > 0).slice(0, 3); + return words.length > 0 ? [words.join(" ")] : []; +} + +function createMockActorExecutors( + manifest: AgentManifest, + input: { + prompt: string; + topologyHint?: string; + simulateValidationNodeIds: Set; + }, +): Record { + const attemptsByNode = new Map(); + const uniqueActorIds = dedupeStrings(manifest.pipeline.nodes.map((node) => node.actorId)); + + const execute: ActorExecutor = async (actorInput) => { + const attempt = (attemptsByNode.get(actorInput.node.id) ?? 0) + 1; + attemptsByNode.set(actorInput.node.id, attempt); + + const shouldValidationFail = + attempt === 1 && input.simulateValidationNodeIds.has(actorInput.node.id); + + const usage = estimateUsage(actorInput.prompt, actorInput.executionContext.allowedTools.length); + await waitWithSignal(Math.min(usage.durationMs, 900), actorInput.signal); + + if (shouldValidationFail) { + const failure: ActorExecutionResult = { + status: "validation_fail", + payload: { + summary: `Node ${actorInput.node.id} requires remediation on first pass.`, + subtasks: extractSubtasks(input.prompt), + security_violation: false, + }, + stateMetadata: { + usage: { + ...usage, + tokenTotal: usage.tokenInput + usage.tokenOutput, + toolCalls: actorInput.executionContext.allowedTools.length, + }, + topologyHint: input.topologyHint ?? "manifest-default", + }, + failureKind: "soft", + failureCode: "ui_mock_validation_required", + }; + + return failure; + } + + return { + status: "success", + payload: { + summary: `Node ${actorInput.node.id} completed in mock mode.`, + prompt: input.prompt, + subtasks: extractSubtasks(input.prompt), + }, + stateMetadata: { + usage: { + ...usage, + tokenTotal: usage.tokenInput + usage.tokenOutput, + toolCalls: actorInput.executionContext.allowedTools.length, + }, + topologyHint: input.topologyHint ?? "manifest-default", + }, + stateFlags: { + [`${actorInput.node.id}_completed`]: true, + }, + }; + }; + + const executors: Record = {}; + for (const actorId of uniqueActorIds) { + executors[actorId] = execute; + } + return executors; +} + +function createSingleExecutorMap(manifest: AgentManifest, executor: ActorExecutor): Record { + const uniqueActorIds = dedupeStrings(manifest.pipeline.nodes.map((node) => node.actorId)); + const executors: Record = {}; + for (const actorId of uniqueActorIds) { + executors[actorId] = executor; + } + return executors; +} + +function toAbortErrorMessage(error: unknown): string { + if (error instanceof Error) { + return error.message; + } + return String(error); +} + +function isAbort(error: unknown): boolean { + if (!(error instanceof Error)) { + return false; + } + return error.name === "AbortError" || error.message.toLowerCase().includes("abort"); +} + +async function loadRuntimeConfig(envPath: string): Promise> { + const parsed = await parseEnvFile(envPath); + return loadConfig({ + ...process.env, + ...parsed.values, + }); +} + +async function writeRunMeta(input: { + stateRoot: string; + sessionId: string; + run: RunRecord; +}): Promise { + const sessionDirectory = resolve(input.stateRoot, input.sessionId); + await mkdir(sessionDirectory, { recursive: true }); + const path = resolve(sessionDirectory, RUN_META_FILE_NAME); + await writeFile(path, `${JSON.stringify(input.run, null, 2)}\n`, "utf8"); +} + +export async function readRunMetaBySession(input: { + stateRoot: string; + sessionId: string; +}): Promise { + const path = resolve(input.stateRoot, input.sessionId, RUN_META_FILE_NAME); + + try { + const content = await readFile(path, "utf8"); + const parsed = JSON.parse(content) as unknown; + if (!parsed || typeof parsed !== "object") { + return undefined; + } + + const record = parsed as Partial; + if ( + typeof record.runId !== "string" || + typeof record.sessionId !== "string" || + typeof record.status !== "string" || + typeof record.startedAt !== "string" + ) { + return undefined; + } + + const normalized: RunRecord = { + runId: record.runId, + sessionId: record.sessionId, + status: + record.status === "running" || + record.status === "success" || + record.status === "failure" || + record.status === "cancelled" + ? record.status + : "failure", + startedAt: record.startedAt, + executionMode: + record.executionMode === "provider" || record.executionMode === "mock" + ? record.executionMode + : "mock", + provider: record.provider === "claude" || record.provider === "codex" ? record.provider : "codex", + ...(typeof record.endedAt === "string" ? { endedAt: record.endedAt } : {}), + ...(typeof record.manifestPath === "string" ? { manifestPath: record.manifestPath } : {}), + ...(typeof record.topologyHint === "string" ? { topologyHint: record.topologyHint } : {}), + ...(typeof record.error === "string" ? { error: record.error } : {}), + }; + + return normalized; + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") { + return undefined; + } + throw error; + } +} + +export class UiRunService { + private readonly workspaceRoot: string; + private readonly envFilePath: string; + private readonly activeRuns = new Map(); + private readonly runHistory = new Map(); + + constructor(input: { + workspaceRoot: string; + envFilePath?: string; + }) { + this.workspaceRoot = resolve(input.workspaceRoot); + this.envFilePath = resolve(this.workspaceRoot, input.envFilePath ?? ".env"); + } + + listRuns(): RunRecord[] { + const output = [...this.runHistory.values()].sort((left, right) => { + return right.startedAt.localeCompare(left.startedAt); + }); + return output; + } + + getRun(runId: string): RunRecord | undefined { + return this.runHistory.get(runId); + } + + async startRun(input: StartRunInput): Promise { + const config = await loadRuntimeConfig(this.envFilePath); + const manifest = parseAgentManifest(input.manifest); + const executionMode = input.executionMode ?? "mock"; + const provider = input.provider ?? "codex"; + const sessionId = input.sessionId?.trim() || toSessionId(); + const runId = randomUUID(); + const controller = new AbortController(); + + const record: RunRecord = { + runId, + sessionId, + status: "running", + startedAt: new Date().toISOString(), + executionMode, + provider, + ...(input.manifestPath ? { manifestPath: input.manifestPath } : {}), + ...(input.topologyHint ? { topologyHint: input.topologyHint } : {}), + }; + this.runHistory.set(runId, record); + + const runPromise = (async () => { + let providerRuntime: Awaited> | undefined; + try { + if (executionMode === "provider") { + providerRuntime = await createProviderRunRuntime({ + provider, + initialPrompt: input.prompt, + config, + }); + } + + const actorExecutors = + executionMode === "provider" && providerRuntime + ? createSingleExecutorMap(manifest, createProviderActorExecutor(providerRuntime)) + : createMockActorExecutors(manifest, { + prompt: input.prompt, + topologyHint: input.topologyHint, + simulateValidationNodeIds: new Set(input.simulateValidationNodeIds ?? []), + }); + + const engine = new SchemaDrivenExecutionEngine({ + manifest, + actorExecutors, + settings: { + workspaceRoot: this.workspaceRoot, + stateRoot: config.orchestration.stateRoot, + projectContextPath: config.orchestration.projectContextPath, + runtimeContext: { + ui_mode: executionMode, + run_provider: provider, + ...(input.runtimeContextOverrides ?? {}), + }, + }, + config, + }); + + await writeRunMeta({ + stateRoot: config.orchestration.stateRoot, + sessionId, + run: record, + }); + + await engine.runSession({ + sessionId, + initialPayload: { + prompt: input.prompt, + }, + initialState: { + flags: { + ...(input.initialFlags ?? {}), + }, + }, + signal: controller.signal, + }); + + const completedRecord = this.runHistory.get(runId); + if (!completedRecord) { + return; + } + + const next: RunRecord = { + ...completedRecord, + status: "success", + endedAt: new Date().toISOString(), + }; + this.runHistory.set(runId, next); + + await writeRunMeta({ + stateRoot: config.orchestration.stateRoot, + sessionId, + run: next, + }); + } catch (error) { + const current = this.runHistory.get(runId); + if (!current) { + return; + } + + const cancelled = controller.signal.aborted || isAbort(error); + const next: RunRecord = { + ...current, + status: cancelled ? "cancelled" : "failure", + endedAt: new Date().toISOString(), + error: toAbortErrorMessage(error), + }; + this.runHistory.set(runId, next); + + await writeRunMeta({ + stateRoot: config.orchestration.stateRoot, + sessionId, + run: next, + }); + } finally { + if (providerRuntime) { + await providerRuntime.close(); + } + this.activeRuns.delete(runId); + } + })(); + + this.activeRuns.set(runId, { + controller, + record, + promise: runPromise, + }); + + return record; + } + + async cancelRun(runId: string): Promise { + const active = this.activeRuns.get(runId); + if (!active) { + const existing = this.runHistory.get(runId); + if (!existing) { + throw new Error(`Run \"${runId}\" does not exist.`); + } + return existing; + } + + active.controller.abort(new Error("Cancelled by operator from UI kill switch.")); + await active.promise; + + const finalRecord = this.runHistory.get(runId); + if (!finalRecord) { + throw new Error(`Run \"${runId}\" cancellation did not produce a final record.`); + } + + return finalRecord; + } +} diff --git a/src/ui/runtime-events-store.ts b/src/ui/runtime-events-store.ts new file mode 100644 index 0000000..d573e6b --- /dev/null +++ b/src/ui/runtime-events-store.ts @@ -0,0 +1,94 @@ +import { readFile } from "node:fs/promises"; +import { resolve } from "node:path"; +import type { RuntimeEvent } from "../telemetry/runtime-events.js"; + +type RuntimeEventFilter = { + sessionId?: string; + types?: string[]; + severities?: Array; + limit?: number; +}; + +function safeParseLine(line: string): RuntimeEvent | undefined { + const trimmed = line.trim(); + if (!trimmed) { + return undefined; + } + + try { + const parsed = JSON.parse(trimmed) as unknown; + if (!parsed || typeof parsed !== "object") { + return undefined; + } + + const record = parsed as Partial; + if ( + typeof record.id !== "string" || + typeof record.timestamp !== "string" || + typeof record.type !== "string" || + typeof record.severity !== "string" || + typeof record.message !== "string" + ) { + return undefined; + } + + return record as RuntimeEvent; + } catch { + return undefined; + } +} + +export async function readRuntimeEvents(logPath: string): Promise { + const absolutePath = resolve(logPath); + let content = ""; + + try { + content = await readFile(absolutePath, "utf8"); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") { + return []; + } + throw error; + } + + const parsed: RuntimeEvent[] = []; + const lines = content.split(/\r?\n/); + for (const line of lines) { + const event = safeParseLine(line); + if (event) { + parsed.push(event); + } + } + + parsed.sort((left, right) => left.timestamp.localeCompare(right.timestamp)); + return parsed; +} + +export function filterRuntimeEvents( + events: readonly RuntimeEvent[], + filter: RuntimeEventFilter, +): RuntimeEvent[] { + const filtered: RuntimeEvent[] = []; + const types = filter.types ? new Set(filter.types) : undefined; + const severities = filter.severities ? new Set(filter.severities) : undefined; + + for (const event of events) { + if (filter.sessionId && event.sessionId !== filter.sessionId) { + continue; + } + if (types && !types.has(event.type)) { + continue; + } + if (severities && !severities.has(event.severity)) { + continue; + } + + filtered.push(event); + } + + if (!filter.limit || filter.limit < 1 || filtered.length <= filter.limit) { + return filtered; + } + + return filtered.slice(-filter.limit); +} diff --git a/src/ui/server.ts b/src/ui/server.ts new file mode 100644 index 0000000..d91155d --- /dev/null +++ b/src/ui/server.ts @@ -0,0 +1,587 @@ +import "dotenv/config"; +import { createServer, type IncomingMessage, type ServerResponse } from "node:http"; +import { pathToFileURL } from "node:url"; +import { resolve } from "node:path"; +import { buildSessionGraphInsight, buildSessionSummaries } from "./session-insights.js"; +import { UiConfigStore, type LimitSettings, type RuntimeNotificationSettings, type SecurityPolicySettings } from "./config-store.js"; +import { ManifestStore } from "./manifest-store.js"; +import { filterRuntimeEvents, readRuntimeEvents } from "./runtime-events-store.js"; +import { parseJsonBody, sendJson, methodNotAllowed, notFound, serveStaticFile } from "./http-utils.js"; +import { readRunMetaBySession, UiRunService, type RunExecutionMode } from "./run-service.js"; +import type { RunProvider } from "./provider-executor.js"; + +type StartRunRequest = { + prompt: string; + manifestPath?: string; + manifest?: unknown; + sessionId?: string; + topologyHint?: string; + initialFlags?: Record; + runtimeContextOverrides?: Record; + simulateValidationNodeIds?: string[]; + executionMode?: RunExecutionMode; + provider?: RunProvider; +}; + +function parsePort(value: string | undefined): number { + const parsed = Number(value ?? "4317"); + if (!Number.isInteger(parsed) || parsed < 1 || parsed > 65535) { + throw new Error("UI server port must be an integer between 1 and 65535."); + } + return parsed; +} + +function parseLimit(value: string | null, fallback: number): number { + if (!value) { + return fallback; + } + const parsed = Number(value); + if (!Number.isInteger(parsed) || parsed < 1) { + return fallback; + } + return parsed; +} + +function toRelativePathFromApi(urlPath: string): string { + return decodeURIComponent(urlPath); +} + +function ensureBooleanRecord(value: unknown): Record { + if (!value || typeof value !== "object") { + return {}; + } + + const output: Record = {}; + for (const [key, raw] of Object.entries(value as Record)) { + if (typeof raw === "boolean") { + output[key] = raw; + } + } + return output; +} + +function ensureStringArray(value: unknown): string[] { + if (!Array.isArray(value)) { + return []; + } + + const output: string[] = []; + for (const item of value) { + if (typeof item !== "string") { + continue; + } + const normalized = item.trim(); + if (!normalized) { + continue; + } + output.push(normalized); + } + + return output; +} + +function ensureRuntimeContext(value: unknown): Record { + if (!value || typeof value !== "object") { + return {}; + } + + const output: Record = {}; + for (const [key, raw] of Object.entries(value as Record)) { + if (typeof raw === "string" || typeof raw === "number" || typeof raw === "boolean") { + output[key] = raw; + } + } + return output; +} + +function ensureExecutionMode(value: unknown): RunExecutionMode { + return value === "provider" ? "provider" : "mock"; +} + +function ensureProvider(value: unknown): RunProvider { + return value === "claude" ? "claude" : "codex"; +} + +async function readRuntimePaths(configStore: UiConfigStore, workspaceRoot: string): Promise<{ + stateRoot: string; + runtimeEventLogPath: string; +}> { + const snapshot = await configStore.readSnapshot(); + return { + stateRoot: resolve(workspaceRoot, snapshot.paths.stateRoot), + runtimeEventLogPath: resolve(workspaceRoot, snapshot.paths.runtimeEventLogPath), + }; +} + +async function handleApiRequest(input: { + request: IncomingMessage; + response: ServerResponse; + workspaceRoot: string; + configStore: UiConfigStore; + manifestStore: ManifestStore; + runService: UiRunService; +}): Promise { + const { request, response, workspaceRoot, configStore, manifestStore, runService } = input; + const requestUrl = new URL(request.url ?? "/", `http://${request.headers.host ?? "localhost"}`); + const { pathname } = requestUrl; + const method = request.method ?? "GET"; + + if (!pathname.startsWith("/api/")) { + return false; + } + + try { + if (pathname === "/api/health") { + if (method !== "GET") { + methodNotAllowed(response); + return true; + } + sendJson(response, 200, { + ok: true, + now: new Date().toISOString(), + }); + return true; + } + + if (pathname === "/api/config") { + if (method !== "GET") { + methodNotAllowed(response); + return true; + } + + const snapshot = await configStore.readSnapshot(); + sendJson(response, 200, { + ok: true, + config: snapshot, + }); + return true; + } + + if (pathname === "/api/config/runtime-events") { + if (method !== "PUT") { + methodNotAllowed(response); + return true; + } + + const body = await parseJsonBody(request); + const snapshot = await configStore.updateRuntimeEvents(body); + sendJson(response, 200, { + ok: true, + config: snapshot, + }); + return true; + } + + if (pathname === "/api/config/security") { + if (method !== "PUT") { + methodNotAllowed(response); + return true; + } + + const body = await parseJsonBody(request); + const snapshot = await configStore.updateSecurityPolicy(body); + sendJson(response, 200, { + ok: true, + config: snapshot, + }); + return true; + } + + if (pathname === "/api/config/limits") { + if (method !== "PUT") { + methodNotAllowed(response); + return true; + } + + const body = await parseJsonBody(request); + const snapshot = await configStore.updateLimits(body); + sendJson(response, 200, { + ok: true, + config: snapshot, + }); + return true; + } + + if (pathname === "/api/manifests") { + if (method !== "GET") { + methodNotAllowed(response); + return true; + } + + const listing = await manifestStore.list(); + sendJson(response, 200, { + ok: true, + manifests: listing.paths, + }); + return true; + } + + if (pathname === "/api/manifests/read") { + if (method !== "GET") { + methodNotAllowed(response); + return true; + } + + const manifestPath = requestUrl.searchParams.get("path"); + if (!manifestPath) { + sendJson(response, 400, { + ok: false, + error: 'Query parameter "path" is required.', + }); + return true; + } + + const record = await manifestStore.read(manifestPath); + sendJson(response, 200, { + ok: true, + manifest: record, + }); + return true; + } + + if (pathname === "/api/manifests/validate") { + if (method !== "POST") { + methodNotAllowed(response); + return true; + } + + const body = await parseJsonBody<{ manifest: unknown }>(request); + const manifest = await manifestStore.validate(body.manifest); + sendJson(response, 200, { + ok: true, + manifest, + }); + return true; + } + + if (pathname === "/api/manifests/save") { + if (method !== "PUT") { + methodNotAllowed(response); + return true; + } + + const body = await parseJsonBody<{ path: string; manifest: unknown }>(request); + if (!body.path || typeof body.path !== "string") { + sendJson(response, 400, { + ok: false, + error: 'Field "path" is required.', + }); + return true; + } + + const record = await manifestStore.save(body.path, body.manifest); + sendJson(response, 200, { + ok: true, + manifest: record, + }); + return true; + } + + if (pathname === "/api/runtime-events") { + if (method !== "GET") { + methodNotAllowed(response); + return true; + } + + const { runtimeEventLogPath } = await readRuntimePaths(configStore, workspaceRoot); + const limit = parseLimit(requestUrl.searchParams.get("limit"), 200); + const sessionId = requestUrl.searchParams.get("sessionId") ?? undefined; + const events = filterRuntimeEvents(await readRuntimeEvents(runtimeEventLogPath), { + ...(sessionId ? { sessionId } : {}), + limit, + }); + + sendJson(response, 200, { + ok: true, + events, + }); + return true; + } + + if (pathname === "/api/sessions") { + if (method !== "GET") { + methodNotAllowed(response); + return true; + } + + const { stateRoot, runtimeEventLogPath } = await readRuntimePaths(configStore, workspaceRoot); + const sessions = await buildSessionSummaries({ + stateRoot, + runtimeEventLogPath, + }); + + sendJson(response, 200, { + ok: true, + sessions, + runs: runService.listRuns(), + }); + return true; + } + + if (pathname === "/api/sessions/graph") { + if (method !== "GET") { + methodNotAllowed(response); + return true; + } + + const sessionId = requestUrl.searchParams.get("sessionId") ?? ""; + if (!sessionId) { + sendJson(response, 400, { + ok: false, + error: 'Query parameter "sessionId" is required.', + }); + return true; + } + + const { stateRoot, runtimeEventLogPath } = await readRuntimePaths(configStore, workspaceRoot); + const explicitManifestPath = requestUrl.searchParams.get("manifestPath"); + const runMeta = await readRunMetaBySession({ stateRoot, sessionId }); + const manifestPath = explicitManifestPath ?? runMeta?.manifestPath; + + if (!manifestPath) { + sendJson(response, 400, { + ok: false, + error: "No manifestPath available for this session. Provide one in query string.", + }); + return true; + } + + const manifestRecord = await manifestStore.read(manifestPath); + const graph = await buildSessionGraphInsight({ + stateRoot, + runtimeEventLogPath, + sessionId, + manifest: manifestRecord.manifest, + }); + + sendJson(response, 200, { + ok: true, + graph, + manifestPath: manifestRecord.path, + }); + return true; + } + + if (pathname === "/api/runs") { + if (method === "GET") { + sendJson(response, 200, { + ok: true, + runs: runService.listRuns(), + }); + return true; + } + + if (method === "POST") { + const body = await parseJsonBody(request); + if (typeof body.prompt !== "string" || body.prompt.trim().length === 0) { + sendJson(response, 400, { + ok: false, + error: 'Field "prompt" is required.', + }); + return true; + } + + const manifestSource = (() => { + if (body.manifest !== undefined) { + return body.manifest; + } + if (typeof body.manifestPath === "string" && body.manifestPath.trim().length > 0) { + return undefined; + } + return undefined; + })(); + + const resolvedManifest = manifestSource ?? (() => { + if (!body.manifestPath) { + return undefined; + } + return body.manifestPath; + })(); + + let manifest: unknown; + if (typeof resolvedManifest === "string") { + manifest = (await manifestStore.read(resolvedManifest)).source; + } else if (resolvedManifest !== undefined) { + manifest = resolvedManifest; + } + + if (!manifest) { + sendJson(response, 400, { + ok: false, + error: "A manifest or manifestPath is required to start a run.", + }); + return true; + } + + const record = await runService.startRun({ + prompt: body.prompt, + manifest, + manifestPath: body.manifestPath, + sessionId: body.sessionId, + topologyHint: body.topologyHint, + initialFlags: ensureBooleanRecord(body.initialFlags), + runtimeContextOverrides: ensureRuntimeContext(body.runtimeContextOverrides), + simulateValidationNodeIds: ensureStringArray(body.simulateValidationNodeIds), + executionMode: ensureExecutionMode(body.executionMode), + provider: ensureProvider(body.provider), + }); + + sendJson(response, 202, { + ok: true, + run: record, + }); + return true; + } + + methodNotAllowed(response); + return true; + } + + if (pathname.startsWith("/api/runs/") && pathname.endsWith("/cancel")) { + if (method !== "POST") { + methodNotAllowed(response); + return true; + } + + const runId = toRelativePathFromApi(pathname.slice("/api/runs/".length, -"/cancel".length)); + const run = await runService.cancelRun(runId); + sendJson(response, 200, { + ok: true, + run, + }); + return true; + } + + if (pathname.startsWith("/api/runs/")) { + if (method !== "GET") { + methodNotAllowed(response); + return true; + } + + const runId = toRelativePathFromApi(pathname.slice("/api/runs/".length)); + const run = runService.getRun(runId); + if (!run) { + sendJson(response, 404, { + ok: false, + error: `Run \"${runId}\" was not found.`, + }); + return true; + } + + sendJson(response, 200, { + ok: true, + run, + }); + return true; + } + + notFound(response); + return true; + } catch (error) { + sendJson(response, 400, { + ok: false, + error: error instanceof Error ? error.message : String(error), + }); + return true; + } +} + +export async function startUiServer(input: { + workspaceRoot: string; + port?: number; + host?: string; +}): Promise<{ + close: () => Promise; +}> { + const workspaceRoot = resolve(input.workspaceRoot); + const staticRoot = resolve(workspaceRoot, "src/ui/public"); + + const configStore = new UiConfigStore({ workspaceRoot }); + const manifestStore = new ManifestStore({ workspaceRoot }); + const runService = new UiRunService({ workspaceRoot }); + + const server = createServer(async (request, response) => { + const handledApi = await handleApiRequest({ + request, + response, + workspaceRoot, + configStore, + manifestStore, + runService, + }); + + if (handledApi) { + return; + } + + const requestUrl = new URL(request.url ?? "/", `http://${request.headers.host ?? "localhost"}`); + const pathname = requestUrl.pathname === "/" ? "/index.html" : requestUrl.pathname; + const cleanPath = pathname.replace(/^\//, ""); + if (cleanPath.includes("..")) { + notFound(response); + return; + } + + const staticPath = resolve(staticRoot, cleanPath); + const served = await serveStaticFile({ + response, + filePath: staticPath, + }); + + if (served) { + return; + } + + const fallbackServed = await serveStaticFile({ + response, + filePath: resolve(staticRoot, "index.html"), + }); + + if (!fallbackServed) { + notFound(response); + } + }); + + const host = input.host ?? "127.0.0.1"; + const port = input.port ?? parsePort(process.env.AGENT_UI_PORT); + + await new Promise((resolveReady, rejectReady) => { + server.once("error", rejectReady); + server.listen(port, host, () => { + server.off("error", rejectReady); + resolveReady(); + }); + }); + + console.log(`AI Ops UI listening at http://${host}:${String(port)}`); + + return { + close: async () => { + await new Promise((resolveClose, rejectClose) => { + server.close((error) => { + if (error) { + rejectClose(error); + return; + } + resolveClose(); + }); + }); + }, + }; +} + +async function main(): Promise { + const port = parsePort(process.env.AGENT_UI_PORT); + await startUiServer({ + workspaceRoot: process.cwd(), + port, + host: process.env.AGENT_UI_HOST ?? "127.0.0.1", + }); +} + +if (process.argv[1] && import.meta.url === pathToFileURL(process.argv[1]).href) { + main().catch((error: unknown) => { + console.error(error instanceof Error ? error.message : String(error)); + process.exitCode = 1; + }); +} diff --git a/src/ui/session-insights.ts b/src/ui/session-insights.ts new file mode 100644 index 0000000..7328b3c --- /dev/null +++ b/src/ui/session-insights.ts @@ -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 { + 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 { + const [sessionDirectories, allEvents] = await Promise.all([ + listSessionDirectories(input.stateRoot), + readRuntimeEvents(input.runtimeEventLogPath), + ]); + + const sessionIds = new Set(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( + (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> { + const handoffDirectory = resolve(stateRoot, sessionId, "handoffs"); + const output: Record = {}; + + 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; + 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(); + 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 { + const output = new Set(); + 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 { + 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(); + const domainEventsByNode = new Map(); + + 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((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 { + 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; + } +} diff --git a/tests/env-store.test.ts b/tests/env-store.test.ts new file mode 100644 index 0000000..d7b99ff --- /dev/null +++ b/tests/env-store.test.ts @@ -0,0 +1,35 @@ +import test from "node:test"; +import assert from "node:assert/strict"; +import { mkdtemp, readFile, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { resolve } from "node:path"; +import { parseEnvFile, writeEnvFileUpdates } from "../src/ui/env-store.js"; + +test("parseEnvFile handles missing files", async () => { + const root = await mkdtemp(resolve(tmpdir(), "ai-ops-env-store-")); + const envPath = resolve(root, ".env"); + + const parsed = await parseEnvFile(envPath); + assert.deepEqual(parsed.values, {}); + assert.deepEqual(parsed.lines, []); +}); + +test("writeEnvFileUpdates merges and appends keys", async () => { + const root = await mkdtemp(resolve(tmpdir(), "ai-ops-env-store-")); + const envPath = resolve(root, ".env"); + + await writeFile(envPath, "FOO=bar\nAGENT_MAX_CONCURRENT=4\n", "utf8"); + + const updated = await writeEnvFileUpdates(envPath, { + AGENT_MAX_CONCURRENT: "9", + AGENT_RUNTIME_DISCORD_MIN_SEVERITY: "warning", + }); + + assert.equal(updated.values.FOO, "bar"); + assert.equal(updated.values.AGENT_MAX_CONCURRENT, "9"); + assert.equal(updated.values.AGENT_RUNTIME_DISCORD_MIN_SEVERITY, "warning"); + + const rendered = await readFile(envPath, "utf8"); + assert.match(rendered, /AGENT_MAX_CONCURRENT=9/); + assert.match(rendered, /AGENT_RUNTIME_DISCORD_MIN_SEVERITY=warning/); +}); diff --git a/tests/provider-executor.test.ts b/tests/provider-executor.test.ts new file mode 100644 index 0000000..05e8e00 --- /dev/null +++ b/tests/provider-executor.test.ts @@ -0,0 +1,66 @@ +import test from "node:test"; +import assert from "node:assert/strict"; +import { parseActorExecutionResultFromModelOutput } from "../src/ui/provider-executor.js"; + +test("parseActorExecutionResultFromModelOutput parses strict JSON payload", () => { + const parsed = parseActorExecutionResultFromModelOutput({ + rawText: JSON.stringify({ + status: "validation_fail", + payload: { + summary: "missing test", + }, + stateFlags: { + needs_fix: true, + }, + stateMetadata: { + stage: "qa", + }, + events: [ + { + type: "validation_failed", + payload: { + summary: "failed", + }, + }, + ], + failureKind: "soft", + failureCode: "missing_test", + }), + }); + + assert.equal(parsed.status, "validation_fail"); + assert.equal(parsed.payload?.summary, "missing test"); + assert.equal(parsed.stateFlags?.needs_fix, true); + assert.equal(parsed.stateMetadata?.stage, "qa"); + assert.equal(parsed.events?.[0]?.type, "validation_failed"); + assert.equal(parsed.failureKind, "soft"); + assert.equal(parsed.failureCode, "missing_test"); +}); + +test("parseActorExecutionResultFromModelOutput parses fenced JSON", () => { + const parsed = parseActorExecutionResultFromModelOutput({ + rawText: [ + "Here is the result", + "```json", + JSON.stringify({ + status: "success", + payload: { + code: "done", + }, + }), + "```", + ].join("\n"), + }); + + assert.equal(parsed.status, "success"); + assert.equal(parsed.payload?.code, "done"); +}); + +test("parseActorExecutionResultFromModelOutput falls back when response is not JSON", () => { + const parsed = parseActorExecutionResultFromModelOutput({ + rawText: "Implemented update successfully.", + }); + + assert.equal(parsed.status, "success"); + assert.equal(parsed.payload?.assistantResponse, "Implemented update successfully."); +}); diff --git a/tests/session-insights.test.ts b/tests/session-insights.test.ts new file mode 100644 index 0000000..1b6b11e --- /dev/null +++ b/tests/session-insights.test.ts @@ -0,0 +1,207 @@ +import test from "node:test"; +import assert from "node:assert/strict"; +import { mkdir, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { resolve } from "node:path"; +import { mkdtemp } from "node:fs/promises"; +import { buildSessionGraphInsight, buildSessionSummaries } from "../src/ui/session-insights.js"; +import { parseAgentManifest } from "../src/agents/manifest.js"; + +function createManifest() { + return parseAgentManifest({ + schemaVersion: "1", + topologies: ["sequential", "retry-unrolled"], + personas: [ + { + id: "planner", + displayName: "Planner", + systemPromptTemplate: "Plan", + toolClearance: { + allowlist: ["read_file"], + banlist: [], + }, + }, + ], + relationships: [], + topologyConstraints: { + maxDepth: 3, + maxRetries: 2, + }, + pipeline: { + entryNodeId: "n1", + nodes: [ + { + id: "n1", + actorId: "a1", + personaId: "planner", + topology: { kind: "sequential" }, + }, + { + id: "n2", + actorId: "a2", + personaId: "planner", + topology: { kind: "retry-unrolled" }, + }, + ], + edges: [ + { + from: "n1", + to: "n2", + event: "validation_failed", + }, + ], + }, + }); +} + +test("buildSessionGraphInsight maps attempts, edge visits, and sandbox payload", async () => { + const root = await mkdtemp(resolve(tmpdir(), "ai-ops-session-insights-")); + const stateRoot = resolve(root, "state"); + const sessionId = "session-1"; + const handoffDir = resolve(stateRoot, sessionId, "handoffs"); + const runtimeLogPath = resolve(root, "runtime-events.ndjson"); + + await mkdir(handoffDir, { recursive: true }); + await writeFile( + resolve(handoffDir, "n2.json"), + `${JSON.stringify({ + nodeId: "n2", + fromNodeId: "n1", + payload: {}, + createdAt: new Date().toISOString(), + })}\n`, + "utf8", + ); + + const lines = [ + { + id: "1", + timestamp: "2026-01-01T00:00:00.000Z", + type: "session.started", + severity: "info", + message: "started", + sessionId, + }, + { + id: "2", + timestamp: "2026-01-01T00:00:01.000Z", + type: "node.attempt.completed", + severity: "info", + message: "n1 success", + sessionId, + nodeId: "n1", + attempt: 1, + usage: { durationMs: 100, costUsd: 0.001 }, + metadata: { + status: "success", + executionContext: { phase: "n1", allowedTools: ["read_file"] }, + }, + }, + { + id: "3", + timestamp: "2026-01-01T00:00:02.000Z", + type: "node.attempt.completed", + severity: "warning", + message: "n2 validation", + sessionId, + nodeId: "n2", + attempt: 1, + usage: { durationMs: 140, costUsd: 0.002 }, + metadata: { + status: "validation_fail", + retrySpawned: true, + subtasks: ["fix tests"], + executionContext: { phase: "n2", allowedTools: ["read_file"] }, + }, + }, + { + id: "4", + timestamp: "2026-01-01T00:00:03.000Z", + type: "node.attempt.completed", + severity: "info", + message: "n2 success", + sessionId, + nodeId: "n2", + attempt: 2, + usage: { durationMs: 120, costUsd: 0.0025 }, + metadata: { + status: "success", + executionContext: { phase: "n2", allowedTools: ["read_file"] }, + }, + }, + { + id: "5", + timestamp: "2026-01-01T00:00:04.000Z", + type: "session.completed", + severity: "info", + message: "completed", + sessionId, + metadata: { + status: "success", + }, + }, + ]; + + await writeFile(runtimeLogPath, `${lines.map((line) => JSON.stringify(line)).join("\n")}\n`, "utf8"); + + const manifest = createManifest(); + const graph = await buildSessionGraphInsight({ + stateRoot, + runtimeEventLogPath: runtimeLogPath, + sessionId, + manifest, + }); + + assert.equal(graph.status, "success"); + assert.equal(graph.nodes.length, 2); + + const node2 = graph.nodes.find((node) => node.nodeId === "n2"); + assert.ok(node2); + assert.equal(node2.attemptCount, 2); + assert.equal(node2.subtaskCount, 1); + assert.equal(node2.sandboxPayload?.phase, "n2"); + + const edge = graph.edges.find((entry) => entry.from === "n1" && entry.to === "n2"); + assert.ok(edge); + assert.equal(edge.visited, true); + assert.equal(edge.trigger, "event:validation_failed"); +}); + +test("buildSessionSummaries reflects aborted failed session", async () => { + const root = await mkdtemp(resolve(tmpdir(), "ai-ops-session-insights-")); + const stateRoot = resolve(root, "state"); + const sessionId = "session-abort"; + const runtimeLogPath = resolve(root, "runtime-events.ndjson"); + + await mkdir(resolve(stateRoot, sessionId), { recursive: true }); + + const lines = [ + { + id: "1", + timestamp: "2026-01-01T00:00:00.000Z", + type: "session.started", + severity: "info", + message: "started", + sessionId, + }, + { + id: "2", + timestamp: "2026-01-01T00:00:01.000Z", + type: "session.failed", + severity: "critical", + message: "Pipeline aborted after hard failures.", + sessionId, + }, + ]; + + await writeFile(runtimeLogPath, `${lines.map((line) => JSON.stringify(line)).join("\n")}\n`, "utf8"); + + const sessions = await buildSessionSummaries({ + stateRoot, + runtimeEventLogPath: runtimeLogPath, + }); + + assert.equal(sessions.length, 1); + assert.equal(sessions[0]?.status, "failure"); + assert.equal(sessions[0]?.aborted, true); +});