fix(devin): read usage from ModelUsageStats and classify cloud failures

src/adapters/devin.ts

@@ -13,6 +13,53 @@ import { getCachedCatalog } from "./devin/cloud-direct/catalog";
 import { buildNonOpenAIToolCatalogNudgeForTools } from "./tool-catalog-nudge";
 import { DEVIN_DEFAULT_API_SERVER, resolveDevinApiServer } from "../oauth/devin";
 
+/**
+ * Combine two usage frames from one turn by keeping the larger count per field.
+ *
+ * Devin's counters are cumulative within a turn, so a frame that reports less
+ * than an earlier one is reporting a subset, not a correction.
+ */
+export function mergeDevinUsage(previous: OcxUsage, next: OcxUsage): OcxUsage {
+  const keys = [
+    "inputTokens", "outputTokens",
+    "cachedInputTokens", "cacheReadInputTokens", "cacheCreationInputTokens",
+    "reasoningOutputTokens",
+  ] as const;
+  const merged: OcxUsage = { ...previous, ...next };
+  for (const key of keys) {
+    const a = previous[key];
+    const b = next[key];
+    if (typeof a === "number" && typeof b === "number") merged[key] = Math.max(a, b);
+    else if (typeof a === "number" && b === undefined) merged[key] = a;
+  }
+  // totalTokens is derived, not merged. Taking the max of two totals alongside
+  // per-field maxima can leave total !== input + output, and the cost and log
+  // paths read the total.
+  const total = (merged.inputTokens ?? 0) + (merged.outputTokens ?? 0);
+  if (total > 0) merged.totalTokens = total;
+  return merged;
+}
+
+/**
+ * The wording `isClientClosedMessage` recognises.
+ *
+ * "Devin turn was aborted." matched nothing, so a cancelled turn fell through to
+ * the default inference and was logged as a 502 upstream failure rather than as
+ * the client hanging up.
+ */
+const DEVIN_CLIENT_CLOSED_MESSAGE = "client closed request";
+
+/** Map a cloud-direct failure onto the structured fields the error event carries. */
+export function devinErrorClassification(error: unknown): { status?: number; errorType?: string; retryable?: boolean } {
+  const status = error instanceof CloudChatError ? error.status : undefined;
+  if (status === undefined) return {};
+  if (status === 401) return { status, errorType: "authentication_error", retryable: false };
+  if (status === 403) return { status, errorType: "permission_error", retryable: false };
+  if (status === 429) return { status, errorType: "rate_limit_error", retryable: true };
+  if (status >= 500) return { status, retryable: true };
+  return { status, retryable: false };
+}
+
 export const DEVIN_API_SERVER = DEVIN_DEFAULT_API_SERVER;
 
 const EFFORT_SUFFIXES = new Set(["low", "medium", "high", "xhigh", "max", "none", "1m", "max-1m", "none-1m", "fast"]);
