fix(openclaw): Anthropic Messages API SSE streaming, usage tracking, and error format

tenequm · tenequm · commit 1d976b76db86 · 2026-04-01T18:09:31.000+01:00
- MPP SSE streaming emits proper `event: {type}` framing for /v1/messages
  and omits `data: [DONE]` sentinel that breaks the Anthropic SDK parser
- Anthropic error responses use correct format for /v1/messages endpoints
- Fix MPP streaming usage tracking (token counts were always 0/0 because
  raw JSON payloads lacked the `data:` prefix the tracker expected)
- Accumulate Anthropic usage from message_start (input) and message_delta
  (output) events instead of relying on a single final chunk
- Extract Anthropic thinking mode (thinking.budget_tokens) for history
- MPP errors now show EVM wallet address instead of Solana address
- Add route.test.ts with 17 tests covering dual-format SSE tracker
diff --git a/packages/x402-proxy/CHANGELOG.md b/packages/x402-proxy/CHANGELOG.md
@@ -7,6 +7,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [0.10.3] - 2026-04-01
+
+### Fixed
+- Anthropic Messages API (`/v1/messages`) SSE streaming - proxy now emits proper `event: {type}` framing and omits the `data: [DONE]` sentinel that breaks the Anthropic SDK parser
+- Anthropic error responses use correct format (`{type:"error",error:{type,message}}`) instead of OpenAI format for `/v1/messages` endpoints
+- MPP streaming usage tracking - token counts were always 0/0 because raw JSON payloads lacked the `data:` prefix the tracker expected
+- Anthropic usage accumulation from `message_start` (input tokens) and `message_delta` (output tokens) events instead of relying on a single final chunk
+- Anthropic thinking mode (`thinking.budget_tokens`) now extracted from request body for history tracking
+- MPP error messages and payment history now show the EVM wallet address instead of the Solana address
+
 ## [0.10.2] - 2026-04-01
 
 ### Fixed
