From a8a83a357cf0104c2c8dd0a072f09f4529ebcfba Mon Sep 17 00:00:00 2001 From: zyssyz123 <916125788@qq.com> Date: Fri, 24 Jul 2026 09:59:57 +0800 Subject: [PATCH] fix(agent): show workflow node runs in logs (#39471) --- api/controllers/console/agent/roster.py | 18 +- api/services/agent/observability_service.py | 304 +++++++++++++--- .../console/agent/test_agent_controllers.py | 14 + .../agent/test_agent_observability_service.py | 326 +++++++++++++++++- 4 files changed, 607 insertions(+), 55 deletions(-) diff --git a/api/controllers/console/agent/roster.py b/api/controllers/console/agent/roster.py index 62393d3bdc4..60fb018915b 100644 --- a/api/controllers/console/agent/roster.py +++ b/api/controllers/console/agent/roster.py @@ -539,9 +539,23 @@ def _parse_observability_time_range(start: str | None, end: str | None, account: def _query_values(name: str, alias_name: str | None = None) -> list[str]: - values = request.args.getlist(name) + def _get_values(field_name: str) -> list[str]: + values = request.args.getlist(field_name) + indexed_values: list[tuple[int, list[str]]] = [] + prefix = f"{field_name}[" + for key in request.args: + if not key.startswith(prefix) or not key.endswith("]"): + continue + index = key[len(prefix) : -1] + if index.isdigit(): + indexed_values.append((int(index), request.args.getlist(key))) + for _, items in sorted(indexed_values): + values.extend(items) + return values + + values = _get_values(name) if alias_name: - values.extend(request.args.getlist(alias_name)) + values.extend(_get_values(alias_name)) return [value.strip() for value in values if value.strip()] diff --git a/api/services/agent/observability_service.py b/api/services/agent/observability_service.py index 8ae77769c00..cabb4c135e5 100644 --- a/api/services/agent/observability_service.py +++ b/api/services/agent/observability_service.py @@ -1,8 +1,11 @@ from __future__ import annotations +import json +from collections.abc import Mapping from dataclasses import dataclass from datetime import datetime from decimal import Decimal +from enum import Enum from typing import Any import sqlalchemy as sa @@ -11,6 +14,7 @@ from sqlalchemy.orm import aliased from configs import dify_config from core.app.entities.app_invoke_entities import InvokeFrom +from graphon.enums import WorkflowNodeExecutionStatus from libs.helper import convert_datetime_to_date, escape_like_pattern, to_timestamp from models.agent import WorkflowAgentNodeBinding from models.enums import CreatorUserRole, MessageStatus @@ -194,11 +198,12 @@ class AgentObservabilityService: self, *, app: App, agent_id: str, conversation_id: str, params: AgentLogQueryParams ) -> dict[str, Any]: source_filters = self.resolve_source_filters(params.sources) - rows: list[Message] = [] + rows: list[dict[str, Any]] = [] for source_filter in source_filters: if source_filter.kind in {"all", "webapp"}: rows.extend( - self._list_webapp_messages( + self.serialize_log_message(message) + for message in self._list_webapp_messages( app=app, conversation_id=conversation_id, params=params, @@ -216,18 +221,18 @@ class AgentObservabilityService: ) ) - deduped = {message.id: message for message in rows} - sort_column = Message.created_at if params.sort_by == "created_at" else Message.updated_at + deduped = {row["id"]: row for row in rows} + sort_key = "created_at" if params.sort_by == "created_at" else "updated_at" sorted_rows = sorted( deduped.values(), - key=lambda message: (getattr(message, sort_column.key), message.id), + key=lambda row: (row[sort_key] or 0, row["id"]), reverse=params.sort_order != "asc", ) total = len(sorted_rows) start = (params.page - 1) * params.limit end = start + params.limit return { - "data": [self.serialize_log_message(message) for message in sorted_rows[start:end]], + "data": sorted_rows[start:end], "page": params.page, "limit": params.limit, "total": total, @@ -284,22 +289,20 @@ class AgentObservabilityService: workflow_app = aliased(App) stmt = ( select( - Conversation, + WorkflowNodeExecutionModel.id.label("node_execution_id"), + WorkflowNodeExecutionModel.title.label("node_title"), + WorkflowNodeExecutionModel.status.label("node_status"), + WorkflowNodeExecutionModel.created_by_role.label("node_created_by_role"), + WorkflowNodeExecutionModel.created_by.label("node_created_by"), + WorkflowNodeExecutionModel.created_at.label("node_created_at"), + WorkflowNodeExecutionModel.finished_at.label("node_finished_at"), workflow_app, WorkflowAgentNodeBinding.workflow_id, WorkflowAgentNodeBinding.workflow_version, WorkflowAgentNodeBinding.node_id, - func.count(sa.distinct(Message.id)).label("message_count"), - func.max(Message.created_at).label("created_at"), - func.max(Message.updated_at).label("updated_at"), - func.sum(sa.case((Message.status == MessageStatus.PAUSED, 1), else_=0)).label("paused_count"), - func.sum( - sa.case((or_(Message.error.is_not(None), Message.status == MessageStatus.ERROR), 1), else_=0) - ).label("failed_count"), ) - .select_from(Message) - .join(Conversation, Conversation.id == Message.conversation_id) - .join(WorkflowRun, WorkflowRun.id == Message.workflow_run_id) + .select_from(WorkflowNodeExecutionModel) + .join(WorkflowRun, WorkflowRun.id == WorkflowNodeExecutionModel.workflow_run_id) .join( WorkflowAgentNodeBinding, and_( @@ -310,40 +313,32 @@ class AgentObservabilityService: WorkflowAgentNodeBinding.workflow_version == WorkflowRun.version, ), ) - .join( - WorkflowNodeExecutionModel, - and_( - WorkflowNodeExecutionModel.workflow_run_id == WorkflowRun.id, - WorkflowNodeExecutionModel.node_id == WorkflowAgentNodeBinding.node_id, - ), - ) .join(workflow_app, workflow_app.id == WorkflowAgentNodeBinding.app_id) - .where(Message.workflow_run_id.is_not(None), Conversation.app_id == WorkflowAgentNodeBinding.app_id) - .group_by( - Conversation.id, - workflow_app.id, - WorkflowAgentNodeBinding.workflow_id, - WorkflowAgentNodeBinding.workflow_version, - WorkflowAgentNodeBinding.node_id, + .where( + WorkflowNodeExecutionModel.tenant_id == app.tenant_id, + WorkflowNodeExecutionModel.app_id == WorkflowAgentNodeBinding.app_id, + WorkflowNodeExecutionModel.workflow_id == WorkflowAgentNodeBinding.workflow_id, + WorkflowNodeExecutionModel.node_id == WorkflowAgentNodeBinding.node_id, ) ) - stmt = self._apply_observability_filters(stmt, params=params, source_filter=source_filter) + stmt = self._apply_workflow_node_filters(stmt, params=params, workflow_app=workflow_app) stmt = self._apply_workflow_source_filter(stmt, source_filter) rows = list(self._session.execute(stmt).all()) return [ - self._serialize_conversation_log( - conversation=row[0], - message_count=row.message_count, - paused_count=row.paused_count, - failed_count=row.failed_count, + self._serialize_workflow_execution_log( + node_execution_id=row.node_execution_id, + title=row.node_title, + status=row.node_status, + created_by_role=row.node_created_by_role, + created_by=row.node_created_by, + created_at=row.node_created_at, + finished_at=row.node_finished_at, source=self._serialize_workflow_source( - app=row[1], + app=row[7], workflow_id=row.workflow_id, workflow_version=row.workflow_version, node_id=row.node_id, ), - created_at=row.created_at, - updated_at=row.updated_at, ) for row in rows ] @@ -363,10 +358,12 @@ class AgentObservabilityService: conversation_id: str, params: AgentLogQueryParams, source_filter: AgentSourceFilter, - ) -> list[Message]: + ) -> list[dict[str, Any]]: + workflow_app = aliased(App) stmt = ( - select(Message) - .join(WorkflowRun, WorkflowRun.id == Message.workflow_run_id) + select(WorkflowNodeExecutionModel) + .select_from(WorkflowNodeExecutionModel) + .join(WorkflowRun, WorkflowRun.id == WorkflowNodeExecutionModel.workflow_run_id) .join( WorkflowAgentNodeBinding, and_( @@ -377,18 +374,23 @@ class AgentObservabilityService: WorkflowAgentNodeBinding.workflow_version == WorkflowRun.version, ), ) - .join( - WorkflowNodeExecutionModel, - and_( - WorkflowNodeExecutionModel.workflow_run_id == WorkflowRun.id, - WorkflowNodeExecutionModel.node_id == WorkflowAgentNodeBinding.node_id, - ), + .join(workflow_app, workflow_app.id == WorkflowAgentNodeBinding.app_id) + .where( + WorkflowNodeExecutionModel.id == conversation_id, + WorkflowNodeExecutionModel.tenant_id == app.tenant_id, + WorkflowNodeExecutionModel.app_id == WorkflowAgentNodeBinding.app_id, + WorkflowNodeExecutionModel.workflow_id == WorkflowAgentNodeBinding.workflow_id, + WorkflowNodeExecutionModel.node_id == WorkflowAgentNodeBinding.node_id, ) - .where(Message.conversation_id == conversation_id) ) - stmt = self._apply_message_filters(stmt, params=params, source_filter=source_filter) + stmt = self._apply_workflow_node_filters(stmt, params=params, workflow_app=workflow_app) stmt = self._apply_workflow_source_filter(stmt, source_filter) - return list(self._session.scalars(stmt.order_by(Message.created_at.desc(), Message.id.desc())).all()) + executions = list( + self._session.scalars( + stmt.order_by(WorkflowNodeExecutionModel.created_at.desc(), WorkflowNodeExecutionModel.id.desc()) + ).all() + ) + return [self.serialize_workflow_node_message(execution) for execution in executions] def _list_workflow_sources(self, *, app: App, agent_id: str) -> list[dict[str, Any]]: workflow_app = aliased(App) @@ -443,6 +445,62 @@ class AgentObservabilityService: stmt = cls._apply_status_filter(stmt, params.statuses) return stmt + @classmethod + def _apply_workflow_node_filters(cls, stmt, *, params: AgentLogQueryParams, workflow_app): + if params.start: + stmt = stmt.where(WorkflowNodeExecutionModel.created_at >= params.start) + if params.end: + stmt = stmt.where(WorkflowNodeExecutionModel.created_at < params.end) + if params.keyword: + escaped_keyword = escape_like_pattern(params.keyword) + pattern = f"%{escaped_keyword}%" + stmt = stmt.where( + or_( + WorkflowNodeExecutionModel.inputs.ilike(pattern, escape="\\"), + WorkflowNodeExecutionModel.outputs.ilike(pattern, escape="\\"), + WorkflowNodeExecutionModel.error.ilike(pattern, escape="\\"), + WorkflowNodeExecutionModel.title.ilike(pattern, escape="\\"), + workflow_app.name.ilike(pattern, escape="\\"), + ) + ) + if params.statuses: + stmt = cls._apply_workflow_node_status_filter(stmt, params.statuses) + return stmt + + @staticmethod + def _apply_workflow_node_status_filter(stmt, statuses: tuple[str, ...]): + conditions = [] + for status in statuses: + normalized = status.strip().lower() + if normalized in {"success", "normal"}: + conditions.append(WorkflowNodeExecutionModel.status == WorkflowNodeExecutionStatus.SUCCEEDED) + elif normalized in {"failed", "error"}: + conditions.append( + WorkflowNodeExecutionModel.status.in_( + ( + WorkflowNodeExecutionStatus.FAILED, + WorkflowNodeExecutionStatus.EXCEPTION, + WorkflowNodeExecutionStatus.STOPPED, + ) + ) + ) + elif normalized == "paused": + conditions.append( + WorkflowNodeExecutionModel.status.in_( + ( + WorkflowNodeExecutionStatus.PAUSED, + WorkflowNodeExecutionStatus.PENDING, + WorkflowNodeExecutionStatus.RUNNING, + WorkflowNodeExecutionStatus.RETRY, + ) + ) + ) + else: + raise ValueError(f"Unsupported status: {status}") + if not conditions: + return stmt + return stmt.where(or_(*conditions)) + @staticmethod def _apply_workflow_source_filter(stmt, source_filter: AgentSourceFilter): if source_filter.app_id: @@ -505,6 +563,148 @@ class AgentObservabilityService: "updated_at": to_timestamp(updated_at or conversation.updated_at), } + @classmethod + def _serialize_workflow_execution_log( + cls, + *, + node_execution_id: str, + title: str, + status: object, + created_by_role: object, + created_by: str, + created_at: datetime, + finished_at: datetime | None, + source: dict[str, Any], + ) -> dict[str, Any]: + created_by_role_value = cls._enum_value(created_by_role) + return { + "id": node_execution_id, + "conversation_id": node_execution_id, + "title": title, + "end_user_id": created_by if created_by_role_value == CreatorUserRole.END_USER.value else None, + "message_count": 1, + "user_rate": None, + "operation_rate": None, + "unread": False, + "source": source, + "status": cls._workflow_node_status(status), + "created_at": to_timestamp(created_at), + "updated_at": to_timestamp(finished_at or created_at), + } + + @classmethod + def serialize_workflow_node_message(cls, node_execution: WorkflowNodeExecutionModel) -> dict[str, Any]: + inputs = cls._json_mapping(node_execution.inputs) + outputs = cls._json_mapping(node_execution.outputs) + metadata = cls._json_mapping(node_execution.execution_metadata) + agent_log = cls._mapping_value(metadata, "agent_log") + agent_backend = cls._mapping_value(agent_log, "agent_backend") + usage = cls._mapping_value(agent_backend, "usage") + + prompt_tokens = cls._int_value(usage.get("prompt_tokens")) + completion_tokens = cls._int_value(usage.get("completion_tokens")) + total_tokens = cls._int_value(usage.get("total_tokens") or metadata.get("total_tokens")) + if not total_tokens: + total_tokens = prompt_tokens + completion_tokens + created_by_role = cls._enum_value(node_execution.created_by_role) + + return { + "id": node_execution.id, + "message_id": node_execution.id, + "conversation_id": node_execution.id, + "query": cls._workflow_node_query(inputs, fallback=node_execution.title), + "answer": cls._workflow_node_answer(outputs), + "status": cls._workflow_node_status(node_execution.status), + "error": node_execution.error, + "from_end_user_id": ( + node_execution.created_by if created_by_role == CreatorUserRole.END_USER.value else None + ), + "from_account_id": ( + node_execution.created_by if created_by_role == CreatorUserRole.ACCOUNT.value else None + ), + "message_tokens": prompt_tokens, + "answer_tokens": completion_tokens, + "total_tokens": total_tokens, + "total_price": str(usage.get("total_price") or metadata.get("total_price") or Decimal(0)), + "currency": str(usage.get("currency") or metadata.get("currency") or ""), + "latency": float(usage.get("latency") or node_execution.elapsed_time or 0), + "created_at": to_timestamp(node_execution.created_at), + "updated_at": to_timestamp(node_execution.finished_at or node_execution.created_at), + } + + @staticmethod + def _json_mapping(value: object) -> Mapping[str, Any]: + if isinstance(value, Mapping): + return value + if not isinstance(value, str) or not value: + return {} + try: + parsed = json.loads(value) + except (TypeError, ValueError): + return {} + return parsed if isinstance(parsed, Mapping) else {} + + @staticmethod + def _mapping_value(value: Mapping[str, Any], key: str) -> Mapping[str, Any]: + nested = value.get(key) + return nested if isinstance(nested, Mapping) else {} + + @staticmethod + def _enum_value(value: object) -> str: + return str(value.value) if isinstance(value, Enum) else str(value) + + @staticmethod + def _int_value(value: object) -> int: + if not isinstance(value, (str, int, float, Decimal)): + return 0 + try: + return int(value) + except (TypeError, ValueError): + return 0 + + @classmethod + def _workflow_node_query(cls, inputs: Mapping[str, Any], *, fallback: str) -> str: + request_data = cls._mapping_value(inputs, "agent_backend_request") + composition = cls._mapping_value(request_data, "composition") + layers = composition.get("layers") + prompts: list[str] = [] + if isinstance(layers, list): + for layer_name in ("workflow_node_job_prompt", "workflow_user_prompt"): + for layer in layers: + if not isinstance(layer, Mapping) or layer.get("name") != layer_name: + continue + config = cls._mapping_value(layer, "config") + user_prompt = config.get("user") + if isinstance(user_prompt, str) and user_prompt.strip(): + prompts.append(user_prompt.strip()) + return "\n\n".join(prompts) or fallback + + @staticmethod + def _workflow_node_answer(outputs: Mapping[str, Any]) -> str: + for key in ("output", "text", "answer"): + value = outputs.get(key) + if isinstance(value, str): + return value + return json.dumps(outputs, ensure_ascii=False) if outputs else "" + + @classmethod + def _workflow_node_status(cls, status: object) -> str: + value = cls._enum_value(status) + if value in { + WorkflowNodeExecutionStatus.FAILED.value, + WorkflowNodeExecutionStatus.EXCEPTION.value, + WorkflowNodeExecutionStatus.STOPPED.value, + }: + return "failed" + if value in { + WorkflowNodeExecutionStatus.PAUSED.value, + WorkflowNodeExecutionStatus.PENDING.value, + WorkflowNodeExecutionStatus.RUNNING.value, + WorkflowNodeExecutionStatus.RETRY.value, + }: + return "paused" + return "success" + @staticmethod def _conversation_status(*, paused_count: int, failed_count: int) -> str: if paused_count: diff --git a/api/tests/unit_tests/controllers/console/agent/test_agent_controllers.py b/api/tests/unit_tests/controllers/console/agent/test_agent_controllers.py index 8f395f5549d..14628b2614d 100644 --- a/api/tests/unit_tests/controllers/console/agent/test_agent_controllers.py +++ b/api/tests/unit_tests/controllers/console/agent/test_agent_controllers.py @@ -69,6 +69,20 @@ def _version_response(version_id: str = "version-1") -> dict: } +def test_query_values_accepts_repeated_and_indexed_arrays() -> None: + app = Flask(__name__) + + with app.test_request_context("/?sources=webapp:app-1&sources%5B1%5D=workflow:app-2&sources%5B0%5D=workflow:app-1"): + assert roster_controller._query_values("sources", "source") == [ + "webapp:app-1", + "workflow:app-1", + "workflow:app-2", + ] + + with app.test_request_context("/?source%5B0%5D=workflow:app-3"): + assert roster_controller._query_values("sources", "source") == ["workflow:app-3"] + + def _workflow_composer_response(**overrides) -> dict: response = { "variant": "workflow", diff --git a/api/tests/unit_tests/services/agent/test_agent_observability_service.py b/api/tests/unit_tests/services/agent/test_agent_observability_service.py index cff60004302..4fcb0466a09 100644 --- a/api/tests/unit_tests/services/agent/test_agent_observability_service.py +++ b/api/tests/unit_tests/services/agent/test_agent_observability_service.py @@ -5,7 +5,8 @@ from types import SimpleNamespace import pytest from core.app.entities.app_invoke_entities import InvokeFrom -from models.enums import ConversationFromSource, MessageStatus +from graphon.enums import WorkflowNodeExecutionStatus +from models.enums import ConversationFromSource, CreatorUserRole, MessageStatus from services.agent import observability_service as observability_service_module from services.agent.observability_service import AgentLogQueryParams, AgentObservabilityService @@ -268,6 +269,183 @@ def test_list_logs_sorts_by_requested_field(monkeypatch: pytest.MonkeyPatch) -> assert [item["id"] for item in payload["data"]] == ["old", "new"] +def test_list_log_messages_merges_deduplicates_and_sorts_sources(monkeypatch: pytest.MonkeyPatch) -> None: + service = AgentObservabilityService(session=None) + webapp_message = SimpleNamespace(id="shared") + webapp_row = {"id": "shared", "created_at": 10, "updated_at": 30} + workflow_rows = [ + {"id": "shared", "created_at": 10, "updated_at": 20}, + {"id": "workflow-only", "created_at": 20, "updated_at": 10}, + ] + monkeypatch.setattr(service, "_list_webapp_messages", lambda **kwargs: [webapp_message]) + monkeypatch.setattr(service, "serialize_log_message", lambda message: webapp_row) + monkeypatch.setattr(service, "_list_workflow_messages", lambda **kwargs: workflow_rows) + + payload = service.list_log_messages( + app=SimpleNamespace(id="agent-app"), # type: ignore[arg-type] + agent_id="agent-1", + conversation_id="execution-1", + params=AgentLogQueryParams( + sources=("webapp:agent-app", "workflow:workflow-app"), + sort_by="created_at", + sort_order="asc", + ), + ) + + assert payload == { + "data": [workflow_rows[0], workflow_rows[1]], + "page": 1, + "limit": 20, + "total": 2, + "has_more": False, + } + + +def test_list_workflow_logs_uses_node_executions_without_messages() -> None: + created_at = datetime(2026, 7, 23, 7, 0, 19, tzinfo=UTC) + node_execution = SimpleNamespace( + id="node-execution-1", + title="Agent", + status=WorkflowNodeExecutionStatus.SUCCEEDED, + created_by_role=CreatorUserRole.END_USER, + created_by="end-user-1", + created_at=created_at, + finished_at=None, + ) + workflow_app = SimpleNamespace( + id="workflow-app-1", + name="Marketing Department", + icon_type=None, + icon=None, + icon_background=None, + ) + + class FakeRow: + node_execution_id = node_execution.id + node_title = node_execution.title + node_status = node_execution.status + node_created_by_role = node_execution.created_by_role + node_created_by = node_execution.created_by + node_created_at = node_execution.created_at + node_finished_at = node_execution.finished_at + workflow_id = "workflow-1" + workflow_version = "v1" + node_id = "node-1" + + def __getitem__(self, index: int): + return (None, None, None, None, None, None, None, workflow_app)[index] + + class FakeResult: + def all(self): + return [FakeRow()] + + class FakeSession: + def __init__(self): + self.query = "" + + def execute(self, stmt): + self.query = str(stmt) + return FakeResult() + + session = FakeSession() + service = AgentObservabilityService(session) + + rows = service._list_workflow_conversation_logs( + app=SimpleNamespace(tenant_id="tenant-1"), # type: ignore[arg-type] + agent_id="agent-1", + params=AgentLogQueryParams(), + source_filter=AgentObservabilityService.resolve_source_filter("workflow:workflow-app-1"), + ) + + assert "FROM workflow_node_executions" in session.query + assert "JOIN messages" not in session.query + assert rows[0]["id"] == "node-execution-1" + assert rows[0]["source"]["app_name"] == "Marketing Department" + + +def test_list_workflow_messages_uses_node_execution_identity(monkeypatch: pytest.MonkeyPatch) -> None: + node_execution = SimpleNamespace(id="node-execution-1") + + class FakeScalarResult: + def all(self): + return [node_execution] + + class FakeSession: + def __init__(self): + self.query = "" + + def scalars(self, stmt): + self.query = str(stmt) + return FakeScalarResult() + + session = FakeSession() + service = AgentObservabilityService(session) + serialized = {"id": "node-execution-1", "conversation_id": "node-execution-1"} + monkeypatch.setattr(service, "serialize_workflow_node_message", lambda execution: serialized) + + rows = service._list_workflow_messages( + app=SimpleNamespace(tenant_id="tenant-1"), # type: ignore[arg-type] + agent_id="agent-1", + conversation_id="node-execution-1", + params=AgentLogQueryParams(), + source_filter=AgentObservabilityService.resolve_source_filter("workflow:workflow-app-1"), + ) + + assert rows == [serialized] + assert "FROM workflow_node_executions" in session.query + assert "workflow_node_executions.id =" in session.query + assert "JOIN messages" not in session.query + + +def test_apply_workflow_node_filters_supports_time_keyword_and_status() -> None: + class FakeStmt: + def __init__(self): + self.conditions = [] + + def where(self, *conditions): + self.conditions.extend(conditions) + return self + + stmt = FakeStmt() + params = AgentLogQueryParams( + start=datetime(2026, 7, 1, tzinfo=UTC), + end=datetime(2026, 8, 1, tzinfo=UTC), + keyword="meeting_100%", + statuses=("success",), + ) + + result = AgentObservabilityService._apply_workflow_node_filters( + stmt, + params=params, + workflow_app=SimpleNamespace(name=observability_service_module.App.name), + ) + + assert result is stmt + assert len(stmt.conditions) == 4 + + +def test_apply_workflow_node_status_filter_supports_all_status_groups() -> None: + class FakeStmt: + def __init__(self): + self.conditions = [] + + def where(self, *conditions): + self.conditions.extend(conditions) + return self + + stmt = FakeStmt() + + result = AgentObservabilityService._apply_workflow_node_status_filter(stmt, ("normal", "error", "paused")) + + assert result is stmt + assert len(stmt.conditions) == 1 + empty_stmt = FakeStmt() + assert AgentObservabilityService._apply_workflow_node_status_filter(empty_stmt, ()) is empty_stmt + assert empty_stmt.conditions == [] + with pytest.raises(ValueError, match="Unsupported status"): + AgentObservabilityService._apply_workflow_node_status_filter(FakeStmt(), ("unknown",)) + + def test_source_serializers_return_structured_frontend_shape() -> None: app = SimpleNamespace( id="app-1", @@ -400,6 +578,152 @@ def test_serialize_log_message_returns_frontend_log_shape() -> None: } +def test_serialize_workflow_node_message_returns_frontend_log_shape() -> None: + created_at = datetime(2026, 7, 23, 7, 0, 19, tzinfo=UTC) + finished_at = datetime(2026, 7, 23, 7, 0, 28, tzinfo=UTC) + node_execution = SimpleNamespace( + id="node-execution-1", + title="Agent", + inputs=( + '{"agent_backend_request":{"composition":{"layers":[' + '{"name":"workflow_node_job_prompt","config":{"user":"Summarize the meeting"}},' + '{"name":"workflow_user_prompt","config":{"user":"Focus on action items"}}]}}}' + ), + outputs='{"output":"Alice owns the follow-up."}', + execution_metadata=( + '{"agent_log":{"agent_backend":{"usage":{"prompt_tokens":10,"completion_tokens":5,' + '"total_tokens":15,"total_price":"0.0015","currency":"USD","latency":1.25}}}}' + ), + status=WorkflowNodeExecutionStatus.SUCCEEDED, + error=None, + elapsed_time=1.5, + created_by_role=CreatorUserRole.END_USER, + created_by="end-user-1", + created_at=created_at, + finished_at=finished_at, + ) + + payload = AgentObservabilityService.serialize_workflow_node_message(node_execution) # type: ignore[arg-type] + + assert payload == { + "id": "node-execution-1", + "message_id": "node-execution-1", + "conversation_id": "node-execution-1", + "query": "Summarize the meeting\n\nFocus on action items", + "answer": "Alice owns the follow-up.", + "status": "success", + "error": None, + "from_end_user_id": "end-user-1", + "from_account_id": None, + "message_tokens": 10, + "answer_tokens": 5, + "total_tokens": 15, + "total_price": "0.0015", + "currency": "USD", + "latency": 1.25, + "created_at": int(created_at.timestamp()), + "updated_at": int(finished_at.timestamp()), + } + + +def test_serialize_workflow_node_message_handles_sparse_runtime_data() -> None: + created_at = datetime(2026, 7, 23, 7, 0, 19, tzinfo=UTC) + node_execution = SimpleNamespace( + id="node-execution-2", + title="Fallback prompt", + inputs={ + "agent_backend_request": { + "composition": { + "layers": [ + None, + {"name": "unrelated", "config": {"user": "ignored"}}, + {"name": "workflow_user_prompt", "config": {"user": " "}}, + ] + } + } + }, + outputs={"output": {"structured": True}}, + execution_metadata={ + "agent_log": { + "agent_backend": { + "usage": { + "prompt_tokens": "2", + "completion_tokens": 3, + } + } + } + }, + status=WorkflowNodeExecutionStatus.PAUSED, + error=None, + elapsed_time=2, + created_by_role=CreatorUserRole.ACCOUNT.value, + created_by="account-1", + created_at=created_at, + finished_at=None, + ) + + payload = AgentObservabilityService.serialize_workflow_node_message(node_execution) # type: ignore[arg-type] + + assert payload["query"] == "Fallback prompt" + assert payload["answer"] == '{"output": {"structured": true}}' + assert payload["status"] == "paused" + assert payload["from_end_user_id"] is None + assert payload["from_account_id"] == "account-1" + assert payload["total_tokens"] == 5 + assert payload["total_price"] == "0" + assert payload["currency"] == "" + assert payload["latency"] == 2.0 + assert payload["updated_at"] == int(created_at.timestamp()) + + +def test_workflow_node_serialization_helpers_handle_invalid_values() -> None: + assert AgentObservabilityService._json_mapping(None) == {} + assert AgentObservabilityService._json_mapping("not-json") == {} + assert AgentObservabilityService._json_mapping("[]") == {} + assert AgentObservabilityService._mapping_value({"value": []}, "value") == {} + assert AgentObservabilityService._int_value(None) == 0 + assert AgentObservabilityService._int_value("not-a-number") == 0 + assert ( + AgentObservabilityService._workflow_node_query( + {"agent_backend_request": {"composition": {"layers": "invalid"}}}, fallback="fallback" + ) + == "fallback" + ) + assert AgentObservabilityService._workflow_node_answer({"output": 1, "text": "fallback text"}) == "fallback text" + assert AgentObservabilityService._workflow_node_answer({}) == "" + + +def test_serialize_workflow_execution_log_uses_node_execution_identity() -> None: + created_at = datetime(2026, 7, 23, 7, 0, 19, tzinfo=UTC) + node_execution = SimpleNamespace( + id="node-execution-1", + title="Agent", + status=WorkflowNodeExecutionStatus.FAILED, + created_by_role=CreatorUserRole.ACCOUNT, + created_by="account-1", + created_at=created_at, + finished_at=None, + ) + + payload = AgentObservabilityService._serialize_workflow_execution_log( + node_execution_id=node_execution.id, + title=node_execution.title, + status=node_execution.status, + created_by_role=node_execution.created_by_role, + created_by=node_execution.created_by, + created_at=node_execution.created_at, + finished_at=node_execution.finished_at, + source={"id": "workflow:app-1:workflow-1:v1:node-1"}, + ) + + assert payload["id"] == "node-execution-1" + assert payload["conversation_id"] == "node-execution-1" + assert payload["message_count"] == 1 + assert payload["end_user_id"] is None + assert payload["status"] == "failed" + assert payload["unread"] is False + + def test_build_charts_and_summary_match_monitoring_metrics() -> None: rows = [ {