Log failed chat turns as chat entries

A turn whose agent raised wrote no user_logs row, so it only surfaced as
the agent's system error row. Every finished turn now writes its chat row,
at level error with the error when it failed, and linked to its trace; the
system row for the same traced activity is no longer listed twice.

docsgpt/api/answer/routes/base.py

@@ -144,6 +144,79 @@ class BaseAnswerResource:
             return missing_fields
         return None
 
+    def _persist_turn_log(
+        self,
+        *,
+        decoded_token: Dict[str, Any],
+        user_api_key: Optional[str],
+        agent_id: Optional[str],
+        question: str,
+        response: str,
+        sources: List[Dict[str, Any]],
+        tool_calls: Any,
+        attachment_ids: Optional[List[str]],
+        request_id: Optional[str],
+        message_id: Optional[str],
+        error: Optional[str] = None,
+        extra: Optional[Dict[str, Any]] = None,
+    ) -> None:
+        """Write the turn's ``user_logs`` row: its Logs entry, linked to its trace.
+
+        Written for every finished turn, failed ones included (``level`` is
+        ``error`` then), so a failed chat shows up as a chat entry. A failure
+        to write is logged, never raised.
+
+        Args:
+            decoded_token: The caller's token.
+            user_api_key: The agent API key the request used, if any.
+            agent_id: The agent that answered.
+            question: The question as stored (after input redaction).
+            response: The answer, or what streamed before a failure.
+            sources: Retrieved sources.
+            tool_calls: The turn's tool calls, before log truncation.
+            attachment_ids: Attached file ids.
+            request_id: The turn's request id (links its trace).
+            message_id: The reserved message id.
+            error: What failed the turn, if it failed.
+            extra: More fields for the row (structured-output details).
+        """
+        log_data: Dict[str, Any] = {
+            "action": "stream_answer",
+            "level": "error" if error else "info",
+            "user": decoded_token.get("sub"),
+            "api_key": user_api_key,
+            "agent_id": agent_id,
+            "question": question,
+            "response": response,
+            "sources": sources,
+            "tool_calls": self._prepare_tool_calls_for_logging(tool_calls),
+            "attachments": attachment_ids,
+            "timestamp": datetime.datetime.now(datetime.timezone.utc),
+            # Links the Logs row to this turn's execution trace.
+            "request_id": request_id,
+            "message_id": message_id,
+        }
+        if error:
+            log_data["error"] = error
+        if extra:
+            log_data.update(extra)
+        # Clean up text fields to be no longer than 10000 characters.
+        for key, value in log_data.items():
+            if isinstance(value, str) and len(value) > 10000:
+                log_data[key] = value[:10000]
+        try:
+            with db_session() as conn:
+                UserLogsRepository(conn).insert(
+                    user_id=log_data.get("user"),
+                    endpoint="stream_answer",
+                    data=log_data,
+                )
+        except Exception as log_err:
+            logger.error(
+                f"Failed to persist stream_answer user log: {log_err}",
+                exc_info=True,
+            )
+
     @staticmethod
     def _prepare_tool_calls_for_logging(
         tool_calls: Optional[List[Dict[str, Any]]], max_chars: int = 10000
@@ -1299,47 +1372,27 @@ class BaseAnswerResource:
                     )
             yield _emit({"type": "id", "id": str(conversation_id)})
 
-            tool_calls_for_logging = self._prepare_tool_calls_for_logging(
-                getattr(agent, "tool_calls", tool_calls) or tool_calls
-            )
-
-            log_data = {
-                "action": "stream_answer",
-                "level": "info",
-                "user": decoded_token.get("sub"),
-                "api_key": user_api_key,
-                "agent_id": agent_id,
-                "question": question,
-                "response": response_full,
-                "sources": source_log_docs,
-                "tool_calls": tool_calls_for_logging,
-                "attachments": attachment_ids,
-                "timestamp": datetime.datetime.now(datetime.timezone.utc),
-                # Links the Logs row to this turn's execution trace.
-                "request_id": request_id,
-                "message_id": reserved_message_id,
-            }
+            extra: Dict[str, Any] = {}
             if is_structured:
-                log_data["structured_output"] = True
+                extra["structured_output"] = True
                 if schema_info:
