test: migrate agent observability sessions and ORM models to SQLite (#40588)

api/tests/unit_tests/services/agent/test_agent_observability_service.py

@@ -1,12 +1,17 @@
+import json
 from datetime import UTC, datetime
 from decimal import Decimal
 from types import SimpleNamespace
-from typing import Protocol
 
 import pytest
+from sqlalchemy import Select, select
+from sqlalchemy.orm import Session
 
 from core.app.entities.app_invoke_entities import InvokeFrom
 from graphon.enums import WorkflowNodeExecutionStatus
+from libs.datetime_utils import naive_utc_now
+from models.agent import WorkflowAgentBindingType, WorkflowAgentNodeBinding
+from models.agent_config_entities import WorkflowNodeJobConfig
 from models.enums import (
     ConversationFromSource,
     CreatorUserRole,
@@ -14,10 +19,186 @@ from models.enums import (
     FeedbackRating,
     MessageStatus,
 )
+from models.model import App, AppMode, Conversation, IconType, Message, MessageFeedback
+from models.workflow import (
+    WorkflowExecutionStatus,
+    WorkflowNodeExecutionModel,
+    WorkflowNodeExecutionTriggeredFrom,
+    WorkflowRun,
+    WorkflowRunTriggeredFrom,
+    WorkflowType,
+)
 from services.agent import observability_service as observability_service_module
 from services.agent.observability_service import AgentLogQueryParams, AgentObservabilityService
 
 
+def _app(*, app_id: str = "app-1", name: str = "Iris", mode: AppMode = AppMode.AGENT_CHAT) -> App:
+    return App(
+        id=app_id,
+        tenant_id="tenant-1",
+        name=name,
+        mode=mode,
+        icon_type=IconType.EMOJI,
+        icon="robot",
+        icon_background="#fff",
+        enable_site=False,
+        enable_api=False,
+    )
+
+
+def _conversation(*, conversation_id: str = "conversation-1", app_id: str = "app-1") -> Conversation:
+    return Conversation(
+        id=conversation_id,
+        app_id=app_id,
+        mode=AppMode.AGENT_CHAT,
+        name="Debug conversation",
+        inputs={},
+        from_source=ConversationFromSource.CONSOLE,
+        from_end_user_id="end-user-1",
+    )
+
+
+def _message(
+    *,
+    message_id: str = "message-1",
+    conversation_id: str = "conversation-1",
+    app_id: str = "app-1",
+    created_at: datetime | None = None,
+) -> Message:
+    timestamp = created_at or naive_utc_now()
+    return Message(
+        id=message_id,
+        app_id=app_id,
+        conversation_id=conversation_id,
+        inputs={},
+        query="hello",
+        message={},
+        answer="hi",
+        status=MessageStatus.NORMAL,
+        message_unit_price=Decimal(0),
+        answer_unit_price=Decimal(0),
+        total_price=Decimal("0.0001"),
+        currency="USD",
+        from_source=ConversationFromSource.CONSOLE,
+        from_account_id="account-1",
+        invoke_from=InvokeFrom.EXPLORE,
+        message_tokens=3,
+        answer_tokens=4,
+        provider_response_latency=1.25,
+        created_at=timestamp,
+        updated_at=timestamp,
+    )
+
+
+def _feedback(
+    *,
+    message_id: str = "message-1",
+    conversation_id: str = "conversation-1",
+    source: FeedbackFromSource = FeedbackFromSource.USER,
+    rating: FeedbackRating = FeedbackRating.LIKE,
+    content: str | None = "Useful",
+) -> MessageFeedback:
+    return MessageFeedback(
+        app_id="app-1",
+        conversation_id=conversation_id,
+        message_id=message_id,
+        rating=rating,
+        from_source=source,
+        content=content,
+    )
+
+
+def _workflow_run(*, workflow_type: WorkflowType = WorkflowType.WORKFLOW) -> WorkflowRun:
+    created_at = datetime(2026, 7, 21, 7, 0, 19)
+    return WorkflowRun(
+        id="workflow-run-1",
+        tenant_id="tenant-1",
+        app_id="workflow-app-1",
+        workflow_id="workflow-1",
+        type=workflow_type,
+        triggered_from=WorkflowRunTriggeredFrom.APP_RUN,
+        version="v1",
+        graph="{}",
+        inputs="{}",
+        status=WorkflowExecutionStatus.SUCCEEDED,
+        outputs="{}",
+        error=None,
+        elapsed_time=59.93,
+        total_tokens=454_064,
+        total_steps=1,
+        created_by_role=CreatorUserRole.END_USER,
+        created_by="end-user-1",
+        created_at=created_at,
+        finished_at=created_at,
+    )
+
+
+def _node_execution(
+    *,
+    execution_id: str = "node-execution-1",
+    status: WorkflowNodeExecutionStatus = WorkflowNodeExecutionStatus.SUCCEEDED,
+) -> WorkflowNodeExecutionModel:
+    created_at = datetime(2026, 7, 23, 7, 0, 19, tzinfo=UTC)
+    return WorkflowNodeExecutionModel(
+        id=execution_id,
+        tenant_id="tenant-1",
+        app_id="workflow-app-1",
+        workflow_id="workflow-1",
+        triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN,
+        workflow_run_id="workflow-run-1",
+        index=1,
+        predecessor_node_id=None,
+        node_execution_id=execution_id,
+        node_id="node-1",
+        node_type="agent",
+        title="Agent",
+        inputs="{}",
+        process_data="{}",
+        outputs="{}",
+        status=status,
+        error=None,
+        elapsed_time=59.93,
+        execution_metadata=json.dumps(
+            {
+                "agent_log": {
+                    "agent_backend": {
+                        "usage": {
+                            "prompt_tokens": 451_938,
+                            "completion_tokens": 2_126,
+                            "total_tokens": 454_064,
+                            "total_price": "2.323470",
+                            "currency": "USD",
+                            "latency": 59.93,
+                        }
+                    }
+                }
+            }
+        ),
+        created_at=created_at,
+        created_by_role=CreatorUserRole.END_USER,
+        created_by="end-user-1",
+        finished_at=None,
+    )
+
+
+def _workflow_binding(
+    *, app_id: str = "workflow-app-1", binding_id: str | None = None, node_id: str = "node-1"
+) -> WorkflowAgentNodeBinding:
+    return WorkflowAgentNodeBinding(
+        id=binding_id or f"binding-{app_id}-{node_id}",
+        tenant_id="tenant-1",
+        app_id=app_id,
+        workflow_id="workflow-1",
+        workflow_version="v1",
+        node_id=node_id,
+        binding_type=WorkflowAgentBindingType.ROSTER_AGENT,
+        agent_id="agent-1",
+        current_snapshot_id="snapshot-1",
+        node_job_config=WorkflowNodeJobConfig(),
+        created_by="account-1",
+    )
+
+
 def test_resolve_source_accepts_frontend_aliases() -> None:
     assert AgentObservabilityService.resolve_source(None) is None
     assert AgentObservabilityService.resolve_source("all") is None
@@ -127,69 +308,37 @@ def test_workflow_metadata_numeric_sql_supports_postgresql_and_mysql(monkeypatch
     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 test_workflow_statistics_include_run_without_message(
+    monkeypatch: pytest.MonkeyPatch, sqlite_session: Session
+) -> None:
+    workflow_app = _app(app_id="workflow-app-1", name="Workflow App", mode=AppMode.WORKFLOW)
+    sqlite_session.add_all([workflow_app, _workflow_run(), _node_execution(), _workflow_binding()])
+    sqlite_session.commit()
 
-        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, "dify_config", SimpleNamespace(DB_TYPE="mysql"))
+    monkeypatch.setattr(observability_service_module, "convert_datetime_to_date", lambda field: f"DATE({field})")
     monkeypatch.setattr(
-        observability_service_module,
-        "convert_datetime_to_date",
-        lambda field: f"DATE({field})",
+        AgentObservabilityService,
+        "_workflow_execution_metadata_numeric_sql",
+        staticmethod(
+            lambda path, numeric_type: (
+                f"CAST(json_extract(wne.execution_metadata, '$.{'.'.join(path)}') AS {numeric_type})"
+            )
+        ),
     )
-    session = FakeSession()
-    service = AgentObservabilityService(session)
+    service = AgentObservabilityService(sqlite_session)
 
     payload = service.get_statistics_summary(
-        app=SimpleNamespace(id="agent-app", tenant_id="tenant-1"),  # type: ignore[arg-type]
+        app=_app(app_id="agent-app"),
         agent_id="agent-1",
-        params=observability_service_module.AgentStatisticsQueryParams(source="workflow:workflow-app"),
+        params=observability_service_module.AgentStatisticsQueryParams(source="workflow:workflow-app-1"),
     )
 
     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]
+    assert Decimal(payload["summary"]["total_price"]) == Decimal("2.323470")
 
 
 def test_merge_daily_statistics_combines_webapp_and_workflow_rows() -> None:
@@ -239,27 +388,19 @@ def test_merge_daily_statistics_combines_webapp_and_workflow_rows() -> None:
 
 
 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()
+    stmt = select(Message)
 
     result = AgentObservabilityService._apply_status_filter(stmt, ("success", "failed", "paused"))
 
-    assert result is stmt
-    assert len(stmt.conditions) == 1
+    assert isinstance(result, Select)
+    assert len(result._where_criteria) == 1
     with pytest.raises(ValueError, match="Unsupported status"):
-        AgentObservabilityService._apply_status_filter(FakeStmt(), ("unknown",))
+        AgentObservabilityService._apply_status_filter(select(Message), ("unknown",))
 
 
 def test_list_logs_sorts_by_requested_field(monkeypatch: pytest.MonkeyPatch) -> None:
     service = AgentObservabilityService(session=None)
-    app = SimpleNamespace(id="app-1")
+    app = _app()
     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},
