from datetime import UTC, datetime
from decimal import Decimal
from types import SimpleNamespace

import pytest

from core.app.entities.app_invoke_entities import InvokeFrom
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


def test_resolve_source_accepts_frontend_aliases() -> None:
    assert AgentObservabilityService.resolve_source(None) is None
    assert AgentObservabilityService.resolve_source("all") is None
    assert AgentObservabilityService.resolve_source("console") == InvokeFrom.EXPLORE
    assert AgentObservabilityService.resolve_source("api") == InvokeFrom.SERVICE_API
    assert AgentObservabilityService.resolve_source("web_app") == InvokeFrom.WEB_APP

    with pytest.raises(ValueError, match="Unsupported source"):
        AgentObservabilityService.resolve_source("unknown")


def test_resolve_source_filter_accepts_structured_sources() -> None:
    assert AgentObservabilityService.resolve_source_filter(None).kind == "all"
    assert AgentObservabilityService.resolve_source_filter("webapp").kind == "webapp"
    assert AgentObservabilityService.resolve_source_filter("webapp:app-1").app_id == "app-1"

    workflow_app_filter = AgentObservabilityService.resolve_source_filter("workflow:app-2")
    assert workflow_app_filter.kind == "workflow"
    assert workflow_app_filter.app_id == "app-2"
    assert workflow_app_filter.workflow_id is None

    workflow_filter = AgentObservabilityService.resolve_source_filter("workflow:app-2:workflow-1:v1:node-1")
    assert workflow_filter.kind == "workflow"
    assert workflow_filter.app_id == "app-2"
    assert workflow_filter.workflow_id == "workflow-1"
    assert workflow_filter.workflow_version == "v1"
    assert workflow_filter.node_id == "node-1"

    timestamp_version_filter = AgentObservabilityService.resolve_source_filter(
        "workflow:app-2:workflow-1:2026-07-06 02:17:12.910515:node-1"
    )
    assert timestamp_version_filter.workflow_version == "2026-07-06 02:17:12.910515"
    assert timestamp_version_filter.node_id == "node-1"

    legacy_filter = AgentObservabilityService.resolve_source_filter("console")
    assert legacy_filter.kind == "webapp"
    assert legacy_filter.invoke_from == InvokeFrom.EXPLORE

    with pytest.raises(ValueError, match="Unsupported source"):
        AgentObservabilityService.resolve_source_filter("workflow:")
    with pytest.raises(ValueError, match="Unsupported source"):
        AgentObservabilityService.resolve_source_filter("workflow:app-2:incomplete")


def test_resolve_source_filters_accepts_multiple_structured_sources() -> None:
    filters = AgentObservabilityService.resolve_source_filters(("webapp:app-1", "workflow:app-2:workflow-1:v1:node-1"))

    assert [source_filter.kind for source_filter in filters] == ["webapp", "workflow"]
    assert filters[0].app_id == "app-1"
    assert filters[1].node_id == "node-1"
    assert AgentObservabilityService.resolve_source_filters(())[0].kind == "all"
    assert AgentObservabilityService.resolve_source_filters(("all", "webapp:app-1"))[0].kind == "all"


def test_statistics_all_source_includes_debugger_messages() -> None:
    source_filter = AgentObservabilityService.resolve_source_filter("all")

    scope_sql = AgentObservabilityService._statistics_webapp_message_scope_sql(source_filter)

    assert "m.app_id = :app_id" in scope_sql
    assert "m.invoke_from != :debugger" not in scope_sql


def test_statistics_explicit_source_filters_invoke_from() -> None:
    source_filter = AgentObservabilityService.resolve_source_filter("debugger")

    scope_sql = AgentObservabilityService._statistics_webapp_message_scope_sql(source_filter)

    assert "m.invoke_from = :source" in scope_sql


def test_statistics_workflow_app_source_covers_all_versions_and_nodes() -> None:
    source_filter = AgentObservabilityService.resolve_source_filter("workflow:app-2")

    scope_sql = AgentObservabilityService._statistics_workflow_binding_filters_sql(source_filter)

    assert "wanb.app_id = :source_app_id" in scope_sql
    assert "wanb.workflow_id = :workflow_id" not in scope_sql
    assert "wanb.workflow_version = :workflow_version" not in scope_sql
    assert "wanb.node_id = :node_id" not in scope_sql


