diff --git a/backend/app/config.py b/backend/app/config.py
index 48a2220c1..e9e154d5e 100644
--- a/backend/app/config.py
+++ b/backend/app/config.py
@@ -159,6 +159,7 @@ class Settings(BaseSettings):
AGENT_RUNTIME_RUN_COMPACT_TOOL_RESULT_BYTES: int | None = Field(default=None, gt=0)
AGENT_RUNTIME_VERIFY_REPAIR_COMPACT_ROUNDS: int | None = Field(default=None, gt=0)
AGENT_RUNTIME_MODEL_CAPABILITY_REFRESH_SECONDS: int = Field(default=86400, gt=0)
+ AGENT_RUNTIME_WEB_STREAMING_ENABLED: bool = True
AGENT_RUNTIME_FALLBACK_CONTEXT_WINDOW_TOKENS: int = Field(default=131072, gt=0)
MULTI_AGENT_COMPACT_MODEL_ID: uuid.UUID | None = None
MULTI_AGENT_PLANNING_MODEL_ID: uuid.UUID | None = None
diff --git a/backend/app/services/agent_runtime/answer_stream.py b/backend/app/services/agent_runtime/answer_stream.py
new file mode 100644
index 000000000..9385b5b2d
--- /dev/null
+++ b/backend/app/services/agent_runtime/answer_stream.py
@@ -0,0 +1,166 @@
+"""Coalesced durable observations for provisional user-visible answer text."""
+
+from __future__ import annotations
+
+import asyncio
+import uuid
+from datetime import UTC, datetime
+
+from loguru import logger
+from sqlalchemy.dialects.postgresql import insert
+
+from app.models.agent_run_event import AgentRunEvent
+from app.services.agent_runtime.command_worker import RuntimeSessionFactory
+
+_DEFAULT_FLUSH_INTERVAL_SECONDS = 0.1
+_DEFAULT_MAX_BUFFER_CHARS = 512
+
+
+class AnswerStreamWriter:
+ """Buffer visible answer deltas and persist short, idempotent observations."""
+
+ def __init__(
+ self,
+ *,
+ session_factory: RuntimeSessionFactory,
+ tenant_id: uuid.UUID,
+ run_id: uuid.UUID,
+ agent_id: uuid.UUID,
+ attempt_id: uuid.UUID | str,
+ flush_interval: float = _DEFAULT_FLUSH_INTERVAL_SECONDS,
+ max_buffer_chars: int = _DEFAULT_MAX_BUFFER_CHARS,
+ ) -> None:
+ if flush_interval <= 0:
+ raise ValueError("flush_interval must be positive")
+ if max_buffer_chars <= 0:
+ raise ValueError("max_buffer_chars must be positive")
+ normalized_attempt_id = str(attempt_id).strip()
+ if not normalized_attempt_id:
+ raise ValueError("attempt_id must be non-empty")
+
+ self._session_factory = session_factory
+ self._tenant_id = tenant_id
+ self._run_id = run_id
+ self._agent_id = agent_id
+ self._attempt_id = normalized_attempt_id
+ self._flush_interval = flush_interval
+ self._max_buffer_chars = max_buffer_chars
+ self._parts: list[str] = []
+ self._buffer_chars = 0
+ self._next_sequence = 1
+ self._closed = False
+ self._visible_started = False
+ self._wake = asyncio.Event()
+ self._worker: asyncio.Task[None] | None = None
+
+ @property
+ def visible_started(self) -> bool:
+ """Whether at least one visible observation was transactionally written."""
+ return self._visible_started
+
+ async def write(self, content: str) -> None:
+ """Accept one visible text delta without waiting for database I/O."""
+ if self._closed:
+ raise RuntimeError("answer stream writer is closed")
+ if not isinstance(content, str):
+ raise TypeError("answer stream content must be text")
+ if not content:
+ return
+
+ self._parts.append(content)
+ self._buffer_chars += len(content)
+ self._ensure_worker()
+ if self._buffer_chars >= self._max_buffer_chars:
+ self._wake.set()
+
+ async def flush(self) -> None:
+ """Flush all currently buffered content."""
+ if self._parts:
+ self._ensure_worker()
+ self._wake.set()
+ worker = self._worker
+ if worker is not None and not worker.done():
+ await worker
+
+ async def close(self) -> None:
+ """Stop accepting content and flush the final buffered delta."""
+ if self._closed:
+ return
+ self._closed = True
+ if self._parts:
+ self._ensure_worker()
+ self._wake.set()
+ worker = self._worker
+ if worker is not None and not worker.done():
+ await worker
+
+ def _ensure_worker(self) -> None:
+ if self._worker is not None and self._worker.done():
+ try:
+ self._worker.result()
+ except Exception as exc:
+ # The failed batch was restored to the buffer and the next
+ # worker retries the same deterministic sequence.
+ logger.warning(
+ "[RuntimeAnswerStream] retrying restored batch after {}",
+ type(exc).__name__,
+ )
+ if self._worker is None or self._worker.done():
+ self._worker = asyncio.create_task(self._run_worker())
+
+ async def _run_worker(self) -> None:
+ while self._parts:
+ try:
+ await asyncio.wait_for(self._wake.wait(), timeout=self._flush_interval)
+ except TimeoutError:
+ pass
+ self._wake.clear()
+ await self._flush_once()
+
+ async def _flush_once(self) -> None:
+ if not self._parts:
+ return
+
+ parts = self._parts
+ content = "".join(parts)
+ sequence = self._next_sequence
+ self._parts = []
+ self._buffer_chars = 0
+ key = f"answer-stream:{self._attempt_id}:{sequence}"
+
+ try:
+ async with self._session_factory() as db, db.begin():
+ await db.execute(
+ insert(AgentRunEvent)
+ .values(
+ id=uuid.uuid5(
+ self._run_id,
+ f"answer-stream-event:{self._attempt_id}:{sequence}",
+ ),
+ tenant_id=self._tenant_id,
+ run_id=self._run_id,
+ agent_id=self._agent_id,
+ event_type="status_changed",
+ summary="Assistant answer streaming",
+ payload={
+ "activity_type": "assistant_delta",
+ "status": "running",
+ "attempt_id": self._attempt_id,
+ "sequence": sequence,
+ "content": content,
+ "reset": sequence == 1,
+ },
+ artifact_refs=[],
+ idempotency_key=key,
+ source_checkpoint_id=None,
+ created_at=datetime.now(UTC),
+ )
+ .on_conflict_do_nothing()
+ )
+ except BaseException:
+ self._parts = [*parts, *self._parts]
+ self._buffer_chars += len(content)
+ raise
+
+ self._next_sequence += 1
+ self._visible_started = True
diff --git a/backend/app/services/agent_runtime/chat_stream.py b/backend/app/services/agent_runtime/chat_stream.py
index 8ec3faf9e..d351ec9c2 100644
--- a/backend/app/services/agent_runtime/chat_stream.py
+++ b/backend/app/services/agent_runtime/chat_stream.py
@@ -143,10 +143,36 @@ async def stream_web_chat_run(
if content is not None:
await send_packet({"type": "thinking", "content": content, **packet_position})
continue
- if event.event_type == "status_changed" and activity_type == "assistant_progress":
- content = _text(payload.get("content"))
+ if event.event_type == "status_changed" and activity_type in {
+ "assistant_progress",
+ "assistant_delta",
+ }:
+ raw_content = payload.get("content")
+ content = (
+ raw_content
+ if activity_type == "assistant_delta"
+ and isinstance(raw_content, str)
+ and raw_content
+ else _text(raw_content)
+ )
if content is not None:
- await send_packet({"type": "chunk", "content": content, **packet_position})
+ packet = {"type": "chunk", "content": content, **packet_position}
+ if activity_type == "assistant_delta":
+ attempt_id = _text(payload.get("attempt_id"))
+ sequence = payload.get("sequence")
+ if attempt_id is None or not isinstance(sequence, int) or sequence <= 0:
+ raise ChatRuntimeStreamError(
+ "invalid_runtime_answer_delta",
+ "Runtime answer delta has no valid attempt position",
+ )
+ packet.update(
+ {
+ "attempt_id": attempt_id,
+ "sequence": sequence,
+ "reset": payload.get("reset") is True,
+ }
+ )
+ await send_packet(packet)
continue
if event.event_type == "status_changed" and activity_type == "tool_call":
tool_name = _text(payload.get("name"))
@@ -258,6 +284,7 @@ async def stream_web_chat_run(
"error": error,
"runtime_status": status,
"delivery_error": error_code,
+ **packet_position,
}
)
return ChatRuntimeStreamOutcome(
@@ -287,8 +314,8 @@ async def stream_web_chat_run(
"role": "assistant",
"content": message.content,
"message_id": str(message.id),
- "run_id": str(handle.run_id),
"runtime_status": status,
+ **packet_position,
}
if status == "failed":
error_code = (
diff --git a/backend/app/services/agent_runtime/checkpoint_side_effects.py b/backend/app/services/agent_runtime/checkpoint_side_effects.py
index 323994206..2f65b9bb3 100644
--- a/backend/app/services/agent_runtime/checkpoint_side_effects.py
+++ b/backend/app/services/agent_runtime/checkpoint_side_effects.py
@@ -344,7 +344,11 @@ def _runtime_observation_events(
)
content = _text_field(message.get("content"))
runtime_intent = message.get("runtime_intent")
- if content is not None and runtime_intent not in {"finish", "wait"}:
+ if (
+ content is not None
+ and runtime_intent not in {"finish", "wait"}
+ and message.get("runtime_answer_streamed") is not True
+ ):
events.append(
(
"status_changed",
diff --git a/backend/app/services/agent_runtime/model_step_service.py b/backend/app/services/agent_runtime/model_step_service.py
index aa9ad2412..d5a3d7744 100644
--- a/backend/app/services/agent_runtime/model_step_service.py
+++ b/backend/app/services/agent_runtime/model_step_service.py
@@ -25,6 +25,7 @@
from app.models.llm import LLMModel
from app.models.participant import Participant
from app.services.agent_context import build_agent_context
+from app.services.agent_runtime.answer_stream import AnswerStreamWriter
from app.services.agent_runtime.command_worker import RuntimeSessionFactory
from app.services.agent_runtime.context_builder import (
ContextBuilder,
@@ -86,7 +87,7 @@
builtin_policy,
is_reserved_custom_tool_name,
)
-from app.services.llm.client import LLMMessage
+from app.services.llm.client import LLMMessage, LLMVisibleStreamInterrupted
from app.services.llm.failover import (
classify_error,
is_retryable_classification,
@@ -323,6 +324,7 @@ async def __call__(
tools: list[dict] | None = None,
agent_id: uuid.UUID | None = None,
supports_vision: bool = False,
+ on_visible_delta: Callable[[str], Awaitable[None]] | None = None,
) -> LLMCompletionStep: ...
@@ -1007,6 +1009,8 @@ def _assistant_message(
message["tool_calls"] = [dict(call) for call in tool_calls]
if step.reasoning_content:
message["reasoning_content"] = step.reasoning_content
+ if step.visible_streamed:
+ message["runtime_answer_streamed"] = True
if runtime_intent:
message["runtime_intent"] = runtime_intent
return message
@@ -1345,6 +1349,7 @@ def __init__(
model_retry_max_delay_seconds: float = _DEFAULT_MODEL_RETRY_MAX_DELAY_SECONDS,
model_retry_jitter_ratio: float = _DEFAULT_MODEL_RETRY_JITTER_RATIO,
retry_sleep: Callable[[float], Awaitable[None]] = asyncio.sleep,
+ answer_stream_enabled: bool = False,
) -> None:
self._session_factory = session_factory
self._context_builder = context_builder
@@ -1369,6 +1374,7 @@ def __init__(
max(0.0, model_retry_jitter_ratio),
)
self._retry_sleep = retry_sleep
+ self._answer_stream_enabled = answer_stream_enabled
async def _load(
self,
@@ -1821,6 +1827,7 @@ async def _call_prepared(
agent: Agent,
messages: list[LLMMessage],
tools: list[dict],
+ on_visible_delta: Callable[[str], Awaitable[None]] | None = None,
) -> LLMCompletionStep:
return await self._completion(
model,
@@ -1828,8 +1835,58 @@ async def _call_prepared(
tools=_provider_tools(tools),
agent_id=agent.id,
supports_vision=bool(model.supports_vision),
+ on_visible_delta=on_visible_delta,
)
+ def _streams_visible_web_answer(
+ self,
+ state: RuntimeGraphState,
+ context: RuntimeContext,
+ ) -> bool:
+ initial_input = state["snapshots"].initial_input
+ return (
+ self._answer_stream_enabled
+ and context.source_type == "chat"
+ and context.session_id is not None
+ and initial_input.get("source_channel") in {None, "web"}
+ and not _is_public_group_chat_run(state)
+ )
+
+ def _answer_stream_writer(
+ self,
+ *,
+ state: RuntimeGraphState,
+ context: RuntimeContext,
+ agent: Agent,
+ ) -> AnswerStreamWriter | None:
+ if not self._streams_visible_web_answer(state, context):
+ return None
+ run_id = uuid.UUID(context.run_id)
+ # This identifies one physical provider invocation, not the logical
+ # model step. A worker crash before checkpoint commitment must create a
+ # new reset boundary instead of replaying sequence numbers from stale
+ # provisional output.
+ attempt_id = uuid.uuid4()
+ return AnswerStreamWriter(
+ session_factory=self._session_factory,
+ tenant_id=uuid.UUID(context.tenant_id),
+ run_id=run_id,
+ agent_id=agent.id,
+ attempt_id=attempt_id,
+ )
+
+ @staticmethod
+ async def _close_answer_stream(writer: AnswerStreamWriter | None) -> None:
+ if writer is None:
+ return
+ try:
+ await writer.close()
+ except Exception as exc:
+ logger.warning(
+ "[RuntimeAnswerStream] provisional observation flush failed: {}",
+ type(exc).__name__,
+ )
+
async def _call_prepared_with_retry(
self,
*,
@@ -1837,18 +1894,31 @@ async def _call_prepared_with_retry(
agent: Agent,
messages: list[LLMMessage],
tools: list[dict],
+ state: RuntimeGraphState,
+ context: RuntimeContext,
) -> LLMCompletionStep:
"""Retry only transient provider failures before model failover."""
total_attempts = self._model_retry_attempts + 1
for attempt in range(1, total_attempts + 1):
+ writer = self._answer_stream_writer(
+ state=state,
+ context=context,
+ agent=agent,
+ )
try:
- return await self._call_prepared(
+ step = await self._call_prepared(
model=model,
agent=agent,
messages=messages,
tools=tools,
+ on_visible_delta=(writer.write if writer is not None else None),
)
except Exception as exc:
+ await self._close_answer_stream(writer)
+ if writer is not None and writer.visible_started:
+ raise LLMVisibleStreamInterrupted(
+ "Provider stream interrupted after visible output was published"
+ ) from exc
classification = classify_error(exc)
is_retryable = is_retryable_classification(classification)
if (
@@ -1890,6 +1960,13 @@ async def _call_prepared_with_retry(
delay,
)
await self._retry_sleep(delay)
+ else:
+ await self._close_answer_stream(writer)
+ return (
+ replace(step, visible_streamed=True)
+ if writer is not None and writer.visible_started
+ else step
+ )
raise AssertionError("model retry loop exhausted without an exception")
@@ -1986,6 +2063,8 @@ async def complete_once(
agent=agent,
messages=prepared,
tools=tools,
+ state=state,
+ context=context,
)
except Exception as primary_error:
primary_classification = classify_error(primary_error)
@@ -2076,6 +2155,8 @@ async def complete_once(
agent=agent,
messages=fallback_prepared,
tools=fallback_tools,
+ state=state,
+ context=context,
)
except Exception as fallback_error:
fallback_classification = classify_error(fallback_error)
diff --git a/backend/app/services/agent_runtime/tool_step_service.py b/backend/app/services/agent_runtime/tool_step_service.py
index 1ca65a47a..8e117666a 100644
--- a/backend/app/services/agent_runtime/tool_step_service.py
+++ b/backend/app/services/agent_runtime/tool_step_service.py
@@ -1070,6 +1070,7 @@ async def _reserve(
lease_owner: str,
reasoning_content: str = "",
assistant_content: str = "",
+ assistant_content_streamed: bool = False,
) -> ToolExecutionReservation:
async with self._session_factory() as db, db.begin():
reservation = await reserve_tool_execution(
@@ -1110,7 +1111,7 @@ async def _reserve(
"message_id": assistant_message_id,
},
)
- if assistant_content.strip():
+ if assistant_content.strip() and not assistant_content_streamed:
await _insert_runtime_activity(
db,
tenant_id=tenant_id,
@@ -1990,6 +1991,11 @@ async def execute_pending(
if isinstance(assistant_message, Mapping)
else ""
)
+ assistant_content_streamed = (
+ assistant_message.get("runtime_answer_streamed") is True
+ if isinstance(assistant_message, Mapping)
+ else False
+ )
try:
step_context = parse_step_tool_context(
state["lifecycle"].get("step_tool_context"),
@@ -2269,6 +2275,7 @@ async def execute_pending(
lease_owner=lease_owner,
reasoning_content=reasoning_content,
assistant_content=assistant_content,
+ assistant_content_streamed=assistant_content_streamed,
)
if reservation.reusable_result is not None:
if reservation.reusable_result.status == "pending":
diff --git a/backend/app/services/agent_runtime/worker_service.py b/backend/app/services/agent_runtime/worker_service.py
index b5f046941..30e919c87 100644
--- a/backend/app/services/agent_runtime/worker_service.py
+++ b/backend/app/services/agent_runtime/worker_service.py
@@ -224,6 +224,7 @@ def build_runtime_worker_components(
model_service = RuntimeModelStepService(
session_factory=session_factory,
context_builder=context_builder,
+ answer_stream_enabled=runtime_settings.AGENT_RUNTIME_WEB_STREAMING_ENABLED,
)
tool_result_store = ToolResultStore(session_factory=session_factory)
tool_result_reconciler = ToolResultReconciler(
diff --git a/backend/app/services/llm/client.py b/backend/app/services/llm/client.py
index 48e8c35bc..53b62ea88 100644
--- a/backend/app/services/llm/client.py
+++ b/backend/app/services/llm/client.py
@@ -30,6 +30,10 @@ class LLMRequestShapeError(LLMError):
"""The final provider request violates a portable message-shape invariant."""
+class LLMVisibleStreamInterrupted(LLMError):
+ """A provider stream failed after user-visible output was published."""
+
+
_LEADING_THINK_TAG = re.compile(r"^\s*", re.IGNORECASE)
_CLOSING_THINK_TAG = re.compile(r"", re.IGNORECASE)
_TEXTUAL_TOOL_CALL = re.compile(
@@ -502,7 +506,7 @@ class LLMStreamChunk:
# Type Definitions
# ============================================================================
-ChunkCallback = Callable[[str], Coroutine[Any, Any, None]]
+ChunkCallback = Callable[[str], Coroutine[Any, Any, bool | None]]
ToolCallback = Callable[[dict], Coroutine[Any, Any, None]]
ThinkingCallback = Callable[[str], Coroutine[Any, Any, None]]
@@ -926,6 +930,7 @@ async def stream(
max_retries = 3
client = await self._get_client()
+ visible_content_emitted = False
for attempt in range(max_retries):
try:
@@ -947,7 +952,10 @@ async def stream(
if chunk.content:
full_content += chunk.content
if on_chunk:
- await on_chunk(chunk.content)
+ published = await on_chunk(chunk.content)
+ visible_content_emitted = (
+ visible_content_emitted or published is not False
+ )
if chunk.reasoning_content:
full_reasoning += chunk.reasoning_content
@@ -992,6 +1000,10 @@ async def stream(
break # Success
except (httpx.ConnectError, httpx.ReadError, httpx.ConnectTimeout) as e:
+ if visible_content_emitted:
+ raise LLMVisibleStreamInterrupted(
+ "Provider stream interrupted after visible output was published"
+ ) from e
if attempt < max_retries - 1:
wait = (attempt + 1) * 1
logger.warning(f"Stream attempt {attempt + 1} failed ({type(e).__name__}), retrying in {wait}s...")
@@ -1842,6 +1854,8 @@ async def stream(
payload = self._build_payload(messages, tools, temperature, max_tokens, **kwargs)
full_text = ""
+ full_reasoning = ""
+ thought_signature: str | None = None
tool_calls: list[dict[str, Any]] = []
seen_tool_calls: set[str] = set()
final_usage: dict[str, int] | None = None
@@ -1891,9 +1905,17 @@ async def stream(
for part in content_obj.get("parts", []) or []:
text = part.get("text")
if text:
- full_text += text
- if on_chunk:
- await on_chunk(text)
+ if part.get("thought") is True:
+ full_reasoning += text
+ if on_thinking:
+ await on_thinking(text)
+ else:
+ full_text += text
+ if on_chunk:
+ await on_chunk(text)
+ signature = part.get("thoughtSignature")
+ if isinstance(signature, str) and signature:
+ thought_signature = signature
function_call = part.get("functionCall")
if function_call:
@@ -1923,6 +1945,8 @@ async def stream(
return LLMResponse(
content=full_text,
tool_calls=tool_calls,
+ reasoning_content=full_reasoning or None,
+ reasoning_signature=thought_signature,
finish_reason=self._normalize_finish_reason(final_finish_reason, tool_calls),
usage=final_usage,
model=self.model,
diff --git a/backend/app/services/llm/failover.py b/backend/app/services/llm/failover.py
index d9b41357d..0cb1f4024 100644
--- a/backend/app/services/llm/failover.py
+++ b/backend/app/services/llm/failover.py
@@ -8,7 +8,7 @@
import re
from enum import Enum
-from .client import LLMError
+from .client import LLMError, LLMVisibleStreamInterrupted
class FailoverErrorType(Enum):
@@ -42,6 +42,9 @@ def classify_error(error: Exception) -> FailoverErrorType:
"""
error_msg = str(error).lower()
+ if isinstance(error, LLMVisibleStreamInterrupted):
+ return FailoverErrorType.NON_RETRYABLE
+
# Non-retryable: an explicit HTTP payment status is deterministic.
if re.search(r"(? None:
+ self._callback = callback
+ self._buffer = ""
+ self._forwarding = False
+ self._held_protocol = False
+ self._blocked_protocol = False
+
+ @staticmethod
+ def _must_hold(value: str) -> bool:
+ probe = value.lstrip().lower()
+ if not probe:
+ return True
+ if probe[0] in "{[":
+ return True
+ return "".startswith(probe) or probe.startswith(" bool:
+ if not delta or self._blocked_protocol:
+ return False
+ self._buffer += delta
+ if not self._forwarding and self._must_hold(self._buffer):
+ self._held_protocol = True
+ return False
+ self._forwarding = True
+ lowered = self._buffer.lower()
+ marker_positions = [
+ position
+ for marker in self._PROTOCOL_MARKERS
+ if (position := lowered.find(marker)) >= 0
+ ]
+ if marker_positions:
+ position = min(marker_positions)
+ safe = self._buffer[:position]
+ self._buffer = self._buffer[position:]
+ self._blocked_protocol = True
+ if safe:
+ await self._callback(safe)
+ return True
+ return False
+ if len(self._buffer) <= self._TAIL_CHARS:
+ return False
+ safe = self._buffer[:-self._TAIL_CHARS]
+ self._buffer = self._buffer[-self._TAIL_CHARS :]
+ await self._callback(safe)
+ return True
+
+ async def finish(
+ self,
+ *,
+ content: str,
+ tool_calls: list[dict],
+ retry_instruction: str | None,
+ ) -> None:
+ if self._blocked_protocol:
+ self._buffer = ""
+ return
+ if (
+ self._held_protocol
+ and not self._forwarding
+ and content
+ and not tool_calls
+ and retry_instruction is None
+ ):
+ await self._callback(content)
+ elif self._forwarding and self._buffer:
+ await self._callback(self._buffer)
+ self._buffer = ""
+
+
@dataclass(frozen=True, slots=True)
class LLMCompletionStep:
"""One normalized provider response with no tool or lifecycle side effects."""
@@ -37,6 +117,7 @@ class LLMCompletionStep:
usage: TokenUsage
retry_tool_name: str | None = None
finish_reason: str | None = None
+ visible_streamed: bool = False
async def complete_llm_once(
@@ -47,6 +128,7 @@ async def complete_llm_once(
agent_id: uuid.UUID | None = None,
supports_vision: bool = False,
max_output_tokens: int | None = None,
+ on_visible_delta: VisibleDeltaCallback | None = None,
) -> LLMCompletionStep:
"""Call one pinned model exactly once and normalize its tool proposals.
@@ -61,21 +143,36 @@ async def complete_llm_once(
base_url=model.base_url,
timeout=_get_model_timeout(model),
)
+ max_tokens = get_max_tokens(
+ model.provider,
+ model.model,
+ (
+ max_output_tokens
+ if max_output_tokens is not None
+ else getattr(model, "max_output_tokens", None)
+ ),
+ )
+ delta_gate = (
+ _VisibleDeltaGate(on_visible_delta)
+ if on_visible_delta and not isinstance(client, OpenAIResponsesClient)
+ else None
+ )
try:
- response = await client.complete(
- messages=api_messages,
- tools=tools or None,
- temperature=model.temperature,
- max_tokens=get_max_tokens(
- model.provider,
- model.model,
- (
- max_output_tokens
- if max_output_tokens is not None
- else getattr(model, "max_output_tokens", None)
- ),
- ),
- )
+ if delta_gate is None:
+ response = await client.complete(
+ messages=api_messages,
+ tools=tools or None,
+ temperature=model.temperature,
+ max_tokens=max_tokens,
+ )
+ else:
+ response = await client.stream(
+ messages=api_messages,
+ tools=tools or None,
+ temperature=model.temperature,
+ max_tokens=max_tokens,
+ on_chunk=delta_gate.push,
+ )
finally:
await client.close()
@@ -105,6 +202,12 @@ async def complete_llm_once(
if textual_retry_instruction is not None:
retry_instruction = textual_retry_instruction
retry_tool_name = None
+ if delta_gate is not None:
+ await delta_gate.finish(
+ content=content or "",
+ tool_calls=list(sanitized_tool_calls or ()),
+ retry_instruction=retry_instruction,
+ )
return LLMCompletionStep(
content=content,
tool_calls=tuple(sanitized_tool_calls or ()),
@@ -119,4 +222,4 @@ async def complete_llm_once(
)
-__all__ = ["LLMCompletionStep", "complete_llm_once"]
+__all__ = ["LLMCompletionStep", "VisibleDeltaCallback", "complete_llm_once"]
diff --git a/backend/tests/test_agent_runtime_answer_stream.py b/backend/tests/test_agent_runtime_answer_stream.py
new file mode 100644
index 000000000..7043e2bba
--- /dev/null
+++ b/backend/tests/test_agent_runtime_answer_stream.py
@@ -0,0 +1,203 @@
+"""Coalesced provisional answer observation tests."""
+
+from __future__ import annotations
+
+import asyncio
+import uuid
+from typing import Self
+
+import pytest
+from sqlalchemy.dialects import postgresql
+
+from app.services.agent_runtime.answer_stream import AnswerStreamWriter
+
+
+class _Transaction:
+ def __init__(self, session: _Session) -> None:
+ self._session = session
+
+ async def __aenter__(self) -> Self:
+ self._session.transaction_entries += 1
+ return self
+
+ async def __aexit__(self, exc_type, exc, traceback) -> bool:
+ self._session.transaction_exits += 1
+ return False
+
+
+class _Session:
+ def __init__(self, *, fail_execute: bool = False) -> None:
+ self.statements: list[object] = []
+ self.transaction_entries = 0
+ self.transaction_exits = 0
+ self.fail_execute = fail_execute
+
+ async def __aenter__(self) -> Self:
+ return self
+
+ async def __aexit__(self, exc_type, exc, traceback) -> bool:
+ return False
+
+ def begin(self) -> _Transaction:
+ return _Transaction(self)
+
+ async def execute(self, statement) -> None:
+ self.statements.append(statement)
+ if self.fail_execute:
+ raise RuntimeError("database unavailable")
+
+
+class _SessionFactory:
+ def __init__(self, *, fail_first: bool = False) -> None:
+ self.sessions: list[_Session] = []
+ self.fail_first = fail_first
+
+ def __call__(self) -> _Session:
+ session = _Session(fail_execute=self.fail_first and not self.sessions)
+ self.sessions.append(session)
+ return session
+
+
+def _params(statement: object) -> dict[str, object]:
+ compiled = statement.compile(dialect=postgresql.dialect())
+ return compiled.params
+
+
+@pytest.mark.asyncio
+async def test_close_coalesces_visible_deltas_into_one_tenant_scoped_event() -> None:
+ tenant_id = uuid.uuid4()
+ run_id = uuid.uuid4()
+ agent_id = uuid.uuid4()
+ attempt_id = uuid.uuid5(run_id, "model-step:2:primary:0")
+ sessions = _SessionFactory()
+ writer = AnswerStreamWriter(
+ session_factory=sessions,
+ tenant_id=tenant_id,
+ run_id=run_id,
+ agent_id=agent_id,
+ attempt_id=attempt_id,
+ flush_interval=60,
+ max_buffer_chars=100,
+ )
+
+ await writer.write("Hello")
+ await writer.write(" world")
+
+ assert sessions.sessions == []
+
+ await writer.close()
+
+ assert len(sessions.sessions) == 1
+ session = sessions.sessions[0]
+ assert session.transaction_entries == session.transaction_exits == 1
+ assert len(session.statements) == 1
+ statement = session.statements[0]
+ params = _params(statement)
+ assert params["tenant_id"] == tenant_id
+ assert params["run_id"] == run_id
+ assert params["agent_id"] == agent_id
+ assert params["event_type"] == "status_changed"
+ assert params["summary"] == "Assistant answer streaming"
+ assert params["payload"] == {
+ "activity_type": "assistant_delta",
+ "status": "running",
+ "attempt_id": str(attempt_id),
+ "sequence": 1,
+ "content": "Hello world",
+ "reset": True,
+ }
+ assert params["artifact_refs"] == []
+ assert params["idempotency_key"] == f"answer-stream:{attempt_id}:1"
+ assert params["source_checkpoint_id"] is None
+ assert params["id"] == uuid.uuid5(
+ run_id,
+ f"answer-stream-event:{attempt_id}:1",
+ )
+ assert "reasoning" not in str(params).lower()
+ assert "tool" not in str(params).lower()
+
+
+@pytest.mark.asyncio
+async def test_size_and_interval_flushes_are_ordered_without_blocking_write() -> None:
+ run_id = uuid.uuid4()
+ attempt_id = uuid.uuid5(run_id, "model-step:1:primary:0")
+ sessions = _SessionFactory()
+ writer = AnswerStreamWriter(
+ session_factory=sessions,
+ tenant_id=uuid.uuid4(),
+ run_id=run_id,
+ agent_id=uuid.uuid4(),
+ attempt_id=attempt_id,
+ flush_interval=0.01,
+ max_buffer_chars=4,
+ )
+
+ await writer.write("ABCD")
+ assert sessions.sessions == []
+ await asyncio.sleep(0.02)
+
+ await writer.write("E")
+ await asyncio.sleep(0.02)
+ await writer.close()
+
+ assert len(sessions.sessions) == 2
+ first = _params(sessions.sessions[0].statements[0])
+ second = _params(sessions.sessions[1].statements[0])
+ assert first["payload"]["content"] == "ABCD"
+ assert first["payload"]["sequence"] == 1
+ assert first["payload"]["reset"] is True
+ assert second["payload"]["content"] == "E"
+ assert second["payload"]["sequence"] == 2
+ assert second["payload"]["reset"] is False
+ assert first["idempotency_key"] == f"answer-stream:{attempt_id}:1"
+ assert second["idempotency_key"] == f"answer-stream:{attempt_id}:2"
+
+
+@pytest.mark.asyncio
+async def test_empty_deltas_are_ignored_and_closed_writer_rejects_more_content() -> None:
+ sessions = _SessionFactory()
+ writer = AnswerStreamWriter(
+ session_factory=sessions,
+ tenant_id=uuid.uuid4(),
+ run_id=uuid.uuid4(),
+ agent_id=uuid.uuid4(),
+ attempt_id=uuid.uuid4(),
+ )
+
+ await writer.write("")
+ await writer.close()
+
+ assert sessions.sessions == []
+ with pytest.raises(RuntimeError, match="closed"):
+ await writer.write("late")
+
+
+@pytest.mark.asyncio
+async def test_failed_flush_retries_the_same_sequence_and_content() -> None:
+ run_id = uuid.uuid4()
+ attempt_id = uuid.uuid4()
+ sessions = _SessionFactory(fail_first=True)
+ writer = AnswerStreamWriter(
+ session_factory=sessions,
+ tenant_id=uuid.uuid4(),
+ run_id=run_id,
+ agent_id=uuid.uuid4(),
+ attempt_id=attempt_id,
+ flush_interval=60,
+ )
+
+ await writer.write("recover me")
+ with pytest.raises(RuntimeError, match="database unavailable"):
+ await writer.flush()
+
+ assert writer.visible_started is False
+ await writer.flush()
+ await writer.close()
+
+ assert len(sessions.sessions) == 2
+ failed = _params(sessions.sessions[0].statements[0])
+ retried = _params(sessions.sessions[1].statements[0])
+ assert failed["id"] == retried["id"]
+ assert failed["idempotency_key"] == retried["idempotency_key"]
+ assert failed["payload"] == retried["payload"]
+ assert writer.visible_started is True
diff --git a/backend/tests/test_agent_runtime_chat_stream.py b/backend/tests/test_agent_runtime_chat_stream.py
index 2613dc96d..00ebc7e97 100644
--- a/backend/tests/test_agent_runtime_chat_stream.py
+++ b/backend/tests/test_agent_runtime_chat_stream.py
@@ -149,7 +149,68 @@ async def send(packet: dict) -> None:
"message_id": str(message.id),
"run_id": str(handle.run_id),
"runtime_status": "completed",
+ "event_id": str(events[-1].event_id),
+ "event_cursor": (
+ f"{events[-1].created_at.isoformat()}|{events[-1].event_id}"
+ ),
+ }
+
+
+@pytest.mark.asyncio
+async def test_answer_delta_maps_attempt_position_without_reasoning() -> None:
+ handle = _handle()
+ event = _event(
+ handle,
+ "status_changed",
+ position=1,
+ payload={
+ "status": "running",
+ "activity_type": "assistant_delta",
+ "attempt_id": "attempt-1",
+ "sequence": 1,
+ "content": " Hello ",
+ "reset": True,
+ },
+ )
+ packets: list[dict] = []
+
+ async def send(packet: dict) -> None:
+ packets.append(packet)
+
+ source = _EventSource([event])
+ source.events.append(
+ _event(
+ handle,
+ "delivery_failed",
+ position=2,
+ payload={
+ "delivery_kind": "terminal",
+ "lifecycle_status": "cancelled",
+ "error_code": "cancelled",
+ },
+ )
+ )
+ await stream_web_chat_run(
+ handle=handle,
+ session_factory=_SessionFactory(), # type: ignore[arg-type]
+ send_packet=send,
+ agent_id=uuid.uuid4(),
+ session_id=uuid.uuid4(),
+ user_id=uuid.uuid4(),
+ event_source=source,
+ )
+
+ assert packets[0] == {
+ "type": "chunk",
+ "content": " Hello ",
+ "run_id": str(handle.run_id),
+ "attempt_id": "attempt-1",
+ "sequence": 1,
+ "reset": True,
+ "event_id": str(event.event_id),
+ "event_cursor": f"{event.created_at.isoformat()}|{event.event_id}",
}
+ assert "reasoning_content" not in packets[0]
@pytest.mark.asyncio
diff --git a/backend/tests/test_agent_runtime_checkpoint_side_effects.py b/backend/tests/test_agent_runtime_checkpoint_side_effects.py
index b8c352d3f..f4e8b976e 100644
--- a/backend/tests/test_agent_runtime_checkpoint_side_effects.py
+++ b/backend/tests/test_agent_runtime_checkpoint_side_effects.py
@@ -292,8 +292,9 @@ async def test_checkpoint_projects_replayable_tool_activity_with_redacted_argume
{
"id": "assistant-1",
"role": "assistant",
- "content": "",
+ "content": "Inspecting the file",
"runtime_run_id": str(run.run_id),
+ "runtime_answer_streamed": True,
"reasoning_content": "Inspect the file",
"tool_calls": [
{
@@ -346,6 +347,10 @@ async def test_checkpoint_projects_replayable_tool_activity_with_redacted_argume
"tool_call",
"tool_call",
]
+ assert all(
+ activity["activity_type"] != "assistant_progress"
+ for activity in activities
+ )
assert activities[1]["status"] == "running"
assert activities[1]["call_instance_id"] == "call-1"
assert activities[1]["provider_call_id"] == "provider-call-1"
diff --git a/backend/tests/test_agent_runtime_config.py b/backend/tests/test_agent_runtime_config.py
index 74f4a2b96..ac90a2dc9 100644
--- a/backend/tests/test_agent_runtime_config.py
+++ b/backend/tests/test_agent_runtime_config.py
@@ -37,6 +37,7 @@ def test_runtime_settings_have_safe_confirmed_defaults() -> None:
assert settings.AGENT_RUNTIME_RUN_COMPACT_TOOL_RESULT_BYTES is None
assert settings.AGENT_RUNTIME_VERIFY_REPAIR_COMPACT_ROUNDS is None
assert settings.AGENT_RUNTIME_MODEL_CAPABILITY_REFRESH_SECONDS == 86400
+ assert settings.AGENT_RUNTIME_WEB_STREAMING_ENABLED is True
assert settings.AGENT_RUNTIME_FALLBACK_CONTEXT_WINDOW_TOKENS == 131072
assert settings.MULTI_AGENT_COMPACT_MODEL_ID is None
assert settings.MULTI_AGENT_PLANNING_MODEL_ID is None
diff --git a/backend/tests/test_agent_runtime_model_step_service.py b/backend/tests/test_agent_runtime_model_step_service.py
index f20bac3cf..82f9e105d 100644
--- a/backend/tests/test_agent_runtime_model_step_service.py
+++ b/backend/tests/test_agent_runtime_model_step_service.py
@@ -531,6 +531,8 @@ def _service(
agent: Agent,
builder: _ContextBuilder,
completion,
+ *,
+ answer_stream_enabled: bool = False,
) -> RuntimeModelStepService:
return RuntimeModelStepService(
session_factory=_session_factory(model, agent),
@@ -540,6 +542,7 @@ def _service(
prompt_builder=_prompt,
model_retry_base_delay_seconds=0,
model_retry_jitter_ratio=0,
+ answer_stream_enabled=answer_stream_enabled,
)
@@ -620,6 +623,8 @@ def _failover_service(
agent: Agent,
builder: _ContextBuilder,
completion,
+ *,
+ answer_stream_enabled: bool = False,
) -> RuntimeModelStepService:
return RuntimeModelStepService(
session_factory=_failover_session_factory(model, agent, fallback),
@@ -629,6 +634,7 @@ def _failover_service(
prompt_builder=_prompt,
model_retry_base_delay_seconds=0,
model_retry_jitter_ratio=0,
+ answer_stream_enabled=answer_stream_enabled,
)
@@ -3096,6 +3102,100 @@ async def complete(model_arg, *args, **kwargs):
assert "runtime_failover_from_model_id" not in result.assistant_message
+@pytest.mark.asyncio
+async def test_visible_stream_failure_never_retries_or_calls_fallback(monkeypatch) -> None:
+ tenant_id = uuid.uuid4()
+ model = _model(tenant_id)
+ fallback = _model(tenant_id)
+ fallback.model = "fallback-model"
+ agent = _agent(tenant_id)
+ agent.fallback_model_id = fallback.id
+ state = _state(tenant_id, model, agent)
+ calls = 0
+
+ class Writer:
+ def __init__(self, **_kwargs) -> None:
+ self.visible_started = False
+
+ async def write(self, _content: str) -> None:
+ self.visible_started = True
+
+ async def close(self) -> None:
+ return None
+
+ monkeypatch.setattr(model_step_service, "AnswerStreamWriter", Writer)
+
+ async def complete(*_args, **kwargs):
+ nonlocal calls
+ calls += 1
+ await kwargs["on_visible_delta"]("partial")
+ raise RuntimeError("connection reset")
+
+ service = _failover_service(
+ model,
+ fallback,
+ agent,
+ _ContextBuilder(_build()),
+ complete,
+ answer_stream_enabled=True,
+ )
+
+ result = await service.complete_once(state, _context(state))
+
+ assert result.intent == "error"
+ assert result.error["code"] == "model_call_failed"
+ assert calls == 1
+
+
+def test_crash_replay_creates_a_fresh_stream_attempt_incarnation() -> None:
+ tenant_id = uuid.uuid4()
+ model = _model(tenant_id)
+ agent = _agent(tenant_id)
+ state = _state(tenant_id, model, agent)
+ context = _context(state)
+ service = _service(
+ model,
+ agent,
+ _ContextBuilder(_build()),
+ AsyncMock(),
+ answer_stream_enabled=True,
+ )
+
+ first = service._answer_stream_writer(
+ state=state,
+ context=context,
+ agent=agent,
+ )
+ replay = service._answer_stream_writer(
+ state=state,
+ context=context,
+ agent=agent,
+ )
+
+ assert first is not None and replay is not None
+ assert first._attempt_id != replay._attempt_id
+
+
+def test_web_answer_stream_can_be_disabled_without_changing_run_state() -> None:
+ tenant_id = uuid.uuid4()
+ model = _model(tenant_id)
+ agent = _agent(tenant_id)
+ state = _state(tenant_id, model, agent)
+ context = _context(state)
+ service = RuntimeModelStepService(
+ session_factory=_session_factory(model, agent),
+ context_builder=_ContextBuilder(_build()), # type: ignore[arg-type]
+ completion=AsyncMock(),
+ answer_stream_enabled=False,
+ )
+
+ assert service._answer_stream_writer(
+ state=state,
+ context=context,
+ agent=agent,
+ ) is None
+
+
@pytest.mark.asyncio
async def test_non_retryable_primary_error_never_calls_configured_fallback() -> None:
tenant_id = uuid.uuid4()
diff --git a/backend/tests/test_llm_failover.py b/backend/tests/test_llm_failover.py
index 317d1c0a5..1211fc930 100644
--- a/backend/tests/test_llm_failover.py
+++ b/backend/tests/test_llm_failover.py
@@ -2,6 +2,7 @@
import pytest
+from app.services.llm.client import LLMVisibleStreamInterrupted
from app.services.llm.failover import (
FailoverErrorType,
classify_error,
@@ -32,6 +33,17 @@ def test_unknown_provider_failure_keeps_retryable_semantics() -> None:
assert is_retryable_classification(classification) is True
+def test_visible_stream_interruption_is_never_retried_or_failed_over() -> None:
+ classification = classify_error(
+ LLMVisibleStreamInterrupted(
+ "Provider stream interrupted after visible output was published"
+ )
+ )
+
+ assert classification is FailoverErrorType.NON_RETRYABLE
+ assert is_retryable_classification(classification) is False
+
+
@pytest.mark.parametrize(
"message",
[
diff --git a/backend/tests/test_llm_single_step.py b/backend/tests/test_llm_single_step.py
index 0325290b3..4ab77171a 100644
--- a/backend/tests/test_llm_single_step.py
+++ b/backend/tests/test_llm_single_step.py
@@ -1,5 +1,6 @@
"""One-call LLM provider boundary tests for the durable Runtime."""
+import asyncio
from types import SimpleNamespace
import uuid
@@ -37,6 +38,16 @@ async def complete(self, **kwargs):
raise self.response
return self.response
+ async def stream(self, **kwargs):
+ self.calls.append(kwargs)
+ if isinstance(self.response, Exception):
+ raise self.response
+ on_chunk = kwargs.get("on_chunk")
+ if on_chunk is not None:
+ await on_chunk("Hello")
+ await on_chunk(" world")
+ return self.response
+
async def close(self) -> None:
self.closed = True
@@ -94,6 +105,141 @@ def _patch_client(monkeypatch, client: _Client) -> None:
monkeypatch.setattr(single_step, "get_max_tokens", lambda *args: 1024)
+@pytest.mark.asyncio
+async def test_visible_delta_callback_uses_provider_stream_and_keeps_final_authority(
+ monkeypatch,
+) -> None:
+ client = _Client(LLMResponse(content="Hello world", finish_reason="stop"))
+ _patch_client(monkeypatch, client)
+ deltas: list[str] = []
+
+ async def collect(delta: str) -> None:
+ deltas.append(delta)
+
+ result = await single_step.complete_llm_once(
+ _model(),
+ [LLMMessage(role="user", content="Say hello")],
+ on_visible_delta=collect,
+ )
+
+ assert "".join(deltas) == "Hello world"
+ assert result.content == "Hello world"
+ assert "on_chunk" in client.calls[0]
+ assert client.closed is True
+
+
+@pytest.mark.asyncio
+async def test_visible_delta_arrives_before_provider_completion(monkeypatch) -> None:
+ response = LLMResponse(content="A sufficiently long streamed answer", finish_reason="stop")
+ client = _Client(response)
+ delta_seen = asyncio.Event()
+ release_provider = asyncio.Event()
+
+ async def blocked_stream(**kwargs):
+ await kwargs["on_chunk"]("A sufficiently long streamed answer")
+ await release_provider.wait()
+ return response
+
+ client.stream = blocked_stream
+ _patch_client(monkeypatch, client)
+
+ async def collect(_delta: str) -> None:
+ delta_seen.set()
+
+ completion = asyncio.create_task(
+ single_step.complete_llm_once(
+ _model(),
+ [LLMMessage(role="user", content="Stream")],
+ on_visible_delta=collect,
+ )
+ )
+ await asyncio.wait_for(delta_seen.wait(), timeout=1)
+
+ assert completion.done() is False
+ release_provider.set()
+ result = await completion
+ assert result.content == response.content
+
+
+@pytest.mark.asyncio
+async def test_protocol_looking_stream_is_held_until_final_normalization(monkeypatch) -> None:
+ response = LLMResponse(
+ content='{"name":"read_file","arguments":{"path":"README.md"}}',
+ finish_reason="stop",
+ )
+ client = _Client(response)
+ published: list[bool | None] = []
+
+ async def protocol_stream(**kwargs):
+ client.calls.append(kwargs)
+ published.append(await kwargs["on_chunk"](""))
+ published.append(
+ await kwargs["on_chunk"](
+ '{"name":"read_file","arguments":{"path":"README.md"}}'
+ )
+ )
+ published.append(await kwargs["on_chunk"](""))
+ return response
+
+ client.stream = protocol_stream
+ _patch_client(monkeypatch, client)
+ deltas: list[str] = []
+
+ async def collect(delta: str) -> None:
+ deltas.append(delta)
+
+ result = await single_step.complete_llm_once(
+ _model(),
+ [LLMMessage(role="user", content="Read it")],
+ tools=[{"type": "function", "function": {"name": "read_file", "parameters": {"type": "object"}}}],
+ on_visible_delta=collect,
+ )
+
+ assert deltas == []
+ assert published == [False, False, False]
+ assert result.content == ""
+ assert result.tool_calls[0]["function"]["name"] == "read_file"
+
+
+@pytest.mark.asyncio
+async def test_mixed_textual_tool_protocol_never_streams_marker_or_arguments(monkeypatch) -> None:
+ response = LLMResponse(
+ content=(
+ 'Let me check.\n{"name":"read_file",'
+ '"arguments":{"path":"private.md"}}'
+ ),
+ finish_reason="stop",
+ )
+ client = _Client(response)
+
+ async def mixed_stream(**kwargs):
+ await kwargs["on_chunk"]("Let me check.\n{"name":"read_file","arguments":{"path":"private.md"}}'
+ )
+ return response
+
+ client.stream = mixed_stream
+ _patch_client(monkeypatch, client)
+ deltas: list[str] = []
+
+ async def collect(delta: str) -> None:
+ deltas.append(delta)
+
+ result = await single_step.complete_llm_once(
+ _model(),
+ [LLMMessage(role="user", content="Read it")],
+ tools=[{"type": "function", "function": {"name": "read_file", "parameters": {"type": "object"}}}],
+ on_visible_delta=collect,
+ )
+
+ streamed = "".join(deltas)
+ assert streamed == "Let me check.\n"
+ assert "tool_call" not in streamed
+ assert "private.md" not in streamed
+ assert result.retry_instruction is not None
+
+
def test_native_gemini_preserves_dynamic_system_context_once() -> None:
client = GeminiClient(api_key="test", model="gemini-test")
diff --git a/frontend/src/pages/agent-detail/AgentDetailPage.tsx b/frontend/src/pages/agent-detail/AgentDetailPage.tsx
index 953f08fe5..9edf764eb 100644
--- a/frontend/src/pages/agent-detail/AgentDetailPage.tsx
+++ b/frontend/src/pages/agent-detail/AgentDetailPage.tsx
@@ -67,7 +67,10 @@ import {
failClosedSessionActiveRun,
mergeSessionToolMessage,
mergeSessionToolMessages,
+ mergeInterruptedStreamMessage,
mergeTerminalAssistantMessage,
+ reduceSessionStreamChunk,
+ shouldPreserveInterruptedStream,
runtimeCompletionNeedsMessageRefresh,
runtimeTerminalPacketNeedsMessageRefresh,
sessionActiveRunFromResponse,
@@ -2462,6 +2465,7 @@ export default function AgentDetailPage() {
const currentAgentIdRef = useRef(id);
const sessionMsgAbortRef = useRef(null);
const sessionLoadSeqRef = useRef(0);
+ const interruptedStreamMessagesRef = useRef>({});
const buildSessionRuntimeKey = (agentId: string, sessionId: string) => `${agentId}:${sessionId}`;
@@ -2502,7 +2506,10 @@ export default function AgentDetailPage() {
}));
const runtimeKey = buildSessionRuntimeKey(agentId, sessionId);
setChatMessages(mergeSessionToolMessages(
- parsed,
+ mergeInterruptedStreamMessage(
+ parsed,
+ interruptedStreamMessagesRef.current[runtimeKey],
+ ),
sessionToolMessagesRef.current[runtimeKey] || [],
));
if (discardSessionToolCacheOnSuccess) {
@@ -2944,7 +2951,7 @@ export default function AgentDetailPage() {
} catch (e: any) { toast.error(t('common.error.saveFailed', '保存失败'), { details: String(e?.message || e) }); }
setExpirySaving(false);
};
- interface ChatMsg { id?: string; role: 'user' | 'assistant' | 'tool_call'; content: string; fileName?: string; toolName?: string; toolCallId?: string; toolArgs?: any; toolStatus?: 'running' | 'done'; toolResult?: string; toolThinking?: string; thinking?: string; imageUrl?: string; timestamp?: string; runtimeError?: ReturnType; }
+ interface ChatMsg { id?: string; role: 'user' | 'assistant' | 'tool_call'; content: string; fileName?: string; toolName?: string; toolCallId?: string; toolArgs?: any; toolStatus?: 'running' | 'done'; toolResult?: string; toolThinking?: string; thinking?: string; imageUrl?: string; timestamp?: string; runtimeError?: ReturnType; _streaming?: boolean; _streamRunId?: string; _streamAttemptId?: string; _streamSequence?: number; }
const [chatMessages, setChatMessages] = useState([]);
const upsertToolCallMessage = (toolMsg: ChatMsg) => {
setChatMessages((previous) => mergeSessionToolMessage(previous, toolMsg));
@@ -3301,6 +3308,7 @@ export default function AgentDetailPage() {
wsMapRef.current = {};
sessionActiveRunRef.current = {};
sessionToolMessagesRef.current = {};
+ interruptedStreamMessagesRef.current = {};
wsRef.current = null;
}, [currentUser?.id, token]);
@@ -3385,7 +3393,10 @@ export default function AgentDetailPage() {
};
ws.onmessage = (e) => {
const d = JSON.parse(e.data);
- if (typeof d.event_cursor === 'string' && d.event_cursor && d.run_id) {
+ const positionedAnswerChunk = d.type === 'chunk'
+ && typeof d.attempt_id === 'string'
+ && Number.isInteger(d.sequence);
+ if (!positionedAnswerChunk && typeof d.event_cursor === 'string' && d.event_cursor && d.run_id) {
runtimeEventCursorRef.current[`${key}:${String(d.run_id)}`] = d.event_cursor;
}
// A completed or already-running pair-scoped onboarding attempt
@@ -3616,11 +3627,40 @@ export default function AgentDetailPage() {
} else if (d.type === 'chunk') {
setChatMessages(prev => {
const last = prev[prev.length - 1];
- if (last && last.role === 'assistant' && (last as any)._streaming) return [...prev.slice(0, -1), { ...last, content: last.content + d.content } as any];
- return [...prev, { role: 'assistant', content: d.content, _streaming: true } as any];
+ const current = last?.role === 'assistant' && last._streaming ? {
+ content: last.content,
+ runId: last._streamRunId,
+ attemptId: last._streamAttemptId,
+ sequence: last._streamSequence,
+ } : null;
+ const next = reduceSessionStreamChunk(current, d);
+ if (next === null) return prev;
+ if (current && next === current) return prev;
+ if (typeof d.event_cursor === 'string' && d.event_cursor && d.run_id) {
+ runtimeEventCursorRef.current[`${key}:${String(d.run_id)}`] = d.event_cursor;
+ }
+ const streamedAssistant: ChatMsg = {
+ ...(current ? last : {}),
+ role: 'assistant',
+ content: next.content,
+ _streaming: true,
+ _streamRunId: next.runId,
+ _streamAttemptId: next.attemptId,
+ _streamSequence: next.sequence,
+ };
+ return current
+ ? [...prev.slice(0, -1), streamedAssistant]
+ : [...prev, streamedAssistant];
});
} else if (d.type === 'done') {
const shouldRefreshCanonicalMessages = runtimeTerminalPacketNeedsMessageRefresh(d.runtime_status);
+ const preserveInterrupted = shouldPreserveInterruptedStream(
+ d.runtime_status,
+ d.delivery_error,
+ );
+ if (!preserveInterrupted) {
+ delete interruptedStreamMessagesRef.current[key];
+ }
if (shouldRefreshCanonicalMessages) {
const existingRun = sessionActiveRunRef.current[key];
if (existingRun && d.run_id && existingRun.runId === String(d.run_id)) {
@@ -3655,7 +3695,22 @@ export default function AgentDetailPage() {
thinking,
timestamp: new Date().toISOString(),
});
- if (last && last.role === 'assistant' && (last as any)._streaming) return [...prev.slice(0, -1), terminalMessage];
+ if (last && last.role === 'assistant' && last._streaming) {
+ if (preserveInterrupted) {
+ const interrupted = {
+ ...last,
+ _streaming: false,
+ ...(runtimeError && { runtimeError }),
+ };
+ interruptedStreamMessagesRef.current[key] = interrupted;
+ return [
+ ...prev.slice(0, -1),
+ interrupted,
+ terminalMessage,
+ ];
+ }
+ return [...prev.slice(0, -1), terminalMessage];
+ }
// Runtime-state polling can observe the committed terminal
// message before its websocket `done` packet arrives. In
// that ordering, refreshSessionMessages already installed
@@ -3743,6 +3798,9 @@ export default function AgentDetailPage() {
currentAgentIdRef.current === runtimeAgentId
&& activeSessionIdRef.current === runtimeSessionId
);
+ // An interrupted partial belongs only to the Run that produced it. A
+ // later non-streaming Run must never inherit it during canonical refresh.
+ delete interruptedStreamMessagesRef.current[runtimeKey];
setSessionUiState(runtimeKey, { isWaiting: true, isStreaming: false });
if (payload.resumeRunId) {
const current = sessionActiveRunRef.current[runtimeKey];
diff --git a/frontend/src/pages/agent-detail/sessionRuntimeState.ts b/frontend/src/pages/agent-detail/sessionRuntimeState.ts
index b6893f350..1c8537b45 100644
--- a/frontend/src/pages/agent-detail/sessionRuntimeState.ts
+++ b/frontend/src/pages/agent-detail/sessionRuntimeState.ts
@@ -130,6 +130,91 @@ const requiredText = (value: unknown): string | null => {
const optionalText = (value: unknown): string | null =>
value == null ? null : requiredText(value);
+export interface SessionStreamChunkPacket {
+ run_id?: unknown;
+ attempt_id?: unknown;
+ sequence?: unknown;
+ content?: unknown;
+ reset?: unknown;
+}
+
+export interface SessionStreamChunkState {
+ content: string;
+ runId?: string;
+ attemptId?: string;
+ sequence?: number;
+}
+
+export const reduceSessionStreamChunk = (
+ current: SessionStreamChunkState | null,
+ packet: SessionStreamChunkPacket,
+): SessionStreamChunkState | null => {
+ const content = typeof packet.content === 'string' ? packet.content : '';
+ const runId = requiredText(packet.run_id);
+ const attemptId = requiredText(packet.attempt_id);
+ const sequence = packet.sequence;
+ const hasAttemptMetadata = runId !== null
+ && attemptId !== null
+ && typeof sequence === 'number'
+ && Number.isInteger(sequence)
+ && sequence > 0;
+
+ if (!hasAttemptMetadata) {
+ return {
+ ...current,
+ content: `${current?.content || ''}${content}`,
+ };
+ }
+
+ const sameAttempt = current?.runId === runId && current.attemptId === attemptId;
+ if (!sameAttempt) {
+ if (sequence !== 1 && packet.reset !== true) return current;
+ return { content, runId, attemptId, sequence };
+ }
+
+ const previousSequence = current.sequence;
+ if (previousSequence !== undefined) {
+ if (sequence <= previousSequence || sequence !== previousSequence + 1) return current;
+ }
+
+ return {
+ content: packet.reset === true || sequence === 1
+ ? content
+ : `${current.content}${content}`,
+ runId,
+ attemptId,
+ sequence,
+ };
+};
+
+export const shouldPreserveInterruptedStream = (
+ runtimeStatus: unknown,
+ deliveryError: unknown,
+): boolean => runtimeStatus === 'failed'
+ || runtimeStatus === 'cancelled'
+ || requiredText(deliveryError) !== null;
+
+export const mergeInterruptedStreamMessage = (
+ messages: T[],
+ interrupted: T | undefined,
+): T[] => {
+ if (!interrupted?.content) return messages;
+ if (messages.some(message => message === interrupted)) return messages;
+ let terminalAssistantIndex = -1;
+ for (let index = messages.length - 1; index >= 0; index -= 1) {
+ if (messages[index].role === 'assistant') {
+ terminalAssistantIndex = index;
+ break;
+ }
+ }
+ if (terminalAssistantIndex < 0) return [...messages, interrupted];
+ return [
+ ...messages.slice(0, terminalAssistantIndex),
+ interrupted,
+ ...messages.slice(terminalAssistantIndex),
+ ];
+};
+
const TOOL_RESOLUTION_STATUSES = new Set([
'checking',
'saved',
diff --git a/frontend/tests/sessionRuntimeState.test.mjs b/frontend/tests/sessionRuntimeState.test.mjs
index 4cbd103d2..6c18999fc 100644
--- a/frontend/tests/sessionRuntimeState.test.mjs
+++ b/frontend/tests/sessionRuntimeState.test.mjs
@@ -7,6 +7,9 @@ import {
failClosedSessionActiveRun,
mergeSessionToolMessage,
mergeSessionToolMessages,
+ mergeInterruptedStreamMessage,
+ reduceSessionStreamChunk,
+ shouldPreserveInterruptedStream,
runtimeCompletionNeedsMessageRefresh,
runtimeTerminalPacketNeedsMessageRefresh,
sessionActiveRunFromResponse,
@@ -43,6 +46,147 @@ test('active run controls are projected only onto their own selected session', (
assert.equal(activeRunForSession(waitingRun, 'session-1'), waitingRun);
});
+test('answer stream starts and resets provisional content by attempt', () => {
+ const firstAttempt = reduceSessionStreamChunk(null, {
+ run_id: 'run-1',
+ attempt_id: 'attempt-1',
+ sequence: 1,
+ content: 'first',
+ reset: true,
+ });
+ assert.deepEqual(firstAttempt, {
+ content: 'first',
+ runId: 'run-1',
+ attemptId: 'attempt-1',
+ sequence: 1,
+ });
+
+ assert.deepEqual(reduceSessionStreamChunk(firstAttempt, {
+ run_id: 'run-1',
+ attempt_id: 'attempt-2',
+ sequence: 1,
+ content: 'replacement',
+ reset: false,
+ }), {
+ content: 'replacement',
+ runId: 'run-1',
+ attemptId: 'attempt-2',
+ sequence: 1,
+ });
+
+ assert.deepEqual(reduceSessionStreamChunk(firstAttempt, {
+ run_id: 'run-1',
+ attempt_id: 'attempt-1',
+ sequence: 2,
+ content: 'explicit reset',
+ reset: true,
+ }), {
+ content: 'explicit reset',
+ runId: 'run-1',
+ attemptId: 'attempt-1',
+ sequence: 2,
+ });
+});
+
+test('answer stream appends only contiguous packets from the active attempt', () => {
+ const first = reduceSessionStreamChunk(null, {
+ run_id: 'run-1',
+ attempt_id: 'attempt-1',
+ sequence: 1,
+ content: 'one',
+ reset: true,
+ });
+ const second = reduceSessionStreamChunk(first, {
+ run_id: 'run-1',
+ attempt_id: 'attempt-1',
+ sequence: 2,
+ content: ' two',
+ reset: false,
+ });
+ assert.equal(second.content, 'one two');
+ assert.equal(second.sequence, 2);
+
+ const gap = reduceSessionStreamChunk(second, {
+ run_id: 'run-1',
+ attempt_id: 'attempt-1',
+ sequence: 4,
+ content: ' four',
+ reset: false,
+ });
+ assert.equal(gap, second);
+
+ const replay = reduceSessionStreamChunk(gap, {
+ run_id: 'run-1',
+ attempt_id: 'attempt-1',
+ sequence: 3,
+ content: ' three',
+ reset: false,
+ });
+ assert.equal(replay.content, 'one two three');
+ assert.equal(replay.sequence, 3);
+ assert.equal(reduceSessionStreamChunk(replay, {
+ run_id: 'run-1',
+ attempt_id: 'attempt-1',
+ sequence: 2,
+ content: ' duplicate',
+ reset: false,
+ }), replay);
+});
+
+test('answer stream rejects a new attempt that starts after sequence one', () => {
+ assert.equal(reduceSessionStreamChunk(null, {
+ run_id: 'run-1',
+ attempt_id: 'attempt-1',
+ sequence: 2,
+ content: 'missing prefix',
+ reset: false,
+ }), null);
+});
+
+test('failed cancelled and delivery-failed terminals preserve provisional output', () => {
+ assert.equal(shouldPreserveInterruptedStream('failed', null), true);
+ assert.equal(shouldPreserveInterruptedStream('cancelled', null), true);
+ assert.equal(shouldPreserveInterruptedStream('completed', 'delivery_failed'), true);
+ assert.equal(shouldPreserveInterruptedStream('completed', null), false);
+ assert.equal(shouldPreserveInterruptedStream('waiting_user', null), false);
+});
+
+test('canonical refresh retains interrupted partial immediately before terminal answer', () => {
+ const partial = { role: 'assistant', content: 'useful partial', _streaming: false };
+ const terminal = { role: 'assistant', content: 'provider failed', runtimeError: { code: 'failed' } };
+ assert.deepEqual(
+ mergeInterruptedStreamMessage(
+ [{ role: 'user', content: 'work' }, terminal],
+ partial,
+ ),
+ [{ role: 'user', content: 'work' }, partial, terminal],
+ );
+});
+
+test('starting a later run clears the prior interrupted stream cache', () => {
+ assert.match(
+ agentDetailSource,
+ /const dispatchChatMessage[\s\S]*delete interruptedStreamMessagesRef\.current\[runtimeKey\]/,
+ );
+});
+
+test('legacy answer chunks remain append-compatible', () => {
+ const first = reduceSessionStreamChunk(null, { content: 'legacy' });
+ assert.deepEqual(first, { content: 'legacy' });
+ assert.deepEqual(reduceSessionStreamChunk(first, { content: ' stream' }), {
+ content: 'legacy stream',
+ });
+});
+
+test('agent detail chunk handler uses attempt-aware reducer while done stays canonical', () => {
+ assert.match(agentDetailSource, /reduceSessionStreamChunk/);
+ assert.match(agentDetailSource, /else if \(d\.type === 'chunk'\)[\s\S]*reduceSessionStreamChunk/);
+ assert.match(
+ agentDetailSource,
+ /else if \(d\.type === 'done'\)[\s\S]*prev\.slice\(0, -1\), terminalMessage/,
+ );
+});
+
test('session Tool cache restores a running card after switching back', () => {
const running = {
role: 'tool_call',