fix(anthropic): retain known usage when streaming is interrupted

Problem: Anthropic error, EOF and cancellation exits could discard usage
already reported by message_start or message_delta. Bounded consumers lost
the counts, while normal recovery reconstructed counts it already knew.

Root cause: the cumulative usage emitter was reachable only after
message_stop; consumers may stop immediately upon receiving ChunkError.

Fix: give each response one cumulative accumulator and delivery attempt.
Flush partial usage before explicit errors with existing Estimated identity,
retain cancellation-aware best-effort delivery on other exits, and compute
the original error classification before accounting changes sink state.
Keep exact committed usage, request construction and retry policy unchanged.

Verification: required builtin/boot plus disjoint rest cover all239 root
packages once,23966 test/subtest passes215 existing skips. Six full owner
race suites2615pass10skip; actualBoot4race events0skip. Exact preceding
decoder compiles and fails14events with4controls passing. Native-step
contract18events/2pkg passes locally;4Win/Linux binaries compile only.
Rootvet/full lint/makecheck/gofmt/diff pass; baseline2663 byte-identical.

Refs #14025

.github/workflows/ci.yml

@@ -421,6 +421,38 @@ jobs:
           retention-days: 7
 
 
+      - name: test (Windows Anthropic interrupted usage)
+        if: env.RUN_STEPS == 'true' && runner.os == 'Windows' && github.event_name == 'pull_request'
+        timeout-minutes: 5
+        shell: bash
+        run: |
+          set -o pipefail
+          go test -count=1 -timeout=3m -json -run '^Test(InterruptedUsage|EffectAnthropicPartialUsageRecovery)' ./internal/model/anthropic ./internal/assembly/boot | tee "$RUNNER_TEMP/anthropic-interrupted-usage.jsonl"
+          jq -se '
+            [.[] | select(.Test == null and .Action == "pass") | .Package] as $packages |
+            [.[] | select(.Test != null and .Action == "pass") | .Test] as $tests |
+            all(.[]; .Action != "skip" and .Action != "fail") and
+            (["reasonix/internal/model/anthropic", "reasonix/internal/assembly/boot"] - $packages | length) == 0 and
+            (["TestInterruptedUsageTerminal", "TestInterruptedUsageTerminal/upstream_error",
+              "TestInterruptedUsageTerminal/payload_error", "TestInterruptedUsageTerminal/premature_eof",
+              "TestInterruptedUsageTerminal/decode_error", "TestInterruptedUsageTerminal/reported_zero",
+              "TestInterruptedUsageTerminal/missing_control", "TestInterruptedUsageTerminal/complete_control",
+              "TestInterruptedUsageBoundedConsumer", "TestInterruptedUsageReadErrorAndCancellation",
+              "TestInterruptedUsageReadErrorAndCancellation/read_error", "TestInterruptedUsageReadErrorAndCancellation/cancelled_buffered_reader",
+              "TestInterruptedUsageAbandonedConsumer", "TestInterruptedUsageIdleTimeout",
+              "TestEffectAnthropicPartialUsageRecovery", "TestEffectAnthropicPartialUsageRecovery/upstream_error",
+              "TestEffectAnthropicPartialUsageRecovery/premature_eof", "TestEffectAnthropicPartialUsageRecovery/complete_control"] - $tests | length) == 0
+          ' "$RUNNER_TEMP/anthropic-interrupted-usage.jsonl"
+
+      - name: upload Windows Anthropic interrupted usage inventory
+        if: always() && env.RUN_STEPS == 'true' && runner.os == 'Windows' && github.event_name == 'pull_request'
+        uses: actions/upload-artifact@cf430e030ddbb5b0abf93d22962f4752f3646cd9 # v7
+        with:
+          name: anthropic-interrupted-usage-windows-${{ github.run_attempt }}
+          path: ${{ runner.temp }}/anthropic-interrupted-usage.jsonl
+          if-no-files-found: ignore
+          retention-days: 7
+
       - name: test (Windows root websearch budget)
         if: env.RUN_STEPS == 'true' && runner.os == 'Windows' && github.event_name == 'pull_request'
         timeout-minutes: 5

docs/ANTHROPIC_INTERRUPTED_USAGE.md