-                    log_data["schema"] = schema_info
-            # Clean up text fields to be no longer than 10000 characters
-
-            for key, value in log_data.items():
-                if isinstance(value, str) and len(value) > 10000:
-                    log_data[key] = value[:10000]
-            try:
-                with db_session() as conn:
-                    UserLogsRepository(conn).insert(
-                        user_id=log_data.get("user"),
-                        endpoint="stream_answer",
-                        data=log_data,
-                    )
-            except Exception as log_err:
-                logger.error(
-                    f"Failed to persist stream_answer user log: {log_err}",
-                    exc_info=True,
-                )
+                    extra["schema"] = schema_info
+            self._persist_turn_log(
+                decoded_token=decoded_token,
+                user_api_key=user_api_key,
+                agent_id=agent_id,
+                question=question,
+                response=response_full,
+                sources=source_log_docs,
+                tool_calls=getattr(agent, "tool_calls", tool_calls) or tool_calls,
+                attachment_ids=attachment_ids,
+                request_id=request_id,
+                message_id=reserved_message_id,
+                # A yielded error (a failed workflow node) ends the turn
+                # normally but still failed it.
+                error=stream_error,
+                extra=extra,
+            )
 
             yield _emit({"type": "end"})
             # Drain the journal buffer so the terminal ``end`` event is
@@ -1617,6 +1670,22 @@ class BaseAnswerResource:
                         f"Failed to finalize errored message: {fin_err}",
                         exc_info=True,
                     )
+            # A failed turn is still a chat turn: log it as one (level
+            # ``error``), with its trace link, instead of leaving only the
+            # agent's system error row.
+            self._persist_turn_log(
+                decoded_token=decoded_token,
+                user_api_key=user_api_key,
+                agent_id=agent_id,
+                question=question,
+                response=response_full,
+                sources=source_log_docs,
+                tool_calls=getattr(agent, "tool_calls", tool_calls) or tool_calls,
+                attachment_ids=attachment_ids,
+                request_id=request_id,
+                message_id=reserved_message_id,
+                error=f"{type(e).__name__}: {e}",
+            )
             yield _emit(
                 {
                     "type": "error",

docsgpt/api/user/analytics/routes.py

@@ -943,6 +943,16 @@ class GetUserLogs(Resource):
                     "WHERE e.activity_id = s.activity_id "
                     "AND e.level = 'error'))"
                 )
+                # A failed chat turn now writes its own chat row (level
+                # ``error``) linked to its trace; the agent's error row for
+                # the same activity would list the failure twice. Rows from
+                # before execution traces, or with tracing off, have no such
+                # trace and still show here.
+                chat_failure_dedupe = (
+                    "NOT EXISTS (SELECT 1 FROM request_traces t "
+                    "WHERE t.activity_id = s.activity_id "
+                    "AND t.source IN ('stream', 'answer', 'v1'))"
+                )
                 if api_key_id:
                     # The owner-scoped lookup gates access, so the
                     # chat/webhook/system branches match on the agent
@@ -978,6 +988,7 @@ class GetUserLogs(Resource):
                         "s.level = 'error'",
                         "COALESCE(s.endpoint, '') NOT IN ('webhook', 'schedule')",
                         stack_agent_match,
+                        chat_failure_dedupe,
                     ]
                     # Owner-gated agent match: drop the user clause so a
                     # shared agent's runs (stamped with the caller's
@@ -1003,6 +1014,7 @@ class GetUserLogs(Resource):
                         "s.user_id = :user_id",
                         "s.level = 'error'",
                         "COALESCE(s.endpoint, '') NOT IN ('webhook', 'schedule')",
+                        chat_failure_dedupe,
                     ]
                     # Terminal statuses only (worker writes ``success`` /
                     # ``failed`` / ``timeout`` / ``skipped``; ``completed``
@@ -1233,6 +1245,7 @@ class GetUserLogs(Resource):
                             "attachments": payload.get("attachments"),
                             "request_id": payload.get("request_id"),
                             "message_id": payload.get("message_id"),
+                            "error": payload.get("error"),
                         }
                     )
                 elif m["event_type"] in ("system", "webhook"):

tests/api/answer/test_trace_wiring.py

