Restore truthful Web answer streaming on v1.11.4

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

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

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 = (

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",