feat: Claude API server fallback stream and continuation

cmd/wasm/README.md

@@ -265,6 +265,7 @@ cancels the previous). Safe to call before the call has started running.
 | --- | --- |
 | `delta` | `content?` or `reasoning?` — a streamed chunk. |
 | `tool_call_delta` | `id, name, arguments` — a streamed chunk of a tool call's arguments. |
+| `tool_calls_retracted` | `ids` — tool calls announced by `tool_call_delta` that will never run (refusal or server fallback). |
 | `tool_call` | `id, name, args` — the call is about to run. |
 | `tool_confirmation` | `id, name, args` — the runtime waits for `confirm()`. |
 | `tool_output` | `id, name, output` — incremental output of a running call. |

cmd/wasm/events.go

@@ -58,6 +58,8 @@ func (p *eventProjector) project(ev embeddedchat.Event) []map[string]any {
 			"name":      e.ToolCall.Function.Name,
 			"arguments": e.ToolCall.Function.Arguments,
 		})
+	case *runtime.ToolCallsRetractedEvent:
+		return one(map[string]any{"type": "tool_calls_retracted", "ids": stringsToAny(e.ToolCallIDs)})
 	case *runtime.ToolCallEvent:
 		return one(toolCallEvent("tool_call", e.ToolCall))
 	case *runtime.ToolCallConfirmationEvent:

cmd/wasm/events_test.go

