root/fusion
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(FN-4388): complete Step 1 — engine cache metrics logs

Browse Source 
Fusion-Task-Id: FN-4388
Fusion-Task-Lineage: 7497a0a1-5edb-4778-b339-118d1fd668f7
This commit is contained in:
Fusion
2026-05-13 22:14:03 -07:00
committed by gsxdsm
parent cf5a681ba2
commit 48f18511d8
3 changed files with 101 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files

96
packages/engine/src/__tests__/session-token-usage.test.ts
View File

@@ -1,10 +1,10 @@
import { describe, it, expect, vi, beforeEach } from "vitest";
import type { Task, TaskStore } from "@fusion/core";
import { accumulateSessionTokenUsage } from "../session-token-usage.js";
import { accumulateSessionTokenUsage, computeCacheHitRatio } from "../session-token-usage.js";
import { TaskExecutor } from "../executor.js";
interface MockSessionStats {
tokens?: { input?: number; output?: number; cacheRead?: number; cacheWrite?: number };
tokens?: { input?: number; output?: number; cacheRead?: number; cacheWrite?: number; total?: number };
}
function createSession(stats: MockSessionStats | undefined) {
@@ -28,13 +28,15 @@ function createStore(initial: Task["tokenUsage"]): TaskStore & { _task: Task; up
describe("accumulateSessionTokenUsage", () => {
beforeEach(() => {
vi.useRealTimers();
vi.restoreAllMocks();
});
it("writes initial token usage when task has none", async () => {
it("writes initial token usage and emits cache metrics log", async () => {
const errorSpy = vi.spyOn(console, "error").mockImplementation(() => {});
const store = createStore(undefined);
const session = createSession({ tokens: { input: 100, output: 30, cacheRead: 5, cacheWrite: 2 } });
await accumulateSessionTokenUsage(store, "FN-1", session);
await accumulateSessionTokenUsage(store, "FN-1", session, { agentId: "agent-1", role: "reviewer" });
expect(store.updateTask).toHaveBeenCalledTimes(1);
const call = store.updateTask.mock.calls[0]![1] as { tokenUsage: Task["tokenUsage"] };
@@ -45,61 +47,22 @@ describe("accumulateSessionTokenUsage", () => {
cacheWriteTokens: 2,
totalTokens: 137,
});
expect(typeof call.tokenUsage!.firstUsedAt).toBe("string");
expect(typeof call.tokenUsage!.lastUsedAt).toBe("string");
});
it("accumulates only the delta on subsequent calls for the same session", async () => {
const store = createStore(undefined);
const session = createSession({ tokens: { input: 100, output: 30, cacheRead: 0, cacheWrite: 0 } });
await accumulateSessionTokenUsage(store, "FN-1", session);
// Second call: session has progressed.
(session as unknown as { getSessionStats: () => MockSessionStats }).getSessionStats = () => ({
tokens: { input: 250, output: 80, cacheRead: 0, cacheWrite: 0 },
});
await accumulateSessionTokenUsage(store, "FN-1", session);
expect(store.updateTask).toHaveBeenCalledTimes(2);
const second = store.updateTask.mock.calls[1]![1] as { tokenUsage: Task["tokenUsage"] };
expect(second.tokenUsage).toMatchObject({
inputTokens: 250,
outputTokens: 80,
cachedTokens: 0,
totalTokens: 330,
const cacheLogCall = errorSpy.mock.calls.find((entry) => String(entry[0]).includes("[token-cache-metrics]"));
expect(cacheLogCall).toBeTruthy();
const payload = JSON.parse(String(cacheLogCall?.[0] ?? "").replace(/^.*\[token-cache-metrics\]\s*/, ""));
expect(payload).toMatchObject({
taskId: "FN-1",
agentId: "agent-1",
role: "reviewer",
inputTokens: 100,
cachedTokens: 5,
cacheWriteTokens: 2,
hitRatio: computeCacheHitRatio(100, 5),
});
});
it("preserves firstUsedAt across updates", async () => {
const store = createStore(undefined);
const session = createSession({ tokens: { input: 10, output: 5, cacheRead: 0, cacheWrite: 0 } });
await accumulateSessionTokenUsage(store, "FN-1", session);
const first = store.updateTask.mock.calls[0]![1] as { tokenUsage: Task["tokenUsage"] };
const initialFirstUsed = first.tokenUsage!.firstUsedAt;
await new Promise((resolve) => setTimeout(resolve, 5));
(session as unknown as { getSessionStats: () => MockSessionStats }).getSessionStats = () => ({
tokens: { input: 20, output: 5, cacheRead: 0, cacheWrite: 0 },
});
await accumulateSessionTokenUsage(store, "FN-1", session);
const second = store.updateTask.mock.calls[1]![1] as { tokenUsage: Task["tokenUsage"] };
expect(second.tokenUsage!.firstUsedAt).toBe(initialFirstUsed);
expect(second.tokenUsage!.lastUsedAt >= initialFirstUsed).toBe(true);
});
it("does nothing when session has no getSessionStats", async () => {
const store = createStore(undefined);
const session = {} as Parameters<typeof accumulateSessionTokenUsage>[2];
await accumulateSessionTokenUsage(store, "FN-1", session);
expect(store.updateTask).not.toHaveBeenCalled();
});
it("does nothing when delta is zero", async () => {
it("does nothing when delta is zero (no write, no metrics log)", async () => {
const errorSpy = vi.spyOn(console, "error").mockImplementation(() => {});
const store = createStore({
inputTokens: 50,
outputTokens: 20,
@@ -112,10 +75,29 @@ describe("accumulateSessionTokenUsage", () => {
const session = createSession({ tokens: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 } });
await accumulateSessionTokenUsage(store, "FN-1", session);
// Calling again with the same zero stats should not produce another update.
await accumulateSessionTokenUsage(store, "FN-1", session);
expect(store.updateTask).not.toHaveBeenCalled();
expect(errorSpy.mock.calls.find((entry) => String(entry[0]).includes("[token-cache-metrics]"))).toBeUndefined();
});
it("emits token-cache-metrics log when executor persists non-zero delta", async () => {
const errorSpy = vi.spyOn(console, "error").mockImplementation(() => {});
const store = createStore(undefined);
const executor = Object.create(TaskExecutor.prototype) as TaskExecutor & {
store: TaskStore;
tokenUsageBaselines: Map<string, { inputTokens: number; outputTokens: number; cachedTokens: number; cacheWriteTokens: number; totalTokens: number }>;
activeSessions: Map<string, { session: unknown }>;
persistTokenUsage: (taskId: string, session?: unknown) => Promise<void>;
};
executor.store = store;
executor.tokenUsageBaselines = new Map();
executor.activeSessions = new Map();
await executor.persistTokenUsage("FN-1", { getSessionStats: () => ({ tokens: { input: 3, output: 2, cacheRead: 1, cacheWrite: 0, total: 6 } }) });
const cacheLogCall = errorSpy.mock.calls.find((entry) => String(entry[0]).includes("[token-cache-metrics]"));
expect(cacheLogCall).toBeTruthy();
});
it("swallows store errors instead of throwing", async () => {

44
packages/engine/src/executor.ts
View File

@@ -46,7 +46,7 @@ import {
inspectBranchConflict,
} from "./branch-conflicts.js";
import { AgentLogger } from "./agent-logger.js";
import { executorLog, reviewerLog, formatError } from "./logger.js";
import { createLogger, executorLog, reviewerLog, formatError } from "./logger.js";
import { TokenCapDetector } from "./token-cap-detector.js";
import { isUsageLimitError, checkSessionError, type UsageLimitPauser } from "./usage-limit-detector.js";
import { isTransientError, isSilentTransientError } from "./transient-error-detector.js";
@@ -121,6 +121,8 @@ export {
taskLogParams,
} from "./agent-tools.js";
const tokenCacheMetricsLog = createLogger("token-cache-metrics");
const STEP_STATUSES: StepStatus[] = ["pending", "in-progress", "done", "skipped"];
function canonicalizePath(path: string): string {
@@ -1984,6 +1986,16 @@ export class TaskExecutor {
const merged = this.accumulateTokenUsage(task.tokenUsage, delta);
if (!merged) return;
tokenCacheMetricsLog.log(JSON.stringify({
taskId,
agentId: task.assignedAgentId ?? undefined,
role: "executor",
inputTokens: merged.inputTokens,
cachedTokens: merged.cachedTokens,
cacheWriteTokens: merged.cacheWriteTokens,
hitRatio: merged.inputTokens + merged.cachedTokens > 0 ? merged.cachedTokens / (merged.inputTokens + merged.cachedTokens) : 0,
}));
await this.store.updateTask(taskId, { tokenUsage: merged });
}
@@ -3368,7 +3380,10 @@ export class TaskExecutor {
// session.prompt() resolves normally even when retries are exhausted —
// the error is stored on session.state.error instead of being thrown.
checkSessionError(session);
await accumulateSessionTokenUsage(this.store, task.id, session);
await accumulateSessionTokenUsage(this.store, task.id, session, {
agentId: task.assignedAgentId ?? undefined,
role: "executor",
});
// Check if proactive context compaction is needed based on token cap setting.
// This runs after the main prompt completes to avoid interrupting active work.
@@ -3426,7 +3441,10 @@ export class TaskExecutor {
await promptWithFallback(session, resumePrompt);
checkSessionError(session);
await accumulateSessionTokenUsage(this.store, task.id, session);
await accumulateSessionTokenUsage(this.store, task.id, session, {
agentId: task.assignedAgentId ?? undefined,
role: "executor",
});
}
// If dependency was added during execution, discard worktree and move to triage
@@ -3665,7 +3683,10 @@ export class TaskExecutor {
stuckDetector?.recordActivity(task.id);
await promptWithFallback(retrySession, retryPrompt);
checkSessionError(retrySession);
await accumulateSessionTokenUsage(this.store, task.id, retrySession);
await accumulateSessionTokenUsage(this.store, task.id, retrySession, {
agentId: task.assignedAgentId ?? undefined,
role: "executor",
});
if (!taskDone) {
const implicitCheck = await this.store.getTask(task.id);
@@ -3907,7 +3928,10 @@ export class TaskExecutor {
await promptWithFallback(activeEntry.session, reducedPrompt);
checkSessionError(activeEntry.session);
await accumulateSessionTokenUsage(this.store, task.id, activeEntry.session);
await accumulateSessionTokenUsage(this.store, task.id, activeEntry.session, {
agentId: task.assignedAgentId ?? undefined,
role: "executor",
});
// Reduced-prompt retry succeeded — return to let the finally block clean up
// without marking the task as failed.
@@ -5462,7 +5486,10 @@ ${failureContext.output.slice(0, VERIFICATION_LOG_MAX_CHARS)}
executorLog.warn(`⏳ ${task.id} executor fix agent rate limited — retry ${attempt} in ${delaySec}s: ${error.message}`);
},
});
await accumulateSessionTokenUsage(this.store, task.id, session);
await accumulateSessionTokenUsage(this.store, task.id, session, {
agentId: task.assignedAgentId ?? undefined,
role: "executor",
});
// Re-run full deterministic verification (test AND build) after the fix attempt
executorLog.log(`${task.id}: re-running deterministic verification after fix attempt ${retryNumber}/${maxRetries}`);
@@ -6478,7 +6505,10 @@ and show an appropriate message to the user.\`
// Completed within the timeout — let any post-completion errors surface.
checkSessionError(session);
await accumulateSessionTokenUsage(this.store, task.id, session);
await accumulateSessionTokenUsage(this.store, task.id, session, {
agentId: task.assignedAgentId ?? undefined,
role: "executor",
});
session.dispose();
await agentLogger.flush();

37
packages/engine/src/session-token-usage.ts
View File

@@ -1,8 +1,9 @@
import type { TaskStore } from "@fusion/core";
import type { AgentRole, TaskStore } from "@fusion/core";
import type { AgentSession } from "@mariozechner/pi-coding-agent";
import { createLogger } from "./logger.js";
const log = createLogger("session-token-usage");
const cacheMetricsLog = createLogger("token-cache-metrics");
interface SessionBaseline {
input: number;
@@ -45,6 +46,7 @@ export async function accumulateSessionTokenUsage(
store: TaskStore,
taskId: string,
session: AgentSession,
options?: { agentId?: string; role?: AgentRole },
): Promise<void> {
try {
const stats = readSessionStats(session);
@@ -78,17 +80,28 @@ export async function accumulateSessionTokenUsage(
const newCached = (task.tokenUsage?.cachedTokens ?? 0) + cachedDelta;
const newCacheWrite = (task.tokenUsage?.cacheWriteTokens ?? 0) + cacheWriteDelta;
await store.updateTask(taskId, {
tokenUsage: {
inputTokens: newInput,
outputTokens: newOutput,
cachedTokens: newCached,
cacheWriteTokens: newCacheWrite,
totalTokens: newInput + newOutput + newCached + newCacheWrite,
firstUsedAt: task.tokenUsage?.firstUsedAt ?? now,
lastUsedAt: now,
},
});
const role = options?.role ?? "executor";
const tokenUsage = {
inputTokens: newInput,
outputTokens: newOutput,
cachedTokens: newCached,
cacheWriteTokens: newCacheWrite,
totalTokens: newInput + newOutput + newCached + newCacheWrite,
firstUsedAt: task.tokenUsage?.firstUsedAt ?? now,
lastUsedAt: now,
};
cacheMetricsLog.log(JSON.stringify({
taskId,
agentId: options?.agentId,
role,
inputTokens: tokenUsage.inputTokens,
cachedTokens: tokenUsage.cachedTokens,
cacheWriteTokens: tokenUsage.cacheWriteTokens,
hitRatio: computeCacheHitRatio(tokenUsage.inputTokens, tokenUsage.cachedTokens),
}));
await store.updateTask(taskId, { tokenUsage });
} catch (err) {
const message = err instanceof Error ? err.message : String(err);
log.warn(`${taskId}: session token usage accumulate failed: ${message}`);