@@ -0,0 +1,89 @@
+---
+owner: @SivanCola
+backup: @esengine
+status: active
+reviewed: 2026-10-26
+---
+
+# Anthropic interrupted usage / Anthropic 中断流用量
+
+## Behavior / 行为
+
+Anthropic `message_start` and `message_delta` can report counters before a
+response ends. Previously, errors and premature EOF bypassed the terminal
+usage emitter. Bounded consumers retained only a request count, while normal
+Agent recovery reconstructed counts instead of retaining the known counters.
+
+Anthropic 可在响应完成前报告输入、输出、缓存和服务端工具计数。此前错误和提前
+EOF 绕过用量出口：有界调用只保留请求数，普通 Agent 恢复则重新估算已有计数。
+
+One response now owns one cumulative accumulator and one delivery attempt.
+Explicit errors deliver known usage before `ChunkError`, because consumers may
+return on that error. Repeated counters still use their maximum, not a sum.
+`message_stop` emits ordinary exact usage once and stops reading immediately.
+
+每个响应现在由同一个累积器和一次交付尝试负责。显式错误先交付已知用量，再交付
+错误；重复累计值取最大值而非相加。`message_stop` 仍只交付一次正常用量并停止读取。
+
+Partial records set the existing `Estimated` flag and `FinishReason=interrupted`.
+The counts retain what the provider reported; missing output is not invented by
+the decoder. The flag describes incomplete accounting, not exact billing.
+Normal Agent recovery may still augment counts under its existing policy.
+
+部分记录使用已有 `Estimated` 和 `interrupted` 标记，表示记账不完整，不能认证完整
+账单。解码器保留供应商已报告的计数，不猜测缺失输出；普通 Agent 恢复仍可按既有
+策略补充估算。金额消费者沿已有不完整报价路径处理，不把它当已知完整金额。
+
+## Source, sink and authority / 来源、出口与权限
+
+| Boundary | Contract / 约定 |
+| --- | --- |
+| Wire input | Existing SSE decoding and nonnegative cumulative maxima; provider counters remain external claims / 沿用解码与非负最大值，供应商计数仍为外部声明 |
+| Usage sink | One record before explicit errors; bounded observer and reporter retain their existing contracts / 错误前交付一次，沿用 observer 和 reporter |
+| Error identity | Classify upstream error before accounting emission so accounting alone cannot disable the before-output retry route / 先分类错误，用量交付不改变原重试资格 |
+| Cancellation | Existing cancellation-aware sink; canceled or abandoned consumers may lose the best-effort record / 沿用可取消出口，消费者离开时不保证必达 |
+| Money and budgets | Partial usage cannot certify a complete quote; positive counts follow existing soft-budget policies / 不认证完整报价，沿用已有软预算策略 |
+
+Known cache counts, native cache-write units and server-tool counts survive.
+Request counts still come from the existing HTTP attempt counter. Recovery sums
+separate body attempts once; repeated deltas within one response are not extra
+requests. Usage delivery does not commit speculative tools or a partial answer.
+
+已知缓存、原生缓存写入计价单位和服务端工具计数均保留。请求数仍取自已有 HTTP
+计数器；恢复只累加各次响应，重复 delta 不算新请求。交付用量不提交推测工具或答案。
+
+No request construction, prefix, tool schema, retry budget, permission, config,
+wire shape or default changes. The change consumes the existing uncertainty
+contract; older persisted complete records are not rewritten. Other providers
+and the distinction between reported usage and a price-table estimate remain.
+
+不改变请求、前缀、工具 schema、重试预算、权限、配置、数据形状或默认值；不回写
+旧持久化账单。其他供应商及供应商用量与价目估算之间的区别保持不变。
+
+## Evidence and limits / 证据与限制
+
+HTTP/SSE regressions cover upstream/payload errors, EOF, malformed JSON, repeated
+partial counters, zero, missing and committed controls. Reader tests cover IO
+cause identity, idle timeout, buffered cancellation and an abandoned consumer.
+Real bounded calls retain the usage at the observer and reporter before failing.
+
+真实 HTTP/SSE 回归覆盖上游和负载错误、EOF、JSON 错误、重复计数、零值、缺失及完成
+对照；读取回归覆盖 IO 原因、空闲超时、可交付的取消和消费者离开。有界调用保留错误
+前的 observer 与 reporter 用量。
+
+Actual Boot and Anthropic HTTP recovery verify three requests, one real file
+read, exact known attempt totals, unchanged replay request JSON and incomplete
+quote/ledger coverage. A normal two-request control remains complete. These are
+protocol fixtures, not a live-provider incident or cache-hit-rate measurement.
+
+真实 Boot 与 Anthropic HTTP 恢复核验三次请求、一次实际文件读取、已知计数、重放
+请求 JSON 不变及不完整报价和账本；两次正常请求对照仍完整。这是协议夹具，不是
+线上事故复现或真实供应商缓存命中率测量。
+
+This does not guarantee accounting after consumer abandonment, process death or
+an unreported provider charge. It adds no durable ledger, hard reservation,
+shared Coordinator budget or physical process completion proof. The complete
+isolation/integration/recovery plan remains open under #14025.
+
+消费者离开、进程退出或供应商未报告的费用仍无必达保证。本单元不提供持久账本、
+硬预留、共享 Coordinator 预算或物理进程完成证明；#14025 完整方案仍未完成。

