fix(providers): prevent crash on cancel, capture thinking signature, nil guard (#287, #188, #566, #335)

- anthropic_stream.go: check ctx.Err() in scanner loop, bounds check toolCallJSON index,
  accumulate signature_delta events into ThinkingSignature
- anthropic_request.go: include signature field in buildRawBlock for thinking blocks
- types.go: add ThinkingSignature field to ChatResponse
- loop_run.go: nil guard on result before FinishTrace access
- gateway.go: defensive ApplyDBSecrets before setupTTS in config reload subscriber

cmd/gateway.go

@@ -10,6 +10,7 @@ import (
 	"path/filepath"
 	"strings"
 	"syscall"
+	"time"
 
 	"github.com/google/uuid"
 
@@ -1049,6 +1050,11 @@ func runGateway() {
 		if !ok {
 			return
 		}
+		if pgStores.ConfigSecrets != nil {
+			if secrets, err := pgStores.ConfigSecrets.GetAll(context.Background()); err == nil && len(secrets) > 0 {
+				updatedCfg.ApplyDBSecrets(secrets)
+			}
+		}
 		newMgr := setupTTS(updatedCfg)
 		if newMgr == nil {
 			return
@@ -1057,6 +1063,21 @@ func runGateway() {
 		slog.Info("tts config reloaded", "provider", newMgr.PrimaryProvider(), "auto", string(newMgr.AutoMode()))
 	})
 
+	// Log orphaned providers on agent deletion. Auto-delete is unsafe because
+	// providers can be referenced by heartbeats (FK), OAuth tokens, media chains.
+	// Users should clean up orphaned providers manually via UI/API.
+	msgBus.Subscribe("agent-deleted-provider-log", func(evt bus.Event) {
+		if evt.Name != bus.TopicAgentDeleted {
+			return
+		}
+		payload, ok := evt.Payload.(bus.AgentDeletedPayload)
+		if !ok || payload.Provider == "" {
+			return
+		}
+		slog.Info("agent deleted, provider may be orphaned — verify via UI",
+			"agent", payload.AgentKey, "provider", payload.Provider)
+	})
+
 	// Contact collector: auto-collect user info from channels with in-memory dedup cache.
 	var contactCollector *store.ContactCollector
 	if pgStores.Contacts != nil {
@@ -1103,6 +1124,12 @@ func runGateway() {
 			sandboxMgr.ReleaseAll(context.Background())
 		}
 
+		if sched != nil {
+			slog.Info("gateway: draining active runs", "timeout", "5s")
+			sched.Stop() // MarkDraining + StopAll
+			time.Sleep(5 * time.Second)
+		}
+
 		cancel()
 	}()
 

internal/agent/loop_run.go

@@ -207,7 +207,11 @@ func (l *Loop) Run(ctx context.Context, req RunRequest) (*RunResult, error) {
 		Payload: completedPayload,
 	})
 	if !isChildTrace && l.traceCollector != nil && traceID != uuid.Nil {
-		l.traceCollector.FinishTrace(ctx, traceID, store.TraceStatusCompleted, "", truncateStr(result.Content, l.traceCollector.PreviewMaxLen()))
+		if result != nil {
+			l.traceCollector.FinishTrace(ctx, traceID, store.TraceStatusCompleted, "", truncateStr(result.Content, l.traceCollector.PreviewMaxLen()))
+		} else {
+			l.traceCollector.FinishTrace(ctx, traceID, store.TraceStatusCompleted, "", "")
+		}
 	}
 	return result, nil
 }

internal/providers/anthropic_request.go

@@ -11,6 +11,9 @@ func (p *AnthropicProvider) buildRawBlock(blockType string, result *ChatResponse
 			"type":     "thinking",
 			"thinking": result.Thinking,
 		}
+		if result.ThinkingSignature != "" {
+			block["signature"] = result.ThinkingSignature
+		}
 		if b, err := json.Marshal(block); err == nil {
 			return b
 		}

internal/providers/anthropic_stream.go

@@ -32,12 +32,16 @@ func (p *AnthropicProvider) ChatStream(ctx context.Context, req ChatRequest, onC
 	var currentBlockType string
 	// Track thinking token count by accumulated chunk size
 	thinkingChars := 0
+	var thinkingSignature string
 
 	scanner := bufio.NewScanner(respBody)
 	scanner.Buffer(make([]byte, 0, SSEScanBufInit), SSEScanBufMax)
 	var currentEvent string
 
 	for scanner.Scan() {
+		if ctx.Err() != nil {
+			return nil, ctx.Err()
+		}
 		line := scanner.Text()
 
 		// Track event type
@@ -102,7 +106,7 @@ func (p *AnthropicProvider) ChatStream(ctx context.Context, req ChatRequest, onC
 						toolCallJSON[idx] += ev.Delta.PartialJSON
 					}
 				case "signature_delta":
-					// Signature is captured in content_block_stop via raw block reconstruction
+					thinkingSignature += ev.Delta.Signature
 				}
 			}
 
@@ -155,7 +159,7 @@ func (p *AnthropicProvider) ChatStream(ctx context.Context, req ChatRequest, onC
 
 	// Parse accumulated tool call JSON arguments
 	for i, rawJSON := range toolCallJSON {
-		if rawJSON != "" {
+		if rawJSON != "" && i < len(result.ToolCalls) {
 			args := make(map[string]any)
 			_ = json.Unmarshal([]byte(rawJSON), &args)
 			result.ToolCalls[i].Arguments = args
@@ -177,6 +181,8 @@ func (p *AnthropicProvider) ChatStream(ctx context.Context, req ChatRequest, onC
 		}
 	}
 
+	result.ThinkingSignature = thinkingSignature
+
 	if onChunk != nil {
 		onChunk(StreamChunk{Done: true})
 	}

internal/providers/anthropic_stream_test.go

@@ -0,0 +1,176 @@
+package providers
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+)
+
+// newAnthropicSSEServer creates a mock SSE server that sends the provided events then closes.
+func newAnthropicSSEServer(t *testing.T, events []string) *httptest.Server {
+	t.Helper()
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "text/event-stream")
+		w.WriteHeader(http.StatusOK)
+		flusher, ok := w.(http.Flusher)
+		if !ok {
+			t.Error("ResponseWriter does not implement http.Flusher")
+			return
+		}
+		for _, ev := range events {
+			fmt.Fprint(w, ev)
+			flusher.Flush()
+		}
+	}))
+	t.Cleanup(server.Close)
+	return server
+}
+
+// newTestAnthropicProvider creates an AnthropicProvider pointing at the given base URL.
+func newTestAnthropicProvider(baseURL string) *AnthropicProvider {
+	p := NewAnthropicProvider("test-key", WithAnthropicBaseURL(baseURL))
+	p.retryConfig.Attempts = 1
+	return p
+}
+
+// TestStreamChat_CancelledContext verifies that cancelling the context mid-stream
+// causes ChatStream to return ctx.Err() rather than continuing to process events.
+func TestStreamChat_CancelledContext(t *testing.T) {
+	// Server sends one chunk then blocks — we cancel before the stream finishes.
+	blocker := make(chan struct{})
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "text/event-stream")
+		w.WriteHeader(http.StatusOK)
+		flusher, ok := w.(http.Flusher)
+		if !ok {
+			t.Error("ResponseWriter does not implement http.Flusher")
+			return
+		}
+		// Send one valid event so the scanner loop starts.
+		fmt.Fprint(w, "event: message_start\ndata: {\"message\":{\"usage\":{\"input_tokens\":5}}}\n\n")
+		flusher.Flush()
+		// Block until the test is done (simulates a slow stream).
+		<-blocker
+	}))
+	defer server.Close()
+	defer close(blocker)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	p := newTestAnthropicProvider(server.URL)
+
+	// Cancel immediately after starting.
+	cancel()
+
+	req := ChatRequest{
+		Model:    "claude-sonnet-4-5-20250929",
+		Messages: []Message{{Role: "user", Content: "hello"}},
+	}
+	_, err := p.ChatStream(ctx, req, nil)
+	if err == nil {
+		t.Fatal("expected error from cancelled context, got nil")
+	}
+	if err != context.Canceled {
+		// Accept either context.Canceled or context.DeadlineExceeded; the HTTP
+		// layer may wrap the error, so we check ctx.Err() as fallback.
+		if ctx.Err() == nil {
+			t.Errorf("expected ctx.Err() to be set, got err=%v", err)
+		}
+	}
+}
+
+// TestStreamChat_ToolCallIndexBounds verifies that receiving input_json_delta
+// without a prior content_block_start:tool_use does not panic (bounds check).
+func TestStreamChat_ToolCallIndexBounds(t *testing.T) {
+	events := []string{
+		// content_block_delta with input_json_delta but no tool_use block started
+		"event: content_block_delta\n",
+		`data: {"index":0,"delta":{"type":"input_json_delta","partial_json":"{\"k\":1}"}}` + "\n\n",
+		// Proper message_stop so the stream ends cleanly
+		"event: message_stop\n",
+		"data: {}\n\n",
+	}
+	server := newAnthropicSSEServer(t, events)
+	p := newTestAnthropicProvider(server.URL)
+
+	req := ChatRequest{
+		Model:    "claude-sonnet-4-5-20250929",
+		Messages: []Message{{Role: "user", Content: "hello"}},
+	}
+	// Should not panic; the fix guards with i < len(result.ToolCalls).
+	result, err := p.ChatStream(context.Background(), req, nil)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if result == nil {
+		t.Fatal("expected non-nil result")
+	}
+	if len(result.ToolCalls) != 0 {
+		t.Errorf("expected 0 tool calls, got %d", len(result.ToolCalls))
+	}
+}
+
+// TestStreamChat_ThinkingSignature verifies that signature_delta events are
+// accumulated and stored in result.ThinkingSignature.
+func TestStreamChat_ThinkingSignature(t *testing.T) {
+	events := []string{
+		"event: message_start\n",
+		`data: {"message":{"usage":{"input_tokens":10}}}` + "\n\n",
+
+		"event: content_block_start\n",
+		`data: {"index":0,"content_block":{"type":"thinking","thinking":""}}` + "\n\n",
+
+		"event: content_block_delta\n",
+		`data: {"index":0,"delta":{"type":"thinking_delta","thinking":"let me think"}}` + "\n\n",
+
+		"event: content_block_delta\n",
+		`data: {"index":0,"delta":{"type":"signature_delta","signature":"sig-part-1"}}` + "\n\n",
+
+		"event: content_block_delta\n",
+		`data: {"index":0,"delta":{"type":"signature_delta","signature":"-part-2"}}` + "\n\n",
+
+		"event: content_block_stop\n",
+		"data: {}\n\n",
+
+		"event: content_block_start\n",
+		`data: {"index":1,"content_block":{"type":"text","text":""}}` + "\n\n",
+
+		"event: content_block_delta\n",
+		`data: {"index":1,"delta":{"type":"text_delta","text":"answer"}}` + "\n\n",
+
+		"event: content_block_stop\n",
+		"data: {}\n\n",
+
+		"event: message_delta\n",
+		`data: {"delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":20}}` + "\n\n",
+
+		"event: message_stop\n",
+		"data: {}\n\n",
+	}
+	server := newAnthropicSSEServer(t, events)
+	p := newTestAnthropicProvider(server.URL)
+
+	req := ChatRequest{
+		Model:    "claude-sonnet-4-5-20250929",
+		Messages: []Message{{Role: "user", Content: "hello"}},
+	}
+	result, err := p.ChatStream(context.Background(), req, nil)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if result == nil {
+		t.Fatal("expected non-nil result")
+	}
+
+	wantSignature := "sig-part-1-part-2"
+	if result.ThinkingSignature != wantSignature {
+		t.Errorf("ThinkingSignature = %q, want %q", result.ThinkingSignature, wantSignature)
+	}
+	if result.Thinking != "let me think" {
+		t.Errorf("Thinking = %q, want %q", result.Thinking, "let me think")
+	}
+	if result.Content != "answer" {
+		t.Errorf("Content = %q, want %q", result.Content, "answer")
+	}
+}

internal/providers/types.go

@@ -84,6 +84,10 @@ type ChatResponse struct {
 	// RawAssistantContent preserves the raw content blocks array from the provider response.
 	// Used by Anthropic to pass thinking blocks back in tool use loops (required by API).
 	RawAssistantContent json.RawMessage `json:"-"`
+
+	// ThinkingSignature is the accumulated signature from streaming thinking blocks.
+	// Required by Anthropic API for tool use passback when thinking is enabled.
+	ThinkingSignature string `json:"-"`
 }
 
 // StreamChunk is a piece of a streaming response.