@@ -4,6 +4,7 @@ package main
 
 import (
 	"errors"
+	"syscall/js"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -34,6 +35,9 @@ func TestProjectToolEvents(t *testing.T) {
 	def := tools.Tool{Name: "echo"}
 
 	assert.Equal(t, []map[string]any{{"type": "tool_call_delta", "id": "c1", "name": "echo", "arguments": `{"text":"x"}`}}, p.project(raw(runtime.PartialToolCall(call, def, "root"))))
+	retracted := p.project(raw(runtime.ToolCallsRetracted([]string{"c1"}, "root")))
+	assert.Equal(t, []map[string]any{{"type": "tool_calls_retracted", "ids": []any{"c1"}}}, retracted)
+	assert.NotPanics(t, func() { js.ValueOf(retracted[0]) }, "emit converts the event for JavaScript")
 	assert.Equal(t, []map[string]any{{"type": "tool_call", "id": "c1", "name": "echo", "args": `{"text":"x"}`}}, p.project(raw(runtime.ToolCall(call, def, "root"))))
 	assert.Equal(t, []map[string]any{{"type": "tool_confirmation", "id": "c1", "name": "echo", "args": `{"text":"x"}`}}, p.project(raw(runtime.ToolCallConfirmation(call, def, "root", nil))))
 	assert.Equal(t, []map[string]any{{"type": "tool_output", "id": "c1", "name": "echo", "output": "line"}}, p.project(raw(runtime.ToolCallOutput("c1", def, "line", "root"))))

cmd/wasm/main.go

@@ -54,6 +54,7 @@
 //
 //	{type: "delta", content?: string, reasoning?: string}
 //	{type: "tool_call_delta", id, name, arguments}
+//	{type: "tool_calls_retracted", ids}           // drop these deltas: they will never run
 //	{type: "tool_call", id, name, args}
 //	{type: "tool_confirmation", id, name, args}   // answer with confirm()
 //	{type: "tool_output", id, name, output}

docs/features/api-server/index.md

@@ -212,6 +212,7 @@ Event types include:
 - `agent_choice` — Streamed text content (partial responses)
 - `agent_choice_reasoning` — Streamed reasoning content
 - `partial_tool_call` — Streamed tool-call arguments, keyed by `tool_call.id`. The first one also carries `tool_definition`
+- `tool_calls_retracted` — Tool calls announced by `partial_tool_call` that will never run, listed in `tool_call_ids`: the model refused, or a server fallback handed the turn to another model
 - `tool_call` — Agent requesting tool execution
 - `tool_call_output` — Streamed output of a running tool, keyed by `tool_call_id`
 - `tool_call_confirmation` — Tool call waiting for user approval

pkg/chat/chat.go

@@ -198,6 +198,16 @@ type MessageDelta struct {
 	// pointer — a scalar field silently dropped every blob but the last. See
 	// [MediaDelta].
 	Media []MediaDelta `json:"media,omitempty"`
+	// ServerFallback marks the point where the provider handed the turn to
+	// a fallback model: work streamed before it is superseded.
+	ServerFallback *ServerFallback `json:"server_fallback,omitempty"`
+}
+
+// ServerFallback is a provider-side handoff from one model to the next
+// within a single response.
+type ServerFallback struct {
+	From string `json:"from"`
+	To   string `json:"to"`
 }
 
 // MessageStreamChoice represents a choice in a streaming response

pkg/leantui/events.go

@@ -87,6 +87,10 @@ func (m *model) handleEvent(ctx context.Context, ev any) {
 		identity := m.contentIdentity.Resolve(sessionID, e.Message.Message.MessageID)
 		m.screen.Transcript.ReconcileAssistantContent(identity, chat.VisibleAssistantContent(e.Message.Message.Content))
 		m.contentIdentity.Finish(sessionID)
+	case *runtime.ToolCallsRetractedEvent:
+		for _, id := range e.ToolCallIDs {
+			m.screen.Transcript.RemoveTool(id)
+		}
 	case *runtime.PartialToolCallEvent:
 		m.screen.Transcript.FlushPending()
 		toolDef := tools.Tool{Name: e.ToolCall.Function.Name}

pkg/leantui/retract_test.go

@@ -0,0 +1,24 @@
+package leantui
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/docker/docker-agent/pkg/leantui/ui"
+	"github.com/docker/docker-agent/pkg/runtime"
+	"github.com/docker/docker-agent/pkg/tools"
+)
+
+func TestRetractedToolCallsAreRemoved(t *testing.T) {
+	t.Parallel()
+	m := &model{screen: ui.NewScreen("", "", "")}
+	for _, id := range []string{"declined", "kept"} {
+		m.handleEvent(t.Context(), runtime.PartialToolCall(tools.ToolCall{ID: id, Function: tools.FunctionCall{Name: "shell"}}, tools.Tool{}, "root"))
+	}
+
+	m.handleEvent(t.Context(), runtime.ToolCallsRetracted([]string{"declined"}, "root"))
+
+	assert.Nil(t, m.screen.Transcript.Tool("declined"))
+	assert.NotNil(t, m.screen.Transcript.Tool("kept"))
+}

pkg/model/provider/anthropic/anthropictest/server.go

@@ -70,7 +70,7 @@ func (s *Server) Requests() []Request {
 // TextReply is a plain text answer reporting inputTokens.
 func TextReply(id, text string, inputTokens int64) string {
 	return body(
-		messageStart(id, inputTokens),
+		messageStart(id, "claude-test", inputTokens),
 		event("content_block_start", map[string]any{"index": 0, "content_block": map[string]any{"type": "text", "text": ""}}),
 		event("content_block_delta", map[string]any{"index": 0, "delta": map[string]any{"type": "text_delta", "text": text}}),
 		event("content_block_stop", map[string]any{"index": 0}),
@@ -82,7 +82,7 @@ func TextReply(id, text string, inputTokens int64) string {
 // ThinkingTextReply is signed thinking followed by a text answer.
 func ThinkingTextReply(id, thinking, signature, text string, inputTokens int64) string {
 	return body(
-		messageStart(id, inputTokens),
+		messageStart(id, "claude-test", inputTokens),
 		event("content_block_start", map[string]any{"index": 0, "content_block": map[string]any{"type": "thinking", "thinking": "", "signature": ""}}),
 		event("content_block_delta", map[string]any{"index": 0, "delta": map[string]any{"type": "thinking_delta", "thinking": thinking}}),
 		event("content_block_delta", map[string]any{"index": 0, "delta": map[string]any{"type": "signature_delta", "signature": signature}}),
@@ -103,7 +103,7 @@ func CompactionReply(id, summary, signature string, iterations ...[2]int64) stri
 		usage = append(usage, map[string]any{"type": "compaction", "input_tokens": it[0], "output_tokens": it[1]})
 	}
 	return body(
-		messageStart(id, 0),
+		messageStart(id, "claude-test", 0),
 		event("content_block_start", map[string]any{"index": 0, "content_block": map[string]any{"type": "compaction", "content": nil, "encrypted_content": "", "signature": nil}}),
 		event("content_block_delta", map[string]any{"index": 0, "delta": map[string]any{"type": "compaction_delta", "content": summary, "encrypted_content": "enc-" + id}}),
 		event("content_block_delta", map[string]any{"index": 0, "delta": map[string]any{"type": "signature_delta", "signature": signature}}),
@@ -113,6 +113,101 @@ func CompactionReply(id, summary, signature string, iterations ...[2]int64) stri
 	)
 }
 
+// FallbackBeforeOutputReply is a turn that from declined before any output:
+// message_start already names to, and the fallback marker comes first.
+func FallbackBeforeOutputReply(id, from, to, text string) string {
+	events := append([]sse{messageStart(id, to, 0)}, fallbackBlock(0, from, to)...)
+	events = append(events, textBlock(1, text)...)
+	events = append(events, messageDelta("end_turn", fallbackUsage(from, to)), event("message_stop", nil))
+	return body(events...)
+}
+
+// FallbackMidOutputReply is a turn that from declined mid-output, after its
+// signed thinking, the text before and a complete client tool call toolID;
+// to continues with the text after.
+func FallbackMidOutputReply(id, from, to, before, after, toolID string) string {
+	events := []sse{
+		messageStart(id, from, 0),
+		event("content_block_start", map[string]any{"index": 0, "content_block": map[string]any{"type": "thinking", "thinking": "", "signature": ""}}),
+		event("content_block_delta", map[string]any{"index": 0, "delta": map[string]any{"type": "thinking_delta", "thinking": "declined plan"}}),
+		event("content_block_delta", map[string]any{"index": 0, "delta": map[string]any{"type": "signature_delta", "signature": "sig-declined"}}),
+		event("content_block_stop", map[string]any{"index": 0}),
+	}
+	events = append(events, textBlock(1, before)...)
+	events = append(events,
+		event("content_block_start", map[string]any{"index": 2, "content_block": map[string]any{"type": "tool_use", "id": toolID, "name": "record", "input": map[string]any{}}}),
+		event("content_block_delta", map[string]any{"index": 2, "delta": map[string]any{"type": "input_json_delta", "partial_json": "{}"}}),
+		event("content_block_stop", map[string]any{"index": 2}),
+	)
+	events = append(events, fallbackBlock(3, from, to)...)
+	events = append(events, textBlock(4, after)...)
+	events = append(events, messageDelta("end_turn", fallbackUsage(from, to)), event("message_stop", nil))
+	return body(events...)
+}
+
+// FallbackChainReply is a turn handed down a chain of models: each one
+// writes a short text, then declines to the next, which ends the turn.
+func FallbackChainReply(id string, models ...string) string {
+	events := []sse{messageStart(id, models[0], 0)}
+	var iterations []any
+	index := 0
+	for i, model := range models {
+		events = append(events, textBlock(index, "from "+model)...)
+		index++
+		if i == len(models)-1 {
+			iterations = append(iterations, iteration("fallback_message", model))
+			break
+		}
+		iterations = append(iterations, iteration("message", model))
+		events = append(events, fallbackBlock(index, model, models[i+1])...)
+		index++
+	}
+	events = append(events, messageDelta("end_turn", map[string]any{"input_tokens": 0, "output_tokens": 7, "iterations": iterations}), event("message_stop", nil))
+	return body(events...)
+}
+
+// StickyReply is a turn sticky routing sent straight to the fallback model:
+// no marker, only the model and a fallback_message iteration show it.
+func StickyReply(id, to, text string) string {
+	events := []sse{messageStart(id, to, 0)}
+	events = append(events, textBlock(0, text)...)
+	events = append(events, messageDelta("end_turn", map[string]any{
+		"input_tokens": 0, "output_tokens": 7,
+		"iterations": []any{iteration("fallback_message", to)},
+	}), event("message_stop", nil))
+	return body(events...)
+}
+
+// fallbackBlock is a fallback marker: a block start and stop with no deltas.
+func fallbackBlock(index int, from, to string) []sse {
+	return []sse{
+		event("content_block_start", map[string]any{"index": index, "content_block": map[string]any{
+			"type": "fallback", "from": map[string]any{"model": from}, "to": map[string]any{"model": to},
+			"trigger": map[string]any{"type": "refusal", "category": "cyber"},
+		}}),
+		event("content_block_stop", map[string]any{"index": index}),
+	}
+}
+
+func textBlock(index int, text string) []sse {
+	return []sse{
+		event("content_block_start", map[string]any{"index": index, "content_block": map[string]any{"type": "text", "text": ""}}),
+		event("content_block_delta", map[string]any{"index": index, "delta": map[string]any{"type": "text_delta", "text": text}}),
+		event("content_block_stop", map[string]any{"index": index}),
+	}
+}
+
+func fallbackUsage(from, to string) map[string]any {
+	return map[string]any{
+		"input_tokens": 0, "output_tokens": 7,
+		"iterations": []any{iteration("message", from), iteration("fallback_message", to)},
+	}
+}
+
+func iteration(typ, model string) map[string]any {
+	return map[string]any{"type": typ, "model": model, "input_tokens": 10, "output_tokens": 7, "cache_read_input_tokens": 0, "cache_creation_input_tokens": 0}
+}
+
 type sse struct {
 	name string
 	data map[string]any
@@ -126,9 +221,9 @@ func event(name string, data map[string]any) sse {
 	return sse{name, data}
 }
 
-func messageStart(id string, inputTokens int64) sse {
+func messageStart(id, model string, inputTokens int64) sse {
 	return event("message_start", map[string]any{"message": map[string]any{
-		"id": id, "type": "message", "role": "assistant", "model": "claude-test", "content": []any{},
+		"id": id, "type": "message", "role": "assistant", "model": model, "content": []any{},
 		"usage": map[string]any{"input_tokens": inputTokens, "output_tokens": 0},
 	}})
 }

pkg/model/provider/anthropic/anthropictest/server_test.go

@@ -6,6 +6,8 @@ import (
 	"strings"
 	"testing"
 
+	"github.com/anthropics/anthropic-sdk-go"
+	"github.com/anthropics/anthropic-sdk-go/option"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
@@ -32,3 +34,43 @@ func TestServerRepliesAndRecords(t *testing.T) {
 	assert.Equal(t, []string{"a", "b"}, got[0].Betas)
 	assert.Equal(t, "m", got[0].Body["model"])
 }
+
+// The fallback replies must parse with the SDK, which relabels the message
+// with the model that served it.
+func TestFallbackRepliesAccumulate(t *testing.T) {
+	t.Parallel()
+
+	for name, tc := range map[string]struct {
+		reply   string
+		markers int
+	}{
+		"before output": {reply: anthropictest.FallbackBeforeOutputReply("m1", "claude-a", "claude-b", "hi"), markers: 1},
+		"mid output":    {reply: anthropictest.FallbackMidOutputReply("m2", "claude-a", "claude-b", "before", "after", "t1"), markers: 1},
+		"sticky":        {reply: anthropictest.StickyReply("m3", "claude-b", "hi")},
+	} {
+		t.Run(name, func(t *testing.T) {
+			t.Parallel()
+			srv := anthropictest.NewServer(t, tc.reply)
+			client := anthropic.NewClient(option.WithBaseURL(srv.URL()), option.WithAPIKey("k"))
+			stream := client.Beta.Messages.NewStreaming(t.Context(), anthropic.BetaMessageNewParams{
+				Model: "claude-a", MaxTokens: 16,
+				Messages: []anthropic.BetaMessageParam{anthropic.NewBetaUserMessage(anthropic.NewBetaTextBlock("q"))},
+			})
+			var msg anthropic.BetaMessage
+			for stream.Next() {
+				require.NoError(t, msg.Accumulate(stream.Current()))
+			}
+			require.NoError(t, stream.Err())
+
+			assert.Equal(t, anthropic.Model("claude-b"), msg.Model)
+			markers := 0
+			for _, block := range msg.Content {
+				if block.Type == "fallback" {
+					markers++
+				}
+			}
+			assert.Equal(t, tc.markers, markers)
+			assert.NotEmpty(t, msg.Usage.Iterations)
+		})
+	}
+}

pkg/model/provider/anthropic/beta_adapter.go

@@ -4,11 +4,15 @@ import (
 	"encoding/json"
 	"fmt"
 	"log/slog"
+	"slices"
+	"sync/atomic"
 
 	"github.com/anthropics/anthropic-sdk-go"
 	"github.com/anthropics/anthropic-sdk-go/packages/ssestream"
 
 	"github.com/docker/docker-agent/pkg/chat"
+	"github.com/docker/docker-agent/pkg/model/provider/base"
+	"github.com/docker/docker-agent/pkg/model/provider/contracts"
 	"github.com/docker/docker-agent/pkg/tools"
 )
 
@@ -27,6 +31,12 @@ type betaStreamAdapter struct {
 	message        anthropic.BetaMessage
 	rawLost        bool
 	requestContext json.RawMessage
+	// config is the dispatching client's; served is the model a server
+	// fallback handed the turn to, unset when the dispatched model served it.
+	// The runtime may read it while the reader goroutine still writes it
+	// (cancellation, idle timeout).
+	config base.Config
+	served atomic.Pointer[string]
 }
 
 // newBetaStreamAdapter creates a new Beta stream adapter
@@ -35,6 +45,7 @@ func (c *Client) newBetaStreamAdapter(stream *ssestream.Stream[anthropic.BetaRaw
 		retryableStream: retryableStream[anthropic.BetaRawMessageStreamEventUnion]{stream: stream},
 		trackUsage:      trackUsage,
 		toolIDByBlock:   map[int64]string{},
+		config:          c.Config,
 	}
 }
 
@@ -77,6 +88,11 @@ func (a *betaStreamAdapter) Recv() (chat.MessageStreamResponse, error) {
 				},
 			}
 			response.Choices[0].Delta.ToolCalls = []tools.ToolCall{toolCall}
+		case anthropic.BetaFallbackBlock:
+			// The declined model's tool calls are superseded.
+			a.toolCall = false
+			a.served.Store(&block.To.Model)
+			response.Choices[0].Delta.ServerFallback = &chat.ServerFallback{From: block.From.Model, To: block.To.Model}
 		case anthropic.BetaThinkingBlock:
 			if block.Thinking != "" {
 				response.Choices[0].Delta.ReasoningContent = block.Thinking
@@ -110,6 +126,16 @@ func (a *betaStreamAdapter) Recv() (chat.MessageStreamResponse, error) {
 		}
 	case anthropic.BetaRawMessageDeltaEvent:
 		a.stopReason = eventVariant.Delta.StopReason
+		// Sticky routing sends no marker: only a fallback_message iteration
+		// shows that a fallback model served the turn.
+		if a.served.Load() == nil {
+			for _, it := range slices.Backward(a.message.Usage.Iterations) {
+				if it.Type == "fallback_message" {
+					a.served.Store(&it.Model)
+					break
+				}
+			}
+		}
 		if a.trackUsage {
 			response.Usage = betaUsageFromMessage(a.message.Usage)
 		}
@@ -146,6 +172,21 @@ func (a *betaStreamAdapter) accumulate(event anthropic.BetaRawMessageStreamEvent
 	}
 }
 
+// ServedBy reports the model a server fallback handed the turn to.
+func (a *betaStreamAdapter) ServedBy() (contracts.ServingModel, bool) {
+	served := a.served.Load()
+	if served == nil {
+		return contracts.ServingModel{}, false
+	}
+	cfg := a.config
+	cfg.ModelConfig.Model = *served
+	// The primary's display name and cost overrides don't describe the
+	// fallback model, which is priced from the catalogue.
+	cfg.ModelConfig.DisplayModel = ""
+	cfg.ModelConfig.Cost = nil
+	return contracts.ServingModel{ID: cfg.ID(), Config: cfg}, true
+}
+
 // Close closes the Beta stream
 func (a *betaStreamAdapter) Close() {
 	a.close()