@@ -268,7 +409,7 @@ def test_list_logs_sorts_by_requested_field(monkeypatch: pytest.MonkeyPatch) ->
     monkeypatch.setattr(service, "_list_workflow_conversation_logs", lambda **kwargs: [])
 
     payload = service.list_logs(
-        app=app,  # type: ignore[arg-type]
+        app=app,
         agent_id="agent-1",
         params=AgentLogQueryParams(sources=("webapp:app-1",), sort_by="created_at", sort_order="asc"),
     )
@@ -278,7 +419,7 @@ def test_list_logs_sorts_by_requested_field(monkeypatch: pytest.MonkeyPatch) ->
 
 def test_list_log_messages_merges_deduplicates_and_sorts_sources(monkeypatch: pytest.MonkeyPatch) -> None:
     service = AgentObservabilityService(session=None)
-    webapp_message = SimpleNamespace(id="shared")
+    webapp_message = _message(message_id="shared")
     webapp_row = {"id": "shared", "created_at": 10, "updated_at": 30}
     workflow_rows = [
         {"id": "shared", "created_at": 10, "updated_at": 20},
@@ -290,7 +431,7 @@ def test_list_log_messages_merges_deduplicates_and_sorts_sources(monkeypatch: py
     monkeypatch.setattr(service, "_list_workflow_messages", lambda **kwargs: workflow_rows)
 
     payload = service.list_log_messages(
-        app=SimpleNamespace(id="agent-app"),  # type: ignore[arg-type]
+        app=_app(app_id="agent-app"),
         agent_id="agent-1",
         conversation_id="execution-1",
         params=AgentLogQueryParams(
@@ -309,52 +450,29 @@ def test_list_log_messages_merges_deduplicates_and_sorts_sources(monkeypatch: py
     }
 
 
-def test_list_webapp_conversation_logs_includes_feedback_rates(monkeypatch: pytest.MonkeyPatch) -> None:
-    timestamp = datetime(2026, 7, 23, 7, 0, 19, tzinfo=UTC)
-    conversation = SimpleNamespace(
-        id="conversation-1",
-        name="Feedback conversation",
-        from_end_user_id="end-user-1",
-        read_at=None,
-    )
-
-    class FakeRow:
-        message_count = 2
-        paused_count = 0
-        failed_count = 0
-        created_at = timestamp
-        updated_at = timestamp
-
-        def __getitem__(self, index: int) -> SimpleNamespace:
-            if index != 0:
-                raise IndexError(index)
-            return conversation
-
-    class FakeResult:
-        def all(self) -> list[FakeRow]:
-            return [FakeRow()]
-
-    class FakeSession:
-        def execute(self, stmt: object) -> FakeResult:
-            str(stmt)
-            return FakeResult()
-
-    app = SimpleNamespace(
-        id="app-1",
-        name="Agent WebApp",
-        icon_type=None,
-        icon=None,
-        icon_background=None,
-    )
-    service = AgentObservabilityService(FakeSession())
-    monkeypatch.setattr(
-        service,
-        "_list_conversation_feedback_rates",
-        lambda **kwargs: {"conversation-1": {"user_rate": 0.5, "operation_rate": 1.0}},
+def test_list_webapp_conversation_logs_includes_feedback_rates(sqlite_session: Session) -> None:
+    timestamp = datetime(2026, 7, 23, 7, 0, 19)
+    app = _app(name="Agent WebApp")
+    conversation = _conversation()
+    conversation.name = "Feedback conversation"
+    first_message = _message(created_at=timestamp)
+    second_message = _message(message_id="message-2", created_at=timestamp)
+    sqlite_session.add_all(
+        [
+            app,
+            conversation,
+            first_message,
+            second_message,
+            _feedback(message_id=first_message.id),
+            _feedback(message_id=second_message.id, rating=FeedbackRating.DISLIKE),
+            _feedback(message_id=first_message.id, source=FeedbackFromSource.ADMIN),
+        ]
     )
+    sqlite_session.commit()
+    service = AgentObservabilityService(sqlite_session)
 
     rows = service._list_webapp_conversation_logs(
-        app=app,  # type: ignore[arg-type]
+        app=app,
         params=AgentLogQueryParams(),
         source_filter=AgentObservabilityService.resolve_source_filter("webapp"),
     )
@@ -363,90 +481,42 @@ def test_list_webapp_conversation_logs_includes_feedback_rates(monkeypatch: pyte
     assert rows[0]["operation_rate"] == 1.0
 
 
-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)
+def test_list_workflow_logs_uses_node_executions_without_messages(sqlite_session: Session) -> None:
+    workflow_app = _app(app_id="workflow-app-1", name="Marketing Department", mode=AppMode.WORKFLOW)
+    sqlite_session.add_all([workflow_app, _workflow_run(), _node_execution(), _workflow_binding()])
+    sqlite_session.commit()
+    service = AgentObservabilityService(sqlite_session)
 
     rows = service._list_workflow_conversation_logs(
-        app=SimpleNamespace(tenant_id="tenant-1"),  # type: ignore[arg-type]
+        app=_app(app_id="agent-app"),
         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)
+def test_list_workflow_messages_uses_node_execution_identity(
+    monkeypatch: pytest.MonkeyPatch, sqlite_session: Session
+) -> None:
+    node_execution = _node_execution()
+    sqlite_session.add_all(
+        [
+            _app(app_id="workflow-app-1", name="Workflow App", mode=AppMode.WORKFLOW),
+            _workflow_run(),
+            node_execution,
+            _workflow_binding(),
+        ]
+    )
+    sqlite_session.commit()
+    service = AgentObservabilityService(sqlite_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]
+        app=_app(app_id="agent-app"),
         agent_id="agent-1",
         conversation_id="node-execution-1",
         params=AgentLogQueryParams(),
@@ -454,21 +524,10 @@ def test_list_workflow_messages_uses_node_execution_identity(monkeypatch: pytest
     )
 
     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()
+    stmt = select(WorkflowNodeExecutionModel)
     params = AgentLogQueryParams(
         start=datetime(2026, 7, 1, tzinfo=UTC),
         end=datetime(2026, 8, 1, tzinfo=UTC),
@@ -479,48 +538,34 @@ def test_apply_workflow_node_filters_supports_time_keyword_and_status() -> None:
     result = AgentObservabilityService._apply_workflow_node_filters(
         stmt,
         params=params,
-        workflow_app=SimpleNamespace(name=observability_service_module.App.name),
+        workflow_app=observability_service_module.App,
     )
 
-    assert result is stmt
-    assert len(stmt.conditions) == 4
+    assert isinstance(result, Select)
+    assert len(result._where_criteria) == 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()
+    stmt = select(WorkflowNodeExecutionModel)
 
     result = AgentObservabilityService._apply_workflow_node_status_filter(stmt, ("normal", "error", "paused"))
 
-    assert result is stmt
-    assert len(stmt.conditions) == 1
-    empty_stmt = FakeStmt()
+    assert isinstance(result, Select)
+    assert len(result._where_criteria) == 1
+    empty_stmt = select(WorkflowNodeExecutionModel)
     assert AgentObservabilityService._apply_workflow_node_status_filter(empty_stmt, ()) is empty_stmt
-    assert empty_stmt.conditions == []
+    assert empty_stmt._where_criteria == ()
     with pytest.raises(ValueError, match="Unsupported status"):
-        AgentObservabilityService._apply_workflow_node_status_filter(FakeStmt(), ("unknown",))
+        AgentObservabilityService._apply_workflow_node_status_filter(select(WorkflowNodeExecutionModel), ("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",
-    )
+    app = _app()
 
-    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]
+    webapp_source = AgentObservabilityService._serialize_webapp_source(app)
+    workflow_app_source = AgentObservabilityService._serialize_workflow_app_source(app=app)
     workflow_source = AgentObservabilityService._serialize_workflow_source(
-        app=app,  # type: ignore[arg-type]
+        app=app,
         workflow_id="workflow-1",
         workflow_version="v1",
         node_id="node-1",
@@ -555,35 +600,23 @@ def test_source_serializers_return_structured_frontend_shape() -> None:
     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,
+def test_list_workflow_sources_deduplicates_versions_and_nodes_by_app(sqlite_session: Session) -> None:
+    app_a = _app(app_id="app-a", name="Alpha", mode=AppMode.WORKFLOW)
+    app_b = _app(app_id="app-b", name="Beta", mode=AppMode.WORKFLOW)
+    sqlite_session.add_all(
+        [
+            app_a,
+            app_b,
+            _workflow_binding(app_id="app-a", node_id="node-a-1"),
+            _workflow_binding(app_id="app-a", node_id="node-a-2"),
+            _workflow_binding(app_id="app-b", node_id="node-b-1"),
+        ]
     )
-    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())
+    sqlite_session.commit()
+    service = AgentObservabilityService(sqlite_session)
 
     sources = service._list_workflow_sources(
-        app=SimpleNamespace(tenant_id="tenant-1"),  # type: ignore[arg-type]
+        app=_app(app_id="agent-app"),
         agent_id="agent-1",
     )
 
@@ -593,44 +626,19 @@ def test_list_workflow_sources_deduplicates_versions_and_nodes_by_app() -> None:
 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")
+    message = _message(created_at=created_at)
+    message.updated_at = updated_at
+    conversation = _conversation()
     feedbacks = [
-        SimpleNamespace(
-            rating=FeedbackRating.LIKE,
-            content="Useful",
-            from_source=FeedbackFromSource.USER,
-        ),
-        SimpleNamespace(
+        _feedback(),
+        _feedback(
             rating=FeedbackRating.DISLIKE,
             content="Needs more detail",
-            from_source=FeedbackFromSource.ADMIN,
+            source=FeedbackFromSource.ADMIN,
         ),
     ]
 
-    payload = AgentObservabilityService.serialize_log_message(  # type: ignore[arg-type]
-        message,
-        conversation,
-        feedbacks,
-    )
+    payload = AgentObservabilityService.serialize_log_message(message, conversation, feedbacks)
 
     assert payload == {
         "id": "message-1",
@@ -664,29 +672,22 @@ 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,
+    node_execution = _node_execution()
+    node_execution.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"}}]}}}'
     )
+    node_execution.outputs = '{"output":"Alice owns the follow-up."}'
+    node_execution.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}}}}'
+    )
+    node_execution.elapsed_time = 1.5
+    node_execution.created_at = created_at
+    node_execution.finished_at = finished_at
 
-    payload = AgentObservabilityService.serialize_workflow_node_message(node_execution)  # type: ignore[arg-type]
+    payload = AgentObservabilityService.serialize_workflow_node_message(node_execution)
 
     assert payload == {
         "id": "node-execution-1",
@@ -713,10 +714,10 @@ def test_serialize_workflow_node_message_returns_frontend_log_shape() -> None:
 
 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={
+    node_execution = _node_execution(execution_id="node-execution-2", status=WorkflowNodeExecutionStatus.PAUSED)
+    node_execution.title = "Fallback prompt"
+    node_execution.inputs = json.dumps(
+        {
             "agent_backend_request": {
                 "composition": {
                     "layers": [
@@ -726,9 +727,11 @@ def test_serialize_workflow_node_message_handles_sparse_runtime_data() -> None:
                     ]
                 }
             }
-        },
-        outputs={"output": {"structured": True}},
-        execution_metadata={
+        }
+    )
+    node_execution.outputs = json.dumps({"output": {"structured": True}})
+    node_execution.execution_metadata = json.dumps(
+        {
             "agent_log": {
                 "agent_backend": {
                     "usage": {
@@ -737,17 +740,15 @@ def test_serialize_workflow_node_message_handles_sparse_runtime_data() -> None:
                     }
                 }
             }
-        },
-        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,
+        }
     )
+    node_execution.elapsed_time = 2
+    node_execution.created_by_role = CreatorUserRole.ACCOUNT
+    node_execution.created_by = "account-1"
+    node_execution.created_at = created_at
+    node_execution.finished_at = None
 
-    payload = AgentObservabilityService.serialize_workflow_node_message(node_execution)  # type: ignore[arg-type]
+    payload = AgentObservabilityService.serialize_workflow_node_message(node_execution)
 
     assert payload["query"] == "Fallback prompt"
     assert payload["answer"] == '{"output": {"structured": true}}'
@@ -767,85 +768,53 @@ def test_positive_feedback_rate_uses_rated_messages_as_denominator() -> None:
     assert AgentObservabilityService._positive_feedback_rate(like_count=None, total_count=0) is None
 
 
-def test_list_message_feedbacks_groups_feedbacks_by_message() -> None:
+def test_list_message_feedbacks_groups_feedbacks_by_message(sqlite_session: Session) -> None:
+    first_message = _message()
+    second_message = _message(message_id="message-2")
     feedbacks = [
-        SimpleNamespace(message_id="message-1"),
-        SimpleNamespace(message_id="message-1"),
-        SimpleNamespace(message_id="message-2"),
+        _feedback(),
+        _feedback(rating=FeedbackRating.DISLIKE),
+        _feedback(message_id="message-2"),
     ]
-
-    class Compilable(Protocol):
-        def compile(self) -> object: ...
-
-    class FakeScalarResult:
-        def all(self) -> list[SimpleNamespace]:
-            return feedbacks
-
-    class FakeSession:
-        def __init__(self) -> None:
-            self.scalar_calls = 0
-
-        def scalars(self, stmt: Compilable) -> FakeScalarResult:
-            stmt.compile()
-            self.scalar_calls += 1
-            return FakeScalarResult()
-
-    session = FakeSession()
-    service = AgentObservabilityService(session)  # type: ignore[arg-type]
+    sqlite_session.add_all([_app(), _conversation(), first_message, second_message, *feedbacks])
+    sqlite_session.commit()
+    service = AgentObservabilityService(sqlite_session)
 
     grouped_feedbacks = service._list_message_feedbacks(
-        app=SimpleNamespace(id="app-1"),  # type: ignore[arg-type]
-        messages=[SimpleNamespace(id="message-1"), SimpleNamespace(id="message-2")],  # type: ignore[list-item]
+        app=_app(),
+        messages=[first_message, second_message],
     )
 
-    assert grouped_feedbacks == {
-        "message-1": feedbacks[:2],
-        "message-2": feedbacks[2:],
+    assert {feedback.id for feedback in grouped_feedbacks["message-1"]} == {
+        feedbacks[0].id,
+        feedbacks[1].id,
     }
-    assert service._list_message_feedbacks(app=SimpleNamespace(id="app-1"), messages=[]) == {}  # type: ignore[arg-type]
-    assert session.scalar_calls == 1
+    assert grouped_feedbacks["message-2"] == [feedbacks[2]]
+    assert service._list_message_feedbacks(app=_app(), messages=[]) == {}
 
 
-def test_list_conversation_feedback_rates_maps_user_and_admin_sources() -> None:
-    class FakeResult:
-        def all(self):
-            return [
-                SimpleNamespace(
-                    conversation_id="conversation-1",
-                    from_source=FeedbackFromSource.USER,
-                    like_count=2,
-                    total_count=4,
-                ),
-                SimpleNamespace(
-                    conversation_id="conversation-1",
-                    from_source=FeedbackFromSource.ADMIN,
-                    like_count=1,
-                    total_count=1,
-                ),
-                SimpleNamespace(
-                    conversation_id="conversation-without-ratings",
-                    from_source=FeedbackFromSource.USER,
-                    like_count=0,
-                    total_count=0,
-                ),
-            ]
-
-    class FakeSession:
-        def execute(self, stmt):
-            stmt.compile()
-            return FakeResult()
-
-    service = AgentObservabilityService(FakeSession())
+def test_list_conversation_feedback_rates_maps_user_and_admin_sources(sqlite_session: Session) -> None:
+    messages = [_message(message_id=f"message-{index}") for index in range(1, 5)]
+    feedbacks = [
+        _feedback(message_id="message-1"),
+        _feedback(message_id="message-2"),
+        _feedback(message_id="message-3", rating=FeedbackRating.DISLIKE),
+        _feedback(message_id="message-4", rating=FeedbackRating.DISLIKE),
+        _feedback(message_id="message-1", source=FeedbackFromSource.ADMIN),
+    ]
+    sqlite_session.add_all([_app(), _conversation(), *messages, *feedbacks])
+    sqlite_session.commit()
+    service = AgentObservabilityService(sqlite_session)
 
     rates = service._list_conversation_feedback_rates(
-        app=SimpleNamespace(id="app-1"),  # type: ignore[arg-type]
+        app=_app(),
         conversation_ids=["conversation-1"],
     )
 
     assert rates == {"conversation-1": {"user_rate": 0.5, "operation_rate": 1.0}}
     assert (
         service._list_conversation_feedback_rates(
-            app=SimpleNamespace(id="app-1"),  # type: ignore[arg-type]
+            app=_app(),
             conversation_ids=[],
         )
         == {}
@@ -871,15 +840,10 @@ def test_workflow_node_serialization_helpers_handle_invalid_values() -> None:
 
 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,
-    )
+    node_execution = _node_execution(status=WorkflowNodeExecutionStatus.FAILED)
+    node_execution.created_by_role = CreatorUserRole.ACCOUNT
+    node_execution.created_by = "account-1"
+    node_execution.created_at = created_at
 
     payload = AgentObservabilityService._serialize_workflow_execution_log(
         node_execution_id=node_execution.id,