diff --git a/application/api/answer/routes/base.py b/application/api/answer/routes/base.py index 380b6f13..df68bae1 100644 --- a/application/api/answer/routes/base.py +++ b/application/api/answer/routes/base.py @@ -246,6 +246,14 @@ class BaseAnswerResource: structured_chunks = [] query_metadata: Dict[str, Any] = {} paused = False + # Set when the agent *yields* a terminal ``error`` event instead of + # raising. Workflow node failures take that route (the engine catches + # the node exception and reports it as an event), so the generator + # returns normally and the ``except`` handler below never runs. Without + # this flag the turn was finalized ``complete`` with an empty response: + # the live client showed an error bubble, but on reload history mapped + # the row to a blank answer with no error text and no retry. + stream_error: Optional[str] = None # A ``tool_calls_pending`` event is held back and only flushed after # continuation state is committed (or the stateless finalize path is # reached): the v1 translator turns it into ``finish_reason:"tool_calls"``, @@ -568,6 +576,7 @@ class BaseAnswerResource: error_text = line.get("error", "An error occurred") if not line.get("user_facing"): error_text = sanitize_api_error(error_text) + stream_error = error_text yield _emit({"type": "error", "error": error_text}) elif line.get("type") == "notice": # Non-fatal, non-terminal notice (e.g. some workflow input @@ -595,6 +604,14 @@ class BaseAnswerResource: } ) + # Record a yielded error before any early return so the pause / + # stateless-tool-round paths persist it too. No producer currently + # emits a non-terminal error and then pauses, but leaving the only + # write below the pause blocks would make that combination lose the + # error silently — the exact shape of the bug being fixed here. + if stream_error: + query_metadata.setdefault("error", stream_error) + # ---- Paused: save continuation state and end stream early ---- if paused: continuation = getattr(agent, "_pending_continuation", None) @@ -848,6 +865,19 @@ class BaseAnswerResource: ) llm._token_usage_source = "title" + # The error was recorded above so the failure stays greppable, but + # it only *fails* the turn when nothing was produced. An error + # arriving after partial output (e.g. a later workflow node) must + # stay ``complete``, since the client only renders ``response`` for + # complete rows — failing it would discard text the user already + # saw. ``structured_chunks`` counts as output for the same reason: + # a structured answer lives there, not in ``response_full``. + errored_empty = ( + bool(stream_error) + and not response_full.strip() + and not structured_chunks + ) + if should_persist: if reserved_message_id is not None: self.conversation_service.finalize_message( @@ -858,7 +888,7 @@ class BaseAnswerResource: tool_calls=tool_calls, model_id=model_id or self.default_model_id, metadata=query_metadata if query_metadata else None, - status="complete", + status="failed" if errored_empty else "complete", title_inputs={ "llm": llm, "question": question, @@ -889,6 +919,7 @@ class BaseAnswerResource: attachment_ids=attachment_ids, metadata=query_metadata if query_metadata else None, visibility=visibility, + status="failed" if errored_empty else "complete", ) # Persist compression metadata/summary if it exists and wasn't saved mid-execution compression_meta = getattr(agent, "compression_metadata", None) diff --git a/application/api/answer/services/conversation_service.py b/application/api/answer/services/conversation_service.py index 97f6d54b..d89b503a 100644 --- a/application/api/answer/services/conversation_service.py +++ b/application/api/answer/services/conversation_service.py @@ -100,9 +100,15 @@ class ConversationService: attachment_ids: Optional[List[str]] = None, metadata: Optional[Dict[str, Any]] = None, visibility: str = "hidden", + status: str = "complete", ) -> str: """Save or update a conversation in Postgres. + ``status`` lets a caller record a turn that failed without producing + an answer. It defaults to ``complete``, matching the column default + this path relied on before; passing ``failed`` is what stops a blank + errored turn from rendering as an empty bubble with no retry. + Returns the string conversation id (PG UUID as string, or the caller-provided id if it was already a UUID). """ @@ -127,6 +133,7 @@ class ConversationService: "attachments": attachment_ids, "model_id": model_id, "timestamp": current_time, + "status": status, } if metadata: message_payload["metadata"] = metadata @@ -410,7 +417,12 @@ class ConversationService: repo.confirm_executed_tool_calls(message_id) # Outside the txn — title-gen is a multi-second LLM round trip. - if title_inputs and status == "complete": + # ``failed`` counts too: the conversation is still listed, and + # ``_maybe_generate_title`` only regenerates while the name is still + # the question-prefix fallback. Skipping it here would strand a + # conversation whose first turn failed with the raw prompt as its + # name forever, because by turn two the fallback no longer matches. + if title_inputs and status in ("complete", "failed"): if async_title_generation: threading.Thread( target=self._generate_title_safely, diff --git a/application/logging.py b/application/logging.py index 5c03f2a2..a244dde1 100644 --- a/application/logging.py +++ b/application/logging.py @@ -34,6 +34,12 @@ class LogContext: self.thought_length = 0 self.source_count = 0 self.tool_call_count = 0 + # Terminal ``error`` events are *yielded*, not raised (workflow node + # failures, agent-reported errors), so they never reach the decorator's + # ``except``. Recording one here keeps ``activity_finished`` from + # reporting ``status="ok"`` on a turn the user saw fail — that gap is + # why blank-answer incidents did not show up in error dashboards. + self.stream_error: str | None = None def build_stack_data( @@ -174,8 +180,12 @@ def _emit_activity_finished( "user_id": context.user, "endpoint": context.endpoint, "duration_ms": duration_ms, - "status": "error" if error is not None else "ok", - "error_class": type(error).__name__ if error is not None else None, + "status": "error" if (error is not None or context.stream_error) else "ok", + "error_class": ( + type(error).__name__ + if error is not None + else ("StreamError" if context.stream_error else None) + ), "answer_length": context.answer_length, "thought_length": context.thought_length, "source_count": context.source_count, @@ -192,6 +202,11 @@ def _accumulate_response_summary(item: Any, context: "LogContext") -> None: """ if not isinstance(item, dict): return + if item.get("type") == "error": + # Fall back to a sentinel: an error event carrying no message would + # otherwise store "" and read as falsy, reporting the activity "ok". + context.stream_error = str(item.get("error") or "")[:200] or "unspecified" + return if "answer" in item: context.answer_length += len(str(item["answer"])) return diff --git a/application/storage/db/repositories/conversations.py b/application/storage/db/repositories/conversations.py index 97f8c56f..a9e46825 100644 --- a/application/storage/db/repositories/conversations.py +++ b/application/storage/db/repositories/conversations.py @@ -720,6 +720,11 @@ class ConversationsRepository: "model_id": message.get("model_id"), "message_metadata": message.get("metadata") or {}, } + # Callers that know the turn failed (e.g. an agent that yielded a + # terminal error) must be able to say so; without this the column + # default silently made every appended row "complete". + if message.get("status") is not None: + values["status"] = message["status"] if message.get("timestamp") is not None: values["timestamp"] = message["timestamp"] @@ -758,7 +763,7 @@ class ConversationsRepository: """ allowed = { "prompt", "response", "thought", "sources", "tool_calls", - "attachments", "model_id", "metadata", "timestamp", + "attachments", "model_id", "metadata", "timestamp", "status", # Feedback can be re-set in rare continuation flows; without # it in the whitelist an upstream re-append that happens to # carry feedback would silently lose it. Mirrors diff --git a/tests/api/answer/test_base_routes.py b/tests/api/answer/test_base_routes.py index 7b6fdaaa..dac6b2c4 100644 --- a/tests/api/answer/test_base_routes.py +++ b/tests/api/answer/test_base_routes.py @@ -909,6 +909,15 @@ def _patch_db_session(conn): # uncommitted writes from this transaction. "application.api.answer.routes.base.db_readonly", _yield, + ), patch( + # The terminal ``stream_answer`` user_logs write opens its own + # ``db_session``. Left unpatched it is a *second* connection that + # blocks on the uncommitted ``users`` row this transaction just + # inserted (via the ``ensure_user_exists`` trigger) until the + # statement timeout fires — ~30s per test, swallowed by the + # caller's except, so it only ever showed up as slowness. + "application.api.answer.routes.base.db_session", + _yield, ): yield @@ -972,6 +981,161 @@ class TestCompleteStreamWalAcceptance: assert "RuntimeError" in msgs[0]["metadata"]["error"] assert "LLM upstream failed" in msgs[0]["metadata"]["error"] + def test_workflow_node_error_persists_as_failed_not_blank_complete( + self, pg_conn, flask_app, + ): + """A workflow node failure must not land as a blank ``complete`` row. + + The engine reports node failures by *yielding* ``{"type": "error"}`` + rather than raising, so the generator returns normally and the turn + used to be finalized ``complete`` with an empty response. Live, the + client shows an error bubble; on reload the row mapped to an empty + answer with no error and no retry affordance — the user saw a blank + message and re-sent the prompt. Reproduces the 2026-08-01 report. + """ + from application.api.answer.routes.base import BaseAnswerResource + from application.storage.db.repositories.conversations import ( + ConversationsRepository, + ) + + with flask_app.app_context(): + resource = BaseAnswerResource() + + mock_agent = MagicMock() + mock_agent.gen.return_value = iter( + [{"type": "error", "error": "No LLM class found for type foundry"}] + ) + + with _patch_db_session(pg_conn): + stream = list( + resource.complete_stream( + question="hello", + agent=mock_agent, + conversation_id=None, + user_api_key=None, + decoded_token={"sub": "u-wf-error"}, + should_persist=True, + model_id="gpt-4", + ) + ) + assert len([s for s in stream if '"type": "error"' in s]) == 1 + + from sqlalchemy import text as sql_text + convs = pg_conn.execute( + sql_text("SELECT id FROM conversations WHERE user_id = :u"), + {"u": "u-wf-error"}, + ).fetchall() + assert len(convs) == 1 + msgs = ConversationsRepository(pg_conn).get_messages(str(convs[0][0])) + assert len(msgs) == 1 + assert msgs[0]["prompt"] == "hello" + assert msgs[0]["status"] == "failed", ( + "a turn that produced no answer and emitted an error must be " + "failed, so history renders the error and a retry button" + ) + assert "foundry" in msgs[0]["metadata"]["error"] + + def test_error_after_partial_answer_keeps_the_answer( + self, pg_conn, flask_app, + ): + """A late error must not discard text the user already received. + + Only the *blank* turn is a failure. If a workflow produced output and + then a downstream node failed, the row stays ``complete`` so the + partial answer still renders; the error was already surfaced live. + """ + from application.api.answer.routes.base import BaseAnswerResource + from application.storage.db.repositories.conversations import ( + ConversationsRepository, + ) + + with flask_app.app_context(): + resource = BaseAnswerResource() + + mock_agent = MagicMock() + mock_agent.gen.return_value = iter( + [ + {"answer": "partial result"}, + {"type": "error", "error": "node 2 blew up"}, + ] + ) + + with _patch_db_session(pg_conn): + list( + resource.complete_stream( + question="run the workflow", + agent=mock_agent, + conversation_id=None, + user_api_key=None, + decoded_token={"sub": "u-wf-partial"}, + should_persist=True, + model_id="gpt-4", + ) + ) + + from sqlalchemy import text as sql_text + convs = pg_conn.execute( + sql_text("SELECT id FROM conversations WHERE user_id = :u"), + {"u": "u-wf-partial"}, + ).fetchall() + msgs = ConversationsRepository(pg_conn).get_messages(str(convs[0][0])) + assert msgs[0]["status"] == "complete" + assert msgs[0]["response"] == "partial result" + # Still recorded, so the failure is greppable in review queries. + assert "node 2 blew up" in msgs[0]["metadata"]["error"] + + def test_workflow_error_fails_the_row_on_the_non_wal_path_too( + self, pg_conn, flask_app, + ): + """The same guarantee when no placeholder row was reserved. + + ``save_conversation`` has its own insert path (used when the WAL + reservation failed, or on a continuation carrying no + ``reserved_message_id``). It took no status, so the row landed on the + column default ``complete`` — leaving exactly the blank bubble this + changeset removes on the other branch. + """ + from application.api.answer.routes.base import BaseAnswerResource + from application.storage.db.repositories.conversations import ( + ConversationsRepository, + ) + + with flask_app.app_context(): + resource = BaseAnswerResource() + + mock_agent = MagicMock() + mock_agent.gen.return_value = iter( + [{"type": "error", "error": "CEL error in node 'Build reply'"}] + ) + + with _patch_db_session(pg_conn), patch.object( + resource.conversation_service, + "save_user_question", + side_effect=RuntimeError("WAL reservation unavailable"), + ): + list( + resource.complete_stream( + question="hello", + agent=mock_agent, + conversation_id=None, + user_api_key=None, + decoded_token={"sub": "u-wf-nonwal"}, + should_persist=True, + model_id="gpt-4", + ) + ) + + from sqlalchemy import text as sql_text + convs = pg_conn.execute( + sql_text("SELECT id FROM conversations WHERE user_id = :u"), + {"u": "u-wf-nonwal"}, + ).fetchall() + assert len(convs) == 1 + msgs = ConversationsRepository(pg_conn).get_messages(str(convs[0][0])) + assert len(msgs) == 1 + assert msgs[0]["status"] == "failed" + assert "CEL error" in msgs[0]["metadata"]["error"] + def test_tool_approval_event_only_fires_when_state_saved( self, pg_conn, flask_app, ): diff --git a/tests/test_logging.py b/tests/test_logging.py index 8bad293b..d10ba27d 100644 --- a/tests/test_logging.py +++ b/tests/test_logging.py @@ -257,6 +257,35 @@ class TestLogActivity: assert finished.status == "error" assert finished.error_class == "ValueError" + def test_log_activity_records_error_status_on_yielded_error(self, caplog): + # Workflow node failures are reported as a yielded ``type: error`` + # event rather than a raised exception, so the decorator's ``except`` + # never runs. Those turns used to log ``status="ok"`` with + # ``answer_length=0``, which kept real user-visible failures out of + # every error dashboard. + import logging as _logging + + from application.logging import log_activity + + class FakeAgent: + endpoint = "stream" + user = "user1" + user_api_key = "" + query = "hello" + + @log_activity() + def erroring(agent, log_context=None): + yield {"type": "error", "error": "No LLM class found for type foundry"} + + with patch("application.logging._log_activity_to_db"), \ + caplog.at_level(_logging.INFO, logger="root"): + list(erroring(FakeAgent())) + + finished = next(r for r in caplog.records if r.message == "activity_finished") + assert finished.status == "error" + assert finished.error_class == "StreamError" + assert finished.answer_length == 0 + def test_log_activity_emits_response_summary_aggregates(self, caplog): # Replaces the ``agent_response`` event that ``run_agent_logic`` # used to emit only on the Celery webhook path: every Flask