diff --git a/src/adapters/devin.ts b/src/adapters/devin.ts index 12d9ab7ee2..5bbd65c480 100644 --- a/src/adapters/devin.ts +++ b/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,7 +352,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 +360,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 +385,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..2c7b7c32a1 100644 --- a/src/adapters/devin/cloud-direct/chat.ts +++ b/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 { +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 + // 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 { // 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 { } }); }); + +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({}); + }); +});