From 27784653ac427f8f49c36c29cbb36efce4a3ff37 Mon Sep 17 00:00:00 2001 From: JUN Date: Sat, 12 Sep 2026 23:54:50 +0900 Subject: [PATCH 1/2] fix(devin): read usage from ModelUsageStats and classify cloud failures A cached Devin turn reported a bare token total with no cached subset, so its log row looked like a smaller request than it was. The decoder was reading GetChatMessageResponse field 28. Field 28 is response_dimension_groups, the rows the IDE renders; field 7 is ModelUsageStats, the per-turn accounting. The old path worked by accident: ResponseDimension.uid is that message's field 5, which the entry walker treats as a metric id, so cache numbers appeared only when the service happened to render cache rows. Field 7 carries cache read and cache write unconditionally. Both fields arrive in the same message and the adapter keeps the last usage event, so decoding both is not enough: field 7 now suppresses field 28 within a message and is yielded last, and it needs its own uint64 decoder because the field-28 walker reads a fixed32 float out of a sub-message. Whether Cognition's input_tokens already includes cache is unsettled, and guessing inclusive is the expensive error: normalizeCostTokens only rejects read + write > input, so an inflated input passes validation and bills cached tokens at the uncached rate. The mapping is therefore derived from the frame. Both branches agree on the 58k-prompt case that prompted this. Two further classification defects. CloudChatError carried no HTTP status, so inferHttpStatusFromAdapterMessage turned an upstream 429 into a 502 and core's failover never rotated or backed off. And a cancelled turn said "Devin turn was aborted.", which isClientClosedMessage does not recognise, so a client hanging up was logged as an upstream failure; it now emits the phrase the classifier knows, with status 499. Usage frames are merged per field instead of replaced, because the counters are cumulative and a later partial frame used to zero an earlier count. --- src/adapters/devin.ts | 65 ++++++++++++++-- src/adapters/devin/cloud-direct/chat.ts | 99 ++++++++++++++++++++++++- tests/providers/devin-hardening.test.ts | 36 +++++++++ 3 files changed, 192 insertions(+), 8 deletions(-) diff --git a/src/adapters/devin.ts b/src/adapters/devin.ts index 12d9ab7ee2..7196718b27 100644 --- a/src/adapters/devin.ts +++ b/src/adapters/devin.ts @@ -13,6 +13,48 @@ 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. + */ +function mergeDevinUsage(previous: OcxUsage, next: OcxUsage): OcxUsage { + const keys = [ + "inputTokens", "outputTokens", "totalTokens", + "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; + } + 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. */ +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 +252,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 +314,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,7 +347,7 @@ 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 } : {}), @@ -313,19 +355,24 @@ export function createDevinAdapter( ...(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 @@ -333,7 +380,13 @@ export function createDevinAdapter( : 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 } : {}), + }); } }, }; diff --git a/src/adapters/devin/cloud-direct/chat.ts b/src/adapters/devin/cloud-direct/chat.ts index 334156daf9..46eb641bb8 100644 --- a/src/adapters/devin/cloud-direct/chat.ts +++ b/src/adapters/devin/cloud-direct/chat.ts @@ -670,6 +670,26 @@ function buildGetChatMessageRequest(args: BuildArgs): Buffer { * 'stop' if no tool_call deltas were emitted). */ function* decodeChatFrame(proto: Buffer): Generator { + // 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, and is yielded last so ordering can + // never invert the precedence. + 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; + } + } 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,10 +760,12 @@ function* decodeChatFrame(proto: Buffer): Generator { // 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; } } + if (authoritativeUsage) yield authoritativeUsage; } /** @@ -834,6 +856,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 +948,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 +1082,7 @@ export async function* streamChatEvents(req: CloudChatRequest): AsyncGenerator { } }); }); + +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(); + }); +}); From 2ddb98fdeab888fb15755344da0544f51c344049 Mon Sep 17 00:00:00 2001 From: JUN Date: Sun, 13 Sep 2026 00:06:10 +0900 Subject: [PATCH 2/2] fix(devin): report field-7 usage ahead of finish and derive the merged total Review follow-ups on the usage decode. The authoritative ModelUsageStats event was yielded after the rest of the frame, so a frame that also carried finish reported usage behind the turn's end. It is now yielded first, which makes the order independent of where the service places the field. mergeDevinUsage took the max of two totals alongside the per-field maxima, which can leave totalTokens different from input + output; the cost and log paths read that total. The total is now derived from the merged counts. Regression coverage for what the change is actually for: field 7 suppressing the display rows within one frame and landing before finish, the display rows still decoding when no field 7 is present, a partial frame not zeroing an earlier count, and an HTTP status becoming a structured classification. A Connect trailer still carries no HTTP status, so a cap delivered that way keeps the older message-inference path. That is noted at the throw site as a follow-up rather than silently left open. --- src/adapters/devin.ts | 11 +++- src/adapters/devin/cloud-direct/chat.ts | 15 +++-- tests/providers/devin-hardening.test.ts | 82 +++++++++++++++++++++++++ 3 files changed, 101 insertions(+), 7 deletions(-) diff --git a/src/adapters/devin.ts b/src/adapters/devin.ts index 7196718b27..5bbd65c480 100644 --- a/src/adapters/devin.ts +++ b/src/adapters/devin.ts @@ -19,9 +19,9 @@ import { DEVIN_DEFAULT_API_SERVER, resolveDevinApiServer } from "../oauth/devin" * 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. */ -function mergeDevinUsage(previous: OcxUsage, next: OcxUsage): OcxUsage { +export function mergeDevinUsage(previous: OcxUsage, next: OcxUsage): OcxUsage { const keys = [ - "inputTokens", "outputTokens", "totalTokens", + "inputTokens", "outputTokens", "cachedInputTokens", "cacheReadInputTokens", "cacheCreationInputTokens", "reasoningOutputTokens", ] as const; @@ -32,6 +32,11 @@ function mergeDevinUsage(previous: OcxUsage, next: OcxUsage): OcxUsage { 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; } @@ -45,7 +50,7 @@ function mergeDevinUsage(previous: OcxUsage, next: OcxUsage): OcxUsage { const DEVIN_CLIENT_CLOSED_MESSAGE = "client closed request"; /** Map a cloud-direct failure onto the structured fields the error event carries. */ -function devinErrorClassification(error: unknown): { status?: number; errorType?: string; retryable?: boolean } { +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 }; diff --git a/src/adapters/devin/cloud-direct/chat.ts b/src/adapters/devin/cloud-direct/chat.ts index 46eb641bb8..2c7b7c32a1 100644 --- a/src/adapters/devin/cloud-direct/chat.ts +++ b/src/adapters/devin/cloud-direct/chat.ts @@ -669,7 +669,7 @@ 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 { +export function* decodeChatFrame(proto: Buffer): Generator { // 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 @@ -681,8 +681,10 @@ function* decodeChatFrame(proto: Buffer): Generator { // // 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, and is yielded last so ordering can - // never invert the precedence. + // 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)) { @@ -690,6 +692,7 @@ function* decodeChatFrame(proto: Buffer): Generator { 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. @@ -765,7 +768,6 @@ function* decodeChatFrame(proto: Buffer): Generator { if (usage) yield usage; } } - if (authoritativeUsage) yield authoritativeUsage; } /** @@ -1082,6 +1084,11 @@ export async function* streamChatEvents(req: CloudChatRequest): AsyncGenerator { 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({}); + }); +});