fix(llm): add stream body idle timeout

Apply a Reactor inter-frame idle timeout at the streaming chat chokepoint so half-open provider body streams surface through the normal retry/failover path. Also keep the HTTP timeout documentation accurate and cover the behavior with focused tests.

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,

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).
+     * <p>
+     * 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<ChatResponse> 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;

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).
+     *
+     * <p><b>Scope caveat (issue #585):</b> {@code setReadTimeout} maps to the
+     * JDK HttpClient's request timeout, which only protects up to the
+     * <em>response headers</em>. Once the headers arrive the clock stops, so a
+     * provider that returns 200 + a first SSE frame and then goes silent is
+     * <b>not</b> 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.
      * <p>
      * Uses the same JDK HttpClient + JdkClientHttpConnector path, so the
      * dependency surface doesn't pull in reactor-netty (excluded by this

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.
+     * <p>
+     * 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);
+    }
 }

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.
+     *
+     * <p><b>Scope caveat (issue #585):</b> {@code setReadTimeout} maps to the
+     * JDK HttpClient's request timeout, which only protects up to the
+     * <em>response headers</em>. Once the headers arrive the clock stops, so
+     * this timeout does <b>not</b> 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.
      *
      * <p>Uses {@link org.springframework.http.client.reactive.JdkClientHttpConnector}
      * with the same {@link HttpClient} so the dependency surface stays clean

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).
+ *
+ * <p>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");
+    }
+}

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));
+    }
 }