diff --git a/src/adapters/opencode/plugin.ts b/src/adapters/opencode/plugin.ts index 5459359c..354ce074 100644 --- a/src/adapters/opencode/plugin.ts +++ b/src/adapters/opencode/plugin.ts @@ -26,7 +26,14 @@ import { fileURLToPath, pathToFileURL } from "node:url"; import { existsSync, readFileSync } from "node:fs"; import { resolveSessionDbPath, SessionDB } from "../../session/db.js"; -import { extractEvents, extractUserEvents, parseOpencodeUsage, buildAgentUsageEvent } from "../../session/extract.js"; +import { + extractEvents, + extractUserEvents, + parseOpencodeUsage, + buildAgentUsageEvent, + toOpencodeUsageStepDelta, + rememberOpencodeCumulativeCost, +} from "../../session/extract.js"; import type { HookInput } from "../../session/extract.js"; import { buildResumeSnapshot } from "../../session/snapshot.js"; import type { SessionEvent } from "../../types.js"; @@ -316,6 +323,14 @@ async function createContextModePlugin(ctx: PluginContext) { // lived plugin process still gets per-session capture exactly once. const agentsMdCaptured = new Set(); + // Per-assistant-message high-water for opencode cumulative cost → step delta + // (issue #1036). Key: `${sessionId}:${messageId}` (or sessionId alone when + // info.id is absent). Value: last observed cumulative `info.cost`. + // Bounded via rememberOpencodeCumulativeCost (insertion-order FIFO ~1000); + // the plugin lifecycle only consumes `message.updated` (no completion event + // to delete-on-finish). + const lastCumulativeCostByMessage = new Map(); + /** * OC-4: Read AGENTS.md (with CLAUDE.md / CONTEXT.md fallbacks) from the * project directory and persist as `rule` + `rule_content` events. Mirrors @@ -542,33 +557,39 @@ async function createContextModePlugin(ctx: PluginContext) { }, // ── event: per-turn token + cost capture (paid-observability) ─── - // The generic bus `event` hook (refs/platforms/opencode/packages/plugin/ - // src/index.ts:224) delivers every Event; we filter `message.updated` - // (published on each assistant-message update incl. step-finish — - // session.ts:673) and read tokens/cost/modelID off properties.info - // (assistant filter via role; refs stream.transport.ts:214-216). - // - // CAVEAT (refs processor.ts:717-718): message-level `.tokens` is the LAST - // step's snapshot (overwritten per step-finish), while `.cost` is - // cumulative for the turn. parseOpencodeUsage passes `.cost` through as - // native_cost_usd so the billed $ stays exact despite the token snapshot - // being last-step only. `message.updated` fires multiple times per turn; - // because tokens are a terminal snapshot and cost is cumulative, the last - // event for a message carries the final figures — re-emitting on each - // update is idempotent at the cost column and merely refreshes the - // last-step token telemetry. db.insertEvent both persists locally AND - // forwards to the platform (the TS-plugin equivalent of the .mjs - // attributeAndInsertEvents path). + // Filter bus `message.updated` (fires per step-finish). Opencode `.cost` + // is cumulative for the turn while `.tokens` is last-step; re-emitting the + // cumulative figure over-counts under additive aggregation (#1036). Convert + // via toOpencodeUsageStepDelta so summed rows equal the true turn cost. + // lastCumulativeCostByMessage tracks per-message high-water (FIFO-capped). event: async (input: EventHookInput) => { try { const ev = input?.event; if (!ev || ev.type !== "message.updated") return; - const sessionId = ev.properties?.info?.sessionID; + const info = ev.properties?.info; + const sessionId = info?.sessionID; if (!sessionId || typeof sessionId !== "string") return; const counts = parseOpencodeUsage(ev); if (!counts) return; - const usageEvent = buildAgentUsageEvent(counts); + + const messageId = typeof info?.id === "string" && info.id.length > 0 ? info.id : ""; + const messageKey = messageId.length > 0 ? `${sessionId}:${messageId}` : sessionId; + const prev = + lastCumulativeCostByMessage.has(messageKey) + ? (lastCumulativeCostByMessage.get(messageKey) as number) + : null; + const stepped = toOpencodeUsageStepDelta(counts, prev); + if (!stepped) return; + if (typeof stepped.nextCumulativeCost === "number") { + rememberOpencodeCumulativeCost( + lastCumulativeCostByMessage, + messageKey, + stepped.nextCumulativeCost, + ); + } + + const usageEvent = buildAgentUsageEvent(stepped.counts); if (!usageEvent) return; db.ensureSession(sessionId, projectDir); diff --git a/src/session/extract.ts b/src/session/extract.ts index 2b399be1..4b43fb8f 100644 --- a/src/session/extract.ts +++ b/src/session/extract.ts @@ -1892,6 +1892,63 @@ export function parseOpencodeUsage(payload: unknown): AgentUsageCounts | null { }; } +/** + * Convert opencode cumulative turn cost to a per-step delta for additive sum + * under multi-fire `message.updated` (#1036). Null on no-progress refresh; + * without native cost, emit only on first observation. High-water is tracked + * per message via previous/next cumulative; map is FIFO-capped (see CAP). + */ +/** FIFO cap for the in-memory cumulative-cost high-water map (#1036). */ +export const OPENCODE_CUMULATIVE_COST_MAP_CAP = 1000; + +/** Remember last cumulative cost per message key (insertion-order FIFO eviction). */ +export function rememberOpencodeCumulativeCost( + map: Map, + key: string, + cost: number, + maxSize: number = OPENCODE_CUMULATIVE_COST_MAP_CAP, +): void { + if (!map.has(key) && map.size >= maxSize) { + const oldest = map.keys().next().value; + if (oldest !== undefined) map.delete(oldest); + } + map.set(key, cost); +} + +export function toOpencodeUsageStepDelta( + counts: AgentUsageCounts, + previousCumulativeCost: number | null, +): { counts: AgentUsageCounts; nextCumulativeCost: number | null } | null { + const current = counts.native_cost_usd; + if (typeof current === "number" && Number.isFinite(current)) { + const prev = + typeof previousCumulativeCost === "number" && Number.isFinite(previousCumulativeCost) + ? previousCumulativeCost + : 0; + const delta = current - prev; + // Same (or lower) cumulative after we've already emitted → no-op refresh. + if (previousCumulativeCost !== null && delta <= 0) { + return null; + } + // First observation emits cumulative; later ones emit the step delta only. + const stepCost = previousCumulativeCost === null ? current : delta; + return { + counts: { ...counts, native_cost_usd: stepCost }, + nextCumulativeCost: current, + }; + } + + // No native cost → emit once per message (first observation only). + if (previousCumulativeCost !== null) { + return null; + } + return { + counts, + // Sentinel "seen" marker so subsequent fires without native cost are skipped. + nextCumulativeCost: 0, + }; +} + /** * Build a structured `agent_usage` event from summed per-model token counts. * Emits the colon-string `data` (human/debug + back-compat) AND the structured diff --git a/tests/session/parse-opencode-usage.test.ts b/tests/session/parse-opencode-usage.test.ts index c8c8be9c..3c31b1e2 100644 --- a/tests/session/parse-opencode-usage.test.ts +++ b/tests/session/parse-opencode-usage.test.ts @@ -24,6 +24,9 @@ import { describe, it, expect } from "vitest"; import { parseOpencodeUsage, buildAgentUsageEvent, + toOpencodeUsageStepDelta, + rememberOpencodeCumulativeCost, + OPENCODE_CUMULATIVE_COST_MAP_CAP, } from "../../src/session/extract.js"; /** Minimal opencode `message.updated` bus-event fixture. */ @@ -169,3 +172,116 @@ describe("parseOpencodeUsage", () => { expect(event?.data).toContain("cost_usd:"); }); }); + +/** + * #1036 — multi-step cumulative cost must become deltas that sum to the final cost. + */ +describe("toOpencodeUsageStepDelta (#1036 cumulative multi-step)", () => { + it("converts a 2-step cumulative turn (0.02 then 0.05) into deltas that sum to 0.05", () => { + const step1 = parseOpencodeUsage( + busEvent({ + cost: 0.02, + tokens: { input: 100, output: 20, cache: { read: 0, write: 0 } }, + }), + ); + const step2 = parseOpencodeUsage( + busEvent({ + cost: 0.05, + tokens: { input: 200, output: 40, cache: { read: 0, write: 0 } }, + }), + ); + expect(step1).not.toBeNull(); + expect(step2).not.toBeNull(); + + // Without delta: naive sum of native costs over-counts (the bug). + expect((step1!.native_cost_usd ?? 0) + (step2!.native_cost_usd ?? 0)).toBeCloseTo(0.07, 10); + + const d1 = toOpencodeUsageStepDelta(step1!, null); + expect(d1).not.toBeNull(); + expect(d1!.counts.native_cost_usd).toBeCloseTo(0.02, 10); + expect(d1!.nextCumulativeCost).toBeCloseTo(0.02, 10); + + const d2 = toOpencodeUsageStepDelta(step2!, d1!.nextCumulativeCost); + expect(d2).not.toBeNull(); + expect(d2!.counts.native_cost_usd).toBeCloseTo(0.03, 10); + expect(d2!.nextCumulativeCost).toBeCloseTo(0.05, 10); + + const e1 = buildAgentUsageEvent(d1!.counts); + const e2 = buildAgentUsageEvent(d2!.counts); + expect(e1?.cost_usd).toBeCloseTo(0.02, 10); + expect(e2?.cost_usd).toBeCloseTo(0.03, 10); + // Additive aggregation across step rows equals the true turn cost. + expect((e1!.cost_usd ?? 0) + (e2!.cost_usd ?? 0)).toBeCloseTo(0.05, 10); + }); + + it("skips a no-op refresh when cumulative cost does not advance", () => { + const first = parseOpencodeUsage(busEvent({ cost: 0.05 })); + const same = parseOpencodeUsage(busEvent({ cost: 0.05 })); + expect(first).not.toBeNull(); + expect(same).not.toBeNull(); + + const d1 = toOpencodeUsageStepDelta(first!, null); + expect(d1).not.toBeNull(); + const d2 = toOpencodeUsageStepDelta(same!, d1!.nextCumulativeCost); + expect(d2).toBeNull(); + }); + + it("emits catalog-priced rows only once when native cost is absent", () => { + const a = parseOpencodeUsage(busEvent({ cost: undefined })); + const b = parseOpencodeUsage(busEvent({ cost: undefined })); + expect(a).not.toBeNull(); + expect(b).not.toBeNull(); + expect(a!.native_cost_usd).toBeNull(); + + const d1 = toOpencodeUsageStepDelta(a!, null); + expect(d1).not.toBeNull(); + const d2 = toOpencodeUsageStepDelta(b!, d1!.nextCumulativeCost); + expect(d2).toBeNull(); + }); +}); + +/** + * #1036 follow-up — bound lastCumulativeCostByMessage so a long-lived plugin + * process cannot unbounded-grow the in-memory Map. Plugin lifecycle has no + * message-completion event to delete-on-finish; insertion-order FIFO cap. + */ +describe("rememberOpencodeCumulativeCost (#1036 map eviction)", () => { + it("evicts the oldest key when inserting past maxSize (insertion-order FIFO)", () => { + const map = new Map(); + const maxSize = 3; + rememberOpencodeCumulativeCost(map, "a", 0.01, maxSize); + rememberOpencodeCumulativeCost(map, "b", 0.02, maxSize); + rememberOpencodeCumulativeCost(map, "c", 0.03, maxSize); + expect(map.size).toBe(3); + expect([...map.keys()]).toEqual(["a", "b", "c"]); + + rememberOpencodeCumulativeCost(map, "d", 0.04, maxSize); + expect(map.size).toBe(3); + expect(map.has("a")).toBe(false); + expect([...map.keys()]).toEqual(["b", "c", "d"]); + expect(map.get("d")).toBe(0.04); + }); + + it("updating an existing key does not grow the map or evict others", () => { + const map = new Map(); + const maxSize = 2; + rememberOpencodeCumulativeCost(map, "a", 0.01, maxSize); + rememberOpencodeCumulativeCost(map, "b", 0.02, maxSize); + rememberOpencodeCumulativeCost(map, "a", 0.05, maxSize); + expect(map.size).toBe(2); + expect(map.get("a")).toBe(0.05); + expect(map.has("b")).toBe(true); + }); + + it(`defaults maxSize to OPENCODE_CUMULATIVE_COST_MAP_CAP (${OPENCODE_CUMULATIVE_COST_MAP_CAP})`, () => { + const map = new Map(); + for (let i = 0; i < OPENCODE_CUMULATIVE_COST_MAP_CAP; i++) { + rememberOpencodeCumulativeCost(map, `k${i}`, i); + } + expect(map.size).toBe(OPENCODE_CUMULATIVE_COST_MAP_CAP); + rememberOpencodeCumulativeCost(map, "overflow", 1); + expect(map.size).toBe(OPENCODE_CUMULATIVE_COST_MAP_CAP); + expect(map.has("k0")).toBe(false); + expect(map.has("overflow")).toBe(true); + }); +});