diff --git a/docsgpt/api/answer/routes/base.py b/docsgpt/api/answer/routes/base.py index 782e11b5..ca92d3cf 100644 --- a/docsgpt/api/answer/routes/base.py +++ b/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", diff --git a/docsgpt/api/user/analytics/routes.py b/docsgpt/api/user/analytics/routes.py index c77d9eb1..3642ed3a 100644 --- a/docsgpt/api/user/analytics/routes.py +++ b/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"): diff --git a/tests/api/answer/test_trace_wiring.py b/tests/api/answer/test_trace_wiring.py index d03dec01..01be1475 100644 --- a/tests/api/answer/test_trace_wiring.py +++ b/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 diff --git a/tests/api/user/test_traces_routes.py b/tests/api/user/test_traces_routes.py index 7c89d864..bbafdd79 100644 --- a/tests/api/user/test_traces_routes.py +++ b/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"]