@@ -210,7 +257,7 @@ export function createDevinAdapter(
 
     async runTurn(parsed: OcxParsedRequest, incoming: IncomingMeta, emit: (event: AdapterEvent) => void) {
       if (incoming.abortSignal?.aborted) {
-        emit({ type: "error", message: "Devin turn was aborted before start." });
+        emit({ type: "error", message: DEVIN_CLIENT_CLOSED_MESSAGE, status: 499, retryable: false });
         return;
       }
       let apiKey: string;
@@ -272,7 +319,7 @@ export function createDevinAdapter(
             // Say what happened instead, the way the other runTurn-only adapter
             // does, and carry any usage already seen.
             closeOpenTool();
-            emit({ type: "error", message: "Devin turn was aborted.", ...(usage ? { usage } : {}) });
+            emit({ type: "error", message: DEVIN_CLIENT_CLOSED_MESSAGE, status: 499, retryable: false, ...(usage ? { usage } : {}) });
             return;
           }
           if (event.kind === "text") {
@@ -305,35 +352,46 @@ export function createDevinAdapter(
           }
           if (event.kind === "usage") {
             const total = event.totalTokens ?? ((event.promptTokens ?? 0) + (event.completionTokens ?? 0));
-            usage = {
+            const next: OcxUsage = {
               inputTokens: event.promptTokens ?? 0,
               outputTokens: event.completionTokens ?? 0,
               ...(total > 0 ? { totalTokens: total } : {}),
               ...(event.cachedInputTokens !== undefined ? { cachedInputTokens: event.cachedInputTokens } : {}),
               ...(event.cacheCreationInputTokens !== undefined ? { cacheCreationInputTokens: event.cacheCreationInputTokens } : {}),
               ...(event.reasoningTokens !== undefined ? { reasoningOutputTokens: event.reasoningTokens } : {}),
             };
+            // Merge rather than replace. A turn can carry more than one usage
+            // frame, and the counters are cumulative, so a later partial frame
+            // that omits a field used to zero a count the earlier frame had
+            // already reported.
+            usage = usage ? mergeDevinUsage(usage, next) : next;
             continue;
           }
         }
         closeOpenTool();
         if (incoming.abortSignal?.aborted) {
-          emit({ type: "error", message: "Devin turn was aborted.", ...(usage ? { usage } : {}) });
+          emit({ type: "error", message: DEVIN_CLIENT_CLOSED_MESSAGE, status: 499, retryable: false, ...(usage ? { usage } : {}) });
         } else {
           emit({ type: "done", ...(usage ? { usage } : {}), ...(stopReason ? { stopReason } : {}) });
         }
       } catch (error) {
         closeOpenTool();
         if (incoming.abortSignal?.aborted) {
-          emit({ type: "error", message: "Devin turn was aborted.", ...(usage ? { usage } : {}) });
+          emit({ type: "error", message: DEVIN_CLIENT_CLOSED_MESSAGE, status: 499, retryable: false, ...(usage ? { usage } : {}) });
           return;
         }
         const message = error instanceof CloudChatError
           ? ("Devin cloud error" + (error.code ? " " + error.code : "") + ": " + error.message)
           : error instanceof Error ? error.message : String(error);
         // Usage that already arrived is still real; dropping it loses the
         // accounting for a turn that did most of its work before failing.
-        emit({ type: "error", message, ...(usage ? { usage } : {}) });
+        emit({
+          type: "error",
+          message,
+          ...devinErrorClassification(error),
+          ...(error instanceof CloudChatError && error.code ? { code: error.code } : {}),
+          ...(usage ? { usage } : {}),
+        });
       }
     },
   };

src/adapters/devin/cloud-direct/chat.ts

