fix(runtime): harden UTF-8 continuations and stream timeouts

crates/runtime/src/server/server_loop_host.rs

@@ -1602,10 +1602,11 @@ fn merge_output_cap_continuation(prefix: &str, continuation: &str) -> String {
     // Providers normally return a suffix, but a small overlap can occur when
     // they restate the final sentence after a continuation request.  Avoid
     // duplicating that overlap while keeping the merge deterministic and
-    // independent of task/domain text.
+    // independent of task/domain text. The byte budget and shorter candidates
+    // can fall inside a UTF-8 code point, so only compare character boundaries.
     let max_overlap = prefix.len().min(512).min(continuation.len());
     for overlap in (1..=max_overlap).rev() {
-        if prefix.ends_with(&continuation[..overlap]) {
+        if continuation.is_char_boundary(overlap) && prefix.ends_with(&continuation[..overlap]) {
             return format!("{}{}", prefix, &continuation[overlap..]);
         }
     }
@@ -36557,6 +36558,80 @@ mod tests {
         );
     }
 
+    #[test]
+    fn output_cap_continuation_preserves_unicode_without_overlap() {
+        let prefix = "p".repeat(169);
+        let continuation = format!("{}—suffix", "x".repeat(167));
+        assert_eq!(
+            merge_output_cap_continuation(&prefix, &continuation),
+            format!("{prefix}\n{continuation}")
+        );
+        assert_eq!(
+            merge_output_cap_continuation("前文", "后续内容"),
+            "前文\n后续内容"
+        );
+    }
+
+    #[test]
+    fn output_cap_continuation_deduplicates_unicode_overlap() {
+        for (prefix, continuation, expected) in [
+            (
+                "前文：实验完成",
+                "实验完成，结果可信",
+                "前文：实验完成，结果可信",
+            ),
+            ("done: 🧪", "🧪✅", "done: 🧪✅"),
+            ("partial—", "—suffix", "partial—suffix"),
+        ] {
+            assert_eq!(
+                merge_output_cap_continuation(prefix, continuation),
+                expected
+            );
+        }
+    }
+
+    #[test]
+    fn output_cap_continuation_preserves_overlap_byte_limit() {
+        for (overlap, expected_deduplication) in [(511, true), (512, true), (513, false)] {
+            let shared = format!("a{}b", "x".repeat(overlap - 2));
+            let prefix = format!("prefix:{shared}");
+            let continuation = format!("{shared}—suffix");
+            let expected = if expected_deduplication {
+                format!("{prefix}—suffix")
+            } else {
+                format!("{prefix}\n{continuation}")
+            };
+            assert_eq!(
+                merge_output_cap_continuation(&prefix, &continuation),
+                expected
+            );
+        }
+
+        let prefix = "p".repeat(520);
+        let continuation = format!("{}—suffix", "x".repeat(511));
+        assert_eq!(
+            merge_output_cap_continuation(&prefix, &continuation),
+            format!("{prefix}\n{continuation}")
+        );
+    }
+
+    #[test]
+    fn output_cap_continuation_preserves_empty_and_replayed_text() {
+        for (prefix, continuation, expected) in [
+            ("", "", ""),
+            ("", "后续🧪", "后续🧪"),
+            ("前文🧪", "", "前文🧪"),
+            ("前文🧪", "前文🧪", "前文🧪"),
+            ("前文🧪", "前文🧪后续", "前文🧪后续"),
+            ("前文", " 后续", "前文 后续"),
+        ] {
+            assert_eq!(
+                merge_output_cap_continuation(prefix, continuation),
+                expected
+            );
+        }
+    }
+
     #[test]
     fn output_cap_continuation_is_a_typed_append_only_authority() {
         let mut messages = vec![json!({"role":"user", "content":"finish the task"})];

crates/runtime/src/turn/llm/client.rs

@@ -173,13 +173,11 @@ const LLM_NONSTREAM_TIMEOUT_S: u64 = 120;
 /// Total budget across all safe retries for a single LLM call (seconds).
 /// Override: `ASTRA_LLM_TOTAL_BUDGET_S`.
 const LLM_TOTAL_BUDGET_S: u64 = 300;
-/// Maximum time a streaming inference may go without yielding or advancing a
-/// deliverable answer/tool call. Reasoning is transport activity, but it does
-/// not postpone the initial actionable yield. Once delivery starts, visible
-/// text and authorized tool JSON fragments roll this deadline so a legitimate
-/// streamed tool payload is not cut off mid-object. Independent physical-idle,
-/// accumulation, and total-call budgets remain additional safety bounds.
-const LLM_SEMANTIC_PROGRESS_TIMEOUT_S: u64 = 120;
+/// Maximum time a streaming inference may go without meaningful text,
+/// reasoning, or authorized tool-call progress. Heartbeats do not refresh it.
+/// The five-minute default tolerates providers buffering long tool payloads;
+/// independent idle, accumulation, and total-call budgets still bound the call.
+const LLM_SEMANTIC_PROGRESS_TIMEOUT_S: u64 = 300;
 /// Maximum slice reserved from a logical inference budget for durable attempt
 /// terminalization. The reserve scales down for very small test/operator
 /// budgets and is never available to HTTP work or retry backoff.
@@ -8846,6 +8844,133 @@ mod tests {
         assert_eq!(dur, std::time::Duration::from_secs(LLM_TOTAL_BUDGET_S));
     }
 
+    #[test]
+    fn default_semantic_progress_timeout_is_300s() {
+        assert_eq!(
+            llm_semantic_progress_timeout(),
+            std::time::Duration::from_secs(300)
+        );
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn default_semantic_progress_timeout_accepts_buffered_tool_delivery() {
+        for anthropic in [false, true] {
+            let arguments = json!({"command": "x".repeat(24 * 1024)}).to_string();
+            let terminal = if anthropic {
+                [
+                    json!({"type": "content_block_start", "index": 0, "content_block": {
+                        "type": "tool_use", "id": "call-1", "name": "bash", "input": {}
+                    }}),
+                    json!({"type": "content_block_delta", "index": 0, "delta": {
+                        "type": "input_json_delta", "partial_json": arguments
+                    }}),
+                    json!({"type": "content_block_stop", "index": 0}),
+                    json!({"type": "message_delta", "delta": {"stop_reason": "tool_use"}}),
+                    json!({"type": "message_stop"}),
+                ]
+                .into_iter()
+                .map(|chunk| format!("data: {chunk}\n\n"))
+                .collect::<String>()
+            } else {
+                let chunk = json!({"choices": [{"delta": {"tool_calls": [{
+                    "index": 0, "id": "call-1", "type": "function",
+                    "function": {"name": "bash", "arguments": arguments}
+                }]}, "finish_reason": "tool_calls"}]});
+                format!("data: {chunk}\n\ndata: [DONE]\n\n")
+            };
+            let source = Box::pin(async_stream::stream! {
+                for _ in 0..12 {
+                    tokio::time::sleep(std::time::Duration::from_secs(15)).await;
+                    yield Ok::<Bytes, reqwest::Error>(Bytes::from_static(b": PING\n\n"));
+                }
+                yield Ok::<Bytes, reqwest::Error>(Bytes::from(terminal));
+            });
+            let started = TokioInstant::now();
+            let result = if anthropic {
+                collect_anthropic_llm_stream(
+                    source,
+                    "test-model",
+                    started.into_std(),
+                    LlmCancel::None,
+                    stream_idle_timeout(),
+                    stream_idle_timeout_after_progress(),
+                    None,
+                )
+                .await
+            } else {
+                collect_llm_stream(
+                    source,
+                    "test-model",
+                    started.into_std(),
+                    LlmCancel::None,
+                    stream_idle_timeout(),
+                    stream_idle_timeout_after_progress(),
+                    None,
+                )
+                .await
+            }
+            .expect(
+                "the default must allow buffered tool delivery after 180 seconds of heartbeats",
+            );
+            assert_eq!(started.elapsed(), std::time::Duration::from_secs(180));
+            assert!(result.full_text.is_empty());
+            assert_eq!(result.tool_calls.len(), 1);
+            assert_eq!(result.tool_calls[0]["function"]["name"], "bash");
+            assert_eq!(result.tool_calls[0]["function"]["arguments"], arguments);
+        }
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn default_semantic_progress_timeout_preserves_provider_work_budget() {
+        for anthropic in [false, true] {
+            let source = Box::pin(async_stream::stream! {
+                loop {
+                    tokio::time::sleep(std::time::Duration::from_secs(15)).await;
+                    yield Ok::<Bytes, reqwest::Error>(Bytes::from_static(b": PING\n\n"));
+                }
+            });
+            let started = TokioInstant::now();
+            let total = llm_total_budget();
+            let work = total - llm_mandatory_settlement_reserve(total);
+            let result = if anthropic {
+                collect_anthropic_llm_stream_with_semantic_progress_deadline_and_surface(
+                    source,
+                    "test-model",
+                    started.into_std(),
+                    work,
+                    LlmCancel::None,
+                    stream_idle_timeout(),
+                    stream_idle_timeout_after_progress(),
+                    llm_semantic_progress_timeout(),
+                    None,
+                    None,
+                )
+                .await
+            } else {
+                collect_llm_stream_with_semantic_progress_deadline_and_surface(
+                    source,
+                    "test-model",
+                    started.into_std(),
+                    work,
+                    LlmCancel::None,
+                    stream_idle_timeout(),
+                    stream_idle_timeout_after_progress(),
+                    llm_semantic_progress_timeout(),
+                    None,
+                    None,
+                )
+                .await
+            };
+            match result.expect_err("heartbeats must not extend the provider work budget") {
+                StreamCollectError::ProviderWorkDeadline {
+                    elapsed_ms: 290000, ..
+                } => {}
+                other => panic!("expected provider-work deadline, got {other:?}"),
+            }
+            assert_eq!(started.elapsed(), work);
+        }
+    }
+
     #[tokio::test]
     async fn streaming_request_without_catalog_timeout_obeys_total_budget() {
         // This case targets an admitted request, not cold TLS/client initialization.

crates/runtime/tests/cache_provider_matrix_e2e.rs

@@ -746,8 +746,8 @@ async fn append_only_metadata_drives_real_output_cap_continuation_and_wal() {
             reuse_scope: Some(CacheReuseScope::ConversationTurns),
         }),
     };
-    let ProviderResponse::OpenAi(mut partial) =
-        response(case, "Partial answer with more detail to follow.", None)
+    // Exercise UTF-8 merging after a real provider length terminal and retry.
+    let ProviderResponse::OpenAi(mut partial) = response(case, "论文实验结果尚未讲完。", None)
     else {
         unreachable!()
     };
@@ -757,7 +757,7 @@ async fn append_only_metadata_drives_real_output_cap_continuation_and_wal() {
         |request| request.path == "/v1/chat/completions" && request.body["model"] == "append-alias",
         vec![
             ProviderResponse::OpenAi(partial),
-            response(case, "The complete explanation is now available.", None),
+            response(case, "—实验结果与结论已补全。", None),
         ],
     )])
     .await;
@@ -787,7 +787,7 @@ async fn append_only_metadata_drives_real_output_cap_continuation_and_wal() {
         .unwrap();
     assert_eq!(
         state.final_text,
-        "Partial answer with more detail to follow.\nThe complete explanation is now available."
+        "论文实验结果尚未讲完。\n—实验结果与结论已补全。"
     );
     assert_eq!(
         state.llm_rounds_completed, 1,
@@ -811,7 +811,7 @@ async fn append_only_metadata_drives_real_output_cap_continuation_and_wal() {
             .messages
             .iter()
             .any(|message| message["role"] == "assistant"
-                && message["content"] == "Partial answer with more detail to follow.")
+                && message["content"] == "论文实验结果尚未讲完。")
     );
     assert!(
         state

docs/reference/configuration.md

@@ -240,7 +240,7 @@ configuration loader reports a warning and skips an invalid user/project layer.
 - `ASTRA_RETRIEVAL_TOP_K`, `ASTRA_MAX_TURN_INPUT_TOKENS`
 - `ASTRA_LLM_PROVIDER_ADMISSION_MODE` — provider admission mode; unset/`disabled` by default, `db_fixed_window` enables MatrixOne-backed RPM/TPM claims before outbound LLM attempts
 - `ASTRA_LLM_PROVIDER_ADMISSION_RPM`, `ASTRA_LLM_PROVIDER_ADMISSION_TPM` — provider budget used by admission; at least one is required when admission is enabled
-- `ASTRA_LLM_CONNECT_TIMEOUT_S`, `ASTRA_LLM_NONSTREAM_TIMEOUT_S`, `ASTRA_LLM_TOTAL_BUDGET_S`, `ASTRA_LLM_ACTION_PROGRESS_TIMEOUT_S` — provider transport/progress bounds. The `300s` total-budget default is per provider call including retries, not an end-to-end session limit; turn profiles and resource policy still bound the overall run. Interactive resource policy is 30s for a single tool execution, while long-session profiles explicitly allow 300s.
+- `ASTRA_LLM_CONNECT_TIMEOUT_S`, `ASTRA_LLM_NONSTREAM_TIMEOUT_S`, `ASTRA_LLM_TOTAL_BUDGET_S`, `ASTRA_LLM_SEMANTIC_PROGRESS_TIMEOUT_S` — provider transport/progress bounds. The semantic-progress timeout defaults to `300s`; non-whitespace text, reasoning, and advancing authorized tool arguments refresh it, but SSE heartbeats and empty deltas do not. Explicit positive overrides remain supported. The independent `300s` total-budget default is per logical provider call including retries, not an end-to-end session limit. Provider work leaves a tail reserve for durable terminalization (10s at the default total budget), so a silent interval is not guaranteed another full five minutes after the last token. Neither heartbeats nor semantic progress extend the total budget. Turn profiles and resource policy still bound the overall run. Interactive resource policy is 30s for a single tool execution, while long-session profiles explicitly allow 300s.
 - `ASTRA_AUX_LLM_POLICY` — policy for bounded auxiliary LLM calls. When unset, Astra uses `capacity_aware`: Work admission is evaluated at an existing typed effect/topology boundary, where the result has an immediate admission consumer; ordinary unbound primary turns stay on the single-request path, while unrelated optional judges remain capacity-gated. `boundary_only` keeps the same Work boundary but also suppresses unrelated optional auxiliary judges. An unavailable auxiliary decision is recorded as typed degradation and does not discard a primary response that already passed the canonical tool/lifecycle boundary; an explicitly `disabled` policy under Auto still fails closed before action or completion, and a client that deliberately omits classification must explicitly request `FixedDefault`. Set `always` only when a deployment deliberately wants speculative Work admission on ordinary turns and all eligible auxiliary calls regardless of capacity policy.
 - `ASTRA_CAPTURE_TRACES`
 - `ASTRA_RUN_CONCURRENCY_LIMIT` — positive agentic loop slots per Astra Server process. For a multi-server deployment, set `ASTRA_CAPACITY_POD_COUNT` to the number of equivalent server processes sharing the same durable admission scope; the cross-pod weighted budget is derived from both values and is fenced when a server presents a different capacity snapshot. Change the declared budget only after active reservations have drained, and use one snapshot across all participating servers.