def test_statistics_workflow_chat_context_only_uses_chat_runs() -> None:
    source_filter = AgentObservabilityService.resolve_source_filter("workflow:app-2")

    scope_sql = AgentObservabilityService._statistics_workflow_message_scope_sql(source_filter)

    assert "wr.id = m.workflow_run_id" in scope_sql
    assert "wr.type = :chat_workflow_type" in scope_sql


def test_workflow_metadata_numeric_sql_supports_postgresql_and_mysql(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setattr(observability_service_module, "dify_config", SimpleNamespace(DB_TYPE="postgresql"))

    postgres_sql = AgentObservabilityService._workflow_execution_metadata_numeric_sql(
        ("agent_log", "agent_backend", "usage", "total_tokens"), "BIGINT"
    )

    assert "CAST(wne.execution_metadata AS JSONB)" in postgres_sql
    assert "#>> '{agent_log,agent_backend,usage,total_tokens}'" in postgres_sql

    monkeypatch.setattr(observability_service_module, "dify_config", SimpleNamespace(DB_TYPE="mysql"))

    mysql_sql = AgentObservabilityService._workflow_execution_metadata_numeric_sql(("total_tokens",), "BIGINT")

    assert "JSON_EXTRACT(wne.execution_metadata, '$.total_tokens')" in mysql_sql
    assert " AS UNSIGNED)" in mysql_sql


def test_workflow_statistics_include_run_without_message(monkeypatch: pytest.MonkeyPatch) -> None:
    class FakeResult:
        def __init__(self, rows):
            self._rows = rows

        def all(self):
            return self._rows

    class FakeSession:
        def __init__(self):
            self.queries: list[str] = []

        def execute(self, stmt, args):
            query = str(stmt)
            self.queries.append(query)
            if "WITH agent_run_usage" in query:
                return FakeResult(
                    [
                        SimpleNamespace(
                            _mapping={
                                "date": "2026-07-21",
                                "message_count": 1,
                                "conversation_count": 1,
                                "end_user_count": 1,
                                "token_count": 454_064,
                                "total_price": Decimal("2.323470"),
                                "avg_latency": 59.93,
                                "latency_sum": 59.93,
                                "answer_tokens": 2_126,
                                "like_count": 0,
                            }
                        )
                    ]
                )
            return FakeResult([])

    monkeypatch.setattr(observability_service_module, "dify_config", SimpleNamespace(DB_TYPE="postgresql"))
    monkeypatch.setattr(
        observability_service_module,
        "convert_datetime_to_date",
        lambda field: f"DATE({field})",
    )
    session = FakeSession()
    service = AgentObservabilityService(session)

    payload = service.get_statistics_summary(
        app=SimpleNamespace(id="agent-app", tenant_id="tenant-1"),  # type: ignore[arg-type]
        agent_id="agent-1",
        params=observability_service_module.AgentStatisticsQueryParams(source="workflow:workflow-app"),
    )

    assert payload["summary"]["total_messages"] == 1
    assert payload["summary"]["total_conversations"] == 1
    assert payload["summary"]["total_end_users"] == 1
    assert payload["summary"]["total_tokens"] == 454_064
    assert payload["summary"]["total_price"] == "2.323470"
    assert len(session.queries) == 2
    assert "FROM workflow_runs wr" in session.queries[0]
    assert "FROM messages m" not in session.queries[0]
    assert "WHERE wr.type != :chat_workflow_type" in session.queries[0]
    assert "FROM messages m" in session.queries[1]
    assert "COUNT(m.id) AS message_count" in session.queries[1]
    assert "SUM(COALESCE(m.message_tokens, 0)" in session.queries[1]


def test_merge_daily_statistics_combines_webapp_and_workflow_rows() -> None:
    rows = [
        {
            "date": "2026-07-21",
            "message_count": 2,
            "conversation_count": 1,
            "end_user_count": 1,
            "token_count": 30,
            "total_price": Decimal("0.003"),
            "avg_latency": 1.5,
            "latency_sum": 3,
            "answer_tokens": 12,
            "like_count": 1,
        },
        {
            "date": "2026-07-21",
            "message_count": 1,
            "conversation_count": 1,
            "end_user_count": 1,
            "token_count": 20,
            "total_price": Decimal("0.002"),
            "avg_latency": 2,
            "latency_sum": 2,
            "answer_tokens": 8,
            "like_count": 0,
        },
    ]

    merged = AgentObservabilityService._merge_daily_statistics(rows)

    assert merged == [
        {
            "date": "2026-07-21",
            "message_count": 3,
            "conversation_count": 2,
            "end_user_count": 2,
            "token_count": 50,
            "total_price": Decimal("0.005"),
            "avg_latency": pytest.approx(5 / 3),
            "latency_sum": 5.0,
            "answer_tokens": 20,
            "like_count": 1,
        }
    ]


def test_apply_status_filter_accepts_multiple_statuses() -> None:
    class FakeStmt:
        def __init__(self):
            self.conditions = []

        def where(self, *conditions):
            self.conditions.extend(conditions)
            return self

    stmt = FakeStmt()

    result = AgentObservabilityService._apply_status_filter(stmt, ("success", "failed", "paused"))

    assert result is stmt
    assert len(stmt.conditions) == 1
    with pytest.raises(ValueError, match="Unsupported status"):
        AgentObservabilityService._apply_status_filter(FakeStmt(), ("unknown",))


def test_list_logs_sorts_by_requested_field(monkeypatch: pytest.MonkeyPatch) -> None:
    service = AgentObservabilityService(session=None)
    app = SimpleNamespace(id="app-1")
    rows = [
        {"id": "old", "source": {"id": "webapp:app-1"}, "created_at": 10, "updated_at": 100},
        {"id": "new", "source": {"id": "webapp:app-1"}, "created_at": 20, "updated_at": 50},
    ]
    monkeypatch.setattr(service, "_list_webapp_conversation_logs", lambda **kwargs: rows)
    monkeypatch.setattr(service, "_list_workflow_conversation_logs", lambda **kwargs: [])

    payload = service.list_logs(
        app=app,  # type: ignore[arg-type]
        agent_id="agent-1",
        params=AgentLogQueryParams(sources=("webapp:app-1",), sort_by="created_at", sort_order="asc"),
    )

    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",
        name="Iris",
        icon_type=SimpleNamespace(value="emoji"),
        icon="robot",
        icon_background="#fff",
    )

    webapp_source = AgentObservabilityService._serialize_webapp_source(app)  # type: ignore[arg-type]
    workflow_app_source = AgentObservabilityService._serialize_workflow_app_source(app=app)  # type: ignore[arg-type]
    workflow_source = AgentObservabilityService._serialize_workflow_source(
        app=app,  # type: ignore[arg-type]
        workflow_id="workflow-1",
        workflow_version="v1",
        node_id="node-1",
    )

    assert webapp_source == {
        "id": "webapp:app-1",
        "type": "webapp",
        "app_id": "app-1",
        "app_name": "Iris",
        "app_icon_type": "emoji",
        "app_icon": "robot",
        "app_icon_background": "#fff",
        "workflow_id": None,
        "workflow_version": None,
        "node_id": None,
    }
    assert workflow_app_source == {
        "id": "workflow:app-1",
        "type": "workflow",
        "app_id": "app-1",
        "app_name": "Iris",
        "app_icon_type": "emoji",
        "app_icon": "robot",
        "app_icon_background": "#fff",
        "workflow_id": None,
        "workflow_version": None,
        "node_id": None,
    }
    assert workflow_source["id"] == "workflow:app-1:workflow-1:v1:node-1"
    assert workflow_source["type"] == "workflow"
    assert workflow_source["workflow_id"] == "workflow-1"