@@ -669,7 +669,30 @@ function buildGetChatMessageRequest(args: BuildArgs): Buffer {
  * any non-zero to 'tool_calls' for now (and let the caller fall back to
  * 'stop' if no tool_call deltas were emitted).
  */
-function* decodeChatFrame(proto: Buffer): Generator<CloudChatEvent> {
+export function* decodeChatFrame(proto: Buffer): Generator<CloudChatEvent> {
+  // Field 7 is `ModelUsageStats`, the authoritative per-turn accounting, and
+  // field 28 is `response_dimension_groups` — the rows the IDE renders. The
+  // decoder below reads 28 because a capture happened to expose metric-looking
+  // strings there (`ResponseDimension.uid` is its field 5, which is what the
+  // entry walker treats as `metric_id`), and that works only when the service
+  // chose to render cache rows. Field 7 carries cache read and cache write
+  // unconditionally, which is why a cached Devin turn used to report a bare
+  // total with no cached subset.
+  //
+  // Both fields arrive in the same message and the adapter keeps the last usage
+  // event it sees, so this cannot be a plain "decode both": field 7 has to
+  // suppress field 28 within the message. It is yielded before the rest of the
+  // frame rather than after it, so a frame that also carries finish (field 5)
+  // still reports usage ahead of the turn's end, and the order does not depend
+  // on where the service happens to place the field.
+  let authoritativeUsage: CloudChatEvent | null = null;
+  for (const f of iterFields(proto)) {
+    if (f.num === 7 && f.wire === 2 && Buffer.isBuffer(f.value)) {
+      authoritativeUsage = decodeModelUsageStats(f.value as Buffer);
+      if (authoritativeUsage) break;
+    }
+  }
+  if (authoritativeUsage) yield authoritativeUsage;
   for (const f of iterFields(proto)) {
     if (f.num === 3 && f.wire === 2 && Buffer.isBuffer(f.value)) {
       // Visible delta_text — what the user should SEE in the chat.
@@ -740,6 +763,7 @@ function* decodeChatFrame(proto: Buffer): Generator<CloudChatEvent> {
       // else stays 'stop' for 0/2/4-9/12/13
       yield { kind: 'finish', reason };
     } else if (f.num === 28 && f.wire === 2 && Buffer.isBuffer(f.value)) {
+      if (authoritativeUsage) continue;
       const usage = decodeUsageBlock(f.value as Buffer);
       if (usage) yield usage;
     }
@@ -834,6 +858,68 @@ function decodeUsageBlock(buf: Buffer): CloudChatEvent | null {
   };
 }
 
+/**
+ * `exa.codeium_common_pb.ModelUsageStats` at GetChatMessageResponse field 7.
+ *
+ *   ModelUsageStats {
+ *     #2 input_tokens        uint64
+ *     #3 output_tokens       uint64
+ *     #4 cache_write_tokens  uint64
+ *     #5 cache_read_tokens   uint64
+ *   }
+ *
+ * Plain varints, so the field-28 entry walker — which descends a
+ * length-delimited sub-message and reads a fixed32 float — cannot read this at
+ * all. It needs its own decoder.
+ *
+ * Whether Cognition's `input_tokens` already includes the cached tokens is not
+ * settled. oh-my-pi sums all four into its total, which suggests exclusive, but
+ * that is their convention rather than a measurement of this field. Guessing
+ * wrong in the inclusive direction is the expensive mistake: `normalizeCostTokens`
+ * only rejects `read + write > input`, so an inflated input passes validation and
+ * bills cached tokens at the uncached rate.
+ *
+ * So the shape is derived from the frame instead of assumed. An input that
+ * already covers the cache is left alone; one that cannot possibly cover it is
+ * folded. Both branches agree on the case that motivated this — a 58k prompt
+ * that is 57k cache read and 1k fresh reads as 58k with a 57k cached subset —
+ * and neither can emit `read + write > input`. Replace the derivation with a
+ * fixed mapping once a live frame settles the question.
+ */
+export function decodeModelUsageStats(buf: Buffer): CloudChatEvent | null {
+  let wireInput: number | undefined;
+  let output: number | undefined;
+  let cacheWrite: number | undefined;
+  let cacheRead: number | undefined;
+  for (const f of iterFields(buf)) {
+    if (f.wire !== 0) continue;
+    const n = Number(f.value);
+    if (!Number.isFinite(n) || n < 0) continue;
+    if (f.num === 2) wireInput = n;
+    else if (f.num === 3) output = n;
+    else if (f.num === 4) cacheWrite = n;
+    else if (f.num === 5) cacheRead = n;
+  }
+  if (wireInput === undefined && output === undefined && cacheRead === undefined && cacheWrite === undefined) {
+    return null;
+  }
+  const read = cacheRead ?? 0;
+  const write = cacheWrite ?? 0;
+  const rawInput = wireInput ?? 0;
+  const promptTokens = rawInput >= read + write ? rawInput : rawInput + read + write;
+  const completionTokens = output ?? 0;
+  const total = promptTokens + completionTokens;
+  return {
+    kind: 'usage',
+    promptTokens,
+    completionTokens,
+    totalTokens: total > 0 ? total : undefined,
+    cachedInputTokens: cacheRead,
+    cacheCreationInputTokens: cacheWrite,
+    reasoningTokens: undefined,
+  };
+}
+
 // ----------------------------------------------------------------------------
 // Public API: streamChat
 // ----------------------------------------------------------------------------
@@ -864,7 +950,18 @@ export interface CloudChatRequest {
 }
 
 export class CloudChatError extends Error {
-  constructor(message: string, public readonly code?: string, public readonly traceId?: string) {
+  constructor(
+    message: string,
+    public readonly code?: string,
+    public readonly traceId?: string,
+    /**
+     * Upstream HTTP status, when the failure was a status line rather than a
+     * Connect trailer. Without it the adapter's message reaches
+     * `inferHttpStatusFromAdapterMessage`, which does not parse `HTTP 429`, so
+     * a live rate limit was classified 502 and core's failover never rotated.
+     */
+    public readonly status?: number,
+  ) {
     super(message);
     this.name = 'CloudChatError';
   }
@@ -987,7 +1084,12 @@ export async function* streamChatEvents(req: CloudChatRequest): AsyncGenerator<C
     // The body is not echoed into the message. This error reaches the adapter's
     // error event and /api/logs, and a Connect error can quote the request that
     // produced it - which is the request holding the api_key.
-    throw new CloudChatError(`GetChatMessage failed (HTTP ${resp.status})`, undefined);
+    //
+    // Only the HTTP status line is carried here. A Connect EOS trailer that
+    // reports resource_exhausted or unavailable still arrives without a status,
+    // so a cap delivered that way keeps the older message-inference path.
+    // Mapping trailer codes onto HTTP statuses is deliberately a follow-up.
+    throw new CloudChatError(`GetChatMessage failed (HTTP ${resp.status})`, undefined, undefined, resp.status);
   }
   if (!resp.body) {
     throw new CloudChatError('GetChatMessage response had no body stream');

tests/providers/devin-hardening.test.ts

@@ -5,6 +5,9 @@ import { DEVIN_DEFAULT_API_SERVER, resolveDevinApiBaseUrl, validateDevinApiBaseU
 import { registerUser } from "../../src/oauth/devin/register-user";
 import { anySignal } from "../../src/lib/abort";
 import { buildGetChatMessageRequestForTests } from "../../src/adapters/devin/cloud-direct/chat";
+import { decodeModelUsageStats } from "../../src/adapters/devin/cloud-direct/chat";
+import { CloudChatError, decodeChatFrame } from "../../src/adapters/devin/cloud-direct/chat";
+import { devinErrorClassification, mergeDevinUsage } from "../../src/adapters/devin";
 import { iterFields } from "../../src/adapters/devin/cloud-direct/wire";
 import { buildMetadata, normalizeDevinSessionToken } from "../../src/adapters/devin/cloud-direct/metadata";
 
@@ -280,3 +283,118 @@ describe("devin session-token normalization", () => {
     }
   });
 });
+
+describe("devin ModelUsageStats decode (response field 7)", () => {
+  function varint(num: number, value: number): Buffer {
+    const out: number[] = [(num << 3) | 0];
+    let v = value;
+    do { const b = v & 0x7f; v = Math.floor(v / 128); out.push(v > 0 ? b | 0x80 : b); } while (v > 0);
+    return Buffer.from(out);
+  }
+  const stats = (input: number, output: number, write: number, read: number) =>
+    Buffer.concat([varint(2, input), varint(3, output), varint(4, write), varint(5, read)]);
+
+  test("an exclusive frame folds cache into the inclusive input this repo reports", () => {
+    // 1k fresh + 57k cache read is the 58k prompt the user sees as one number.
+    const u = decodeModelUsageStats(stats(1_000, 200, 0, 57_000));
+    expect(u?.promptTokens).toBe(58_000);
+    expect(u?.cachedInputTokens).toBe(57_000);
+    expect(u?.totalTokens).toBe(58_200);
+  });
+
+  test("an already-inclusive frame is left alone rather than inflated", () => {
+    const u = decodeModelUsageStats(stats(58_000, 200, 0, 57_000));
+    expect(u?.promptTokens).toBe(58_000);
+    expect(u?.cachedInputTokens).toBe(57_000);
+    // normalizeCostTokens only rejects read + write > input, so an inflated
+    // input would pass validation and bill cache at the uncached rate.
+    expect(u!.cachedInputTokens! + (u!.cacheCreationInputTokens ?? 0)).toBeLessThanOrEqual(u!.promptTokens!);
+  });
+
+  test("cache write counts as prompt too, and an empty message decodes to nothing", () => {
+    const u = decodeModelUsageStats(stats(1_000, 0, 4_000, 0));
+    expect(u?.promptTokens).toBe(5_000);
+    expect(u?.cacheCreationInputTokens).toBe(4_000);
+    expect(decodeModelUsageStats(Buffer.alloc(0))).toBeNull();
+  });
+});
+
+describe("devin frame-level usage precedence and classification", () => {
+  // Tags above 15 need a multi-byte varint: field 28 wire 2 is 226, and
+  // writing that as one raw byte sets the continuation bit and swallows the
+  // next byte.
+  function uvarint(value: number): number[] {
+    const out: number[] = [];
+    let v = value;
+    do { const b = v & 0x7f; v = Math.floor(v / 128); out.push(v > 0 ? b | 0x80 : b); } while (v > 0);
+    return out;
+  }
+  function varint(num: number, value: number): Buffer {
+    return Buffer.from([...uvarint((num << 3) | 0), ...uvarint(value)]);
+  }
+  function lenDelim(num: number, payload: Buffer): Buffer {
+    return Buffer.concat([Buffer.from([...uvarint((num << 3) | 2), ...uvarint(payload.length)]), payload]);
+  }
+  // ResponseDimensionGroup carrying a cumulative metric whose uid reads like a
+  // metric id — the shape the old decoder mined for usage.
+  function displayGroup(uid: string, value: number): Buffer {
+    const f32 = Buffer.alloc(5);
+    f32.writeUInt8((2 << 3) | 5, 0);
+    f32.writeFloatLE(value, 1);
+    const entry = Buffer.concat([lenDelim(4, f32), lenDelim(5, Buffer.from(uid, "utf8"))]);
+    return lenDelim(2, entry);
+  }
+
+  test("field 7 suppresses the display rows and is reported before finish", () => {
+    const stats = Buffer.concat([varint(2, 1_000), varint(3, 200), varint(4, 0), varint(5, 57_000)]);
+    const frame = Buffer.concat([
+      lenDelim(7, stats),
+      varint(5, 2),                                   // stop_reason STOP_PATTERN
+      lenDelim(28, displayGroup("input_tokens", 999)), // the wrong, display-derived number
+    ]);
+    const events = [...decodeChatFrame(frame)];
+    const usages = events.filter(e => e.kind === "usage");
+    expect(usages).toHaveLength(1);
+    expect(usages[0]!.promptTokens).toBe(58_000);
+    expect(usages[0]!.cachedInputTokens).toBe(57_000);
+    // Ahead of finish, so ordering does not depend on where the service puts
+    // the field.
+    expect(events.findIndex(e => e.kind === "usage"))
+      .toBeLessThan(events.findIndex(e => e.kind === "finish"));
+  });
+
+  test("a frame with no field 7 still falls back to the display rows", () => {
+    const frame = lenDelim(28, Buffer.concat([
+      displayGroup("input_tokens", 4_000),
+      displayGroup("output_tokens", 100),
+    ]));
+    const usages = [...decodeChatFrame(frame)].filter(e => e.kind === "usage");
+    expect(usages).toHaveLength(1);
+    expect(usages[0]!.promptTokens).toBe(4_000);
+  });
+});
+
+describe("devin usage merging and error classification", () => {
+  test("a later partial frame cannot zero an earlier count, and the total stays derived", () => {
+    const merged = mergeDevinUsage(
+      { inputTokens: 58_000, outputTokens: 200, totalTokens: 58_200, cachedInputTokens: 57_000 },
+      { inputTokens: 58_000, outputTokens: 900 },
+    );
+    expect(merged.cachedInputTokens).toBe(57_000);
+    expect(merged.outputTokens).toBe(900);
+    // Taking the max of two totals alongside per-field maxima would leave
+    // 58,200 here, which no longer equals input + output.
+    expect(merged.totalTokens).toBe(58_900);
+  });
+
+  test("an HTTP status on the cloud error becomes a structured classification", () => {
+    expect(devinErrorClassification(new CloudChatError("x", undefined, undefined, 429)))
+      .toEqual({ status: 429, errorType: "rate_limit_error", retryable: true });
+    expect(devinErrorClassification(new CloudChatError("x", undefined, undefined, 401)))
+      .toEqual({ status: 401, errorType: "authentication_error", retryable: false });
+    expect(devinErrorClassification(new CloudChatError("x", undefined, undefined, 503)))
+      .toEqual({ status: 503, retryable: true });
+    // A Connect trailer carries no status, so it keeps the older inference path.
+    expect(devinErrorClassification(new CloudChatError("x", "resource_exhausted"))).toEqual({});
+  });
+});