From 248eeff8d0295dc0a0d8afbc89769e528cbade7b Mon Sep 17 00:00:00 2001 From: Y1fe1Zh0u Date: Fri, 21 Aug 2026 13:46:21 +0800 Subject: [PATCH 1/2] Restore truthful answer streaming without weakening Runtime authority Connect existing provider streaming callbacks to tenant-scoped, coalesced Runtime observations and the existing Web Chat chunk consumer. Attempt identity, ordered replay, protocol filtering, canonical terminal replacement, and a default-off rollout gate keep provisional output separate from execution and delivery truth. Constraint: LangGraph checkpoint and delivered ChatMessage remain the only lifecycle and final-result authorities Constraint: No new dependencies or database migration Rejected: Write every token into checkpoint state | violates Runtime ownership and amplifies persistence cost Rejected: Push provider callbacks directly to WebSocket | couples Worker progress to one process and client Rejected: Retry or fail over after visible output | can splice incompatible provider attempts Confidence: high Scope-risk: moderate Directive: Keep AGENT_RUNTIME_WEB_STREAMING_ENABLED disabled for fleet rollout until real-provider concurrency and database load are measured Tested: Backend 2557 passed; focused 245 passed; frontend 122 passed; production build; scoped Ruff; Architecture Guard P0; git diff --check Not-tested: Real provider TTFT, multi-worker crash canary, enabled-path load, 3010 deployment --- backend/app/config.py | 1 + .../services/agent_runtime/answer_stream.py | 166 ++++++++++++++ .../app/services/agent_runtime/chat_stream.py | 35 ++- .../agent_runtime/checkpoint_side_effects.py | 6 +- .../agent_runtime/model_step_service.py | 85 +++++++- .../agent_runtime/tool_step_service.py | 9 +- .../services/agent_runtime/worker_service.py | 1 + backend/app/services/llm/client.py | 34 ++- backend/app/services/llm/failover.py | 5 +- backend/app/services/llm/single_step.py | 133 ++++++++++-- .../tests/test_agent_runtime_answer_stream.py | 203 ++++++++++++++++++ .../tests/test_agent_runtime_chat_stream.py | 61 ++++++ ...t_agent_runtime_checkpoint_side_effects.py | 7 +- .../test_agent_runtime_model_step_service.py | 100 +++++++++ backend/tests/test_llm_failover.py | 12 ++ backend/tests/test_llm_single_step.py | 146 +++++++++++++ .../pages/agent-detail/AgentDetailPage.tsx | 70 +++++- .../pages/agent-detail/sessionRuntimeState.ts | 85 ++++++++ frontend/tests/sessionRuntimeState.test.mjs | 144 +++++++++++++ 19 files changed, 1267 insertions(+), 36 deletions(-) create mode 100644 backend/app/services/agent_runtime/answer_stream.py create mode 100644 backend/tests/test_agent_runtime_answer_stream.py diff --git a/backend/app/config.py b/backend/app/config.py index 48a2220c1..9dcbd96ad 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 = False 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_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', From 76921c361258fd9bd71f67ead5fb6f9c475dd7f0 Mon Sep 17 00:00:00 2001 From: Y1fe1Zh0u Date: Fri, 21 Aug 2026 13:49:58 +0800 Subject: [PATCH 2/2] Make restored Web streaming active in v1.11.4 Enable the reviewed Runtime streaming path by default while preserving AGENT_RUNTIME_WEB_STREAMING_ENABLED=false as an operational rollback. Constraint: User explicitly selected default-on behavior and approved backporting the validated change to v1.11.4 Confidence: high Scope-risk: moderate Directive: Keep the false override available until broader real-provider concurrency and database load are validated Tested: Backend 2557 passed; frontend 122 passed; production build; scoped Ruff; Architecture Guard P0; git diff --check Not-tested: Additional v1.11.4 deployment beyond the existing 3010 canary --- backend/app/config.py | 2 +- backend/tests/test_agent_runtime_config.py | 1 + 2 files changed, 2 insertions(+), 1 deletion(-) diff --git a/backend/app/config.py b/backend/app/config.py index 9dcbd96ad..e9e154d5e 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -159,7 +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 = False + 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/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