fix: keep long extended-thinking streams intact

internal/proxy/models.go

@@ -1,6 +1,7 @@
 package proxy
 
 import (
+	"context"
 	"encoding/json"
 	"fmt"
 	"net/http"
@@ -12,6 +13,8 @@ import (
 	"github.com/lich0821/ccNexus/internal/logger"
 )
 
+const modelsRequestTimeout = 300 * time.Second
+
 // ModelInfo represents a single model information
 type ModelInfo struct {
 	ID         string `json:"id"`
@@ -122,8 +125,11 @@ func (p *Proxy) fetchModelsFromEndpoint(ep config.Endpoint) ([]ModelInfo, error)
 	// Set User-Agent
 	req.Header.Set("User-Agent", "ccNexus/1.0")
 
-	// Execute request
-	resp, err := p.httpClient.Do(req)
+	// Execute request. The shared client has no total timeout (it serves long
+	// streams), so keep this short request bounded.
+	ctx, cancel := context.WithTimeout(context.Background(), modelsRequestTimeout)
+	defer cancel()
+	resp, err := p.httpClient.Do(req.WithContext(ctx))
 	if err != nil {
 		return nil, fmt.Errorf("failed to fetch models: %s", redactEndpointMessage(err.Error(), ep))
 	}

internal/proxy/proxy.go

@@ -54,9 +54,10 @@ func New(cfg *config.Config, statsStorage StatsStorage, sqliteStorage *storage.S
 	stats := NewStats(statsStorage, deviceID)
 
 	// Create a reusable HTTP client with connection pool
-	// Enhanced configuration for large SSE streaming and HTTP/2 support
+	// Enhanced configuration for large SSE streaming and HTTP/2 support.
+	// No total Timeout: it would cut off healthy long-running streams. Stalled
+	// upstreams are aborted by upstreamIdleTimeout in sendRequest instead.
 	httpClient := &http.Client{
-		Timeout: 300 * time.Second,
 		Transport: &http.Transport{
 			MaxIdleConns:           100,
 			MaxIdleConnsPerHost:    10,

internal/proxy/proxy_request.go

@@ -55,6 +55,12 @@ const (
 )
 
 func (p *Proxy) handleProxyRequest(w http.ResponseWriter, r *http.Request) {
+	// Proxied responses (e.g. long extended-thinking streams) can legitimately outlive
+	// the server's WriteTimeout, so lift the write deadline for this route only.
+	if err := http.NewResponseController(w).SetWriteDeadline(time.Time{}); err != nil {
+		logger.Debug("Failed to clear write deadline for proxied request: %v", err)
+	}
+
 	reqCtx, err := p.newProxyRequestContext(w, r)
 	if err != nil {
 		return

internal/proxy/request.go

@@ -4,7 +4,9 @@ import (
 	"bytes"
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
+	"io"
 	"net/http"
 	"net/url"
 	"path"
@@ -334,10 +336,24 @@ func overrideModelInPayload(payload []byte, model string) []byte {
 	return updated
 }
 
+// upstreamIdleTimeout aborts an upstream request that makes no progress (no response
+// headers or body data) for this long. Unlike a total timeout, it never cuts off a
+// healthy stream that keeps sending data, however long it runs.
+const upstreamIdleTimeout = 10 * time.Minute
+
+var errUpstreamIdleTimeout = errors.New("upstream idle timeout")
+
 // sendRequest sends the HTTP request and returns the response
 func sendRequest(ctx context.Context, proxyReq *http.Request, httpClient *http.Client, cfg *config.Config) (*http.Response, error) {
+	return sendRequestWithIdleTimeout(ctx, proxyReq, httpClient, cfg, upstreamIdleTimeout)
+}
+
+func sendRequestWithIdleTimeout(ctx context.Context, proxyReq *http.Request, httpClient *http.Client, cfg *config.Config, idleTimeout time.Duration) (*http.Response, error) {
+	ctx, cancel := context.WithCancelCause(ctx)
+	idleTimer := time.AfterFunc(idleTimeout, func() { cancel(errUpstreamIdleTimeout) })
 	proxyReq = proxyReq.WithContext(ctx)
 
+	client := httpClient
 	proxyURL := resolveProxyURLForRequest(cfg, proxyReq.URL)
 	// Apply proxy if configured
 	if strings.TrimSpace(proxyURL) != "" {
@@ -354,10 +370,50 @@ func sendRequest(ctx context.Context, proxyReq *http.Request, httpClient *http.C
 			clientWithProxy.Transport = transport
 		}
 
-		return clientWithProxy.Do(proxyReq)
+		client = clientWithProxy
+	}
+
+	resp, err := client.Do(proxyReq)
+	if err != nil {
+		idleTimer.Stop()
+		if errors.Is(context.Cause(ctx), errUpstreamIdleTimeout) {
+			err = fmt.Errorf("%w: no response within %s: %v", errUpstreamIdleTimeout, idleTimeout, err)
+		}
+		cancel(nil)
+		return nil, err
 	}
 
-	return httpClient.Do(proxyReq)
+	idleTimer.Reset(idleTimeout)
+	resp.Body = &idleTimeoutBody{ReadCloser: resp.Body, ctx: ctx, cancel: cancel, timer: idleTimer, timeout: idleTimeout}
+	return resp, nil
+}
+
+// idleTimeoutBody re-arms the idle timer whenever upstream data arrives, and
+// releases the request context when the body is closed.
+type idleTimeoutBody struct {
+	io.ReadCloser
+	ctx     context.Context
+	cancel  context.CancelCauseFunc
+	timer   *time.Timer
+	timeout time.Duration
+}
+
+func (b *idleTimeoutBody) Read(p []byte) (int, error) {
+	n, err := b.ReadCloser.Read(p)
+	if n > 0 {
+		b.timer.Reset(b.timeout)
+	}
+	if err != nil && err != io.EOF && errors.Is(context.Cause(b.ctx), errUpstreamIdleTimeout) {
+		err = fmt.Errorf("%w: no data received for %s: %v", errUpstreamIdleTimeout, b.timeout, err)
+	}
+	return n, err
+}
+
+func (b *idleTimeoutBody) Close() error {
+	b.timer.Stop()
+	err := b.ReadCloser.Close()
+	b.cancel(nil)
+	return err
 }
 
 func resolveProxyURLForRequest(cfg *config.Config, targetURL *url.URL) string {

internal/proxy/upstream_timeout_test.go

@@ -0,0 +1,151 @@
+package proxy
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/lich0821/ccNexus/internal/config"
+	"github.com/lich0821/ccNexus/internal/storage"
+)
+
+const testIdleTimeout = 200 * time.Millisecond
+
+// streamSSE writes count events spaced by interval, so the whole stream outlives
+// any timeout shorter than count*interval while never going idle for long.
+func streamSSE(w http.ResponseWriter, count int, interval time.Duration) {
+	w.Header().Set("Content-Type", "text/event-stream")
+	flusher := w.(http.Flusher)
+	for i := 0; i < count; i++ {
+		fmt.Fprintf(w, "event: ping\ndata: {\"type\":\"ping\",\"n\":%d}\n\n", i)
+		flusher.Flush()
+		time.Sleep(interval)
+	}
+	fmt.Fprint(w, "event: message_stop\ndata: {\"type\":\"message_stop\"}\n\n")
+	flusher.Flush()
+}
+
+func TestSendRequestKeepsActiveStreamPastIdleTimeout(t *testing.T) {
+	upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		streamSSE(w, 8, testIdleTimeout/4)
+	}))
+	defer upstream.Close()
+
+	req, err := http.NewRequest(http.MethodPost, upstream.URL, nil)
+	if err != nil {
+		t.Fatalf("new request: %v", err)
+	}
+	resp, err := sendRequestWithIdleTimeout(context.Background(), req, &http.Client{}, nil, testIdleTimeout)
+	if err != nil {
+		t.Fatalf("send request: %v", err)
+	}
+	body, err := io.ReadAll(resp.Body)
+	_ = resp.Body.Close()
+	if err != nil {
+		t.Fatalf("active stream was cut off: %v", err)
+	}
+	if !strings.Contains(string(body), "message_stop") {
+		t.Fatalf("stream incomplete: %q", body)
+	}
+}
+
+func TestSendRequestAbortsStalledStream(t *testing.T) {
+	release := make(chan struct{})
+	upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		fmt.Fprint(w, "data: first\n\n")
+		w.(http.Flusher).Flush()
+		select {
+		case <-r.Context().Done():
+		case <-release:
+		}
+	}))
+	defer upstream.Close()
+	defer close(release)
+
+	req, err := http.NewRequest(http.MethodPost, upstream.URL, nil)
+	if err != nil {
+		t.Fatalf("new request: %v", err)
+	}
+	resp, err := sendRequestWithIdleTimeout(context.Background(), req, &http.Client{}, nil, testIdleTimeout)
+	if err != nil {
+		t.Fatalf("send request: %v", err)
+	}
+	defer resp.Body.Close()
+
+	start := time.Now()
+	_, err = io.ReadAll(resp.Body)
+	if !errors.Is(err, errUpstreamIdleTimeout) {
+		t.Fatalf("expected idle timeout error, got %v", err)
+	}
+	if elapsed := time.Since(start); elapsed > 2*time.Second {
+		t.Fatalf("stalled stream took %s to abort", elapsed)
+	}
+}
+
+func TestSendRequestAbortsWhenNoResponseHeaders(t *testing.T) {
+	release := make(chan struct{})
+	upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		select {
+		case <-r.Context().Done():
+		case <-release:
+		}
+	}))
+	defer upstream.Close()
+	defer close(release)
+
+	req, err := http.NewRequest(http.MethodPost, upstream.URL, nil)
+	if err != nil {
+		t.Fatalf("new request: %v", err)
+	}
+	_, err = sendRequestWithIdleTimeout(context.Background(), req, &http.Client{}, nil, testIdleTimeout)
+	if !errors.Is(err, errUpstreamIdleTimeout) {
+		t.Fatalf("expected idle timeout error, got %v", err)
+	}
+}
+
+func TestProxiedStreamOutlivesServerWriteTimeout(t *testing.T) {
+	const writeTimeout = 200 * time.Millisecond
+	upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		streamSSE(w, 8, writeTimeout/4)
+	}))
+	defer upstream.Close()
+
+	db, err := storage.NewSQLiteStorage(filepath.Join(t.TempDir(), "write-timeout.db"))
+	if err != nil {
+		t.Fatalf("create storage: %v", err)
+	}
+	t.Cleanup(func() { _ = db.Close() })
+
+	cfg := config.DefaultConfig()
+	cfg.UpdateEndpoints([]config.Endpoint{{
+		Name: "Claude", APIUrl: upstream.URL, APIKey: "x", AuthMode: config.AuthModeAPIKey,
+		Enabled: true, Transformer: "claude", Model: "claude-test",
+	}})
+	p := New(cfg, storage.NewStatsStorageAdapter(db), db, "test")
+
+	server := httptest.NewUnstartedServer(http.HandlerFunc(p.handleProxy))
+	server.Config.WriteTimeout = writeTimeout
+	server.Start()
+	defer server.Close()
+
+	resp, err := http.Post(server.URL+"/v1/messages", "application/json",
+		strings.NewReader(`{"model":"claude-test","stream":true,"messages":[{"role":"user","content":"hi"}]}`))
+	if err != nil {
+		t.Fatalf("post: %v", err)
+	}
+	body, err := io.ReadAll(resp.Body)
+	_ = resp.Body.Close()
+	if err != nil {
+		t.Fatalf("proxied stream was cut off: %v", err)
+	}
+	if !strings.Contains(string(body), "message_stop") {
+		t.Fatalf("proxied stream incomplete: %q", body)
+	}
+}

