fix(cost): emit opencode multi-step usage as deltas (#1036)

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<string>();
 
+  // 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<string, number>();
+
   /**
    * 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);

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<string, number>,
+  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

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<string, number>();
+    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<string, number>();
+    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<string, number>();
+    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);
+  });
+});