internal/assembly/boot/effect_anthropic_partial_usage_test.go

@@ -0,0 +1,142 @@
+package boot
+
+import (
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"reasonix/internal/contract/event"
+	"reasonix/internal/contract/pricing"
+	"reasonix/internal/contract/provider"
+	"reasonix/internal/session/control"
+)
+
+func TestEffectAnthropicPartialUsageRecovery(t *testing.T) {
+	for _, mode := range []string{"upstream_error", "premature_eof", "complete_control"} {
+		t.Run(mode, func(t *testing.T) {
+			home, root := statedBootHome(t), t.TempDir()
+			writeFile(t, root, "fixture.txt", "partial usage fixture")
+			partial := mode != "complete_control"
+			var calls, reads atomic.Int32
+			var mu sync.Mutex
+			var requests []string
+			srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				body, err := io.ReadAll(r.Body)
+				if err != nil {
+					t.Error(err)
+					w.WriteHeader(http.StatusBadRequest)
+					return
+				}
+				mu.Lock()
+				requests = append(requests, string(body))
+				mu.Unlock()
+				n := calls.Add(1)
+				w.Header().Set("Content-Type", "text/event-stream")
+				if partial && n == 1 {
+					_, _ = io.WriteString(w, `data: {"type":"message_start","message":{"usage":{"input_tokens":1000,"cache_creation_input_tokens":20,"cache_read_input_tokens":30,"output_tokens":2}}}`+"\n\n")
+					_, _ = io.WriteString(w, `data: {"type":"message_delta","usage":{"output_tokens":7}}`+"\n\n")
+					if mode == "upstream_error" {
+						_, _ = io.WriteString(w, `data: {"type":"error","error":{"type":"overloaded_error","message":"fixture overload"}}`+"\n\n")
+					} else {
+						_, _ = io.WriteString(w, `data: {"type":"content_block_delta","delta":{"type":"text_delta","text":"partial"}}`+"\n\n")
+					}
+					return
+				}
+				_, _ = io.WriteString(w, `data: {"type":"message_start","message":{"usage":{"input_tokens":1000}}}`+"\n\n")
+				toolRound := n == 1 || partial && n == 2
+				reason := "end_turn"
+				if toolRound {
+					reason = "tool_use"
+					_, _ = io.WriteString(w, `data: {"type":"content_block_start","index":0,"content_block":{"type":"tool_use","id":"read-1","name":"read_file"}}`+"\n\n")
+					_, _ = io.WriteString(w, `data: {"type":"content_block_delta","index":0,"delta":{"type":"input_json_delta","partial_json":"{\"path\":\"fixture.txt\"}"}}`+"\n\n")
+					_, _ = io.WriteString(w, `data: {"type":"content_block_stop","index":0}`+"\n\n")
+				} else {
+					_, _ = io.WriteString(w, `data: {"type":"content_block_delta","delta":{"type":"text_delta","text":"Fixture read complete."}}`+"\n\n")
+				}
+				_, _ = fmt.Fprintf(w, "data: {\"type\":\"message_delta\",\"delta\":{\"stop_reason\":%q},\"usage\":{\"output_tokens\":100}}\n\n", reason)
+				_, _ = io.WriteString(w, "data: {\"type\":\"message_stop\"}\n\n")
+			}))
+			t.Cleanup(srv.Close)
+			writeFile(t, root, "reasonix.toml", `default_model = "relay"
+[codegraph]
+enabled = false
+[[providers]]
+name = "relay"
+kind = "anthropic"
+base_url = "`+srv.URL+`"
+model = "fixture-model"
+api_key_env = "ANTHROPIC_PARTIAL_EFFECT_KEY"
+price = { input = 1, output = 2, currency = "USD" }
+`)
+			t.Setenv("ANTHROPIC_PARTIAL_EFFECT_KEY", "test")
+			approveWorkspaceAt(t, home, root)
+			var usages []provider.Usage
+			var quotes []pricing.CostQuote
+			ledger := pricing.NewLedger()
+			sink := event.FuncSink(func(e event.Event) {
+				if e.Kind == event.ToolResult && e.Tool.Name == "read_file" && e.Tool.Err == "" {
+					reads.Add(1)
+				}
+				if e.Kind == event.Usage && e.Usage != nil && e.CostQuote != nil {
+					mu.Lock()
+					defer mu.Unlock()
+					usages = append(usages, *e.Usage)
+					quotes = append(quotes, *e.CostQuote)
+					ledger.Add(*e.CostQuote, pricing.UsageTokens{PromptTokens: e.Usage.PromptTokens, CompletionTokens: e.Usage.CompletionTokens, Estimated: e.Usage.Estimated}, time.Time{})
+				}
+			})
+			ctrl, err := Build(t.Context(), Options{Home: home, WorkspaceRoot: root, Sink: sink})
+			if err != nil {
+				t.Fatal(err)
+			}
+			t.Cleanup(ctrl.Close)
+			ctrl.SetToolApprovalMode(control.ToolApprovalYolo)
+			if err := ctrl.Run(t.Context(), "read fixture.txt"); err != nil {
+				t.Fatal(err)
+			}
+			mu.Lock()
+			defer mu.Unlock()
+			wantCalls, wantPrompt, wantOutput, wantFirstRequests := 2, 1000, 100, 1
+			if partial {
+				wantCalls++
+				wantPrompt += 1050
+				wantOutput += 7
+				wantFirstRequests++
+			}
+			if calls.Load() != int32(wantCalls) || reads.Load() != 1 || len(usages) != 2 || len(quotes) != 2 {
+				t.Fatalf("HTTP=%d reads=%d usages=%+v quotes=%+v", calls.Load(), reads.Load(), usages, quotes)
+			}
+			u := usages[0]
+			if u.PromptTokens != wantPrompt || u.CompletionTokens != wantOutput || u.RequestCount != wantFirstRequests || u.Estimated != partial || usages[1].Estimated || usages[1].RequestCount != 1 {
+				t.Fatalf("attempt usage lost or duplicated: %+v", usages)
+			}
+			if quotes[0].Original.IsZero() || quotes[0].CostComplete == partial || !quotes[1].CostComplete {
+				t.Fatalf("partial counts certified as a bill: %+v", quotes)
+			}
+			wantCoverage := pricing.CoverageComplete
+			if partial {
+				wantCoverage = pricing.CoveragePartial
+			}
+			if q := ledger.Total("USD"); q.Coverage != wantCoverage || q.CostComplete == partial {
+				t.Fatalf("ledger=%+v", q)
+			}
+			if partial {
+				var first, replay any
+				if json.Unmarshal([]byte(requests[0]), &first) != nil || json.Unmarshal([]byte(requests[1]), &replay) != nil {
+					t.Fatal("invalid request JSON")
+				}
+				f, _ := json.Marshal(first)
+				replayBytes, _ := json.Marshal(replay)
+				if string(f) != string(replayBytes) {
+					t.Fatal("sampling recovery changed the frozen request")
+				}
+			}
+		})
+	}
+}

