diff --git a/mateclaw-server/src/main/java/vip/mate/agent/AgentGraphBuilder.java b/mateclaw-server/src/main/java/vip/mate/agent/AgentGraphBuilder.java index 4f62dffa..903842bc 100644 --- a/mateclaw-server/src/main/java/vip/mate/agent/AgentGraphBuilder.java +++ b/mateclaw-server/src/main/java/vip/mate/agent/AgentGraphBuilder.java @@ -660,6 +660,15 @@ public class AgentGraphBuilder { contextWindowResolver.noteContextLimitError( primaryModelConfig.getProvider(), primaryModelConfig.getModelName(), errorMessage)); + // Issue #585: drive the streaming inter-frame idle timeout from + // the per-model read-timeout knob so a stalled provider can't + // hang the body Flux after the response headers arrive. Only + // override when the model explicitly sets a value — otherwise + // the helper keeps its 180s default. + Integer perModelTimeout = primaryModelConfig.getRequestTimeoutSeconds(); + if (perModelTimeout != null) { + streamingHelper.setStreamIdleTimeoutSec(perModelTimeout); + } } ToolExecutionExecutor executor = new ToolExecutionExecutor( toolSet, toolGuardService, approvalService, streamTracker, @@ -970,6 +979,15 @@ public class AgentGraphBuilder { contextWindowResolver.noteContextLimitError( primaryModelConfig.getProvider(), primaryModelConfig.getModelName(), errorMessage)); + // Issue #585: drive the streaming inter-frame idle timeout from + // the per-model read-timeout knob so a stalled provider can't + // hang the body Flux after the response headers arrive. Only + // override when the model explicitly sets a value — otherwise + // the helper keeps its 180s default. + Integer perModelTimeout = primaryModelConfig.getRequestTimeoutSeconds(); + if (perModelTimeout != null) { + streamingHelper.setStreamIdleTimeoutSec(perModelTimeout); + } } ToolExecutionExecutor executor = new ToolExecutionExecutor( toolSet, toolGuardService, approvalService, streamTracker, diff --git a/mateclaw-server/src/main/java/vip/mate/agent/graph/NodeStreamingChatHelper.java b/mateclaw-server/src/main/java/vip/mate/agent/graph/NodeStreamingChatHelper.java index 1e69bdea..52ae444d 100644 --- a/mateclaw-server/src/main/java/vip/mate/agent/graph/NodeStreamingChatHelper.java +++ b/mateclaw-server/src/main/java/vip/mate/agent/graph/NodeStreamingChatHelper.java @@ -17,7 +17,9 @@ import vip.mate.llm.chatmodel.ReasoningContentCache; import vip.mate.llm.chatmodel.ThinkingLevelHolder; import reactor.core.Disposable; +import reactor.core.publisher.Flux; +import java.time.Duration; import java.time.Instant; import java.time.ZonedDateTime; import java.time.format.DateTimeFormatter; @@ -29,6 +31,7 @@ import java.util.concurrent.CancellationException; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ThreadLocalRandom; import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; @@ -92,6 +95,26 @@ public class NodeStreamingChatHelper { */ private final vip.mate.llm.failover.AvailableProviderPool providerPool; + /** + * Inter-frame idle timeout (seconds) applied to every streaming LLM call. + * The JDK HttpClient request timeout (which {@code setReadTimeout} maps to) + * only protects up to the response headers; once they arrive the clock + * stops, so a provider that returns 200 + a first SSE frame then goes + * silent hangs the body Flux forever — no exception, so health tracking / + * failover never engage (issue #585). A reactor {@code .timeout()} on the + * delta Flux fills that gap: total silence for this long propagates a + * {@code TimeoutException} down the existing error path (classifyError + * buckets it as a retryable SERVER_ERROR). + *