diff --git a/packages/x402-proxy/package.json b/packages/x402-proxy/package.json
@@ -1,6 +1,6 @@
 {
   "name": "x402-proxy",
-  "version": "0.10.2",
+  "version": "0.10.3",
   "description": "curl for x402 paid APIs. Auto-pays any endpoint on Base, Solana, and Tempo. Also works as an OpenClaw plugin.",
   "type": "module",
   "sideEffects": false,
diff --git a/packages/x402-proxy/src/openclaw/route.test.ts b/packages/x402-proxy/src/openclaw/route.test.ts
@@ -0,0 +1,351 @@
+import type { ServerResponse } from "node:http";
+import { describe, expect, it } from "vitest";
+import { createSseTracker, writeErrorResponse } from "./route.js";
+
+// ── createSseTracker ────────────────────────────────────────────────
+
+describe("createSseTracker", () => {
+  // ── OpenAI format ─────────────────────────────────────────────────
+
+  describe("OpenAI format via pushJson (MPP path)", () => {
+    it("extracts usage from final chunk with usage field", () => {
+      const tracker = createSseTracker();
+      tracker.pushJson('{"choices":[{"delta":{"content":"Hi"}}],"model":"gpt-4"}');
+      tracker.pushJson(
+        '{"choices":[],"model":"gpt-4","usage":{"prompt_tokens":10,"completion_tokens":5}}',
+      );
+      expect(tracker.result).toEqual({
+        model: "gpt-4",
+        inputTokens: 10,
+        outputTokens: 5,
+        reasoningTokens: undefined,
+        cacheRead: undefined,
+        cacheWrite: undefined,
+      });
+    });
+
+    it("extracts reasoning tokens and cache details", () => {
+      const tracker = createSseTracker();
+      tracker.pushJson(
+        JSON.stringify({
+          model: "o1",
+          usage: {
+            prompt_tokens: 100,
+            completion_tokens: 50,
+            prompt_tokens_details: { cached_tokens: 80, cache_creation_input_tokens: 20 },
+            completion_tokens_details: { reasoning_tokens: 30 },
+          },
+        }),
+      );
+      expect(tracker.result).toEqual({
+        model: "o1",
+        inputTokens: 100,
+        outputTokens: 50,
+        reasoningTokens: 30,
+        cacheRead: 80,
+        cacheWrite: 20,
+      });
+    });
+  });
+
+  describe("OpenAI format via push (x402 raw SSE path)", () => {
+    it("extracts usage from SSE-framed data lines", () => {
+      const tracker = createSseTracker();
+      tracker.push('data: {"choices":[{"delta":{"content":"Hi"}}],"model":"gpt-4"}\n\n');
+      tracker.push(
+        'data: {"choices":[],"model":"gpt-4","usage":{"prompt_tokens":10,"completion_tokens":5}}\n\ndata: [DONE]\n\n',
+      );
+      expect(tracker.result).toEqual({
+        model: "gpt-4",
+        inputTokens: 10,
+        outputTokens: 5,
+        reasoningTokens: undefined,
+        cacheRead: undefined,
+        cacheWrite: undefined,
+      });
+    });
+
+    it("handles chunks split across push calls", () => {
+      const tracker = createSseTracker();
+      // Split in the middle of a data line
+      tracker.push('data: {"model":"gpt-4","usa');
+      tracker.push('ge":{"prompt_tokens":7,"completion_tokens":3}}\n\ndata: [DONE]\n\n');
+      expect(tracker.result).toEqual({
+        model: "gpt-4",
+        inputTokens: 7,
+        outputTokens: 3,
+        reasoningTokens: undefined,
+        cacheRead: undefined,
+        cacheWrite: undefined,
+      });
+    });
+  });
+
+  // ── Anthropic format ──────────────────────────────────────────────
+
+  describe("Anthropic streaming via pushJson (MPP path)", () => {
+    it("accumulates usage from message_start and message_delta", () => {
+      const tracker = createSseTracker();
+      tracker.pushJson(
+        JSON.stringify({
+          type: "message_start",
+          message: {
+            model: "claude-opus-4-6",
+            usage: { input_tokens: 100, output_tokens: 0 },
+          },
+        }),
+      );
+      tracker.pushJson(
+        JSON.stringify({
+          type: "content_block_start",
+          index: 0,
+          content_block: { type: "text", text: "" },
+        }),
+      );
+      tracker.pushJson(
+        JSON.stringify({
+          type: "content_block_delta",
+          index: 0,
+          delta: { type: "text_delta", text: "Hello" },
+        }),
+      );
+      tracker.pushJson(
+        JSON.stringify({
+          type: "message_delta",
+          delta: { stop_reason: "end_turn" },
+          usage: { output_tokens: 15 },
+        }),
+      );
+      tracker.pushJson(JSON.stringify({ type: "message_stop" }));
+
+      expect(tracker.result).toEqual({
+        model: "claude-opus-4-6",
+        inputTokens: 100,
+        outputTokens: 15,
+        cacheRead: undefined,
+        cacheWrite: undefined,
+      });
+    });
+
+    it("extracts cache fields from message_start", () => {
+      const tracker = createSseTracker();
+      tracker.pushJson(
+        JSON.stringify({
+          type: "message_start",
+          message: {
+            model: "claude-sonnet-4-6",
+            usage: {
+              input_tokens: 200,
+              output_tokens: 0,
+              cache_creation_input_tokens: 50,
+              cache_read_input_tokens: 150,
+            },
+          },
+        }),
+      );
+      tracker.pushJson(
+        JSON.stringify({
+          type: "message_delta",
+          delta: { stop_reason: "end_turn" },
+          usage: { output_tokens: 20 },
+        }),
+      );
+
+      expect(tracker.result).toEqual({
+        model: "claude-sonnet-4-6",
+        inputTokens: 200,
+        outputTokens: 20,
+        cacheRead: 150,
+        cacheWrite: 50,
+      });
+    });
+  });
+
+  describe("Anthropic streaming via push (x402 raw SSE path)", () => {
+    it("parses event-framed Anthropic SSE", () => {
+      const tracker = createSseTracker();
+      tracker.push(
+        [
+          "event: message_start",
+          'data: {"type":"message_start","message":{"model":"claude-opus-4-6","usage":{"input_tokens":50}}}',
+          "",
+          "event: content_block_delta",
+          'data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"Hi"}}',
+          "",
+          "event: message_delta",
+          'data: {"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":8}}',
+          "",
+          "event: message_stop",
+          'data: {"type":"message_stop"}',
+          "",
+          "",
+        ].join("\n"),
+      );
+
+      expect(tracker.result).toEqual({
+        model: "claude-opus-4-6",
+        inputTokens: 50,
+        outputTokens: 8,
+        cacheRead: undefined,
+        cacheWrite: undefined,
+      });
+    });
+  });
+
+  describe("Anthropic non-streaming via pushJson", () => {
+    it("extracts usage from complete message response", () => {
+      const tracker = createSseTracker();
+      tracker.pushJson(
+        JSON.stringify({
+          type: "message",
+          id: "msg_123",
+          role: "assistant",
+          model: "claude-opus-4-6",
+          content: [{ type: "text", text: "Hello" }],
+          usage: { input_tokens: 42, output_tokens: 7 },
+        }),
+      );
+
+      expect(tracker.result).toEqual({
+        model: "claude-opus-4-6",
+        inputTokens: 42,
+        outputTokens: 7,
+        cacheRead: undefined,
+        cacheWrite: undefined,
+      });
+    });
+
+    it("extracts cache fields from non-streaming response", () => {
+      const tracker = createSseTracker();
+      tracker.pushJson(
+        JSON.stringify({
+          type: "message",
+          model: "claude-sonnet-4-6",
+          usage: {
+            input_tokens: 100,
+            output_tokens: 10,
+            cache_creation_input_tokens: 30,
+            cache_read_input_tokens: 70,
+          },
+        }),
+      );
+
+      expect(tracker.result).toEqual({
+        model: "claude-sonnet-4-6",
+        inputTokens: 100,
+        outputTokens: 10,
+        cacheRead: 70,
+        cacheWrite: 30,
+      });
+    });
+  });
+
+  // ── Edge cases ────────────────────────────────────────────────────
+
+  describe("edge cases", () => {
+    it("returns undefined for empty tracker", () => {
+      const tracker = createSseTracker();
+      expect(tracker.result).toBeUndefined();
+    });
+
+    it("returns undefined for malformed JSON", () => {
+      const tracker = createSseTracker();
+      tracker.pushJson("not json at all");
+      expect(tracker.result).toBeUndefined();
+    });
+
+    it("returns undefined for JSON without usage or type fields", () => {
+      const tracker = createSseTracker();
+      tracker.pushJson('{"foo":"bar"}');
+      expect(tracker.result).toBeUndefined();
+    });
+
+    it("does not misdetect OpenAI chunks as Anthropic", () => {
+      const tracker = createSseTracker();
+      // Full realistic OpenAI streaming sequence
+      tracker.pushJson(
+        '{"id":"chatcmpl-abc","object":"chat.completion.chunk","model":"gpt-4","choices":[{"index":0,"delta":{"role":"assistant"},"finish_reason":null}]}',
+      );
+      tracker.pushJson(
+        '{"id":"chatcmpl-abc","object":"chat.completion.chunk","model":"gpt-4","choices":[{"index":0,"delta":{"content":"Hi"},"finish_reason":null}]}',
+      );
+      tracker.pushJson(
+        '{"id":"chatcmpl-abc","object":"chat.completion.chunk","model":"gpt-4","choices":[],"usage":{"prompt_tokens":12,"completion_tokens":3}}',
+      );
+      // Should use OpenAI path (has reasoningTokens field), not Anthropic
+      expect(tracker.result).toEqual({
+        model: "gpt-4",
+        inputTokens: 12,
+        outputTokens: 3,
+        reasoningTokens: undefined,
+        cacheRead: undefined,
+        cacheWrite: undefined,
+      });
+    });
+
+    it("ignores [DONE] sentinel in SSE stream", () => {
+      const tracker = createSseTracker();
+      tracker.push(
+        'data: {"model":"gpt-4","usage":{"prompt_tokens":5,"completion_tokens":2}}\n\ndata: [DONE]\n\n',
+      );
+      expect(tracker.result?.inputTokens).toBe(5);
+    });
+  });
+});
+
+// ── writeErrorResponse ──────────────────────────────────────────────
+
+describe("writeErrorResponse", () => {
+  function mockRes() {
+    let body = "";
+    let headStatus = 0;
+    let headHeaders: Record<string, string> = {};
+    return {
+      writeHead(status: number, headers: Record<string, string>) {
+        headStatus = status;
+        headHeaders = headers;
+      },
+      end(data: string) {
+        body = data;
+      },
+      get _status() {
+        return headStatus;
+      },
+      get _headers() {
+        return headHeaders;
+      },
+      get _body() {
+        return body;
+      },
+    } as unknown as ServerResponse & {
+      _status: number;
+      _headers: Record<string, string>;
+      _body: string;
+    };
+  }
+
+  it("writes OpenAI error format when isAnthropicFormat is false", () => {
+    const res = mockRes();
+    writeErrorResponse(res, 402, "Payment failed", "payment_error", "payment_failed", false);
+    expect(res._status).toBe(402);
+    expect(JSON.parse(res._body)).toEqual({
+      error: { message: "Payment failed", type: "payment_error", code: "payment_failed" },
+    });
+  });
+
+  it("writes Anthropic error format when isAnthropicFormat is true", () => {
+    const res = mockRes();
+    writeErrorResponse(res, 402, "Payment failed", "payment_error", "payment_failed", true);
+    expect(res._status).toBe(402);
+    expect(JSON.parse(res._body)).toEqual({
+      type: "error",
+      error: { type: "payment_error", message: "Payment failed" },
+    });
+  });
+
+  it("sets Content-Type to application/json", () => {
+    const res = mockRes();
+    writeErrorResponse(res, 500, "err", "t", "c", false);
+    expect(res._headers["Content-Type"]).toBe("application/json");
+  });
+});
diff --git a/packages/x402-proxy/src/openclaw/route.ts b/packages/x402-proxy/src/openclaw/route.ts

Original file line number	Diff line number	Diff line change
`@@ -1,6 +1,6 @@`
`1`	`1`	`{`
`2`	`2`	`"name": "x402-proxy",`
`3`		`- "version": "0.10.2",`
	`3`	`+ "version": "0.10.3",`
`4`	`4`	`"description": "curl for x402 paid APIs. Auto-pays any endpoint on Base, Solana, and Tempo. Also works as an OpenClaw plugin.",`
`5`	`5`	`"type": "module",`
`6`	`6`	`"sideEffects": false,`