internal/contract/provider/usage.go

@@ -14,8 +14,8 @@ import (
 // CompletionTokens reported by thinking-capable models. FinishReason carries
 // the model's last reported choices[0].finish_reason so the agent can surface
 // abnormal terminations ("length", "content_filter", "repetition_truncation").
-// Estimated marks counts reconstructed locally because the provider's terminal
-// usage record did not arrive; exact provider usage leaves it false.
+// Estimated marks reconstructed counts or partial interrupted-response usage.
+// Neither certifies a complete bill; exact terminal provider usage leaves it false.
 type Usage struct {
 	PromptTokens           int
 	CompletionTokens       int

internal/model/anthropic/anthropic.go

@@ -544,24 +544,19 @@ func (c *client) readStream(ctx context.Context, resp *http.Response, out chan<-
 	tools := map[int]*provider.ToolCall{} // tool_use blocks, keyed by content index
 	argBuckets := map[int]int{}           // last emitted 2KB progress bucket per block
 	server := serverBlock{index: -1}      // the provider-run call a result block answers
-	var inTok, outTok, cacheCreate, cacheRead, serverTools int
+	var usage streamUsage
 	var stopReason string
-	haveUsage := false
-	mergeUsage := func(usage *wireUsage) {
-		if usage == nil {
-			return
+	finishUsage := func(partial bool) bool {
+		return usage.finish(ctx, c.nativeAnthropic, stopReason, partial, send)
+	}
+	// Explicit errors flush usage first because consumers may stop at ChunkError.
+	// Other exits (including canceled/abandoned consumers) make one best-effort
+	// delivery through the same cancellation-aware sink, never an unbounded wait.
+	defer func() { finishUsage(true) }()
+	fail := func(err error) {
+		if finishUsage(true) {
+			send(provider.Chunk{Type: provider.ChunkError, Err: err})
 		}
-		// The native Anthropic stream reports input/cache counters in
-		// message_start and output_tokens in message_delta. Compatible gateways
-		// such as LongCat report all counters in message_delta instead. Counters
-		// are cumulative and non-negative, so retaining the largest value also
-		// tolerates gateways that repeat partial usage in both events.
-		inTok = max(inTok, usage.InputTokens)
-		outTok = max(outTok, usage.OutputTokens)
-		cacheCreate = max(cacheCreate, usage.CacheCreationInputTokens)
-		cacheRead = max(cacheRead, usage.CacheReadInputTokens)
-		serverTools = max(serverTools, usage.ServerToolUse.WebSearchRequests)
-		haveUsage = true
 	}
 
 	scanner := bufio.NewScanner(resp.Body)
@@ -585,14 +580,14 @@ func (c *client) readStream(ctx context.Context, resp *http.Response, out chan<-
 
 		var ev streamEvent
 		if err := json.Unmarshal([]byte(data), &ev); err != nil {
-			send(provider.Chunk{Type: provider.ChunkError, Err: provider.StreamDecodeError(c.name, data, err)})
+			fail(provider.StreamDecodeError(c.name, data, err))
 			return
 		}
 
 		switch ev.Type {
 		case "message_start":
 			if ev.Message != nil && ev.Message.Usage != nil {
-				mergeUsage(ev.Message.Usage)
+				usage.merge(ev.Message.Usage)
 			}
 		case "content_block_start":
 			if ev.ContentBlock != nil {
@@ -663,7 +658,7 @@ func (c *client) readStream(ctx context.Context, resp *http.Response, out chan<-
 			if ev.Delta != nil && ev.Delta.StopReason != "" {
 				stopReason = ev.Delta.StopReason
 			}
-			mergeUsage(ev.Usage)
+			usage.merge(ev.Usage)
 		case "message_stop":
 			// Anthropic's terminal event. Tool blocks may already have closed;
 			// without this, the attempt stays speculative and is not committed.
@@ -671,7 +666,9 @@ func (c *client) readStream(ctx context.Context, resp *http.Response, out chan<-
 			// reclassify a complete response as interrupted.
 			goto finalize
 		case "error":
-			send(provider.Chunk{Type: provider.ChunkError, Err: anthropicStreamError(c.name, ev.Error, sink.emitted)})
+			// Classify before usage delivery: accounting alone must not disable
+			// the existing before-output retry route.
+			fail(anthropicStreamError(c.name, ev.Error, sink.emitted))
 			return
 		}
 	}
@@ -681,42 +678,24 @@ func (c *client) readStream(ctx context.Context, resp *http.Response, out chan<-
 	}
 	if stalled.Load() {
 		err := fmt.Errorf("%s: stream stalled — no data for %s, connection likely dropped; raise idle_timeout_seconds to allow a longer silence", c.name, idleTimeout)
-		send(provider.Chunk{Type: provider.ChunkError, Err: provider.StreamInterrupt(err, provider.StreamInterruptIdleTimeout)})
+		fail(provider.StreamInterrupt(err, provider.StreamInterruptIdleTimeout))
 		return
 	}
 	if err := scanner.Err(); err != nil {
-		send(provider.Chunk{Type: provider.ChunkError, Err: provider.ReadCut(c.name, err)})
+		fail(provider.ReadCut(c.name, err))
 		return
 	}
 	// EOF / clean close before message_stop is an uncommitted attempt. Complete
 	// ChunkToolCall blocks that arrived earlier remain speculative.
-	send(provider.Chunk{Type: provider.ChunkError, Err: provider.StreamInterrupt(
+	fail(provider.StreamInterrupt(
 		fmt.Errorf("%s: stream ended before message_stop: %w", c.name, io.ErrUnexpectedEOF),
 		provider.StreamInterruptPrematureEOF,
-	)})
+	))
 	return
 
 finalize:
-	if haveUsage {
-		cacheWriteBilledTokens := 0.0
-		if cacheCreate > 0 && c.nativeAnthropic {
-			cacheWriteBilledTokens = float64(cacheCreate) * cacheWrite5MinuteInputMultiplier
-		}
-		usage := &provider.Usage{
-			PromptTokens:           inTok + cacheCreate + cacheRead,
-			CompletionTokens:       outTok,
-			TotalTokens:            inTok + cacheCreate + cacheRead + outTok,
-			CacheHitTokens:         cacheRead,
-			CacheMissTokens:        inTok + cacheCreate,
-			CacheWriteTokens:       cacheCreate,
-			CacheWriteBilledTokens: cacheWriteBilledTokens,
-			FinishReason:           mapStopReason(stopReason),
-			ServerToolRequests:     serverTools,
-		}
-		provider.ApplyRequestAttemptCount(ctx, usage)
-		if !send(provider.Chunk{Type: provider.ChunkUsage, Usage: usage}) {
-			return
-		}
+	if !finishUsage(false) {
+		return
 	}
 	send(provider.Chunk{Type: provider.ChunkDone})
 }

internal/model/anthropic/interrupted_usage_test.go

@@ -0,0 +1,241 @@
+package anthropic
+
+import (
+	"context"
+	"errors"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+	"time"
+
+	"reasonix/internal/contract/event"
+	"reasonix/internal/contract/provider"
+	"reasonix/internal/model/boundedllm"
+)
+
+const partialUsageStart = `data: {"type":"message_start","message":{"usage":{"input_tokens":1000,"output_tokens":2,"cache_creation_input_tokens":20,"cache_read_input_tokens":30,"server_tool_use":{"web_search_requests":1}}}}` + "\n\n"
+const partialUsageDelta = `data: {"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"input_tokens":900,"output_tokens":7,"cache_creation_input_tokens":10,"cache_read_input_tokens":25}}` + "\n\n"
+const partialUsageText = `data: {"type":"content_block_delta","delta":{"type":"text_delta","text":"answer"}}` + "\n\n"
+
+func TestInterruptedUsageTerminal(t *testing.T) {
+	for _, tc := range []struct {
+		name, tail                    string
+		text, complete, missing, zero bool
+	}{
+		{name: "upstream_error", tail: overloadedEvent},
+		{name: "payload_error", tail: overloadedEvent, text: true},
+		{name: "premature_eof"},
+		{name: "decode_error", tail: "data: {bad\n\n"},
+		{name: "reported_zero", zero: true},
+		{name: "missing_control", missing: true},
+		{name: "complete_control", complete: true, tail: "data: {\"type\":\"message_stop\"}\n\ndata: {bad\n\n"},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			var wire strings.Builder
+			if !tc.missing {
+				if tc.zero {
+					wire.WriteString("data: {\"type\":\"message_start\",\"message\":{\"usage\":{}}}\n\n")
+				} else {
+					wire.WriteString(partialUsageStart + partialUsageDelta + partialUsageDelta)
+				}
+			}
+			if tc.text {
+				wire.WriteString(partialUsageText)
+			}
+			wire.WriteString(tc.tail)
+			srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+				w.Header().Set("Content-Type", "text/event-stream")
+				_, _ = io.WriteString(w, wire.String())
+			}))
+			t.Cleanup(srv.Close)
+			p, err := New(provider.Config{Name: "fixture", BaseURL: srv.URL, Model: "fixture", APIKey: "test"})
+			if err != nil {
+				t.Fatal(err)
+			}
+			ctx := provider.WithRequestAttemptCounter(t.Context())
+			ch, err := p.Stream(ctx, provider.Request{Messages: []provider.Message{{Role: provider.RoleUser, Content: "hi"}}})
+			if err != nil {
+				t.Fatal(err)
+			}
+			var usage *provider.Usage
+			var terminal error
+			var usages, dones int
+			for chunk := range ch {
+				switch chunk.Type {
+				case provider.ChunkUsage:
+					if terminal != nil {
+						t.Fatal("usage arrived after error")
+					}
+					usage = chunk.Usage
+					usages++
+				case provider.ChunkError:
+					terminal = chunk.Err
+				case provider.ChunkDone:
+					dones++
+				}
+			}
+			if tc.missing {
+				if usage != nil || usages != 0 || terminal == nil {
+					t.Fatalf("missing usage=%+v count=%d err=%v", usage, usages, terminal)
+				}
+				return
+			}
+			if usages != 1 || usage == nil || usage.RequestCount != 1 || usage.Estimated == tc.complete {
+				t.Fatalf("usage=%+v count=%d", usage, usages)
+			}
+			if !tc.zero && (usage.PromptTokens != 1050 || usage.CompletionTokens != 7 || usage.TotalTokens != 1057 || usage.CacheHitTokens != 30 || usage.CacheMissTokens != 1020 || usage.CacheWriteTokens != 20 || usage.ServerToolRequests != 1) {
+				t.Fatalf("cumulative wire counters lost or added twice: %+v", usage)
+			}
+			if tc.zero && usage.TotalTokens != 0 {
+				t.Fatalf("invented tokens: %+v", usage)
+			}
+			if tc.complete {
+				if terminal != nil || dones != 1 || usage.FinishReason != "stop" {
+					t.Fatalf("committed terminal usage=%+v done=%d err=%v", usage, dones, terminal)
+				}
+			} else if terminal == nil || dones != 0 || usage.FinishReason != "interrupted" {
+				t.Fatalf("interruption certified as complete: usage=%+v done=%d err=%v", usage, dones, terminal)
+			}
+			if tc.tail == overloadedEvent {
+				var payload *provider.StreamPayloadError
+				if !errors.As(terminal, &payload) || payload.Type != "overloaded_error" || provider.IsStreamInterrupted(terminal) == tc.text {
+					t.Fatalf("usage changed original retry/payload classification: %v", terminal)
+				}
+			}
+		})
+	}
+}
+
+func TestInterruptedUsageBoundedConsumer(t *testing.T) {
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		w.Header().Set("Content-Type", "text/event-stream")
+		_, _ = io.WriteString(w, partialUsageStart+partialUsageDelta+overloadedEvent)
+	}))
+	t.Cleanup(srv.Close)
+	p, err := New(provider.Config{Name: "fixture", BaseURL: srv.URL, Model: "fixture", APIKey: "test"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	var observed, reported *provider.Usage
+	var observeCalls, reportCalls int
+	ctx := provider.WithAuxiliaryUsageObserver(t.Context(), func(u *provider.Usage) { observed = u; observeCalls++ })
+	_, err = boundedllm.Call(ctx, boundedllm.Config{Provider: p, UsageSource: event.UsageSourceAdvisor, Sink: event.FuncSink(func(e event.Event) { reported = e.Usage; reportCalls++ })}, "policy", "evidence")
+	if !provider.IsStreamInterrupted(err) {
+		t.Fatalf("lost interruption: %v", err)
+	}
+	if observeCalls != 1 || reportCalls != 1 || observed == nil || reported == nil || observed.PromptTokens != 1050 || reported.CompletionTokens != 7 || !reported.Estimated || observed.RequestCount != 1 {
+		t.Fatalf("observer=%+v reporter=%+v calls=%d/%d", observed, reported, observeCalls, reportCalls)
+	}
+}
+
+type interruptedUsageBody struct {
+	io.Reader
+	end error
+}
+
+func (b interruptedUsageBody) Read(p []byte) (int, error) {
+	n, err := b.Reader.Read(p)
+	if err == io.EOF {
+		err = b.end
+	}
+	return n, err
+}
+func (interruptedUsageBody) Close() error { return nil }
+
+func TestInterruptedUsageReadErrorAndCancellation(t *testing.T) {
+	for _, cancelFirst := range []bool{false, true} {
+		t.Run(map[bool]string{false: "read_error", true: "cancelled_buffered_reader"}[cancelFirst], func(t *testing.T) {
+			ctx, cancel := context.WithCancel(t.Context())
+			defer cancel()
+			cause := errors.New("fixture body cut")
+			body := interruptedUsageBody{Reader: strings.NewReader(partialUsageStart), end: cause}
+			if cancelFirst {
+				cancel()
+			}
+			out := make(chan provider.Chunk, 4)
+			(&client{name: "fixture", nativeAnthropic: true}).readStream(ctx, &http.Response{Body: body}, out)
+			var usage *provider.Usage
+			var terminal error
+			for chunk := range out {
+				if chunk.Type == provider.ChunkUsage {
+					if usage != nil {
+						t.Fatal("duplicate usage")
+					}
+					usage = chunk.Usage
+				}
+				if chunk.Type == provider.ChunkError {
+					terminal = chunk.Err
+				}
+			}
+			if usage == nil || !usage.Estimated || usage.PromptTokens != 1050 || usage.CompletionTokens != 2 || usage.CacheWriteBilledTokens != 25 {
+				t.Fatalf("usage=%+v", usage)
+			}
+			if !cancelFirst && !errors.Is(terminal, cause) {
+				t.Fatalf("lost read cause: %v", terminal)
+			}
+		})
+	}
+}
+
+func TestInterruptedUsageAbandonedConsumer(t *testing.T) {
+	ctx, cancel := context.WithCancel(t.Context())
+	defer cancel()
+	out := make(chan provider.Chunk)
+	done := make(chan struct{})
+	go func() {
+		(&client{name: "fixture"}).readStream(ctx, &http.Response{Body: io.NopCloser(strings.NewReader(partialUsageStart + partialUsageText))}, out)
+		close(done)
+	}()
+	// Receiving text proves message_start was parsed. Leave the subsequent
+	// usage send without a consumer, then cancel its authority.
+	if chunk := <-out; chunk.Type != provider.ChunkText {
+		t.Fatalf("first chunk=%+v", chunk)
+	}
+	cancel()
+	select {
+	case <-done:
+	case <-time.After(5 * time.Second):
+		t.Fatal("partial accounting held an abandoned stream open")
+	}
+}
+
+func TestInterruptedUsageIdleTimeout(t *testing.T) {
+	release := make(chan struct{})
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		w.Header().Set("Content-Type", "text/event-stream")
+		_, _ = io.WriteString(w, partialUsageStart)
+		w.(http.Flusher).Flush()
+		<-release
+	}))
+	t.Cleanup(srv.Close)
+	t.Cleanup(func() { close(release) })
+	p, err := New(provider.Config{Name: "fixture", BaseURL: srv.URL, Model: "fixture", APIKey: "test"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	p.(*client).idleTimeout = 20 * time.Millisecond
+	ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second)
+	defer cancel()
+	ch, err := p.Stream(ctx, provider.Request{Messages: []provider.Message{{Role: provider.RoleUser, Content: "hi"}}})
+	if err != nil {
+		t.Fatal(err)
+	}
+	var usage *provider.Usage
+	var terminal error
+	for chunk := range ch {
+		if chunk.Type == provider.ChunkUsage {
+			if usage != nil {
+				t.Fatal("duplicate usage")
+			}
+			usage = chunk.Usage
+		}
+		if chunk.Type == provider.ChunkError {
+			terminal = chunk.Err
+		}
+	}
+	if usage == nil || usage.PromptTokens != 1050 || !usage.Estimated || provider.StreamInterruptReason(terminal) != provider.StreamInterruptIdleTimeout {
+		t.Fatalf("usage=%+v error=%v", usage, terminal)
+	}
+}