@@ -185,6 +185,52 @@ class TestTraceWithPersistence:
         assert row[0]["request_id"] == "req-p"
         assert row[0]["message_id"] == trace.message_id
 
+    def test_failed_turn_is_logged_as_a_chat_row(self, pg_conn, flask_app):
+        """A raised failure still writes the turn's chat row, at level error."""
+        from docsgpt.api.answer.routes.base import BaseAnswerResource
+        from tests.api.answer.test_base_routes import _patch_db_session
+
+        agent = MagicMock()
+        agent.gen.side_effect = RuntimeError("upstream down")
+        agent.tool_calls = []
+        with flask_app.app_context(), _patch_db_session(pg_conn), _captured_flushes():
+            _run(
+                BaseAnswerResource(),
+                agent,
+                should_persist=True,
+                model_id="gpt-4",
+                request_id="req-failed",
+            )
+        from sqlalchemy import text as sql_text
+
+        rows = pg_conn.execute(
+            sql_text("SELECT data FROM user_logs WHERE user_id = 'u-trace'")
+        ).fetchall()
+        assert len(rows) == 1
+        data = rows[0][0]
+        assert data["level"] == "error"
+        assert data["request_id"] == "req-failed"
+        assert data["error"] == "RuntimeError: upstream down"
+
+    def test_yielded_error_logs_the_chat_row_at_error_level(self, pg_conn, flask_app):
+        from docsgpt.api.answer.routes.base import BaseAnswerResource
+        from tests.api.answer.test_base_routes import _patch_db_session
+
+        with flask_app.app_context(), _patch_db_session(pg_conn), _captured_flushes():
+            _run(
+                BaseAnswerResource(),
+                _agent([{"type": "error", "error": "node failed"}]),
+                should_persist=True,
+                model_id="gpt-4",
+            )
+        from sqlalchemy import text as sql_text
+
+        data = pg_conn.execute(
+            sql_text("SELECT data FROM user_logs WHERE user_id = 'u-trace'")
+        ).fetchone()[0]
+        assert data["level"] == "error"
+        assert data["error"] == "node failed"
+
     def test_paused_turn_is_flushed_paused(self, pg_conn, flask_app):
         from docsgpt.api.answer.routes.base import BaseAnswerResource
         from tests.api.answer.test_base_routes import _patch_db_session

tests/api/user/test_traces_routes.py

@@ -212,3 +212,39 @@ class TestSummaryFailure:
         assert response.status_code == 200
         (row,) = response.json["logs"]
         assert "trace" not in row
+
+
+class TestFailedChatTurns:
+    def _failed_turn(self, pg_conn, *, with_trace):
+        from docsgpt.storage.db.repositories.stack_logs import StackLogsRepository
+        from docsgpt.storage.db.repositories.user_logs import UserLogsRepository
+
+        StackLogsRepository(pg_conn).insert(
+            activity_id="act-chat", endpoint="stream", level="error", user_id="owner", query="q"
+        )
+        UserLogsRepository(pg_conn).insert(
+            user_id="owner",
+            endpoint="stream_answer",
+            data={
+                "question": "q",
+                "level": "error",
+                "error": "RuntimeError: upstream down",
+                "request_id": "req-f",
+            },
+        )
+        if with_trace:
+            _trace(pg_conn, request_id="req-f", activity_id="act-chat", status="error")
+
+    def test_failed_turn_is_one_chat_row_not_also_a_system_row(self, app, pg_conn):
+        self._failed_turn(pg_conn, with_trace=True)
+        rows = _logs(app, pg_conn, "owner", {}).json["logs"]
+        assert [r["event_type"] for r in rows] == ["chat"]
+        assert rows[0]["level"] == "error"
+        assert rows[0]["error"] == "RuntimeError: upstream down"
+        assert rows[0]["trace"]["status"] == "error"
+
+    def test_untraced_failures_keep_their_system_row(self, app, pg_conn):
+        """Rows from before traces (or with tracing off) have nothing to dedupe against."""
+        self._failed_turn(pg_conn, with_trace=False)
+        rows = _logs(app, pg_conn, "owner", {}).json["logs"]
+        assert sorted(r["event_type"] for r in rows) == ["chat", "system"]