FN-7135: Record triage token usage by model
Record triage and spec-review session token usage so Command Center model analytics includes planning models.\n\n- Add fail-soft token usage recording for triage session completion, fallback disposal, pause/delete, and engine stop paths.\n- Capture spec-review subagent token deltas before unregistering or force-disposing subagent sessions.\n- Add regression coverage for primary, fallback, and spec-review model token attribution.\n- Add a patch changeset for published CLI release notes.\n\nFiles changed:\n .changeset/fn-7135-triage-token-usage.md | 7 +\n .../src/__tests__/triage-token-usage.test.ts | 206 +++++++++++++++++++++\n packages/engine/src/triage.ts | 63 +++++++\n 3 files changed, 276 insertions(+) Fusion-Task-Id: FN-7135 Fusion-Task-Lineage: abbff18a-44e8-41fd-9012-388115914fc7 Co-authored-by: Fusion (runfusion.ai) <noreply@runfusion.ai>
This commit is contained in:
gsxdsm
7
.changeset/fn-7135-triage-token-usage.md
Normal file
7
.changeset/fn-7135-triage-token-usage.md
Normal file
@@ -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.
|
||||
206
packages/engine/src/__tests__/triage-token-usage.test.ts
Normal file
206
packages/engine/src/__tests__/triage-token-usage.test.ts
Normal file
@@ -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<typeof vi.fn> } {
|
||||
const task = {
|
||||
id: taskId,
|
||||
title: "Triage token usage regression",
|
||||
tokenUsage: undefined,
|
||||
} as Task;
|
||||
const updateTask = vi.fn(async (_id: string, updates: Partial<Task>) => {
|
||||
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<typeof vi.fn> };
|
||||
}
|
||||
|
||||
type TriageTokenRecorder = TriageProcessor & {
|
||||
recordTriageSessionTokenUsage: (taskId: string, session: AgentSession, options?: { agentId?: string }) => Promise<void>;
|
||||
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<void> {
|
||||
await new Promise<void>((resolve) => setImmediate(resolve));
|
||||
await new Promise<void>((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 }),
|
||||
]);
|
||||
});
|
||||
});
|
||||
@@ -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<void> {
|
||||
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();
|
||||
}
|
||||
};
|
||||
|
||||
Reference in New Issue
Block a user