diff --git a/.changeset/fn-7135-triage-token-usage.md b/.changeset/fn-7135-triage-token-usage.md new file mode 100644 index 0000000000..280b4ee79b --- /dev/null +++ b/.changeset/fn-7135-triage-token-usage.md @@ -0,0 +1,7 @@ +--- +"@runfusion/fusion": patch +--- + +summary: Include triage/planning model usage in Command Center Tokens by model. +category: fix +dev: Records token usage for triage primary, fallback, and spec-review subagent sessions. diff --git a/packages/engine/src/__tests__/triage-token-usage.test.ts b/packages/engine/src/__tests__/triage-token-usage.test.ts new file mode 100644 index 0000000000..ecfb5e6de5 --- /dev/null +++ b/packages/engine/src/__tests__/triage-token-usage.test.ts @@ -0,0 +1,206 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { mkdtempSync } from "node:fs"; +import { rm } from "node:fs/promises"; +import { join } from "node:path"; +import { tmpdir } from "node:os"; +import type { AgentSession } from "@earendil-works/pi-coding-agent"; +import { aggregateTokenAnalytics, Database, type Task, type TaskStore } from "@fusion/core"; +import { TriageProcessor } from "../triage.js"; + +interface MockSessionStats { + tokens?: { input?: number; output?: number; cacheRead?: number; cacheWrite?: number }; +} + +function createSession( + stats: MockSessionStats, + model: { provider: string; id: string }, +): AgentSession { + return { + getSessionStats: vi.fn(() => stats), + dispose: vi.fn(), + model, + } as unknown as AgentSession; +} + +function createStore(taskId = "FN-7135"): TaskStore & { _task: Task; updateTask: ReturnType } { + const task = { + id: taskId, + title: "Triage token usage regression", + tokenUsage: undefined, + } as Task; + const updateTask = vi.fn(async (_id: string, updates: Partial) => { + if (updates.tokenUsage !== undefined) { + task.tokenUsage = updates.tokenUsage; + } + return task; + }); + return { + _task: task, + getTask: vi.fn(async () => task), + updateTask, + on: vi.fn(), + off: vi.fn(), + } as unknown as TaskStore & { _task: Task; updateTask: ReturnType }; +} + +type TriageTokenRecorder = TriageProcessor & { + recordTriageSessionTokenUsage: (taskId: string, session: AgentSession, options?: { agentId?: string }) => Promise; + registerSubagentSession: (taskId: string, session: AgentSession) => void; + unregisterSubagentSession: (taskId: string, session: AgentSession) => void; + disposeSubagentsForTask: (taskId: string, reason: string) => void; +}; + +function createProcessor(store: TaskStore): TriageTokenRecorder { + return new TriageProcessor(store, "/test/root") as TriageTokenRecorder; +} + +async function flushAsyncRecorders(): Promise { + await new Promise((resolve) => setImmediate(resolve)); + await new Promise((resolve) => setImmediate(resolve)); +} + +function insertUsageTask(db: Database, task: Task): void { + const usage = task.tokenUsage; + if (!usage) throw new Error("expected token usage"); + db.prepare( + `INSERT INTO tasks + (id, description, "column", createdAt, updatedAt, + tokenUsageInputTokens, tokenUsageOutputTokens, tokenUsageCachedTokens, + tokenUsageCacheWriteTokens, tokenUsageTotalTokens, tokenUsageLastUsedAt, + modelProvider, modelId, tokenUsageModelProvider, tokenUsageModelId, tokenUsagePerModel) + VALUES (?, 'desc', 'todo', '2026-01-01T00:00:00.000Z', '2026-01-01T00:00:00.000Z', + ?, ?, ?, ?, ?, ?, null, null, ?, ?, ?)`, + ).run( + task.id, + usage.inputTokens, + usage.outputTokens, + usage.cachedTokens, + usage.cacheWriteTokens, + usage.totalTokens, + usage.lastUsedAt, + usage.modelProvider ?? null, + usage.modelId ?? null, + JSON.stringify(usage.perModel ?? []), + ); +} + +describe("triage session token usage recording", () => { + let tmpDir: string | undefined; + let db: Database | undefined; + + beforeEach(() => { + vi.useRealTimers(); + vi.restoreAllMocks(); + }); + + afterEach(async () => { + db?.close(); + if (tmpDir) { + await rm(tmpDir, { recursive: true, force: true }); + } + tmpDir = undefined; + db = undefined; + }); + + it("records a triage-only Anthropic planning model and surfaces it in by-model analytics", async () => { + const store = createStore("FN-TRIAGE-ANTHROPIC"); + const processor = createProcessor(store); + const session = createSession( + { tokens: { input: 120, output: 30, cacheRead: 10, cacheWrite: 5 } }, + { provider: "anthropic", id: "claude-sonnet-4-5" }, + ); + + // Symptom baseline: before the triage recording path runs, the Anthropic bucket is absent. + expect(store._task.tokenUsage?.perModel?.some((bucket) => bucket.modelProvider === "anthropic")).toBeFalsy(); + + await processor.recordTriageSessionTokenUsage(store._task.id, session, { agentId: "triage" }); + + expect(store._task.tokenUsage).toMatchObject({ + inputTokens: 120, + outputTokens: 30, + cachedTokens: 10, + cacheWriteTokens: 5, + totalTokens: 165, + modelProvider: "anthropic", + modelId: "claude-sonnet-4-5", + }); + expect(store._task.tokenUsage?.perModel).toEqual([ + expect.objectContaining({ + modelProvider: "anthropic", + modelId: "claude-sonnet-4-5", + inputTokens: 120, + outputTokens: 30, + cachedTokens: 10, + cacheWriteTokens: 5, + totalTokens: 165, + }), + ]); + + tmpDir = mkdtempSync(join(tmpdir(), "kb-triage-token-analytics-")); + db = new Database(join(tmpDir, ".fusion")); + db.init(); + insertUsageTask(db, store._task); + + const byModel = aggregateTokenAnalytics(db, { groupBy: "model" }); + expect(byModel.totals).toMatchObject({ totalTokens: 165, nTasks: 1 }); + expect(byModel.groups).toEqual([ + expect.objectContaining({ key: "claude-sonnet-4-5", totalTokens: 165, nTasks: 1 }), + ]); + + const byProvider = aggregateTokenAnalytics(db, { groupBy: "provider" }); + expect(byProvider.groups).toEqual([ + expect.objectContaining({ key: "anthropic", totalTokens: 165, nTasks: 1 }), + ]); + }); + + it("records primary and fallback planning sessions into distinct model buckets on the same task", async () => { + const store = createStore("FN-TRIAGE-FALLBACK"); + const processor = createProcessor(store); + const primary = createSession( + { tokens: { input: 50, output: 20, cacheRead: 0, cacheWrite: 0 } }, + { provider: "anthropic", id: "claude-sonnet-4-5" }, + ); + const fallback = createSession( + { tokens: { input: 25, output: 15, cacheRead: 3, cacheWrite: 2 } }, + { provider: "openai", id: "gpt-5" }, + ); + + await processor.recordTriageSessionTokenUsage(store._task.id, primary, { agentId: "triage" }); + await processor.recordTriageSessionTokenUsage(store._task.id, fallback, { agentId: "triage" }); + + expect(store._task.tokenUsage).toMatchObject({ inputTokens: 75, outputTokens: 35, cachedTokens: 3, cacheWriteTokens: 2, totalTokens: 115 }); + expect(store._task.tokenUsage?.perModel).toEqual([ + expect.objectContaining({ modelProvider: "anthropic", modelId: "claude-sonnet-4-5", totalTokens: 70 }), + expect.objectContaining({ modelProvider: "openai", modelId: "gpt-5", totalTokens: 45 }), + ]); + }); + + it("records spec-review subagent usage on normal completion and forced disposal", async () => { + const store = createStore("FN-TRIAGE-SUBAGENT"); + const processor = createProcessor(store); + const normalReview = createSession( + { tokens: { input: 10, output: 6, cacheRead: 1, cacheWrite: 0 } }, + { provider: "anthropic", id: "claude-reviewer" }, + ); + const forcedReview = createSession( + { tokens: { input: 7, output: 3, cacheRead: 0, cacheWrite: 0 } }, + { provider: "openai", id: "gpt-reviewer" }, + ); + + processor.registerSubagentSession(store._task.id, normalReview); + processor.unregisterSubagentSession(store._task.id, normalReview); + await flushAsyncRecorders(); + + processor.registerSubagentSession(store._task.id, forcedReview); + processor.disposeSubagentsForTask(store._task.id, "test forced disposal"); + await flushAsyncRecorders(); + + expect(normalReview.dispose).not.toHaveBeenCalled(); + expect(forcedReview.dispose).toHaveBeenCalledTimes(1); + expect(store._task.tokenUsage).toMatchObject({ totalTokens: 27 }); + expect(store._task.tokenUsage?.perModel).toEqual([ + expect.objectContaining({ modelProvider: "anthropic", modelId: "claude-reviewer", totalTokens: 17 }), + expect.objectContaining({ modelProvider: "openai", modelId: "gpt-reviewer", totalTokens: 10 }), + ]); + }); +}); diff --git a/packages/engine/src/triage.ts b/packages/engine/src/triage.ts index 4ada3420c2..8a2d3b7dcc 100644 --- a/packages/engine/src/triage.ts +++ b/packages/engine/src/triage.ts @@ -131,6 +131,7 @@ import { evaluateReleaseAuthorizationGate } from "./triage-release-authorization import { archiveAsGhostBug } from "./self-healing.js"; import { createRunAuditor, generateSyntheticRunId } from "./run-audit.js"; import { resolveAndEmitGoalContext } from "./goal-injection-diagnostics.js"; +import { accumulateSessionTokenUsage } from "./session-token-usage.js"; export interface TriageProcessorOptions { @@ -255,6 +256,11 @@ export class TriageProcessor { planLog.warn(`Failed to abort triage session for ${task.id}: ${err}`); }); } + /* + FNXC:TokenAnalytics 2026-06-27-14:52: + Task delete may dispose the live triage session before agentWork reaches its finally; fire a fail-soft delta snapshot now, with the finally call serving as a zero-delta backstop when it unwinds. + */ + this.recordTriageSessionTokenUsageSoon(task.id, session as AgentSession, { agentId: task.assignedAgentId ?? "triage" }); session.dispose(); this.activeSessions.delete(task.id); } @@ -281,6 +287,11 @@ export class TriageProcessor { planLog.warn(`Failed to abort triage session for ${task.id}: ${err}`); }); } + /* + FNXC:TokenAnalytics 2026-06-27-14:52: + Task pause can force resource disposal before the normal triage finally runs; record the current model token delta immediately and rely on delta baselines to avoid double-counting. + */ + this.recordTriageSessionTokenUsageSoon(task.id, session as AgentSession, { agentId: task.assignedAgentId ?? "triage" }); session.dispose(); this.activeSessions.delete(task.id); } @@ -372,6 +383,11 @@ export class TriageProcessor { planLog.warn(`Failed to abort triage session for ${taskId}: ${err}`); }); } + /* + FNXC:TokenAnalytics 2026-06-27-14:52: + Engine stop/global pause force-disposes active triage sessions synchronously, so snapshot token deltas before disposal while preserving the existing non-blocking abort behavior. + */ + this.recordTriageSessionTokenUsageSoon(taskId, session as AgentSession); session.dispose(); } } @@ -399,8 +415,40 @@ export class TriageProcessor { set.add(session); } + /** + * FNXC:TokenAnalytics 2026-06-27-14:52: + * Triage and spec-review subagent sessions are AI lanes that must snapshot the actually-used model before resource teardown so Command Center Tokens by model includes triage-only models such as Anthropic. + * Use one shared recorder for normal completion, fallback swaps, and abort disposal; the token helper is delta-based and fail-soft, so repeated emergency/finally calls do not inflate totals. + */ + private async recordTriageSessionTokenUsage( + taskId: string, + session: AgentSession, + options?: { agentId?: string }, + ): Promise { + await accumulateSessionTokenUsage(this.store, taskId, session, { + agentId: options?.agentId, + role: "triage", + }); + } + + private recordTriageSessionTokenUsageSoon( + taskId: string, + session: AgentSession, + options?: { agentId?: string }, + ): void { + void this.recordTriageSessionTokenUsage(taskId, session, options).catch((err: unknown) => { + const msg = err instanceof Error ? err.message : String(err); + planLog.warn(`${taskId}: failed to record triage session token usage before disposal: ${msg}`); + }); + } + /** Deregister a reviewer subagent that finished naturally. */ private unregisterSubagentSession(taskId: string, session: AgentSession): void { + /* + FNXC:TokenAnalytics 2026-06-27-14:52: + The spec-review subagent disposes inside reviewer.ts before this callback; record its retained session stats here before dropping the reference so normal APPROVE/REVISE/RETHINK reviews count in per-model analytics. + */ + this.recordTriageSessionTokenUsageSoon(taskId, session); const set = this.activeSubagentSessions.get(taskId); if (!set) return; set.delete(session); @@ -414,6 +462,11 @@ export class TriageProcessor { planLog.log(`${taskId}: disposing ${set.size} subagent session(s) — ${reason}`); for (const session of set) { try { + /* + FNXC:TokenAnalytics 2026-06-27-14:52: + Pause/delete/stop can force-dispose spec-review subagents outside the normal reviewer callback, so record their model token delta before disposal without blocking the synchronous abort path. + */ + this.recordTriageSessionTokenUsageSoon(taskId, session); session.dispose(); } catch (err) { planLog.warn(`${taskId}: failed to dispose subagent session: ${err}`); @@ -1142,6 +1195,11 @@ export class TriageProcessor { `Primary planning model produced no approved spec (${verdictDesc}) — retrying with fallback ${fallbackDesc}`, ); + /* + FNXC:TokenAnalytics 2026-06-27-14:52: + Planning fallback replaces the primary triage session, so record the primary model's token delta before disposal; the shared finally records the fallback session separately. + */ + await this.recordTriageSessionTokenUsage(task.id, session, { agentId: triageRunContext.agentId }); session.dispose(); this.activeSessions.delete(task.id); stuckDetector?.untrackTask(task.id); @@ -1319,6 +1377,11 @@ export class TriageProcessor { this.activeSessions.delete(task.id); stuckDetector?.untrackTask(task.id); await agentLogger.flush(); + /* + FNXC:TokenAnalytics 2026-06-27-14:52: + Every triage planning exit path, including APPROVE, retry, pause/stuck abort, split/delete, and rate-limit wrapper attempts, records the active session's actual model before disposal so by-model analytics do not collapse triage usage to missing buckets. + */ + await this.recordTriageSessionTokenUsage(task.id, session, { agentId: triageRunContext.agentId }); session.dispose(); } };