merge: a filtered or empty completion fails the send (#24)

Author: gluck

internal/provider/anthropic/anthropic.go

@@ -476,14 +476,7 @@ func decodeNativeMessage(nm nativeMessage) message.Message {
 			})
 		}
 	}
-	switch nm.StopReason {
-	case "end_turn", "stop":
-		m.StopReason = message.StopEnd
-	case "max_tokens", "length":
-		m.StopReason = message.StopLength
-	case "tool_use":
-		m.StopReason = message.StopToolInvoke
-	}
+	m.StopReason = provider.AnthropicStopReason(nm.StopReason)
 	if nm.Usage != nil {
 		m.Usage = &message.Usage{
 			InputTokens:      nm.Usage.InputTokens,
@@ -1047,50 +1040,60 @@ func (a *Anthropic) Send(ctx context.Context, in provider.SendInput, bus provide
 		return fmt.Errorf("anthropic: %s", apiErrorMessage(resp.StatusCode, errBody))
 	}
 
-	nm, err := a.drainSSE(ctx, resp.Body, model, bus)
-	if err != nil {
-		// A CANCELLED TURN IS A PREMATURE CLOSE, NOT A BROKEN STREAM. What the
-		// accumulator holds is real: it was produced by this provider from
-		// this wire, so the message and its native payload agree by
-		// construction. Dropping it made figaro synthesise a partial from its
-		// OWN text accumulator -- a message with no native payload, whose
-		// translation had to be re-encoded and lost every signed thinking
-		// block. Anything else that ends a stream (a scanner fault, a
-		// truncated body) is still a broken stream and still drops.
-		if !errors.Is(err, context.Canceled) || len(nm.Content) == 0 {
+	return a.land(a.drainSSE(ctx, resp.Body, model, bus))(providerName, bus)
+}
+
+// land is what happens to a drained stream, and there is ONE of it: Send and
+// SendWithTransport used to carry two copies of this tail, and a rule
+// learned on one path was a rule the other could forget. It returns a
+// function so the drain call reads as one expression at both sites.
+//
+// Three outcomes:
+//
+//   - A cancelled turn is a premature close, not a broken stream. What the
+//     accumulator holds is real: it was produced by this provider from this
+//     wire, so the message and its native payload agree by construction.
+//     Dropping it made figaro synthesise a partial from its OWN text, a
+//     message with no native payload whose re-encoded translation lost every
+//     signed thinking block. Anything else that ends a stream (a scanner
+//     fault, a truncated body) is still a broken stream and still drops.
+//   - A stream that closed cleanly but is not a reply (nothing in it, or a
+//     stop reason that is not a way of finishing: refusal, content_filter,
+//     a word from the future) keeps whatever the model produced, marked
+//     StopError, and FAILS THE SEND with the wire's word. This used to land
+//     as a normal stop, or as nothing at all, and the aria went quiet
+//     (jack-work/figaro#24).
+//   - A reply lands as one.
+func (a *Anthropic) land(nm nativeMessage, err error) func(label string, bus provider.Bus) error {
+	return func(label string, bus provider.Bus) error {
+		if err != nil {
+			if !errors.Is(err, context.Canceled) || len(nm.Content) == 0 {
+				return err
+			}
+			nm.StopReason = string(message.StopAborted)
+			if perr := a.handOver(nm, bus); perr != nil {
+				return perr
+			}
 			return err
 		}
-		nm.StopReason = string(message.StopAborted)
-		if perr := a.handOver(nm, bus); perr != nil {
-			return perr
+		if fault := provider.CheckCompletion(label, decodeNativeMessage(nm), nm.StopReason); fault != nil {
+			if len(nm.Content) == 0 {
+				return fault
+			}
+			if perr := a.handOver(nm, bus); perr != nil {
+				return perr
+			}
+			return fault
 		}
-		return err
-	}
-	if len(nm.Content) == 0 {
-		return nil
-	}
-
-	// Land the assistant message: figLog → push figaro → cache.
-	msg := decodeNativeMessage(nm)
-	if msg.Timestamp == 0 {
-		msg.Timestamp = time.Now().UnixMilli()
-	}
-	// THE PROVIDER DOES NOT OWN THE LOG. It hands the message to the bus and
-	// the fig IR side appends it: only that side has the LT, so "the fig IR
-	// entry exists before anything that names it" is a SHAPE rather than a
-	// rule five call sites had to remember.
-	bus.PushMessageEnd(string(msg.StopReason))
-	native, err := a.assistantCacheNative(nm)
-	if err != nil {
-		return fmt.Errorf("anthropic cache assistant: %w", err)
+		return a.handOver(nm, bus)
 	}
-	bus.PushFigaro(msg, native)
-	return nil
 }
 
 // handOver gives the fig IR side the message and the native payload from ONE
-// accumulator. The normal close and the premature close both go through it,
-// so a partial message cannot be shaped differently from a whole one.
+// accumulator. The normal close, the premature close and the faulted close
+// all go through it, so a partial message cannot be shaped differently from
+// a whole one. THE PROVIDER DOES NOT OWN THE LOG: it hands the message to
+// the bus and the fig IR side appends it, since only that side has the LT.
 func (a *Anthropic) handOver(nm nativeMessage, bus provider.Bus) error {
 	msg := decodeNativeMessage(nm)
 	if msg.Timestamp == 0 {
@@ -1149,44 +1152,7 @@ func (a *Anthropic) SendWithTransport(ctx context.Context, in provider.SendInput
 		return fmt.Errorf("copilot: %s", apiErrorMessage(resp.StatusCode, errBody))
 	}
 
-	nm, err := a.drainSSE(ctx, resp.Body, model, bus)
-	if err != nil {
-		// A CANCELLED TURN IS A PREMATURE CLOSE, NOT A BROKEN STREAM. What the
-		// accumulator holds is real: it was produced by this provider from
-		// this wire, so the message and its native payload agree by
-		// construction. Dropping it made figaro synthesise a partial from its
-		// OWN text accumulator -- a message with no native payload, whose
-		// translation had to be re-encoded and lost every signed thinking
-		// block. Anything else that ends a stream (a scanner fault, a
-		// truncated body) is still a broken stream and still drops.
-		if !errors.Is(err, context.Canceled) || len(nm.Content) == 0 {
-			return err
-		}
-		nm.StopReason = string(message.StopAborted)
-		if perr := a.handOver(nm, bus); perr != nil {
-			return perr
-		}
-		return err
-	}
-	if len(nm.Content) == 0 {
-		return nil
-	}
-
-	msg := decodeNativeMessage(nm)
-	if msg.Timestamp == 0 {
-		msg.Timestamp = time.Now().UnixMilli()
-	}
-	// THE PROVIDER DOES NOT OWN THE LOG. It hands the message to the bus and
-	// the fig IR side appends it: only that side has the LT, so "the fig IR
-	// entry exists before anything that names it" is a SHAPE rather than a
-	// rule five call sites had to remember.
-	bus.PushMessageEnd(string(msg.StopReason))
-	native, err := a.assistantCacheNative(nm)
-	if err != nil {
-		return fmt.Errorf("anthropic cache assistant: %w", err)
-	}
-	bus.PushFigaro(msg, native)
-	return nil
+	return a.land(a.drainSSE(ctx, resp.Body, model, bus))("copilot", bus)
 }
 
 func (a *Anthropic) assistantCacheNative(msg nativeMessage) (provider.AssistantCache, error) {
@@ -1317,10 +1283,12 @@ func (a *Anthropic) drainSSE(ctx context.Context, body io.ReadCloser, model stri
 		}()
 	}
 
+	// message_stop without a message_delta naming a stop reason is NOT a
+	// clean finish, and it is not this function's place to say it was: an
+	// empty stop reason used to become "end_turn" here, so a cut or filtered
+	// stream was byte-identical to a completion downstream. The empty string
+	// travels, and the decoder reads it as the fault it is.
 	finalizeClean := func() (nativeMessage, error) {
-		if stopReason == "" {
-			stopReason = "end_turn"
-		}
 		nm.StopReason = stopReason
 		nm.Usage = &usage
 		return nm, nil

internal/provider/anthropic/completion_fault_test.go

@@ -0,0 +1,154 @@
+package anthropic
+
+import (
+	"context"
+	"encoding/json"
+	"io"
+	"net/http"
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/jack-work/figaro/api/message"
+	"github.com/jack-work/figaro/internal/provider"
+	"github.com/jack-work/figaro/internal/store"
+)
+
+// jack-work/figaro#24. A proxy that filters a completion answers 200 with a
+// healthy-looking stream: no content, or a thinking block and then a
+// stop_reason figaro had no case for. Every one of those landed in the IR as
+// a normal reply and the aria went quiet with nothing in the log to say why.
+// The turn must END IN ERROR, with the wire's own word for it in the text.
+
+type recordingBus struct {
+	msgs  []message.Message
+	stops []string
+}
+
+func (b *recordingBus) PushDelta(message.Content)          {}
+func (b *recordingBus) PushToolInvokeStart(string, string) {}
+func (b *recordingBus) PushToolInvokeDelta(string, string) {}
+func (b *recordingBus) PushToolReady(message.Content)      {}
+func (b *recordingBus) PushMessageEnd(s string)            { b.stops = append(b.stops, s) }
+func (b *recordingBus) PushFigaro(m message.Message, _ ...provider.AssistantCache) {
+	b.msgs = append(b.msgs, m)
+}
+
+type nilResolver struct{}
+
+func (nilResolver) Resolve() (string, error) { return "tok", nil }
+func (nilResolver) Invalidate(string) error  { return nil }
+
+func sseBody(events ...string) string {
+	var b strings.Builder
+	for _, ev := range events {
+		b.WriteString(ev)
+		b.WriteString("\n\n")
+	}
+	return b.String()
+}
+
+const (
+	evStart    = "event: message_start\ndata: " + `{"type":"message_start","message":{"id":"msg_1","role":"assistant","model":"claude-test","content":[],"stop_reason":null,"usage":{"input_tokens":3,"output_tokens":0}}}`
+	evThinking = "event: content_block_start\ndata: " + `{"type":"content_block_start","index":0,"content_block":{"type":"thinking"}}` +
+		"\n\nevent: content_block_delta\ndata: " + `{"type":"content_block_delta","index":0,"delta":{"type":"thinking_delta","thinking":"hm"}}` +
+		"\n\nevent: content_block_delta\ndata: " + `{"type":"content_block_delta","index":0,"delta":{"type":"signature_delta","signature":"sig"}}` +
+		"\n\nevent: content_block_stop\ndata: " + `{"type":"content_block_stop","index":0}`
+	evText = "event: content_block_start\ndata: " + `{"type":"content_block_start","index":1,"content_block":{"type":"text","text":""}}` +
+		"\n\nevent: content_block_delta\ndata: " + `{"type":"content_block_delta","index":1,"delta":{"type":"text_delta","text":"ciao"}}` +
+		"\n\nevent: content_block_stop\ndata: " + `{"type":"content_block_stop","index":1}`
+	evStop = "event: message_stop\ndata: " + `{"type":"message_stop"}`
+)
+
+func evDelta(stop string) string {
+	return "event: message_delta\ndata: " + `{"type":"message_delta","delta":{"stop_reason":"` + stop + `"},"usage":{"output_tokens":1}}`
+}
+
+func sendSSE(t *testing.T, body string) (*recordingBus, error) {
+	t.Helper()
+	be, aria := store.NewTestAria(t, "d", message.Patch{})
+	figLog, err := be.OpenFigIR(aria)
+	require.NoError(t, err)
+	_, err = figLog.Append(store.Entry[message.Message]{Payload: message.Message{
+		Role: message.RoleInput, Content: []message.Content{message.TextContent("say ciao")},
+	}})
+	require.NoError(t, err)
+	rows, err := be.OpenTranslator(aria, "anthropic")
+	require.NoError(t, err)
+
+	a, err := New(provider.Knobs{Model: "claude-test", MaxTokens: 64}, nilResolver{},
+		func(string) (store.Log[[]json.RawMessage], error) { return rows, nil })
+	require.NoError(t, err)
+
+	bus := &recordingBus{}
+	err = a.SendWithTransport(context.Background(), provider.SendInput{
+		AriaID: aria, FigLog: figLog, MaxTokens: 64,
+	}, bus, false, func(context.Context, *provider.RequestBody) (*http.Response, error) {
+		return &http.Response{
+			StatusCode: http.StatusOK,
+			Header:     http.Header{"Content-Type": []string{"text/event-stream"}},
+			Body:       io.NopCloser(strings.NewReader(body)),
+		}, nil
+	})
+	return bus, err
+}
+
+func TestAFilteredCompletionEndsTheSendInError(t *testing.T) {
+	cases := []struct {
+		name string
+		body string
+		// What the wire said, which must appear in the error so the user
+		// learns the proxy's word for what happened.
+		wireWord string
+		// Whether a message still reaches the IR: content the model paid
+		// for is kept, marked as the error it is.
+		kept bool
+	}{
+		{"refusal, thinking only", sseBody(evStart, evThinking, evDelta("refusal"), evStop), "refusal", true},
+		{"content_filter, thinking only", sseBody(evStart, evThinking, evDelta("content_filter"), evStop), "content_filter", true},
+		{"refusal, nothing", sseBody(evStart, evDelta("refusal"), evStop), "refusal", false},
+		{"end_turn, nothing", sseBody(evStart, evDelta("end_turn"), evStop), "end_turn", false},
+		{"message_stop with no stop reason", sseBody(evStart, evText, evStop), "no stop reason", true},
+		{"a stop reason from the future", sseBody(evStart, evText, evDelta("pause_turn"), evStop), "pause_turn", true},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			bus, err := sendSSE(t, tc.body)
+			require.Error(t, err, "the send returned nil: the turn will record a normal stop")
+			assert.Contains(t, err.Error(), tc.wireWord)
+			if !tc.kept {
+				assert.Empty(t, bus.msgs, "an empty completion is not a message")
+				return
+			}
+			require.Len(t, bus.msgs, 1, "the content the model produced is kept")
+			assert.Equal(t, message.StopError, bus.msgs[0].StopReason)
+			assert.Equal(t, []string{string(message.StopError)}, bus.stops)
+		})
+	}
+}
+
+// The other side of the same instrument: a healthy stream still lands as a
+// healthy message, or the guard is a tautology.
+func TestAHealthyCompletionStillLands(t *testing.T) {
+	for _, stop := range []string{"end_turn", "stop_sequence"} {
+		t.Run(stop, func(t *testing.T) {
+			bus, err := sendSSE(t, sseBody(evStart, evThinking, evText, evDelta(stop), evStop))
+			require.NoError(t, err)
+			require.Len(t, bus.msgs, 1)
+			assert.Equal(t, message.StopEnd, bus.msgs[0].StopReason)
+		})
+	}
+}
+
+// An SSE error event already failed the send. It must keep doing so and
+// carry the event's own text, since the stop reason it sets is "error" and
+// the message would otherwise be decoded as if it had finished.
+func TestAnSSEErrorEventStillFailsTheSend(t *testing.T) {
+	bus, err := sendSSE(t, sseBody(evStart, evText,
+		"event: error\ndata: "+`{"type":"error","error":{"type":"overloaded_error","message":"Overloaded"}}`))
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "overloaded_error")
+	assert.Empty(t, bus.msgs)
+}

internal/provider/anthropicsdk/anthropicsdk.go

@@ -280,12 +280,24 @@ func (p *Provider) Send(ctx context.Context, in provider.SendInput, bus provider
 		}
 		return annotateRateLimit(cleanAPIError(err), note)
 	}
-	if len(msg.Content) == 0 {
-		return nil
-	}
 	if msg.Timestamp == 0 {
 		msg.Timestamp = time.Now().UnixMilli()
 	}
+	// A STREAM THAT CLOSED CLEANLY IS NOT YET A REPLY. Nothing in it, or a
+	// stop reason that is not a way of finishing (refusal, content_filter, a
+	// word from the future), keeps what the model produced marked StopError
+	// and fails the send with the wire's word. An empty completion used to
+	// return nil here and the turn recorded a normal stop with no content
+	// (jack-work/figaro#24).
+	if fault := provider.CheckCompletion(providerName, msg, string(acc.StopReason)); fault != nil {
+		if len(msg.Content) == 0 {
+			return fault
+		}
+		if perr := p.handOver(msg, acc, bus); perr != nil {
+			return perr
+		}
+		return fault
+	}
 
 	return p.handOver(msg, acc, bus)
 }