+ * Defaults to {@link vip.mate.llm.chatmodel.HttpTimeouts#DEFAULT_STREAM_IDLE_TIMEOUT} + * (180s). {@code 0} or negative disables it (for tests / opt-out). + * Production wiring sets it from {@code ModelConfigEntity.requestTimeoutSeconds} + * so a single per-model knob governs both the connect-level read timeout + * and the body-level idle timeout. + */ + private long streamIdleTimeoutSec = + vip.mate.llm.chatmodel.HttpTimeouts.DEFAULT_STREAM_IDLE_TIMEOUT.toSeconds(); + public NodeStreamingChatHelper(ChatStreamTracker streamTracker) { this(streamTracker, List.of(), null, null, null, null); } @@ -445,6 +468,17 @@ public class NodeStreamingChatHelper { this.maxTotalDurationMs = maxTotalDurationMs; } + /** + * Override the streaming inter-frame idle timeout (seconds). Wired from + * {@code ModelConfigEntity.requestTimeoutSeconds} by AgentGraphBuilder so a + * single per-model knob governs both the connect-level read timeout and + * the body-level idle timeout. {@code 0} or negative disables the idle + * timeout (used by tests / opt-out). See {@link #streamIdleTimeoutSec}. + */ + public void setStreamIdleTimeoutSec(long seconds) { + this.streamIdleTimeoutSec = seconds; + } + private static final ObjectMapper TOOL_ARG_JSON_MAPPER = new ObjectMapper(); /** @@ -1220,7 +1254,26 @@ public class NodeStreamingChatHelper { CountDownLatch latch = new CountDownLatch(1); - Disposable subscription = chatModel.stream(prompt) + // Issue #585: inter-frame idle timeout on the streaming body Flux. + // The JDK HttpClient request timeout (which setReadTimeout maps to) + // only protects up to the response headers; once they arrive the + // clock stops, so a provider that returns 200 + a first frame then + // goes silent hangs the body forever. This reactor timeout measures + // the gap between successive stream elements, so total silence for + // streamIdleTimeoutSec propagates an error down the existing path. + // The fallback Flux carries a descriptive message so classifyError's + // "timeout" pattern matches it (vanilla TimeoutException.getMessage() + // is null) and the health tracker / failover chain engage. + Flux streamWithIdleGuard = + streamIdleTimeoutSec > 0 + ? chatModel.stream(prompt).timeout( + Duration.ofSeconds(streamIdleTimeoutSec), + Flux.error(new TimeoutException( + "LLM stream idle timeout after " + streamIdleTimeoutSec + + "s with no delta — provider half-open or stalled"))) + : chatModel.stream(prompt); + + Disposable subscription = streamWithIdleGuard .doOnNext(chatResponse -> { if (chatResponse == null || chatResponse.getResults() == null || chatResponse.getResults().isEmpty()) { return; diff --git a/mateclaw-server/src/main/java/vip/mate/llm/chatmodel/AnthropicChatModelBuilder.java b/mateclaw-server/src/main/java/vip/mate/llm/chatmodel/AnthropicChatModelBuilder.java index 3abe7ca0..ef07ac48 100644 --- a/mateclaw-server/src/main/java/vip/mate/llm/chatmodel/AnthropicChatModelBuilder.java +++ b/mateclaw-server/src/main/java/vip/mate/llm/chatmodel/AnthropicChatModelBuilder.java @@ -256,10 +256,17 @@ public class AnthropicChatModelBuilder implements ChatModelBuilder { /** * Streaming counterpart of {@link #applyHttpTimeouts(RestClient.Builder)}. - * Without this, Spring AI's AnthropicApi would back its streaming chat - * call by a default WebClient with neither connect nor read timeout — a - * stalled provider could hang the agent thread indefinitely while the - * failover chain idles (no exception = no signal). + * + *

Scope caveat (issue #585): {@code setReadTimeout} maps to the + * JDK HttpClient's request timeout, which only protects up to the + * response headers. Once the headers arrive the clock stops, so a + * provider that returns 200 + a first SSE frame and then goes silent is + * not caught here — the body Flux would hang indefinitely. The + * body-level gap is closed by a reactor inter-frame idle timeout applied + * at the streaming chokepoint ({@code NodeStreamingChatHelper}, driven by + * {@link vip.mate.llm.chatmodel.HttpTimeouts#DEFAULT_STREAM_IDLE_TIMEOUT}). + * This WebClient timeout still catches the "provider never sends headers" + * case (connection accepted, no response), so both layers are kept. *

* Uses the same JDK HttpClient + JdkClientHttpConnector path, so the * dependency surface doesn't pull in reactor-netty (excluded by this diff --git a/mateclaw-server/src/main/java/vip/mate/llm/chatmodel/HttpTimeouts.java b/mateclaw-server/src/main/java/vip/mate/llm/chatmodel/HttpTimeouts.java index 7334ea9f..b23ccfd2 100644 --- a/mateclaw-server/src/main/java/vip/mate/llm/chatmodel/HttpTimeouts.java +++ b/mateclaw-server/src/main/java/vip/mate/llm/chatmodel/HttpTimeouts.java @@ -26,6 +26,25 @@ public final class HttpTimeouts { */ public static final Duration DEFAULT_READ_TIMEOUT = Duration.ofSeconds(180); + /** + * Default inter-frame idle timeout for streaming LLM responses + * (the reactor {@code .timeout(Duration)} applied on the chat model's + * delta Flux). Distinct from {@link #DEFAULT_READ_TIMEOUT}: the JDK + * HttpClient request timeout (which is what {@code setReadTimeout} + * ultimately maps to) only protects up to the response headers — once + * the headers arrive it stops the clock, so a provider that accepts the + * connection, returns 200 + a first SSE frame, then goes silent hangs + * the body Flux forever with no exception and no failover signal + * (issue #585). The reactor idle timeout fills that gap: it measures the + * gap between successive stream elements, so total silence for this long + * propagates a {@code TimeoutException} down the existing error path. + *

+ * Defaults to the same 180s as the read timeout — long-thinking models + * can legitimately sit between frames for a while, but complete silence + * for three minutes is a dead provider, not a slow one. + */ + public static final Duration DEFAULT_STREAM_IDLE_TIMEOUT = Duration.ofSeconds(180); + private HttpTimeouts() {} /** @@ -40,4 +59,19 @@ public final class HttpTimeouts { } return Duration.ofSeconds(override); } + + /** + * Resolve the effective streaming inter-frame idle timeout. Same fallback + * semantics as {@link #resolveReadTimeout(Integer)}: a positive override + * wins, otherwise the canonical 180s default applies. Callers can pass + * {@code modelConfig.getRequestTimeoutSeconds()} directly so the per-model + * knob governs both the connect-level read timeout and the body-level + * idle timeout from a single config field. + */ + public static Duration resolveStreamIdleTimeout(Integer override) { + if (override == null || override <= 0) { + return DEFAULT_STREAM_IDLE_TIMEOUT; + } + return Duration.ofSeconds(override); + } } diff --git a/mateclaw-server/src/main/java/vip/mate/llm/chatmodel/OpenAiCompatibleChatModelBuilder.java b/mateclaw-server/src/main/java/vip/mate/llm/chatmodel/OpenAiCompatibleChatModelBuilder.java index afde9e21..f2fc16a9 100644 --- a/mateclaw-server/src/main/java/vip/mate/llm/chatmodel/OpenAiCompatibleChatModelBuilder.java +++ b/mateclaw-server/src/main/java/vip/mate/llm/chatmodel/OpenAiCompatibleChatModelBuilder.java @@ -440,9 +440,19 @@ public class OpenAiCompatibleChatModelBuilder implements ChatModelBuilder { /** * Apply equivalent timeouts to the WebClient backing OpenAI-compatible - * STREAMING calls. Without this the streaming path uses a default WebClient - * with neither connect nor read timeout, so a stalled provider can hang the - * call indefinitely while the failover chain idles (no exception thrown). + * STREAMING calls. + * + *

Scope caveat (issue #585): {@code setReadTimeout} maps to the + * JDK HttpClient's request timeout, which only protects up to the + * response headers. Once the headers arrive the clock stops, so + * this timeout does not prevent a provider that returns 200 + a + * first SSE frame and then goes silent from hanging the body Flux. The + * body-level gap is closed by a reactor inter-frame idle timeout applied + * at the streaming chokepoint ({@code NodeStreamingChatHelper}, driven by + * {@link HttpTimeouts#DEFAULT_STREAM_IDLE_TIMEOUT}) — that is what actually + * surfaces a stalled provider to the error path / failover chain. Both + * layers are needed: this one catches a provider that never sends headers + * at all, the reactor one catches a provider that sends headers then stalls. * *

Uses {@link org.springframework.http.client.reactive.JdkClientHttpConnector} * with the same {@link HttpClient} so the dependency surface stays clean diff --git a/mateclaw-server/src/test/java/vip/mate/agent/graph/NodeStreamingChatHelperStreamIdleTimeoutTest.java b/mateclaw-server/src/test/java/vip/mate/agent/graph/NodeStreamingChatHelperStreamIdleTimeoutTest.java new file mode 100644 index 00000000..dbdfe75f --- /dev/null +++ b/mateclaw-server/src/test/java/vip/mate/agent/graph/NodeStreamingChatHelperStreamIdleTimeoutTest.java @@ -0,0 +1,124 @@ +package vip.mate.agent.graph; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; +import org.springframework.ai.chat.messages.AssistantMessage; +import org.springframework.ai.chat.messages.UserMessage; +import org.springframework.ai.chat.model.ChatModel; +import org.springframework.ai.chat.model.ChatResponse; +import org.springframework.ai.chat.model.Generation; +import org.springframework.ai.chat.metadata.ChatGenerationMetadata; +import org.springframework.ai.chat.prompt.Prompt; +import reactor.core.publisher.Flux; +import vip.mate.channel.web.ChatStreamTracker; +import vip.mate.llm.failover.ProviderHealthProperties; +import vip.mate.llm.failover.ProviderHealthTracker; + +import java.util.List; + +import static org.junit.jupiter.api.Assertions.assertNotEquals; +import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; +import static org.assertj.core.api.Assertions.assertThat; + +import java.time.Duration; + +/** + * Regression test for issue #585: the streaming body Flux must have an + * inter-frame idle timeout so a provider that accepts the connection and then + * goes silent cannot hang the call forever (no exception → no failover). + * + *

The JDK HttpClient request timeout (what {@code setReadTimeout} maps to) + * only protects up to the response headers; once they arrive the clock stops. + * A reactor {@code .timeout()} on the delta Flux closes the body-level gap. + * This test wires a {@link ChatModel} whose {@code stream()} returns + * {@link Flux#never()} (a provider that sends headers then stalls) and asserts + * the call surfaces an error within a bounded time instead of hanging until + * the 10-minute latch deadline. + */ +class NodeStreamingChatHelperStreamIdleTimeoutTest { + + private ChatStreamTracker streamTracker; + private ProviderHealthTracker healthTracker; + + @BeforeEach + void setUp() { + streamTracker = mock(ChatStreamTracker.class); + when(streamTracker.isStopRequested(any())).thenReturn(false); + healthTracker = new ProviderHealthTracker(new ProviderHealthProperties()); + } + + /** A chat model whose stream() never emits — a stalled provider. */ + private static ChatModel stalledModel() { + ChatModel m = mock(ChatModel.class); + when(m.stream(any(Prompt.class))).thenReturn(Flux.never()); + return m; + } + + private NodeStreamingChatHelper helperWithIdle(ChatModel primary, long idleSec) { + NodeStreamingChatHelper h = new NodeStreamingChatHelper( + streamTracker, List.of(), null, healthTracker, "stalled-provider"); + // Shrink the retry backoff so the full retry loop stays fast even + // though each attempt waits `idleSec` for the idle timeout to fire. + h.setRetryTimingForTest(1L, 1L, 5_000L); + h.setStreamIdleTimeoutSec(idleSec); + return h; + } + + private static Prompt smallPrompt() { + return new Prompt(List.of(new UserMessage("hi"))); + } + + @Test + @DisplayName("Stalled stream (Flux.never) surfaces an error via the idle timeout, not the 10-min latch") + void stalledStreamSurfacesErrorViaIdleTimeout() { + ChatModel primary = stalledModel(); + // 1s idle timeout; the retry loop exhausts well inside the 60s bound. + NodeStreamingChatHelper helper = helperWithIdle(primary, 1L); + + // assertTimeoutPreemptively fails the test (and unwedges it) if the + // idle timeout did NOT wire — the call would otherwise block on the + // 10-minute latch deadline. + var result = assertTimeoutPreemptively(Duration.ofSeconds(60), () -> + helper.streamCall(primary, smallPrompt(), "conv-stall", "reasoning")); + + // The stalled provider produced no text and a non-NONE error type — + // the idle timeout fired (otherwise the latch would have timed out + // and the result would still carry a generic timeout message, but + // far slower; the bounded duration above is the real assertion). + assertNotEquals(NodeStreamingChatHelper.ErrorType.NONE, result.errorType(), + "a stalled stream must surface a non-NONE error type via the idle timeout"); + // The primary was retried (idle-timeout error is retryable), proving + // the timeout propagated through the normal error path rather than + // hanging the subscription. atLeast(2) is enough — the exact count + // depends on the retry time budget, which the test shrinks. + verify(primary, org.mockito.Mockito.atLeast(2)).stream(any(Prompt.class)); + } + + @Test + @DisplayName("idle timeout disabled (<=0) keeps the legacy behavior: stream completes normally when not stalled") + void disabledIdleTimeoutDoesNotBreakNormalStream() { + // A fast-completing stream must still work when the idle timeout is + // turned off — guards against the .timeout() wiring accidentally + // short-circuiting the happy path. + ChatModel m = mock(ChatModel.class); + var gen = new Generation( + new AssistantMessage("ok"), ChatGenerationMetadata.NULL); + var resp = mock(ChatResponse.class); + when(resp.getResults()).thenReturn(List.of(gen)); + when(resp.getResult()).thenReturn(gen); + when(resp.getMetadata()).thenReturn(null); + when(m.stream(any(Prompt.class))).thenReturn(Flux.just(resp)); + + NodeStreamingChatHelper helper = helperWithIdle(m, 0L); + + var result = assertTimeoutPreemptively(Duration.ofSeconds(20), () -> + helper.streamCall(m, smallPrompt(), "conv-ok", "reasoning")); + + assertThat(result.text()).contains("ok"); + } +} diff --git a/mateclaw-server/src/test/java/vip/mate/llm/chatmodel/HttpTimeoutsTest.java b/mateclaw-server/src/test/java/vip/mate/llm/chatmodel/HttpTimeoutsTest.java index 5b469fd9..149ef1e2 100644 --- a/mateclaw-server/src/test/java/vip/mate/llm/chatmodel/HttpTimeoutsTest.java +++ b/mateclaw-server/src/test/java/vip/mate/llm/chatmodel/HttpTimeoutsTest.java @@ -68,4 +68,37 @@ class HttpTimeoutsTest { void defaultMatchesLegacy() { assertEquals(Duration.ofSeconds(180), HttpTimeouts.DEFAULT_READ_TIMEOUT); } + + // ===== Streaming inter-frame idle timeout (issue #585) ===== + + @Test + @DisplayName("default stream idle timeout is 180s, aligned with the read timeout") + void defaultStreamIdleTimeout() { + assertEquals(Duration.ofSeconds(180), HttpTimeouts.DEFAULT_STREAM_IDLE_TIMEOUT); + } + + @Test + @DisplayName("null override → default 180s stream idle timeout") + void streamIdleNullFallsBack() { + assertEquals(Duration.ofSeconds(180), + HttpTimeouts.resolveStreamIdleTimeout(null)); + } + + @Test + @DisplayName("zero / negative override → default 180s (treated as unset)") + void streamIdleZeroOrNegativeFallsBack() { + assertEquals(Duration.ofSeconds(180), + HttpTimeouts.resolveStreamIdleTimeout(0)); + assertEquals(Duration.ofSeconds(180), + HttpTimeouts.resolveStreamIdleTimeout(-5)); + } + + @Test + @DisplayName("positive override → exact seconds (per-model knob governs body idle too)") + void streamIdlePositiveHonored() { + assertEquals(Duration.ofSeconds(60), + HttpTimeouts.resolveStreamIdleTimeout(60)); + assertEquals(Duration.ofSeconds(300), + HttpTimeouts.resolveStreamIdleTimeout(300)); + } }