def test_list_workflow_sources_deduplicates_versions_and_nodes_by_app() -> None:
    app_a = SimpleNamespace(
        id="app-a",
        name="Alpha",
        icon_type=None,
        icon=None,
        icon_background=None,
    )
    app_b = SimpleNamespace(
        id="app-b",
        name="Beta",
        icon_type=None,
        icon=None,
        icon_background=None,
    )

    class FakeResult:
        def all(self):
            return [(app_a,), (app_a,), (app_a,), (app_b,)]

    class FakeSession:
        def execute(self, stmt):
            stmt.compile()
            return FakeResult()

    service = AgentObservabilityService(FakeSession())

    sources = service._list_workflow_sources(
        app=SimpleNamespace(tenant_id="tenant-1"),  # type: ignore[arg-type]
        agent_id="agent-1",
    )

    assert [source["id"] for source in sources] == ["workflow:app-a", "workflow:app-b"]


def test_serialize_log_message_returns_frontend_log_shape() -> None:
    created_at = datetime(2026, 6, 17, 1, 2, 3, tzinfo=UTC)
    updated_at = datetime(2026, 6, 17, 1, 3, 3, tzinfo=UTC)
    message = SimpleNamespace(
        id="message-1",
        conversation_id="conversation-1",
        query="hello",
        answer="hi",
        error=None,
        status=MessageStatus.NORMAL,
        invoke_from=InvokeFrom.EXPLORE,
        from_source=ConversationFromSource.CONSOLE,
        from_end_user_id=None,
        from_account_id="account-1",
        message_tokens=3,
        answer_tokens=4,
        total_price=Decimal("0.0001"),
        currency="USD",
        provider_response_latency=1.25,
        created_at=created_at,
        updated_at=updated_at,
    )
    conversation = SimpleNamespace(name="Debug conversation")

    payload = AgentObservabilityService.serialize_log_message(message, conversation)  # type: ignore[arg-type]

    assert payload == {
        "id": "message-1",
        "message_id": "message-1",
        "conversation_id": "conversation-1",
        "conversation_name": "Debug conversation",
        "query": "hello",
        "answer": "hi",
        "status": "success",
        "error": None,
        "source": "explore",
        "from_source": "console",
        "from_end_user_id": None,
        "from_account_id": "account-1",
        "message_tokens": 3,
        "answer_tokens": 4,
        "total_tokens": 7,
        "total_price": "0.0001",
        "currency": "USD",
        "latency": 1.25,
        "created_at": int(created_at.timestamp()),
        "updated_at": int(updated_at.timestamp()),
    }


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 = [
        {
            "date": "2026-06-16",
            "message_count": 2,
            "conversation_count": 1,
            "end_user_count": 1,
            "token_count": 30,
            "total_price": Decimal("0.003"),
            "avg_latency": 1.5,
            "latency_sum": 3,
            "answer_tokens": 12,
            "like_count": 1,
        },
        {
            "date": "2026-06-17",
            "message_count": 1,
            "conversation_count": 1,
            "end_user_count": 1,
            "token_count": 20,
            "total_price": Decimal("0.002"),
            "avg_latency": 2,
            "latency_sum": 2,
            "answer_tokens": 8,
            "like_count": 1,
        },
    ]

    charts = AgentObservabilityService._build_charts(rows)
    summary = AgentObservabilityService._build_summary(rows)

    assert charts["token_usage"] == [
        {"date": "2026-06-16", "token_count": 30, "total_price": "0.003", "currency": "USD"},
        {"date": "2026-06-17", "token_count": 20, "total_price": "0.002", "currency": "USD"},
    ]
    assert charts["average_response_time"] == [
        {"date": "2026-06-16", "latency": 1500.0},
        {"date": "2026-06-17", "latency": 2000.0},
    ]
    assert summary == {
        "total_messages": 3,
        "total_conversations": 2,
        "total_end_users": 2,
        "total_tokens": 50,
        "total_price": "0.005",
        "currency": "USD",
        "average_session_interactions": 1.5,
        "average_response_time": 1666.6667,
        "tokens_per_second": 4.0,
        "user_satisfaction_rate": 66.67,
    }
