From 0aaae47b9ac96e50113223fa61bc87fac67f3155 Mon Sep 17 00:00:00 2001 From: NarwhalChen Date: Sun, 2 Aug 2026 19:35:08 +0800 Subject: [PATCH 1/2] feat(app): add local agent trace DAG and evals --- packages/app/package.json | 1 + packages/app/scripts/trace-eval.mts | 135 +++++ packages/app/scripts/web-bridge-dev.mts | 20 +- .../app/src/electron/agent-host/executor.ts | 56 +- .../app/src/electron/agent-host/turn-log.ts | 185 ------- packages/app/src/electron/main.ts | 20 +- .../app/src/electron/trace/agent-host.test.ts | 189 +++++++ packages/app/src/electron/trace/agent-host.ts | 493 ++++++++++++++++++ .../app/src/electron/trace/evaluation.test.ts | 68 +++ packages/app/src/electron/trace/evaluation.ts | 150 ++++++ .../app/src/electron/trace/projector.test.ts | 91 ++++ packages/app/src/electron/trace/projector.ts | 137 +++++ packages/app/src/electron/trace/schema.ts | 61 +++ packages/app/src/electron/trace/store.test.ts | 99 ++++ packages/app/src/electron/trace/store.ts | 254 +++++++++ packages/app/src/shared/types/trace.ts | 101 ++++ 16 files changed, 1833 insertions(+), 227 deletions(-) create mode 100644 packages/app/scripts/trace-eval.mts delete mode 100644 packages/app/src/electron/agent-host/turn-log.ts create mode 100644 packages/app/src/electron/trace/agent-host.test.ts create mode 100644 packages/app/src/electron/trace/agent-host.ts create mode 100644 packages/app/src/electron/trace/evaluation.test.ts create mode 100644 packages/app/src/electron/trace/evaluation.ts create mode 100644 packages/app/src/electron/trace/projector.test.ts create mode 100644 packages/app/src/electron/trace/projector.ts create mode 100644 packages/app/src/electron/trace/schema.ts create mode 100644 packages/app/src/electron/trace/store.test.ts create mode 100644 packages/app/src/electron/trace/store.ts create mode 100644 packages/app/src/shared/types/trace.ts diff --git a/packages/app/package.json b/packages/app/package.json index 4a6e362d..a4936db6 100644 --- a/packages/app/package.json +++ b/packages/app/package.json @@ -27,6 +27,7 @@ "test:all": "vitest run", "memory:eval": "tsx scripts/memory-eval.mts", "memory:eval:real": "tsx scripts/memory-eval.mts --real", + "trace:eval": "tsx scripts/trace-eval.mts", "automation": "WDIO_LOG_LEVEL=silent tsx automation/server.ts", "automation:prepare": "tsx automation/check-runtime.ts && electron-forge package && tsx automation/prepare-driver.ts", "automation:typecheck": "tsc --noEmit -p automation/tsconfig.json" diff --git a/packages/app/scripts/trace-eval.mts b/packages/app/scripts/trace-eval.mts new file mode 100644 index 00000000..7c78f28d --- /dev/null +++ b/packages/app/scripts/trace-eval.mts @@ -0,0 +1,135 @@ +import { randomUUID } from "node:crypto"; +import { mkdir, writeFile } from "node:fs/promises"; +import { resolve } from "node:path"; +import { evaluateTrace } from "@/electron/trace/evaluation"; +import { projectTrace } from "@/electron/trace/projector"; +import { LocalTraceStore } from "@/electron/trace/store"; +import type { + TraceEvent, + TraceEventInput, + TraceSpanKind, +} from "@/shared/types/trace"; +import { TRACE_SCHEMA_VERSION } from "@/shared/types/trace"; + +const pathArgument = process.argv.find((argument) => + argument.startsWith("--path="), +); +const requestedPath = pathArgument?.slice("--path=".length); +const runId = new Date().toISOString().replaceAll(/[:.]/g, "-"); +const outputDirectory = resolve( + ".automation", + "artifacts", + "trace-eval", + runId, +); + +function syntheticTrace( + name: string, + parallelTasks: number, + terminalStatus: "ok" | "interrupted" = "ok", +): TraceEvent[] { + const traceId = `synthetic:${name}`; + const occurredAt = "2026-01-01T00:00:00.000Z"; + const inputs: TraceEventInput[] = []; + const add = ( + spanId: string, + spanKind: TraceSpanKind, + type: "span.start" | "span.end", + parentSpanId?: string, + metrics?: Record, + ) => { + inputs.push({ + eventId: `${spanId}:${type}`, + traceId, + spanId, + parentSpanId, + occurredAt, + emitter: + spanKind === "model" + ? "provider" + : spanKind === "tool" + ? "tool" + : "main", + type, + spanKind, + name: spanKind, + status: type === "span.end" ? terminalStatus : undefined, + attributes: + spanKind === "tool" ? { toolName: `tool-${spanId.at(-1)}` } : undefined, + metrics, + classification: "P0", + }); + }; + + add("mission", "mission", "span.start"); + for (let index = 0; index < parallelTasks; index += 1) { + const suffix = String(index + 1); + add(`task-${suffix}`, "task", "span.start", "mission"); + add(`run-${suffix}`, "run", "span.start", `task-${suffix}`); + add(`turn-${suffix}`, "turn", "span.start", `run-${suffix}`); + add(`model-${suffix}`, "model", "span.start", `turn-${suffix}`); + add(`tool-${suffix}`, "tool", "span.start", `model-${suffix}`); + add(`tool-${suffix}`, "tool", "span.end", `model-${suffix}`); + add(`model-${suffix}`, "model", "span.end", `turn-${suffix}`, { + inputTokens: 100, + outputTokens: 50, + totalTokens: 150, + }); + add(`turn-${suffix}`, "turn", "span.end", `run-${suffix}`); + add(`run-${suffix}`, "run", "span.end", `task-${suffix}`); + } + return inputs.map((input, index) => ({ + ...input, + eventId: input.eventId ?? randomUUID(), + schemaVersion: TRACE_SCHEMA_VERSION, + sequence: index + 1, + recordedAt: occurredAt, + })) as TraceEvent[]; +} + +const cases = []; +if (requestedPath) { + const store = new LocalTraceStore(resolve(requestedPath)); + for (const traceId of await store.listTraceIds()) { + cases.push(evaluateTrace(await store.graph(traceId))); + } +} else { + for (const [name, tasks, terminalStatus] of [ + ["single-task", 1, "ok"], + ["parallel-two-task", 2, "ok"], + ["parallel-four-task", 4, "ok"], + ["interrupted-recovery", 1, "interrupted"], + ] as const) { + const events = syntheticTrace(name, tasks, terminalStatus); + cases.push(evaluateTrace(projectTrace(events, `synthetic:${name}`))); + } +} + +const report = { + schemaVersion: 1, + runId, + generatedAt: new Date().toISOString(), + source: requestedPath + ? resolve(requestedPath) + : "built-in synthetic scenarios", + summary: { + traces: cases.length, + passed: cases.filter((entry) => entry.summary.score === 1).length, + meanScore: + cases.length === 0 + ? 0 + : Math.round( + (cases.reduce((total, entry) => total + entry.summary.score, 0) / + cases.length) * + 10_000, + ) / 10_000, + }, + cases, +}; + +await mkdir(outputDirectory, { recursive: true }); +const outputPath = resolve(outputDirectory, "report.json"); +await writeFile(outputPath, `${JSON.stringify(report, null, 2)}\n`, "utf8"); +console.log(JSON.stringify({ ...report.summary, outputPath }, null, 2)); + +if (report.summary.passed !== report.summary.traces) process.exitCode = 1; diff --git a/packages/app/scripts/web-bridge-dev.mts b/packages/app/scripts/web-bridge-dev.mts index feb7021e..1001aa6d 100644 --- a/packages/app/scripts/web-bridge-dev.mts +++ b/packages/app/scripts/web-bridge-dev.mts @@ -23,6 +23,8 @@ import { JsonAgentHostJobRepository } from "@/electron/agent-host/repository"; import { AgentHostRendererBridge } from "@/electron/agent-host/renderer-bridge"; import { LocalAiAgentHostExecutor } from "@/electron/agent-host/executor"; import { withAgentHostTools } from "@/electron/ai/agent-host-tools"; +import { AgentHostTraceRecorder } from "@/electron/trace/agent-host"; +import { LocalTraceStore } from "@/electron/trace/store"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { @@ -67,8 +69,12 @@ const runtime = new LocalAiRuntime({ // The bridge owns one sender per connected tab; the host talks to whichever // one is live. -/** Readable from outside the browser, unlike anything in the renderer's DB. */ -const TURN_LOG_PATH = join(tmpdir(), "convera-dev-agent-turns.jsonl"); +const traceStore = new LocalTraceStore( + join(tmpdir(), "convera-dev-agent-traces.jsonl"), +); +const traceRecorder = new AgentHostTraceRecorder(traceStore, { + onError: (error) => console.warn("[agent-trace] write failed", error), +}); const agentHostBridge = new AgentHostRendererBridge(() => { // The newest live tab, not the first ever seen: a stale sender from a closed @@ -89,12 +95,18 @@ const agentHost = new AgentHost({ executor: new LocalAiAgentHostExecutor( runtime, agentHostBridge, - TURN_LOG_PATH, + traceRecorder, ), startPaused: true, }); -agentHost.subscribe((event) => agentHostBridge.emit(event)); +agentHost.subscribe((event) => { + void traceRecorder.record(event); + agentHostBridge.emit(event); +}); await agentHost.initialize(); +await Promise.all( + (await agentHost.listJobs()).map((job) => traceRecorder.recordJob(job)), +); const recordingIPC = createRecordingIpcMain({ handle: () => {}, diff --git a/packages/app/src/electron/agent-host/executor.ts b/packages/app/src/electron/agent-host/executor.ts index 8a21d318..69bc3c6c 100644 --- a/packages/app/src/electron/agent-host/executor.ts +++ b/packages/app/src/electron/agent-host/executor.ts @@ -10,9 +10,9 @@ import type { LocalAIStreamEvent, } from "@/shared/types/local-ai"; import { WORKSPACE_QUERY_INTERACTION } from "@/shared/types/workspace-perception"; +import type { AgentHostTraceRecorder } from "@/electron/trace/agent-host"; import type { AgentHostExecutor } from "./host"; import type { AgentHostRendererBridge } from "./renderer-bridge"; -import { TurnLogger } from "./turn-log"; /** * A turn that ends without calling the speech tool said nothing at all: the @@ -69,8 +69,7 @@ export class LocalAiAgentHostExecutor implements AgentHostExecutor { constructor( private readonly runtime: LocalAIRuntimeService, private readonly bridge: AgentHostRendererBridge, - /** Where each turn's tool sequence is written; omit to record nothing. */ - private readonly turnLogPath?: string, + private readonly trace?: AgentHostTraceRecorder, ) {} async execute( @@ -98,37 +97,19 @@ export class LocalAiAgentHostExecutor implements AgentHostExecutor { }, }; this.activeRequests.set(job.id, request.requestId); - const log = this.turnLogPath - ? new TurnLogger(this.turnLogPath) - : undefined; - const flush = () => - log?.flush({ - id: job.id, - requestId: request.requestId, - conversationId: job.conversationId, - memberId: job.agentMemberId, - mode: job.mode, - }); - if (await this.runTurn(request, job, emit, log)) { - await flush(); - return; - } + if (await this.runTurn(request, job, emit)) return; // Silence is a complete answer on an open floor; only a direct question // left unanswered is worth a second ask. if (job.mode !== "direct") { - await flush(); return; } // Announced before the second turn starts, so the renderer can hold the // typing indicator across the gap rather than retiring it and lighting // it again for the same reply. emit({ type: "retrying", jobId: job.id }); - log?.noteRetry(); - if (await this.runTurn(remind(request), job, emit, log)) { - await flush(); + if (await this.runTurn(remind(request), job, emit, request.turnId)) { return; } - await flush(); throw new Error( "The agent completed a direct offer without sending a message.", ); @@ -142,20 +123,27 @@ export class LocalAiAgentHostExecutor implements AgentHostExecutor { request: LocalAIChatRequest, job: AgentHostJob, emit: (event: AgentHostEvent) => void, - log?: TurnLogger, + previousTurnId?: string, ): Promise { let streamError: Error | undefined; let spoke = false; - await this.runtime.startChat(request, (event) => { - log?.record(event); - if (event.type === "error") { - streamError = new Error(event.error.message); - } - if (isSpeech(event)) spoke = true; - emit({ type: "stream", jobId: job.id, event }); - }); - if (streamError) throw streamError; - return spoke; + const trace = await this.trace?.beginTurn(job, request, previousTurnId); + try { + await this.runtime.startChat(request, (event) => { + trace?.record(event); + if (event.type === "error") { + streamError = new Error(event.error.message); + } + if (isSpeech(event)) spoke = true; + emit({ type: "stream", jobId: job.id, event }); + }); + if (streamError) throw streamError; + await trace?.complete(); + return spoke; + } catch (error) { + await trace?.complete(error); + throw error; + } } async cancel(job: AgentHostJob): Promise { diff --git a/packages/app/src/electron/agent-host/turn-log.ts b/packages/app/src/electron/agent-host/turn-log.ts deleted file mode 100644 index a40adf38..00000000 --- a/packages/app/src/electron/agent-host/turn-log.ts +++ /dev/null @@ -1,185 +0,0 @@ -/** - * What each agent turn actually did, on disk. - * - * "Three colleagues showed as typing and one answered" is a question about - * events that leave no trace anywhere else: the room shows a message or it - * shows nothing, and a turn that opened the speech tool and abandoned it - * looks identical to one that was never asked. - * - * This lives in the main process and writes a plain JSONL file, deliberately: - * the renderer's IndexedDB is reachable only from inside the browser, so a - * log kept there cannot be read while diagnosing — by a script, by a support - * request, or by anyone who is not sitting in front of the open devtools. - */ -import { appendFile, readFile, writeFile } from "node:fs/promises"; -import type { LocalAIStreamEvent } from "@/shared/types/local-ai"; - -export interface TurnLogStep { - /** Bare tool name — `send_message`, `read_channel` — or `(re-asked)`. */ - tool: string; - /** Milliseconds from the turn's first event. */ - at: number; - outcome: "started" | "completed" | "error" | "denied" | "note"; - detail?: string; -} - -export interface TurnLogEntry { - at: string; - jobId: string; - requestId: string; - conversationId: string; - memberId: string; - mode: string; - steps: TurnLogStep[]; - spoke: boolean; - /** The turn's own output: invisible to the room, and usually the reason. */ - turnText?: string; - error?: string; - durationMs: number; -} - -const TURN_TEXT_MAX = 400; -const DETAIL_MAX = 160; -/** Rewritten when it grows past this, keeping the newest half. */ -const MAX_LINES = 500; - -function toolNameOf(chunk: unknown): string | undefined { - const name = (chunk as { toolName?: string }).toolName; - return typeof name === "string" ? name.replace(/^workspace:/, "") : undefined; -} - -/** - * Accumulates one turn. The executor owns the lifetime: it sees every stream - * event already, so nothing extra has to be plumbed through. - */ -export class TurnLogger { - private readonly startedAt = Date.now(); - private readonly steps: TurnLogStep[] = []; - private readonly toolNames = new Map(); - private turnText = ""; - private spoke = false; - private error?: string; - - constructor(private readonly path: string) {} - - record(event: LocalAIStreamEvent): void { - const at = Date.now() - this.startedAt; - - if (event.type === "error") { - this.error = event.error.message; - return; - } - - if (event.type === "interaction") { - const input = event.input as { - kind?: string; - content?: string; - emoji?: string; - } | null; - if (input?.kind === "send_message") { - this.spoke = true; - const step = this.lastStarted("send_message"); - if (step) step.detail = (input.content ?? "").slice(0, DETAIL_MAX); - } - // Same rule as the executor: a reaction is an act, not silence. - if (input?.kind === "add_reaction") { - this.spoke = true; - const step = this.lastStarted("add_reaction"); - if (step) step.detail = input.emoji ?? ""; - } - return; - } - - if (event.type !== "ui-message") return; - const chunk = event.chunk as { - type?: string; - delta?: string; - toolCallId?: string; - errorText?: string; - }; - - if (chunk.type === "text-delta" && typeof chunk.delta === "string") { - this.turnText = (this.turnText + chunk.delta).slice(0, TURN_TEXT_MAX); - return; - } - - const callId = chunk.toolCallId; - if (typeof callId !== "string" || !callId) return; - - if (chunk.type === "tool-input-start") { - const tool = toolNameOf(chunk) ?? "unknown"; - this.toolNames.set(callId, tool); - this.steps.push({ tool, at, outcome: "started" }); - return; - } - - const tool = this.toolNames.get(callId); - if (!tool) return; - const step = this.lastStarted(tool); - if (!step) return; - - if (chunk.type === "tool-output-available") step.outcome = "completed"; - else if (chunk.type === "tool-output-denied") step.outcome = "denied"; - else if ( - chunk.type === "tool-output-error" || - chunk.type === "tool-input-error" - ) { - step.outcome = "error"; - if (typeof chunk.errorText === "string") { - step.detail = chunk.errorText.slice(0, DETAIL_MAX); - } - } - } - - /** Marks where a silent direct offer was asked again. */ - noteRetry(): void { - this.steps.push({ - tool: "(re-asked)", - at: Date.now() - this.startedAt, - outcome: "note", - detail: "ended without speaking, so it was asked once more", - }); - } - - async flush(job: { - id: string; - requestId: string; - conversationId: string; - memberId: string; - mode: string; - }): Promise { - const entry: TurnLogEntry = { - at: new Date().toISOString(), - jobId: job.id, - requestId: job.requestId, - conversationId: job.conversationId, - memberId: job.memberId, - mode: job.mode, - steps: this.steps, - spoke: this.spoke, - ...(this.turnText.trim() ? { turnText: this.turnText.trim() } : {}), - ...(this.error ? { error: this.error } : {}), - durationMs: Date.now() - this.startedAt, - }; - try { - await appendFile(this.path, `${JSON.stringify(entry)}\n`, "utf8"); - await this.trim(); - } catch { - // A diagnostic must never break the turn it was watching. - } - } - - private lastStarted(tool: string): TurnLogStep | undefined { - for (let index = this.steps.length - 1; index >= 0; index -= 1) { - const step = this.steps[index]; - if (step.tool === tool && step.outcome === "started") return step; - } - return undefined; - } - - private async trim(): Promise { - const lines = (await readFile(this.path, "utf8")).split("\n"); - if (lines.length <= MAX_LINES) return; - await writeFile(this.path, lines.slice(-MAX_LINES / 2).join("\n"), "utf8"); - } -} diff --git a/packages/app/src/electron/main.ts b/packages/app/src/electron/main.ts index 079d5217..846622e4 100644 --- a/packages/app/src/electron/main.ts +++ b/packages/app/src/electron/main.ts @@ -23,6 +23,8 @@ import { JsonAgentHostJobRepository } from "@/electron/agent-host/repository"; import { AgentHostRendererBridge } from "@/electron/agent-host/renderer-bridge"; import { LocalAiAgentHostExecutor } from "@/electron/agent-host/executor"; import { withAgentHostTools } from "@/electron/ai/agent-host-tools"; +import { AgentHostTraceRecorder } from "@/electron/trace/agent-host"; +import { LocalTraceStore } from "@/electron/trace/store"; import { SANDBOX_LAYOUT } from "@/shared/types/workspace"; import { getCurrentShortcut } from "@/electro-bridge/ipc/ipc-handlers"; @@ -55,6 +57,7 @@ let memoryCoordinator: MemoryIntegrationCoordinator | undefined; let agentHost: AgentHost | undefined; let agentHostBridge: AgentHostRendererBridge | undefined; let unsubscribeAgentHost: (() => void) | undefined; +let traceStore: LocalTraceStore | undefined; let webBridge: WebBridgeHandle | undefined; let localAICleanup: Promise | undefined; let quitAfterCleanup = false; @@ -76,6 +79,7 @@ function cleanupLocalAI(): Promise { await memoryCoordinator?.dispose().catch((error) => { logger.error("Memory coordinator cleanup failed:", error); }); + await traceStore?.idle(); })(); return localAICleanup; } @@ -207,6 +211,10 @@ app.whenReady().then(async () => { ? mcpToolCall(toolName, input) : callTool(serverName, toolName, input), }); + traceStore = new LocalTraceStore(join(userDataPath, "agent-traces.jsonl")); + const traceRecorder = new AgentHostTraceRecorder(traceStore, { + onError: (error) => logger.warn("Agent trace write failed:", error), + }); agentHost = new AgentHost({ repository: new JsonAgentHostJobRepository({ path: join(userDataPath, "agent-host-jobs.json"), @@ -214,14 +222,18 @@ app.whenReady().then(async () => { executor: new LocalAiAgentHostExecutor( localAIRuntime, agentHostBridge, - join(userDataPath, "agent-turns.jsonl"), + traceRecorder, ), startPaused: true, }); - unsubscribeAgentHost = agentHost.subscribe((event) => - agentHostBridge?.emit(event), - ); + unsubscribeAgentHost = agentHost.subscribe((event) => { + void traceRecorder.record(event); + agentHostBridge?.emit(event); + }); await agentHost.initialize(); + await Promise.all( + (await agentHost.listJobs()).map((job) => traceRecorder.recordJob(job)), + ); // Initialize MCP Hub asynchronously but don't block startup initializeMCPHub() diff --git a/packages/app/src/electron/trace/agent-host.test.ts b/packages/app/src/electron/trace/agent-host.test.ts new file mode 100644 index 00000000..c9ef7f88 --- /dev/null +++ b/packages/app/src/electron/trace/agent-host.test.ts @@ -0,0 +1,189 @@ +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { describe, expect, it } from "vitest"; +import type { AgentHostJob } from "@/shared/types/agent-host"; +import type { + LocalAIChatRequest, + LocalAIStreamEvent, +} from "@/shared/types/local-ai"; +import { AgentHostTraceRecorder } from "./agent-host"; +import { LocalTraceStore } from "./store"; + +const job: AgentHostJob = { + id: "run-1", + taskId: "task-1", + channelId: "channel-1", + channelKind: "channel", + conversationId: "conversation-1", + triggerMessageId: "message-1", + contextMessageIds: ["message-1"], + mode: "direct", + offeredAgentMemberIds: ["agent:one"], + agentId: "one", + agentMemberId: "agent:one", + chain: { hops: 0, invoked: ["agent:one"] }, + controlInstructions: [], + status: "running", + attempts: 1, + createdAt: "2026-01-01T00:00:00.000Z", + updatedAt: "2026-01-01T00:00:01.000Z", + startedAt: "2026-01-01T00:00:01.000Z", +}; + +const request: LocalAIChatRequest = { + requestId: "request-1", + turnId: "turn-1", + conversationId: "conversation-1", + providerId: "codex-cli", + modelId: "gpt-test", + operation: { kind: "bootstrap", messages: [] }, +}; + +function chunk(value: Record): LocalAIStreamEvent { + return { + type: "ui-message", + requestId: request.requestId, + chunk: value, + } as unknown as LocalAIStreamEvent; +} + +describe("AgentHostTraceRecorder", () => { + it("records the canonical hierarchy and correlates parallel same-name tools by call id", async () => { + const directory = await mkdtemp(join(tmpdir(), "convera-agent-trace-")); + try { + const store = new LocalTraceStore(join(directory, "traces.jsonl")); + const recorder = new AgentHostTraceRecorder(store); + const turn = await recorder.beginTurn(job, request); + turn.record( + chunk({ + type: "tool-input-start", + toolCallId: "call-1", + toolName: "workspace:read_channel", + }), + ); + turn.record( + chunk({ + type: "tool-input-start", + toolCallId: "call-2", + toolName: "workspace:read_channel", + }), + ); + turn.record( + chunk({ type: "tool-output-available", toolCallId: "call-1" }), + ); + turn.record( + chunk({ type: "tool-output-available", toolCallId: "call-2" }), + ); + turn.record({ + type: "finish", + requestId: request.requestId, + finishReason: "stop", + usage: { inputTokens: 100, outputTokens: 20, totalTokens: 120 }, + }); + await turn.complete(); + await recorder.recordJob({ + ...job, + status: "completed", + completedAt: "2026-01-01T00:00:02.000Z", + updatedAt: "2026-01-01T00:00:02.000Z", + }); + + const graph = await store.graph("trace:message-1"); + expect(graph.spans.map((span) => span.spanKind)).toEqual( + expect.arrayContaining([ + "mission", + "task", + "run", + "turn", + "model", + "tool", + "tool", + ]), + ); + expect( + graph.spans + .filter((span) => span.spanKind === "tool") + .map((span) => ({ + id: span.attributes.toolCallId, + status: span.status, + })), + ).toEqual([ + { id: "call-1", status: "ok" }, + { id: "call-2", status: "ok" }, + ]); + expect( + graph.spans.find((span) => span.spanKind === "model")?.metrics, + ).toEqual({ inputTokens: 100, outputTokens: 20, totalTokens: 120 }); + } finally { + await rm(directory, { recursive: true, force: true }); + } + }); + + it("links a silent retry to the turn it supersedes", async () => { + const directory = await mkdtemp(join(tmpdir(), "convera-agent-trace-")); + try { + const store = new LocalTraceStore(join(directory, "traces.jsonl")); + const recorder = new AgentHostTraceRecorder(store); + const first = await recorder.beginTurn(job, request); + await first.complete(); + const second = await recorder.beginTurn( + job, + { ...request, turnId: "turn-2" }, + request.turnId, + ); + await second.complete(); + const graph = await store.graph("trace:message-1"); + expect(graph.edges).toContainEqual({ + fromSpanId: "turn:turn-1", + toSpanId: "turn:turn-2", + relation: "supersedes", + }); + } finally { + await rm(directory, { recursive: true, force: true }); + } + }); + + it("closes open runtime spans when an interrupted job is recovered", async () => { + const directory = await mkdtemp(join(tmpdir(), "convera-agent-trace-")); + try { + const store = new LocalTraceStore(join(directory, "traces.jsonl")); + const recorder = new AgentHostTraceRecorder(store); + const turn = await recorder.beginTurn(job, request); + turn.record( + chunk({ + type: "tool-input-start", + toolCallId: "call-before-crash", + toolName: "workspace:read_channel", + }), + ); + await store.idle(); + + await recorder.recordJob({ + ...job, + status: "interrupted", + completedAt: "2026-01-01T00:00:02.000Z", + updatedAt: "2026-01-01T00:00:02.000Z", + }); + + const graph = await store.graph("trace:message-1"); + const runtimeSpans = graph.spans.filter((span) => + ["turn", "model", "tool"].includes(span.spanKind), + ); + expect(runtimeSpans).toHaveLength(3); + expect(runtimeSpans.every((span) => span.status === "interrupted")).toBe( + true, + ); + expect(runtimeSpans.every((span) => span.endedAt !== undefined)).toBe( + true, + ); + expect( + runtimeSpans.every( + (span) => span.attributes.recoveredAfterRestart === true, + ), + ).toBe(true); + } finally { + await rm(directory, { recursive: true, force: true }); + } + }); +}); diff --git a/packages/app/src/electron/trace/agent-host.ts b/packages/app/src/electron/trace/agent-host.ts new file mode 100644 index 00000000..adb6b4b5 --- /dev/null +++ b/packages/app/src/electron/trace/agent-host.ts @@ -0,0 +1,493 @@ +import type { AgentHostEvent, AgentHostJob } from "@/shared/types/agent-host"; +import type { + LocalAIChatRequest, + LocalAIStreamEvent, +} from "@/shared/types/local-ai"; +import type { + TraceEventInput, + TraceSpan, + TraceStatus, +} from "@/shared/types/trace"; +import { WORKSPACE_QUERY_INTERACTION } from "@/shared/types/workspace-perception"; +import type { TraceEventSink, TraceEventStore } from "./store"; + +function traceIdFor(job: AgentHostJob): string { + return `trace:${job.triggerMessageId}`; +} + +function missionSpanId(job: AgentHostJob): string { + return `mission:${job.triggerMessageId}`; +} + +function taskSpanId(job: AgentHostJob): string { + return `task:${job.taskId}`; +} + +function runSpanId(job: AgentHostJob): string { + return `run:${job.id}`; +} + +function terminalStatus(job: AgentHostJob): TraceStatus | undefined { + if (job.status === "completed") return "ok"; + if (job.status === "failed") return "error"; + if (job.status === "cancelled") return "cancelled"; + if (job.status === "interrupted") return "interrupted"; + return undefined; +} + +function toolNameOf(chunk: unknown): string | undefined { + const name = (chunk as { toolName?: string }).toolName; + return typeof name === "string" ? name.replace(/^workspace:/, "") : undefined; +} + +export class AgentHostTraceRecorder { + constructor( + private readonly sink: TraceEventStore, + private readonly options: { + now?: () => Date; + onError?: (error: unknown) => void; + } = {}, + ) {} + + async record(event: AgentHostEvent): Promise { + if (event.type === "job") await this.recordJob(event.job); + } + + async recordJob(job: AgentHostJob): Promise { + const traceId = traceIdFor(job); + const missionId = missionSpanId(job); + const taskId = taskSpanId(job); + const runId = runSpanId(job); + const common = { + traceId, + emitter: "main" as const, + classification: "P0" as const, + }; + const inputs: TraceEventInput[] = [ + { + ...common, + eventId: `${missionId}:start`, + spanId: missionId, + occurredAt: job.createdAt, + type: "span.start", + spanKind: "mission", + name: "Agent collaboration mission", + attributes: { + conversationId: job.conversationId, + channelId: job.channelId, + triggerMessageId: job.triggerMessageId, + mode: job.mode, + }, + }, + { + ...common, + eventId: `${taskId}:start`, + spanId: taskId, + parentSpanId: missionId, + occurredAt: job.createdAt, + type: "span.start", + spanKind: "task", + name: "Agent task", + attributes: { + taskId: job.taskId, + agentId: job.agentId, + agentMemberId: job.agentMemberId, + }, + }, + { + ...common, + eventId: `${runId}:start`, + spanId: runId, + parentSpanId: taskId, + links: job.parentJobId + ? [{ spanId: `run:${job.parentJobId}`, relation: "supersedes" }] + : undefined, + occurredAt: job.createdAt, + type: "span.start", + spanKind: "run", + name: "Agent run", + attributes: { + runId: job.id, + attempt: job.attempts, + jobStatus: job.status, + agentMemberId: job.agentMemberId, + }, + }, + ]; + const status = terminalStatus(job); + inputs.push({ + ...common, + eventId: `${taskId}:status:${job.status}:${job.updatedAt}`, + spanId: taskId, + parentSpanId: missionId, + occurredAt: job.updatedAt, + type: "span.event", + spanKind: "task", + name: "Task status changed", + status, + attributes: { jobStatus: job.status, currentRunId: job.id }, + }); + inputs.push({ + ...common, + eventId: `${runId}:${status ? "end" : "status"}:${job.status}:${job.updatedAt}`, + spanId: runId, + parentSpanId: taskId, + occurredAt: job.completedAt ?? job.updatedAt, + type: status ? "span.end" : "span.event", + spanKind: "run", + name: status ? "Agent run ended" : "Run status changed", + status, + attributes: { + jobStatus: job.status, + attempt: job.attempts, + ...(job.error ? { errorType: "agent-run-error" } : {}), + }, + }); + await this.safeAppend(inputs); + if (job.status === "interrupted") await this.recoverInterruptedRun(job); + } + + async beginTurn( + job: AgentHostJob, + request: LocalAIChatRequest, + previousTurnId?: string, + ): Promise { + await this.recordJob(job); + const turn = new AgentTurnTrace(this.sink, job, request, { + now: this.options.now, + onError: this.options.onError, + previousTurnId, + }); + await turn.start(); + return turn; + } + + private async safeAppend(input: TraceEventInput[]): Promise { + try { + await this.sink.append(input); + } catch (error) { + this.options.onError?.(error); + } + } + + private async recoverInterruptedRun(job: AgentHostJob): Promise { + try { + const traceId = traceIdFor(job); + const runId = runSpanId(job); + const graph = await this.sink.graph(traceId); + const children = new Map(); + for (const span of graph.spans) { + if (!span.parentSpanId) continue; + const siblings = children.get(span.parentSpanId); + if (siblings) siblings.push(span); + else children.set(span.parentSpanId, [span]); + } + const descendants: Array<{ span: TraceSpan; depth: number }> = []; + const visit = (parentSpanId: string, depth: number) => { + for (const span of children.get(parentSpanId) ?? []) { + descendants.push({ span, depth }); + visit(span.spanId, depth + 1); + } + }; + visit(runId, 1); + const occurredAt = job.completedAt ?? job.updatedAt; + const inputs = descendants + .filter( + ({ span }) => + !span.endedAt && ["turn", "model", "tool"].includes(span.spanKind), + ) + .sort((left, right) => right.depth - left.depth) + .map( + ({ span }): TraceEventInput => ({ + eventId: `${span.spanId}:recovered:interrupted`, + traceId, + spanId: span.spanId, + parentSpanId: span.parentSpanId, + occurredAt, + emitter: "main", + type: "span.end", + spanKind: span.spanKind, + name: span.name, + status: "interrupted", + attributes: { recoveredAfterRestart: true }, + classification: "P0", + }), + ); + if (inputs.length > 0) await this.safeAppend(inputs); + } catch (error) { + this.options.onError?.(error); + } + } +} + +export class AgentTurnTrace { + private readonly writes: Promise[] = []; + private readonly traceId: string; + private readonly runId: string; + private readonly turnId: string; + private readonly modelId: string; + private readonly startedAt: string; + private readonly toolNames = new Map(); + private ended = false; + private spoke = false; + + constructor( + private readonly sink: TraceEventSink, + job: AgentHostJob, + private readonly request: LocalAIChatRequest, + private readonly options: { + now?: () => Date; + onError?: (error: unknown) => void; + previousTurnId?: string; + } = {}, + ) { + this.traceId = traceIdFor(job); + this.runId = runSpanId(job); + this.turnId = `turn:${request.turnId}`; + this.modelId = `model:${request.turnId}`; + this.startedAt = this.now(); + } + + async start(): Promise { + await this.safeAppend([ + { + eventId: `${this.turnId}:start`, + traceId: this.traceId, + spanId: this.turnId, + parentSpanId: this.runId, + links: this.options.previousTurnId + ? [ + { + spanId: `turn:${this.options.previousTurnId}`, + relation: "supersedes", + }, + ] + : undefined, + occurredAt: this.startedAt, + emitter: "main", + type: "span.start", + spanKind: "turn", + name: "Agent turn", + attributes: { + turnId: this.request.turnId, + requestId: this.request.requestId, + operation: this.request.operation.kind, + conversationId: this.request.conversationId, + }, + classification: "P0", + }, + { + eventId: `${this.modelId}:start`, + traceId: this.traceId, + spanId: this.modelId, + parentSpanId: this.turnId, + occurredAt: this.startedAt, + emitter: "provider", + type: "span.start", + spanKind: "model", + name: this.request.providerId, + attributes: { + providerId: this.request.providerId, + ...(this.request.modelId ? { modelId: this.request.modelId } : {}), + }, + classification: "P0", + }, + ]); + } + + record(event: LocalAIStreamEvent): void { + if (event.type === "interaction") { + const input = event.input as { kind?: string } | null; + if ( + event.name === WORKSPACE_QUERY_INTERACTION && + (input?.kind === "send_message" || input?.kind === "add_reaction") + ) { + this.spoke = true; + } + this.queue([ + { + eventId: `${this.turnId}:interaction:${event.interactionId}`, + traceId: this.traceId, + spanId: this.turnId, + parentSpanId: this.runId, + occurredAt: this.now(), + emitter: "tool", + type: "span.event", + spanKind: "turn", + name: "Tool interaction requested", + attributes: { + interactionName: event.name, + interactionKind: event.kind, + ...(input?.kind ? { effectKind: input.kind } : {}), + }, + classification: "P0", + }, + ]); + return; + } + if (event.type === "finish") { + this.endFromStream( + ["error", "content-filter"].includes(event.finishReason) + ? "error" + : event.finishReason === "aborted" + ? "cancelled" + : "ok", + event.finishReason, + event.usage, + ); + return; + } + if (event.type === "error") { + this.endFromStream("error", "error", undefined, event.error.name); + return; + } + const chunk = event.chunk as { + type?: string; + toolCallId?: string; + errorText?: string; + }; + const callId = chunk.toolCallId; + if (!callId) return; + const toolId = `tool:${this.request.turnId}:${callId}`; + if (chunk.type === "tool-input-start") { + const toolName = toolNameOf(chunk) ?? "unknown"; + this.toolNames.set(callId, toolName); + this.queue([ + { + eventId: `${toolId}:start`, + traceId: this.traceId, + spanId: toolId, + parentSpanId: this.modelId, + occurredAt: this.now(), + emitter: "tool", + type: "span.start", + spanKind: "tool", + name: toolName, + attributes: { toolCallId: callId, toolName }, + classification: "P0", + }, + ]); + return; + } + const toolName = this.toolNames.get(callId); + if (!toolName) return; + const outcome = + chunk.type === "tool-output-available" + ? "ok" + : chunk.type === "tool-output-denied" + ? "cancelled" + : chunk.type === "tool-output-error" || + chunk.type === "tool-input-error" + ? "error" + : undefined; + if (!outcome) return; + this.queue([ + { + eventId: `${toolId}:end`, + traceId: this.traceId, + spanId: toolId, + parentSpanId: this.modelId, + occurredAt: this.now(), + emitter: "tool", + type: "span.end", + spanKind: "tool", + name: toolName, + status: outcome, + attributes: { + toolCallId: callId, + toolName, + ...(chunk.errorText ? { errorType: "tool-error" } : {}), + }, + classification: "P0", + }, + ]); + } + + async complete(error?: unknown): Promise { + if (!this.ended) { + this.endFromStream( + error ? "error" : "ok", + error ? "error" : "unknown", + undefined, + error instanceof Error ? error.name : undefined, + ); + } + await Promise.all(this.writes); + } + + private endFromStream( + status: TraceStatus, + finishReason: string, + usage?: { + inputTokens?: number; + outputTokens?: number; + totalTokens?: number; + }, + errorName?: string, + ): void { + if (this.ended) return; + this.ended = true; + const occurredAt = this.now(); + const metrics = { + ...(usage?.inputTokens !== undefined + ? { inputTokens: usage.inputTokens } + : {}), + ...(usage?.outputTokens !== undefined + ? { outputTokens: usage.outputTokens } + : {}), + ...(usage?.totalTokens !== undefined + ? { totalTokens: usage.totalTokens } + : {}), + }; + this.queue([ + { + eventId: `${this.modelId}:end`, + traceId: this.traceId, + spanId: this.modelId, + parentSpanId: this.turnId, + occurredAt, + emitter: "provider", + type: "span.end", + spanKind: "model", + name: this.request.providerId, + status, + attributes: { + finishReason, + ...(errorName ? { errorName } : {}), + }, + metrics, + classification: "P0", + }, + { + eventId: `${this.turnId}:end`, + traceId: this.traceId, + spanId: this.turnId, + parentSpanId: this.runId, + occurredAt, + emitter: "main", + type: "span.end", + spanKind: "turn", + name: "Agent turn", + status, + attributes: { spoke: this.spoke }, + classification: "P0", + }, + ]); + } + + private queue(inputs: TraceEventInput[]): void { + this.writes.push(this.safeAppend(inputs)); + } + + private async safeAppend(inputs: TraceEventInput[]): Promise { + try { + await this.sink.append(inputs); + } catch (error) { + this.options.onError?.(error); + } + } + + private now(): string { + return (this.options.now?.() ?? new Date()).toISOString(); + } +} diff --git a/packages/app/src/electron/trace/evaluation.test.ts b/packages/app/src/electron/trace/evaluation.test.ts new file mode 100644 index 00000000..3ce7a199 --- /dev/null +++ b/packages/app/src/electron/trace/evaluation.test.ts @@ -0,0 +1,68 @@ +import { describe, expect, it } from "vitest"; +import type { TraceEvent } from "@/shared/types/trace"; +import { TRACE_SCHEMA_VERSION } from "@/shared/types/trace"; +import { evaluateTrace } from "./evaluation"; +import { projectTrace } from "./projector"; + +function lifecycleEvents(): TraceEvent[] { + const definitions = [ + ["mission", "mission", undefined], + ["task", "task", "mission"], + ["run", "run", "task"], + ["turn", "turn", "run"], + ["model", "model", "turn"], + ["tool", "tool", "model"], + ] as const; + let sequence = 0; + return definitions.flatMap(([spanId, spanKind, parentSpanId]) => + (["span.start", "span.end"] as const).map((type) => ({ + schemaVersion: TRACE_SCHEMA_VERSION, + eventId: `${spanId}:${type}`, + traceId: "trace", + spanId, + parentSpanId, + sequence: ++sequence, + occurredAt: "2026-01-01T00:00:00.000Z", + recordedAt: "2026-01-01T00:00:00.000Z", + emitter: spanKind === "model" ? "provider" : "main", + type, + spanKind, + name: spanKind, + status: type === "span.end" ? ("ok" as const) : undefined, + attributes: spanKind === "tool" ? { toolName: "read_file" } : undefined, + metrics: + spanKind === "model" && type === "span.end" + ? { inputTokens: 10, outputTokens: 5, totalTokens: 15 } + : undefined, + classification: "P0" as const, + })), + ); +} + +describe("evaluateTrace", () => { + it("scores a complete canonical trace and reports process metrics", () => { + const report = evaluateTrace(projectTrace(lifecycleEvents(), "trace")); + expect(report.summary).toEqual({ checks: 5, passed: 5, score: 1 }); + expect(report.metrics).toMatchObject({ + spans: 6, + turns: 1, + modelCalls: 1, + toolCalls: 1, + repeatedToolCalls: 0, + totalTokens: 15, + }); + }); + + it("fails incomplete runtime evidence", () => { + const events = lifecycleEvents().filter( + (event) => !(event.spanId === "tool" && event.type === "span.end"), + ); + const report = evaluateTrace(projectTrace(events, "trace")); + expect( + report.checks.find((entry) => entry.id === "runtime-terminal"), + ).toMatchObject({ + passed: false, + actual: "1 incomplete runtime spans", + }); + }); +}); diff --git a/packages/app/src/electron/trace/evaluation.ts b/packages/app/src/electron/trace/evaluation.ts new file mode 100644 index 00000000..db6b34e5 --- /dev/null +++ b/packages/app/src/electron/trace/evaluation.ts @@ -0,0 +1,150 @@ +import type { TraceGraph, TraceSpan } from "@/shared/types/trace"; + +export interface TraceEvaluationCheck { + id: string; + label: string; + passed: boolean; + expected: string; + actual: string; +} + +export interface TraceEvaluationReport { + schemaVersion: 1; + traceId: string; + generatedAt: string; + summary: { + checks: number; + passed: number; + score: number; + }; + metrics: { + spans: number; + turns: number; + modelCalls: number; + toolCalls: number; + repeatedToolCalls: number; + inputTokens?: number; + outputTokens?: number; + totalTokens?: number; + }; + checks: TraceEvaluationCheck[]; +} + +const expectedParent: Partial< + Record +> = { + task: "mission", + run: "task", + turn: "run", + model: "turn", + tool: "model", +}; + +function check( + id: string, + label: string, + passed: boolean, + expected: string, + actual: string, +): TraceEvaluationCheck { + return { id, label, passed, expected, actual }; +} + +function optionalSum( + spans: TraceSpan[], + metric: "inputTokens" | "outputTokens" | "totalTokens", +): number | undefined { + const values = spans + .map((span) => span.metrics[metric]) + .filter((value): value is number => value !== undefined); + return values.length > 0 + ? values.reduce((total, value) => total + value, 0) + : undefined; +} + +export function evaluateTrace( + graph: TraceGraph, + generatedAt = new Date().toISOString(), +): TraceEvaluationReport { + const byId = new Map(graph.spans.map((span) => [span.spanId, span])); + const missionRoots = graph.rootSpanIds.filter( + (spanId) => byId.get(spanId)?.spanKind === "mission", + ); + const hierarchyViolations = graph.spans.filter((span) => { + const parentKind = expectedParent[span.spanKind]; + if (!parentKind) return false; + return ( + !span.parentSpanId || byId.get(span.parentSpanId)?.spanKind !== parentKind + ); + }); + const runtimeSpans = graph.spans.filter((span) => + ["run", "turn", "model", "tool"].includes(span.spanKind), + ); + const incompleteRuntimeSpans = runtimeSpans.filter( + (span) => !span.startedAt || !span.endedAt, + ); + const toolNames = graph.spans + .filter((span) => span.spanKind === "tool") + .map((span) => String(span.attributes.toolName ?? span.name)); + const repeatedToolCalls = toolNames.length - new Set(toolNames).size; + const checks = [ + check( + "mission-root", + "Trace has exactly one Mission root", + missionRoots.length === 1 && graph.rootSpanIds.length === 1, + "1 Mission root", + `${missionRoots.length} Mission roots, ${graph.rootSpanIds.length} total roots`, + ), + check( + "no-orphans", + "Every parent and local link resolves", + graph.orphanSpanIds.length === 0, + "0 orphan spans", + `${graph.orphanSpanIds.length} orphan spans`, + ), + check( + "acyclic", + "Parent and causal links form a DAG", + graph.cycleSpanIds.length === 0, + "0 cyclic spans", + `${graph.cycleSpanIds.length} cyclic spans`, + ), + check( + "canonical-hierarchy", + "Task, Run, Turn, Model, and Tool use the canonical parent chain", + hierarchyViolations.length === 0, + "0 hierarchy violations", + `${hierarchyViolations.length} hierarchy violations`, + ), + check( + "runtime-terminal", + "Every runtime span has start and end evidence", + incompleteRuntimeSpans.length === 0, + "0 incomplete runtime spans", + `${incompleteRuntimeSpans.length} incomplete runtime spans`, + ), + ]; + const passed = checks.filter((entry) => entry.passed).length; + const modelSpans = graph.spans.filter((span) => span.spanKind === "model"); + return { + schemaVersion: 1, + traceId: graph.traceId, + generatedAt, + summary: { + checks: checks.length, + passed, + score: Math.round((passed / checks.length) * 10_000) / 10_000, + }, + metrics: { + spans: graph.spans.length, + turns: graph.spans.filter((span) => span.spanKind === "turn").length, + modelCalls: modelSpans.length, + toolCalls: toolNames.length, + repeatedToolCalls, + inputTokens: optionalSum(modelSpans, "inputTokens"), + outputTokens: optionalSum(modelSpans, "outputTokens"), + totalTokens: optionalSum(modelSpans, "totalTokens"), + }, + checks, + }; +} diff --git a/packages/app/src/electron/trace/projector.test.ts b/packages/app/src/electron/trace/projector.test.ts new file mode 100644 index 00000000..dce3b278 --- /dev/null +++ b/packages/app/src/electron/trace/projector.test.ts @@ -0,0 +1,91 @@ +import { describe, expect, it } from "vitest"; +import type { TraceEvent } from "@/shared/types/trace"; +import { TRACE_SCHEMA_VERSION } from "@/shared/types/trace"; +import { projectTrace } from "./projector"; + +function event( + sequence: number, + input: Partial & Pick, +): TraceEvent { + return { + schemaVersion: TRACE_SCHEMA_VERSION, + eventId: `event-${sequence}`, + traceId: "trace", + sequence, + occurredAt: "2026-01-01T00:00:00.000Z", + recordedAt: "2026-01-01T00:00:00.000Z", + emitter: "main", + type: "span.start", + name: input.spanKind, + classification: "P0", + ...input, + }; +} + +describe("projectTrace", () => { + it("projects concurrent parents and causal links into a DAG", () => { + const graph = projectTrace( + [ + event(1, { spanId: "mission", spanKind: "mission" }), + event(2, { + spanId: "task-a", + spanKind: "task", + parentSpanId: "mission", + }), + event(3, { + spanId: "task-b", + spanKind: "task", + parentSpanId: "mission", + }), + event(4, { + spanId: "join", + spanKind: "task", + parentSpanId: "mission", + links: [ + { spanId: "task-a", relation: "joins" }, + { spanId: "task-b", relation: "joins" }, + ], + }), + ], + "trace", + ); + + expect(graph.rootSpanIds).toEqual(["mission"]); + expect(graph.edges).toEqual( + expect.arrayContaining([ + { fromSpanId: "mission", toSpanId: "task-a", relation: "parent" }, + { fromSpanId: "mission", toSpanId: "task-b", relation: "parent" }, + { fromSpanId: "task-a", toSpanId: "join", relation: "joins" }, + { fromSpanId: "task-b", toSpanId: "join", relation: "joins" }, + ]), + ); + expect(graph.orphanSpanIds).toEqual([]); + expect(graph.cycleSpanIds).toEqual([]); + }); + + it("reports unresolved parents and cycles", () => { + const graph = projectTrace( + [ + event(1, { + spanId: "mission", + spanKind: "mission", + links: [{ spanId: "task", relation: "triggered_by" }], + }), + event(2, { + spanId: "task", + spanKind: "task", + parentSpanId: "mission", + }), + event(3, { + spanId: "orphan", + spanKind: "tool", + parentSpanId: "missing", + }), + ], + "trace", + ); + + expect(graph.orphanSpanIds).toEqual(["orphan"]); + expect(graph.cycleSpanIds).toEqual(["mission", "task"]); + }); +}); diff --git a/packages/app/src/electron/trace/projector.ts b/packages/app/src/electron/trace/projector.ts new file mode 100644 index 00000000..65010dba --- /dev/null +++ b/packages/app/src/electron/trace/projector.ts @@ -0,0 +1,137 @@ +import type { + TraceEdge, + TraceEvent, + TraceGraph, + TraceLink, + TraceSpan, +} from "@/shared/types/trace"; + +function uniqueLinks(links: TraceLink[]): TraceLink[] { + const seen = new Set(); + return links.filter((link) => { + const key = `${link.traceId ?? ""}:${link.spanId}:${link.relation}`; + if (seen.has(key)) return false; + seen.add(key); + return true; + }); +} + +function addEdge(edges: TraceEdge[], seen: Set, edge: TraceEdge): void { + const key = `${edge.fromSpanId}:${edge.toSpanId}:${edge.relation}`; + if (seen.has(key)) return; + seen.add(key); + edges.push(edge); +} + +function cycleMembers(spanIds: string[], edges: TraceEdge[]): string[] { + const adjacency = new Map(); + for (const spanId of spanIds) adjacency.set(spanId, []); + for (const edge of edges) { + if (!adjacency.has(edge.fromSpanId) || !adjacency.has(edge.toSpanId)) { + continue; + } + adjacency.get(edge.fromSpanId)?.push(edge.toSpanId); + } + const visiting = new Set(); + const visited = new Set(); + const cyclic = new Set(); + const stack: string[] = []; + + const visit = (spanId: string): void => { + if (visited.has(spanId)) return; + if (visiting.has(spanId)) { + const index = stack.lastIndexOf(spanId); + for (const member of stack.slice(Math.max(index, 0))) cyclic.add(member); + cyclic.add(spanId); + return; + } + visiting.add(spanId); + stack.push(spanId); + for (const child of adjacency.get(spanId) ?? []) visit(child); + stack.pop(); + visiting.delete(spanId); + visited.add(spanId); + }; + + for (const spanId of spanIds) visit(spanId); + return [...cyclic].sort(); +} + +export function projectTrace( + events: TraceEvent[], + traceId: string, +): TraceGraph { + const selected = events + .filter((event) => event.traceId === traceId) + .sort((left, right) => left.sequence - right.sequence); + const bySpan = new Map(); + + for (const event of selected) { + const existing = bySpan.get(event.spanId); + const span = existing ?? { + traceId, + spanId: event.spanId, + parentSpanId: event.parentSpanId, + spanKind: event.spanKind, + name: event.name, + attributes: {}, + metrics: {}, + links: [], + events: [], + }; + span.parentSpanId ??= event.parentSpanId; + span.attributes = { ...span.attributes, ...(event.attributes ?? {}) }; + span.metrics = { ...span.metrics, ...(event.metrics ?? {}) }; + span.links = uniqueLinks([...span.links, ...(event.links ?? [])]); + span.events.push(event); + if (event.type === "span.start") span.startedAt ??= event.occurredAt; + if (event.type === "span.end") span.endedAt = event.occurredAt; + if (event.status) span.status = event.status; + bySpan.set(event.spanId, span); + } + + const spans = [...bySpan.values()]; + const spanIds = new Set(spans.map((span) => span.spanId)); + const edges: TraceEdge[] = []; + const seenEdges = new Set(); + const orphanSpanIds = new Set(); + + for (const span of spans) { + if (span.parentSpanId) { + if (spanIds.has(span.parentSpanId)) { + addEdge(edges, seenEdges, { + fromSpanId: span.parentSpanId, + toSpanId: span.spanId, + relation: "parent", + }); + } else { + orphanSpanIds.add(span.spanId); + } + } + for (const link of span.links) { + if (link.traceId && link.traceId !== traceId) continue; + if (!spanIds.has(link.spanId)) { + orphanSpanIds.add(span.spanId); + continue; + } + const predecessor = link.relation !== "produces"; + addEdge(edges, seenEdges, { + fromSpanId: predecessor ? link.spanId : span.spanId, + toSpanId: predecessor ? span.spanId : link.spanId, + relation: link.relation, + }); + } + } + + return { + traceId, + spans, + edges, + rootSpanIds: spans + .filter((span) => !span.parentSpanId) + .map((span) => span.spanId), + orphanSpanIds: [...orphanSpanIds].sort(), + cycleSpanIds: cycleMembers([...spanIds], edges), + eventCount: selected.length, + }; +} diff --git a/packages/app/src/electron/trace/schema.ts b/packages/app/src/electron/trace/schema.ts new file mode 100644 index 00000000..dd28c876 --- /dev/null +++ b/packages/app/src/electron/trace/schema.ts @@ -0,0 +1,61 @@ +import { z } from "zod"; +import { TRACE_SCHEMA_VERSION, type TraceEvent } from "@/shared/types/trace"; + +const attributeValueSchema = z.union([ + z.string(), + z.number().finite(), + z.boolean(), + z.null(), + z.array(z.string()), +]); + +const linkSchema = z.object({ + traceId: z.string().min(1).optional(), + spanId: z.string().min(1), + relation: z.enum([ + "triggered_by", + "consumes", + "produces", + "joins", + "supersedes", + "handoff", + ]), +}); + +export const traceEventSchema = z.object({ + schemaVersion: z.literal(TRACE_SCHEMA_VERSION), + eventId: z.string().min(1), + traceId: z.string().min(1), + spanId: z.string().min(1), + parentSpanId: z.string().min(1).optional(), + links: z.array(linkSchema).optional(), + sequence: z.number().int().positive(), + occurredAt: z.string().datetime(), + recordedAt: z.string().datetime(), + emitter: z.enum(["main", "provider", "tool", "memory", "automation", "eval"]), + type: z.enum(["span.start", "span.event", "span.end"]), + spanKind: z.enum([ + "mission", + "task", + "run", + "turn", + "model", + "tool", + "handoff", + "memory.mutation", + "artifact", + "automation.action", + "eval.grader", + ]), + name: z.string().min(1), + status: z + .enum(["ok", "error", "cancelled", "interrupted", "uncertain"]) + .optional(), + attributes: z.record(z.string(), attributeValueSchema).optional(), + metrics: z.record(z.string(), z.number().finite()).optional(), + classification: z.enum(["P0", "P1", "P2", "P3"]), +}); + +export function parseTraceEvent(value: unknown): TraceEvent { + return traceEventSchema.parse(value) as TraceEvent; +} diff --git a/packages/app/src/electron/trace/store.test.ts b/packages/app/src/electron/trace/store.test.ts new file mode 100644 index 00000000..d51e1834 --- /dev/null +++ b/packages/app/src/electron/trace/store.test.ts @@ -0,0 +1,99 @@ +import { appendFile, mkdir, mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { describe, expect, it } from "vitest"; +import type { TraceEventInput } from "@/shared/types/trace"; +import { LocalTraceStore } from "./store"; + +function input(traceId: string, eventId: string): TraceEventInput { + return { + eventId, + traceId, + spanId: `span:${eventId}`, + occurredAt: "2026-01-01T00:00:00.000Z", + emitter: "main", + type: "span.start", + spanKind: "mission", + name: "mission", + classification: "P0", + }; +} + +describe("LocalTraceStore", () => { + it("serializes concurrent writes and deduplicates stable event ids", async () => { + const directory = await mkdtemp(join(tmpdir(), "convera-trace-store-")); + try { + const store = new LocalTraceStore(join(directory, "traces.jsonl")); + await Promise.all( + Array.from({ length: 20 }, (_, index) => + store.append(input("trace", `event-${index}`)), + ), + ); + await store.append(input("trace", "event-0")); + const events = await store.read(); + expect(events).toHaveLength(20); + expect(events.map((event) => event.sequence)).toEqual( + Array.from({ length: 20 }, (_, index) => index + 1), + ); + } finally { + await rm(directory, { recursive: true, force: true }); + } + }); + + it("repairs a corrupt crash tail before appending more events", async () => { + const directory = await mkdtemp(join(tmpdir(), "convera-trace-store-")); + const path = join(directory, "traces.jsonl"); + try { + await new LocalTraceStore(path).append(input("trace", "first")); + await appendFile(path, "{partial", "utf8"); + const recovered = new LocalTraceStore(path); + await recovered.append(input("trace", "second")); + expect((await recovered.read()).map((event) => event.eventId)).toEqual([ + "first", + "second", + ]); + expect(await recovered.health()).toEqual({ + validEvents: 2, + corruptLines: 1, + }); + } finally { + await rm(directory, { recursive: true, force: true }); + } + }); + + it("can retry the same event after a disk write failure", async () => { + const directory = await mkdtemp(join(tmpdir(), "convera-trace-store-")); + const path = join(directory, "traces.jsonl"); + try { + await mkdir(path); + const store = new LocalTraceStore(path); + await expect(store.append(input("trace", "retry-me"))).rejects.toThrow(); + await rm(path, { recursive: true }); + await store.append(input("trace", "retry-me")); + expect((await store.read()).map((event) => event.eventId)).toEqual([ + "retry-me", + ]); + } finally { + await rm(directory, { recursive: true, force: true }); + } + }); + + it("retains whole recent traces instead of cutting a graph in half", async () => { + const directory = await mkdtemp(join(tmpdir(), "convera-trace-store-")); + try { + const store = new LocalTraceStore(join(directory, "traces.jsonl"), { + maxEvents: 4, + }); + await store.append([ + input("old", "old-1"), + input("old", "old-2"), + input("old", "old-3"), + ]); + await store.append([input("new", "new-1"), input("new", "new-2")]); + expect(await store.listTraceIds()).toEqual(["new"]); + expect(await store.read()).toHaveLength(2); + } finally { + await rm(directory, { recursive: true, force: true }); + } + }); +}); diff --git a/packages/app/src/electron/trace/store.ts b/packages/app/src/electron/trace/store.ts new file mode 100644 index 00000000..bbddb015 --- /dev/null +++ b/packages/app/src/electron/trace/store.ts @@ -0,0 +1,254 @@ +import { randomUUID } from "node:crypto"; +import { mkdir, open, readFile, rename, rm } from "node:fs/promises"; +import { dirname } from "node:path"; +import { SerialTaskQueue } from "@/electron/memory/serial-queue"; +import type { + TraceEvent, + TraceEventInput, + TraceGraph, +} from "@/shared/types/trace"; +import { TRACE_SCHEMA_VERSION } from "@/shared/types/trace"; +import { projectTrace } from "./projector"; +import { traceEventSchema } from "./schema"; + +export interface TraceEventSink { + append(input: TraceEventInput | TraceEventInput[]): Promise; +} + +export interface TraceEventStore extends TraceEventSink { + graph(traceId: string): Promise; +} + +export interface TraceStoreHealth { + validEvents: number; + corruptLines: number; +} + +export const DEFAULT_MAX_TRACE_EVENTS = 50_000; + +function isMissingFile(error: unknown): boolean { + return ( + typeof error === "object" && + error !== null && + "code" in error && + error.code === "ENOENT" + ); +} + +export class LocalTraceStore implements TraceEventStore { + private readonly writes = new SerialTaskQueue(); + private readonly maxEvents: number; + private readonly now: () => Date; + private readonly createId: () => string; + private initialized = false; + private sequence = 0; + private eventIds = new Set(); + private eventCount = 0; + private recoveredCorruptLines = 0; + + constructor( + readonly path: string, + options: { + maxEvents?: number; + now?: () => Date; + createId?: () => string; + } = {}, + ) { + this.maxEvents = options.maxEvents ?? DEFAULT_MAX_TRACE_EVENTS; + this.now = options.now ?? (() => new Date()); + this.createId = options.createId ?? randomUUID; + if (!Number.isInteger(this.maxEvents) || this.maxEvents < 1) { + throw new RangeError("maxEvents must be a positive integer."); + } + } + + async append( + input: TraceEventInput | TraceEventInput[], + ): Promise { + return this.writes.run(async () => { + await this.initialize(); + const inputs = Array.isArray(input) ? input : [input]; + const created: TraceEvent[] = []; + const pendingEventIds = new Set(this.eventIds); + let nextSequence = this.sequence; + for (const candidate of inputs) { + const eventId = candidate.eventId ?? this.createId(); + if (pendingEventIds.has(eventId)) continue; + const event = traceEventSchema.parse({ + ...candidate, + schemaVersion: TRACE_SCHEMA_VERSION, + eventId, + sequence: ++nextSequence, + recordedAt: this.now().toISOString(), + }) as TraceEvent; + pendingEventIds.add(eventId); + created.push(event); + } + if (created.length === 0) return []; + await mkdir(dirname(this.path), { recursive: true }); + const handle = await open(this.path, "a", 0o600); + try { + await handle.writeFile( + created.map((event) => `${JSON.stringify(event)}\n`).join(""), + "utf8", + ); + await handle.sync(); + } finally { + await handle.close(); + } + this.sequence = nextSequence; + for (const event of created) this.eventIds.add(event.eventId); + this.eventCount += created.length; + if (this.eventCount > this.maxEvents) await this.prune(); + return structuredClone(created); + }); + } + + async read(traceId?: string): Promise { + return this.writes.run(async () => { + await this.initialize(); + const { events } = await this.readFile(); + return structuredClone( + traceId ? events.filter((event) => event.traceId === traceId) : events, + ); + }); + } + + async listTraceIds(): Promise { + const events = await this.read(); + const latest = new Map(); + for (const event of events) latest.set(event.traceId, event.sequence); + return [...latest.entries()] + .sort((left, right) => right[1] - left[1]) + .map(([traceId]) => traceId); + } + + async graph(traceId: string): Promise { + return projectTrace(await this.read(traceId), traceId); + } + + async health(): Promise { + return this.writes.run(async () => { + await this.initialize(); + const { events, corruptLines } = await this.readFile(); + return { + validEvents: events.length, + corruptLines: corruptLines + this.recoveredCorruptLines, + }; + }); + } + + async idle(): Promise { + await this.writes.idle(); + } + + private async initialize(): Promise { + if (this.initialized) return; + const { events, corruptLines } = await this.readFile(); + if (corruptLines > 0) { + await this.rewrite(events); + this.recoveredCorruptLines += corruptLines; + } + this.sequence = events.reduce( + (maximum, event) => Math.max(maximum, event.sequence), + 0, + ); + this.eventIds = new Set(events.map((event) => event.eventId)); + this.eventCount = events.length; + this.initialized = true; + } + + private async readFile(): Promise<{ + events: TraceEvent[]; + corruptLines: number; + }> { + let content: string; + try { + content = await readFile(this.path, "utf8"); + } catch (error) { + if (isMissingFile(error)) return { events: [], corruptLines: 0 }; + throw error; + } + const events: TraceEvent[] = []; + let corruptLines = 0; + for (const line of content.split("\n")) { + if (!line.trim()) continue; + try { + const parsed = traceEventSchema.safeParse(JSON.parse(line)); + if (parsed.success) events.push(parsed.data as TraceEvent); + else corruptLines += 1; + } catch { + corruptLines += 1; + } + } + return { events, corruptLines }; + } + + private async prune(): Promise { + const { events } = await this.readFile(); + const byTrace = new Map(); + for (const event of events) { + const trace = byTrace.get(event.traceId); + if (trace) trace.push(event); + else byTrace.set(event.traceId, [event]); + } + const newestFirst = [...byTrace.values()].sort( + (left, right) => + Math.max(...right.map((event) => event.sequence)) - + Math.max(...left.map((event) => event.sequence)), + ); + const kept: TraceEvent[] = []; + for (const trace of newestFirst) { + if (kept.length > 0 && kept.length + trace.length > this.maxEvents) { + continue; + } + kept.push(...trace); + } + kept.sort((left, right) => left.sequence - right.sequence); + await this.rewrite(kept); + this.eventCount = kept.length; + this.eventIds = new Set(kept.map((event) => event.eventId)); + } + + private async rewrite(events: TraceEvent[]): Promise { + await mkdir(dirname(this.path), { recursive: true }); + const temporaryPath = `${this.path}.${process.pid}.${randomUUID()}.tmp`; + let handle: Awaited> | undefined; + try { + handle = await open(temporaryPath, "wx", 0o600); + await handle.writeFile( + events.map((event) => `${JSON.stringify(event)}\n`).join(""), + "utf8", + ); + await handle.sync(); + await handle.close(); + handle = undefined; + await rename(temporaryPath, this.path); + await this.syncParentDirectory(); + } finally { + await handle?.close().catch(() => undefined); + await rm(temporaryPath, { force: true }).catch(() => undefined); + } + } + + private async syncParentDirectory(): Promise { + let directory: Awaited> | undefined; + try { + directory = await open(dirname(this.path), "r"); + await directory.sync(); + } catch (error) { + const code = + typeof error === "object" && + error !== null && + "code" in error && + typeof error.code === "string" + ? error.code + : undefined; + if (!["ENOENT", "EINVAL", "EPERM", "EISDIR"].includes(code ?? "")) { + throw error; + } + } finally { + await directory?.close().catch(() => undefined); + } + } +} diff --git a/packages/app/src/shared/types/trace.ts b/packages/app/src/shared/types/trace.ts new file mode 100644 index 00000000..91bbd689 --- /dev/null +++ b/packages/app/src/shared/types/trace.ts @@ -0,0 +1,101 @@ +export const TRACE_SCHEMA_VERSION = 1 as const; + +export type TraceSpanKind = + | "mission" + | "task" + | "run" + | "turn" + | "model" + | "tool" + | "handoff" + | "memory.mutation" + | "artifact" + | "automation.action" + | "eval.grader"; + +export type TraceEventType = "span.start" | "span.event" | "span.end"; + +export type TraceStatus = + | "ok" + | "error" + | "cancelled" + | "interrupted" + | "uncertain"; + +export type TraceClassification = "P0" | "P1" | "P2" | "P3"; + +export type TraceLinkRelation = + | "triggered_by" + | "consumes" + | "produces" + | "joins" + | "supersedes" + | "handoff"; + +export type TraceAttributeValue = string | number | boolean | null | string[]; + +export type TraceAttributes = Record; + +export interface TraceLink { + traceId?: string; + spanId: string; + relation: TraceLinkRelation; +} + +export interface TraceEvent { + schemaVersion: typeof TRACE_SCHEMA_VERSION; + eventId: string; + traceId: string; + spanId: string; + parentSpanId?: string; + links?: TraceLink[]; + sequence: number; + occurredAt: string; + recordedAt: string; + emitter: "main" | "provider" | "tool" | "memory" | "automation" | "eval"; + type: TraceEventType; + spanKind: TraceSpanKind; + name: string; + status?: TraceStatus; + attributes?: TraceAttributes; + metrics?: Record; + classification: TraceClassification; +} + +export type TraceEventInput = Omit< + TraceEvent, + "schemaVersion" | "sequence" | "recordedAt" | "eventId" +> & { + eventId?: string; +}; + +export interface TraceEdge { + fromSpanId: string; + toSpanId: string; + relation: "parent" | TraceLinkRelation; +} + +export interface TraceSpan { + traceId: string; + spanId: string; + parentSpanId?: string; + spanKind: TraceSpanKind; + name: string; + startedAt?: string; + endedAt?: string; + status?: TraceStatus; + attributes: TraceAttributes; + metrics: Record; + links: TraceLink[]; + events: TraceEvent[]; +} + +export interface TraceGraph { + traceId: string; + spans: TraceSpan[]; + edges: TraceEdge[]; + rootSpanIds: string[]; + orphanSpanIds: string[]; + cycleSpanIds: string[]; + eventCount: number; +} From 40684191870b1e7d6ae96d5a21147577ca44d0f5 Mon Sep 17 00:00:00 2001 From: NarwhalChen Date: Sun, 2 Aug 2026 19:39:38 +0800 Subject: [PATCH 2/2] feat(app): trace structured delegation and handoff --- packages/app/scripts/trace-eval.mts | 52 +++++++ .../app/src/electron/trace/agent-host.test.ts | 83 +++++++++++ packages/app/src/electron/trace/agent-host.ts | 131 +++++++++++++++++- .../app/src/electron/trace/evaluation.test.ts | 49 +++++++ packages/app/src/electron/trace/evaluation.ts | 61 ++++++-- 5 files changed, 360 insertions(+), 16 deletions(-) diff --git a/packages/app/scripts/trace-eval.mts b/packages/app/scripts/trace-eval.mts index 7c78f28d..85b8ceac 100644 --- a/packages/app/scripts/trace-eval.mts +++ b/packages/app/scripts/trace-eval.mts @@ -87,6 +87,52 @@ function syntheticTrace( })) as TraceEvent[]; } +function structuredCollaborationTrace(): TraceEvent[] { + const events = syntheticTrace("structured-collaboration", 1); + let sequence = events.length; + const add = ( + spanId: string, + spanKind: "task" | "handoff", + parentSpanId: string, + attributes: Record, + ) => { + for (const type of ["span.start", "span.end"] as const) { + events.push({ + schemaVersion: TRACE_SCHEMA_VERSION, + eventId: `${spanId}:${type}`, + traceId: "synthetic:structured-collaboration", + spanId, + parentSpanId, + links: + spanKind === "task" + ? [{ spanId: "run-1", relation: "triggered_by" }] + : [{ spanId: "run-1", relation: "handoff" }], + sequence: ++sequence, + occurredAt: "2026-01-01T00:00:01.000Z", + recordedAt: "2026-01-01T00:00:01.000Z", + emitter: "main", + type, + spanKind, + name: spanKind, + status: type === "span.end" ? "ok" : undefined, + attributes, + classification: "P0", + }); + } + }; + add("task-delegated", "task", "task-1", { + collaborationKind: "delegation", + collaborationOperationId: "delegation-1", + taskDepth: 1, + resultMessageCount: 1, + }); + add("handoff-1", "handoff", "task-delegated", { + operationId: "handoff-1", + committed: true, + }); + return events; +} + const cases = []; if (requestedPath) { const store = new LocalTraceStore(resolve(requestedPath)); @@ -103,6 +149,12 @@ if (requestedPath) { const events = syntheticTrace(name, tasks, terminalStatus); cases.push(evaluateTrace(projectTrace(events, `synthetic:${name}`))); } + const structured = structuredCollaborationTrace(); + cases.push( + evaluateTrace( + projectTrace(structured, "synthetic:structured-collaboration"), + ), + ); } const report = { diff --git a/packages/app/src/electron/trace/agent-host.test.ts b/packages/app/src/electron/trace/agent-host.test.ts index c9ef7f88..0e0e5e80 100644 --- a/packages/app/src/electron/trace/agent-host.test.ts +++ b/packages/app/src/electron/trace/agent-host.test.ts @@ -144,6 +144,89 @@ describe("AgentHostTraceRecorder", () => { } }); + it("projects PR #218 delegation and handoff provenance into the task DAG", async () => { + const directory = await mkdtemp(join(tmpdir(), "convera-agent-trace-")); + try { + const store = new LocalTraceStore(join(directory, "traces.jsonl")); + const recorder = new AgentHostTraceRecorder(store); + await recorder.recordJob(job); + await recorder.recordJob({ + ...job, + id: "run-child", + taskId: "task-child", + parentTaskId: job.taskId, + agentId: "two", + agentMemberId: "agent:two", + collaboration: { + kind: "delegation", + operationId: "delegation-1", + sourceTaskId: job.taskId, + sourceJobId: job.id, + fromMemberId: job.agentMemberId, + depth: 1, + path: [job.agentMemberId, "agent:two"], + }, + outputMessageIds: ["message-result"], + } as AgentHostJob); + await recorder.recordJob({ + ...job, + id: "run-handoff", + parentJobId: job.id, + agentId: "three", + agentMemberId: "agent:three", + collaboration: { + kind: "handoff", + operationId: "handoff-1", + sourceTaskId: job.taskId, + sourceJobId: job.id, + fromMemberId: job.agentMemberId, + depth: 1, + path: [job.agentMemberId, "agent:three"], + }, + } as AgentHostJob); + + const graph = await store.graph("trace:message-1"); + expect(graph.edges).toEqual( + expect.arrayContaining([ + { + fromSpanId: "task:task-1", + toSpanId: "task:task-child", + relation: "parent", + }, + { + fromSpanId: "run:run-1", + toSpanId: "task:task-child", + relation: "triggered_by", + }, + { + fromSpanId: "run:run-1", + toSpanId: "handoff:handoff-1", + relation: "handoff", + }, + { + fromSpanId: "handoff:handoff-1", + toSpanId: "run:run-handoff", + relation: "triggered_by", + }, + ]), + ); + expect( + graph.spans.find((span) => span.spanId === "task:task-child") + ?.attributes, + ).toMatchObject({ + collaborationKind: "delegation", + collaborationOperationId: "delegation-1", + resultMessageCount: 1, + taskDepth: 1, + }); + expect( + graph.spans.find((span) => span.spanId === "handoff:handoff-1"), + ).toMatchObject({ status: "ok", attributes: { committed: true } }); + } finally { + await rm(directory, { recursive: true, force: true }); + } + }); + it("closes open runtime spans when an interrupted job is recovered", async () => { const directory = await mkdtemp(join(tmpdir(), "convera-agent-trace-")); try { diff --git a/packages/app/src/electron/trace/agent-host.ts b/packages/app/src/electron/trace/agent-host.ts index adb6b4b5..dee383c3 100644 --- a/packages/app/src/electron/trace/agent-host.ts +++ b/packages/app/src/electron/trace/agent-host.ts @@ -4,6 +4,7 @@ import type { LocalAIStreamEvent, } from "@/shared/types/local-ai"; import type { + TraceAttributes, TraceEventInput, TraceSpan, TraceStatus, @@ -11,6 +12,32 @@ import type { import { WORKSPACE_QUERY_INTERACTION } from "@/shared/types/workspace-perception"; import type { TraceEventSink, TraceEventStore } from "./store"; +/** + * Forward-compatible view of the structured collaboration provenance added by + * PR #218. Keeping this local means tracing can land independently and starts + * recording the extra fields as soon as that AgentHost shape is present. + */ +interface StructuredCollaboration { + kind: "delegation" | "handoff"; + operationId: string; + sourceTaskId: string; + sourceJobId: string; + fromMemberId: string; + depth: number; + path: string[]; + expiresAt?: string; +} + +type TraceableAgentHostJob = AgentHostJob & { + parentTaskId?: string; + collaboration?: StructuredCollaboration; + outputMessageIds?: string[]; +}; + +function structuredJob(job: AgentHostJob): TraceableAgentHostJob { + return job as TraceableAgentHostJob; +} + function traceIdFor(job: AgentHostJob): string { return `trace:${job.triggerMessageId}`; } @@ -27,6 +54,20 @@ function runSpanId(job: AgentHostJob): string { return `run:${job.id}`; } +function collaborationAttributes(job: TraceableAgentHostJob): TraceAttributes { + const collaboration = job.collaboration; + return collaboration + ? { + collaborationKind: collaboration.kind, + collaborationOperationId: collaboration.operationId, + sourceTaskId: collaboration.sourceTaskId, + sourceJobId: collaboration.sourceJobId, + fromMemberId: collaboration.fromMemberId, + taskDepth: collaboration.depth, + } + : {}; +} + function terminalStatus(job: AgentHostJob): TraceStatus | undefined { if (job.status === "completed") return "ok"; if (job.status === "failed") return "error"; @@ -54,10 +95,19 @@ export class AgentHostTraceRecorder { } async recordJob(job: AgentHostJob): Promise { + const traceable = structuredJob(job); + const collaboration = traceable.collaboration; const traceId = traceIdFor(job); const missionId = missionSpanId(job); const taskId = taskSpanId(job); const runId = runSpanId(job); + const taskParentId = traceable.parentTaskId + ? `task:${traceable.parentTaskId}` + : missionId; + const handoffId = + collaboration?.kind === "handoff" + ? `handoff:${collaboration.operationId}` + : undefined; const common = { traceId, emitter: "main" as const, @@ -83,7 +133,16 @@ export class AgentHostTraceRecorder { ...common, eventId: `${taskId}:start`, spanId: taskId, - parentSpanId: missionId, + parentSpanId: taskParentId, + links: + collaboration?.kind === "delegation" + ? [ + { + spanId: `run:${collaboration.sourceJobId}`, + relation: "triggered_by", + }, + ] + : undefined, occurredAt: job.createdAt, type: "span.start", spanKind: "task", @@ -92,6 +151,10 @@ export class AgentHostTraceRecorder { taskId: job.taskId, agentId: job.agentId, agentMemberId: job.agentMemberId, + ...(traceable.parentTaskId + ? { parentTaskId: traceable.parentTaskId } + : {}), + ...collaborationAttributes(traceable), }, }, { @@ -99,9 +162,19 @@ export class AgentHostTraceRecorder { eventId: `${runId}:start`, spanId: runId, parentSpanId: taskId, - links: job.parentJobId - ? [{ spanId: `run:${job.parentJobId}`, relation: "supersedes" }] - : undefined, + links: [ + ...(job.parentJobId + ? [ + { + spanId: `run:${job.parentJobId}`, + relation: "supersedes" as const, + }, + ] + : []), + ...(handoffId + ? [{ spanId: handoffId, relation: "triggered_by" as const }] + : []), + ], occurredAt: job.createdAt, type: "span.start", spanKind: "run", @@ -111,21 +184,67 @@ export class AgentHostTraceRecorder { attempt: job.attempts, jobStatus: job.status, agentMemberId: job.agentMemberId, + ...collaborationAttributes(traceable), }, }, ]; + if (handoffId && collaboration) { + inputs.push( + { + ...common, + eventId: `${handoffId}:start`, + spanId: handoffId, + parentSpanId: taskId, + links: [ + { + spanId: `run:${collaboration.sourceJobId}`, + relation: "handoff", + }, + ], + occurredAt: job.createdAt, + type: "span.start", + spanKind: "handoff", + name: "Task ownership handoff", + attributes: { + operationId: collaboration.operationId, + taskId: job.taskId, + sourceJobId: collaboration.sourceJobId, + fromMemberId: collaboration.fromMemberId, + toMemberId: job.agentMemberId, + taskDepth: collaboration.depth, + }, + }, + { + ...common, + eventId: `${handoffId}:end`, + spanId: handoffId, + parentSpanId: taskId, + occurredAt: job.createdAt, + type: "span.end", + spanKind: "handoff", + name: "Task ownership handoff", + status: "ok", + attributes: { committed: true }, + }, + ); + } const status = terminalStatus(job); inputs.push({ ...common, eventId: `${taskId}:status:${job.status}:${job.updatedAt}`, spanId: taskId, - parentSpanId: missionId, + parentSpanId: taskParentId, occurredAt: job.updatedAt, type: "span.event", spanKind: "task", name: "Task status changed", status, - attributes: { jobStatus: job.status, currentRunId: job.id }, + attributes: { + jobStatus: job.status, + currentRunId: job.id, + resultMessageCount: traceable.outputMessageIds?.length ?? 0, + ...collaborationAttributes(traceable), + }, }); inputs.push({ ...common, diff --git a/packages/app/src/electron/trace/evaluation.test.ts b/packages/app/src/electron/trace/evaluation.test.ts index 3ce7a199..cf9b8e8b 100644 --- a/packages/app/src/electron/trace/evaluation.test.ts +++ b/packages/app/src/electron/trace/evaluation.test.ts @@ -65,4 +65,53 @@ describe("evaluateTrace", () => { actual: "1 incomplete runtime spans", }); }); + + it("counts delegated tasks, handoffs, depth, and result receipts", () => { + const events = lifecycleEvents(); + let sequence = events.length; + const add = ( + spanId: string, + spanKind: "task" | "handoff", + parentSpanId: string, + attributes: Record, + ) => { + for (const type of ["span.start", "span.end"] as const) { + events.push({ + schemaVersion: TRACE_SCHEMA_VERSION, + eventId: `${spanId}:${type}`, + traceId: "trace", + spanId, + parentSpanId, + sequence: ++sequence, + occurredAt: "2026-01-01T00:00:01.000Z", + recordedAt: "2026-01-01T00:00:01.000Z", + emitter: "main", + type, + spanKind, + name: spanKind, + status: type === "span.end" ? "ok" : undefined, + attributes, + classification: "P0", + }); + } + }; + add("task-child", "task", "task", { + collaborationKind: "delegation", + collaborationOperationId: "delegation-1", + taskDepth: 1, + resultMessageCount: 2, + }); + add("handoff", "handoff", "task-child", { committed: true }); + + const report = evaluateTrace(projectTrace(events, "trace")); + expect(report.summary.score).toBe(1); + expect(report.metrics).toMatchObject({ + tasks: 2, + delegationOperations: 1, + delegatedTasks: 1, + handoffs: 1, + maxTaskDepth: 1, + resultReceipts: 2, + }); + }); }); diff --git a/packages/app/src/electron/trace/evaluation.ts b/packages/app/src/electron/trace/evaluation.ts index db6b34e5..413185b9 100644 --- a/packages/app/src/electron/trace/evaluation.ts +++ b/packages/app/src/electron/trace/evaluation.ts @@ -19,6 +19,12 @@ export interface TraceEvaluationReport { }; metrics: { spans: number; + tasks: number; + delegationOperations: number; + delegatedTasks: number; + handoffs: number; + maxTaskDepth: number; + resultReceipts: number; turns: number; modelCalls: number; toolCalls: number; @@ -33,13 +39,29 @@ export interface TraceEvaluationReport { const expectedParent: Partial< Record > = { - task: "mission", run: "task", turn: "run", model: "turn", tool: "model", + handoff: "task", }; +function hasCanonicalParent( + span: TraceSpan, + byId: Map, +): boolean { + if (span.spanKind === "task") { + const parentKind = span.parentSpanId + ? byId.get(span.parentSpanId)?.spanKind + : undefined; + return parentKind === "mission" || parentKind === "task"; + } + const parentKind = expectedParent[span.spanKind]; + return ( + !parentKind || byId.get(span.parentSpanId ?? "")?.spanKind === parentKind + ); +} + function check( id: string, label: string, @@ -70,15 +92,11 @@ export function evaluateTrace( const missionRoots = graph.rootSpanIds.filter( (spanId) => byId.get(spanId)?.spanKind === "mission", ); - const hierarchyViolations = graph.spans.filter((span) => { - const parentKind = expectedParent[span.spanKind]; - if (!parentKind) return false; - return ( - !span.parentSpanId || byId.get(span.parentSpanId)?.spanKind !== parentKind - ); - }); + const hierarchyViolations = graph.spans.filter( + (span) => !hasCanonicalParent(span, byId), + ); const runtimeSpans = graph.spans.filter((span) => - ["run", "turn", "model", "tool"].includes(span.spanKind), + ["run", "turn", "model", "tool", "handoff"].includes(span.spanKind), ); const incompleteRuntimeSpans = runtimeSpans.filter( (span) => !span.startedAt || !span.endedAt, @@ -87,6 +105,19 @@ export function evaluateTrace( .filter((span) => span.spanKind === "tool") .map((span) => String(span.attributes.toolName ?? span.name)); const repeatedToolCalls = toolNames.length - new Set(toolNames).size; + const taskSpans = graph.spans.filter((span) => span.spanKind === "task"); + const delegatedTasks = taskSpans.filter( + (span) => span.attributes.collaborationKind === "delegation", + ); + const delegationOperations = new Set( + delegatedTasks + .map((span) => span.attributes.collaborationOperationId) + .filter((value): value is string => typeof value === "string"), + ).size; + const taskDepths = taskSpans.map((span) => { + const depth = span.attributes.taskDepth; + return typeof depth === "number" ? depth : 0; + }); const checks = [ check( "mission-root", @@ -111,7 +142,7 @@ export function evaluateTrace( ), check( "canonical-hierarchy", - "Task, Run, Turn, Model, and Tool use the canonical parent chain", + "Task, Run, Turn, Model, Tool, and Handoff use the canonical parent chain", hierarchyViolations.length === 0, "0 hierarchy violations", `${hierarchyViolations.length} hierarchy violations`, @@ -137,6 +168,16 @@ export function evaluateTrace( }, metrics: { spans: graph.spans.length, + tasks: taskSpans.length, + delegationOperations, + delegatedTasks: delegatedTasks.length, + handoffs: graph.spans.filter((span) => span.spanKind === "handoff") + .length, + maxTaskDepth: Math.max(0, ...taskDepths), + resultReceipts: taskSpans.reduce((total, span) => { + const count = span.attributes.resultMessageCount; + return total + (typeof count === "number" ? count : 0); + }, 0), turns: graph.spans.filter((span) => span.spanKind === "turn").length, modelCalls: modelSpans.length, toolCalls: toolNames.length,