internal/transformer/cc/claude.go

@@ -53,6 +53,10 @@ func (t *ClaudeTransformer) TransformResponseWithContext(resp []byte, isStreamin
 	}
 
 	scanner := bufio.NewScanner(bytes.NewReader(resp))
+	// A line can never be longer than the whole event, so size the buffer to fit it.
+	// The default 64KB limit silently dropped large lines such as the signature_delta
+	// of long extended-thinking blocks.
+	scanner.Buffer(make([]byte, 0, len(resp)+1), len(resp)+1)
 	var result bytes.Buffer
 
 	for scanner.Scan() {
@@ -104,6 +108,10 @@ func (t *ClaudeTransformer) TransformResponseWithContext(resp []byte, isStreamin
 		result.WriteString(line)
 		result.WriteString("\n")
 	}
+	if err := scanner.Err(); err != nil {
+		// Never forward a partially scanned event; pass the original through untouched.
+		return resp, nil
+	}
 
 	return result.Bytes(), nil
 }

internal/transformer/cc/claude_test.go

@@ -0,0 +1,23 @@
+package cc
+
+import (
+	"strings"
+	"testing"
+
+	"github.com/lich0821/ccNexus/internal/transformer"
+)
+
+func TestClaudeTransformerForwardsLargeSignatureDelta(t *testing.T) {
+	// Long extended-thinking blocks produce signature_delta lines well over 64KB.
+	signature := strings.Repeat("s", 100*1024)
+	event := "event: content_block_delta\n" +
+		`data: {"type":"content_block_delta","index":0,"delta":{"type":"signature_delta","signature":"` + signature + `"}}` + "\n\n"
+
+	out, err := NewClaudeTransformer().TransformResponseWithContext([]byte(event), true, transformer.NewStreamContext())
+	if err != nil {
+		t.Fatalf("transform: %v", err)
+	}
+	if string(out) != event {
+		t.Fatalf("large signature_delta event was not forwarded unchanged: got %d bytes, want %d", len(out), len(event))
+	}
+}