diff --git a/cmd/gateway.go b/cmd/gateway.go index e796f190..dda07bf1 100644 --- a/cmd/gateway.go +++ b/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() }() diff --git a/internal/agent/loop_run.go b/internal/agent/loop_run.go index 619f3f80..4ff8603f 100644 --- a/internal/agent/loop_run.go +++ b/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 } diff --git a/internal/providers/anthropic_request.go b/internal/providers/anthropic_request.go index 67aa3692..14cb0d55 100644 --- a/internal/providers/anthropic_request.go +++ b/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 } diff --git a/internal/providers/anthropic_stream.go b/internal/providers/anthropic_stream.go index 633bd29c..96946c72 100644 --- a/internal/providers/anthropic_stream.go +++ 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}) } diff --git a/internal/providers/anthropic_stream_test.go b/internal/providers/anthropic_stream_test.go new file mode 100644 index 00000000..297fb20e --- /dev/null +++ b/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") + } +} diff --git a/internal/providers/types.go b/internal/providers/types.go index 4a3a7168..598b15eb 100644 --- a/internal/providers/types.go +++ b/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.