fix(anthropic): report token usage for streaming responses

[vjsai]

AgentScope-Java Version

2.0.0-SNAPSHOT (reproduced on main; reported against 2.0.0-RC4/RC5)

Description

Fixes #2094.

Background. Anthropic reports token usage in SSE events that carry no content:

• message_start → message.usage.input_tokens
• message_delta → usage.output_tokens

The bug. parseStreamEvents ends with a filter that drops any response with empty content:

.filter(response -> response != null && !response.getContent().isEmpty());

Both usage-bearing events have empty content, so both were discarded and ChatResponse.getUsage() was always null when streaming. On top of that, parseStreamEvent read only the message id from message_start and never its input_tokens, and because it is a stateless static method, the message_delta branch could only ever build a ChatUsage with outputTokens — the input count had nowhere to live. Non-streaming (parseMessage) was unaffected.

The change.

1. Carry the input token count across the stream and attach it to the usage-bearing message_delta response, so the caller sees both counts on one response (inputTokens=5, outputTokens=10 for the issue's example stream). The counter is held per-subscription via Flux.defer, so concurrent streams don't share state.
2. Let responses that report usage through the filter (!content.isEmpty() || usage != null).

message_start itself is still filtered out — it has neither content nor usage of its own — so the existing stream-filtering contract is unchanged. This keeps parseStreamEvent's signature intact.

The issue proposed either attaching usage to the final event or emitting a usage-only response; this does both at once — the message_delta response is usage-only, and it reports the full input+output counts rather than output alone.

How to test

New regression test AnthropicResponseParserTest#testParseStreamEventsReportsUsage feeds a message_start (input_tokens=5) followed by a message_delta (output_tokens=10) and asserts the caller receives a response reporting both.

Fails before the change (the usage-bearing response never reaches the subscriber):

[ERROR] AnthropicResponseParserTest.testParseStreamEventsReportsUsage:338
        expectation "assertNext" failed (expected: onNext(); actual: onComplete())

Passes after. The full module is green — note the two pre-existing stream tests (testParseStreamEventsMessageStart, testParseStreamEventsFiltersEmptyContent) still pass unmodified:

mvn -pl agentscope-extensions/.../agentscope-extensions-model-anthropic verify
[INFO] Tests run: 146, Failures: 0, Errors: 0, Skipped: 1
[INFO] BUILD SUCCESS

Checklist

• Code has been formatted with mvn spotless:apply
• All tests are passing (mvn test)
• Javadoc comments are complete and follow project conventions
• Related documentation has been updated (e.g. links, examples, etc.) — not applicable, no user-facing API change
• Code is ready for review

[CLAassistant]

All committers have signed the CLA.

[codecov]

Codecov Report

❌ Patch coverage is 89.18919% with 4 lines in your changes missing coverage. Please review.

Files with missing lines	Patch %	Lines
...l/anthropic/formatter/AnthropicResponseParser.java	89.18%	0 Missing and 4 partials ⚠️

📢 Thoughts on this report? Let us know!

[AgentScopeJavaBot]

🤖 AI Review

Fixes #2094: Anthropic streaming responses had null usage because parseStreamEvents filter dropped usage-bearing events. Carries input token count via Flux.defer and lets usage responses through filter. Sound approach with regression test. Three recommended improvements: (1) flatMap should be concatMap for ordering guarantees; (2) parseStreamEvent lacks null guards on event fields; (3) withInputTokens doesn't copy metadata/finishReason fields.

(inline comments could not be attached — line numbers fell outside PR hunks. See archived report.)

[oss-maintainer]

⚠️ CLA Not Signed

Please sign the CLA at: https://cla-assistant.io/agentscope-ai/agentscope-java?pullRequest=2112

---

Code Review (pending CLA, draft PR)

Summary: Fixes Anthropic streaming token usage reporting — carries input token count across the stream and attaches it to the usage-bearing message_delta response.

Findings:

• [Info] Per-subscription counter via Flux.defer prevents concurrent stream state sharing. Correct.
• [Info] Lets usage-bearing responses through the filter. Good.
• [Info] Regression test added. Good coverage.

Verdict: Well-designed fix. Ready to merge once CLA is signed and PR is marked ready for review.

---

Automated review by github-manager