internal/model/anthropic/stream_usage.go

@@ -0,0 +1,59 @@
+package anthropic
+
+import (
+	"context"
+
+	"reasonix/internal/contract/provider"
+)
+
+// streamUsage owns cumulative counters for one HTTP response. An interrupted
+// response retains known counts without certifying that its bill is complete.
+type streamUsage struct {
+	input, output, cacheCreate, cacheRead, serverTools int
+	have, attempted                                    bool
+}
+
+func (u *streamUsage) merge(w *wireUsage) {
+	if w == nil {
+		return
+	}
+	// Native input/cache counts arrive in message_start; some gateways send
+	// everything in message_delta. Repeated cumulative counts are not additive.
+	u.input = max(u.input, w.InputTokens)
+	u.output = max(u.output, w.OutputTokens)
+	u.cacheCreate = max(u.cacheCreate, w.CacheCreationInputTokens)
+	u.cacheRead = max(u.cacheRead, w.CacheReadInputTokens)
+	u.serverTools = max(u.serverTools, w.ServerToolUse.WebSearchRequests)
+	u.have = true
+}
+
+func (u *streamUsage) finish(ctx context.Context, native bool, reason string, partial bool, send func(provider.Chunk) bool) bool {
+	if u.attempted {
+		return true
+	}
+	u.attempted = true
+	if !u.have {
+		return true
+	}
+	writeBilled := 0.0
+	if u.cacheCreate > 0 && native {
+		writeBilled = float64(u.cacheCreate) * cacheWrite5MinuteInputMultiplier
+	}
+	usage := &provider.Usage{
+		PromptTokens:           u.input + u.cacheCreate + u.cacheRead,
+		CompletionTokens:       u.output,
+		TotalTokens:            u.input + u.cacheCreate + u.cacheRead + u.output,
+		CacheHitTokens:         u.cacheRead,
+		CacheMissTokens:        u.input + u.cacheCreate,
+		CacheWriteTokens:       u.cacheCreate,
+		CacheWriteBilledTokens: writeBilled,
+		ServerToolRequests:     u.serverTools,
+		FinishReason:           mapStopReason(reason),
+		Estimated:              partial,
+	}
+	if partial {
+		usage.FinishReason = "interrupted"
+	}
+	provider.ApplyRequestAttemptCount(ctx, usage)
+	return send(provider.Chunk{Type: provider.ChunkUsage, Usage: usage})
+}

release-notes/studio/unreleased/14025-anthropic-interrupted-usage.md

@@ -0,0 +1,2 @@
+- 修复：Anthropic 流中断时保留已知用量，并标记记账不完整；重试分类、请求计数和取消边界保持不变。关联 #14025。
+- Fixed: Anthropic interruptions retain known usage as incomplete accounting, preserving retry classification, request counts and cancellation boundaries. Refs #14025.