fix(agent): attribute nested tool activity to its source

Resolve Agent logs, details and usage statistics from the executed source workflow and binding while preserving the outer workflow run relationship.

Author: laipz8200

api/services/agent/observability_service.py

@@ -14,12 +14,19 @@
 
 from configs import dify_config
 from core.app.entities.app_invoke_entities import InvokeFrom
+from core.workflow.node_execution_process_data import WORKFLOW_AGENT_BINDING_ID_KEY
 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, FeedbackFromSource, FeedbackRating, MessageStatus
 from models.model import App, Conversation, Message, MessageFeedback
-from models.workflow import WorkflowNodeExecutionModel, WorkflowRun, WorkflowType
+from models.workflow import (
+    Workflow,
+    WorkflowNodeExecutionModel,
+    WorkflowNodeExecutionTriggeredFrom,
+    WorkflowRun,
+    WorkflowType,
+)
 
 
 @dataclass(frozen=True)
@@ -326,17 +333,15 @@ def _list_workflow_conversation_logs(
                 and_(
                     WorkflowAgentNodeBinding.tenant_id == app.tenant_id,
                     WorkflowAgentNodeBinding.agent_id == agent_id,
-                    WorkflowAgentNodeBinding.app_id == WorkflowRun.app_id,
-                    WorkflowAgentNodeBinding.workflow_id == WorkflowRun.workflow_id,
-                    WorkflowAgentNodeBinding.workflow_version == WorkflowRun.version,
+                    self._workflow_node_binding_condition(),
                 ),
             )
-            .join(workflow_app, workflow_app.id == WorkflowAgentNodeBinding.app_id)
+            .join(
+                workflow_app,
+                and_(workflow_app.id == WorkflowAgentNodeBinding.app_id, workflow_app.tenant_id == app.tenant_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_workflow_node_filters(stmt, params=params, workflow_app=workflow_app)
@@ -443,18 +448,16 @@ def _list_workflow_messages(
                 and_(
                     WorkflowAgentNodeBinding.tenant_id == app.tenant_id,
                     WorkflowAgentNodeBinding.agent_id == agent_id,
-                    WorkflowAgentNodeBinding.app_id == WorkflowRun.app_id,
-                    WorkflowAgentNodeBinding.workflow_id == WorkflowRun.workflow_id,
-                    WorkflowAgentNodeBinding.workflow_version == WorkflowRun.version,
+                    self._workflow_node_binding_condition(),
                 ),
             )
-            .join(workflow_app, workflow_app.id == WorkflowAgentNodeBinding.app_id)
+            .join(
+                workflow_app,
+                and_(workflow_app.id == WorkflowAgentNodeBinding.app_id, workflow_app.tenant_id == app.tenant_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,
             )
         )
         stmt = self._apply_workflow_node_filters(stmt, params=params, workflow_app=workflow_app)
@@ -466,6 +469,65 @@ def _list_workflow_messages(
         )
         return [self.serialize_workflow_node_message(execution) for execution in executions]
 
+    def _workflow_node_binding_condition(self, *, statistics: bool = False) -> sa.ColumnElement[bool]:
+        """Resolve source ownership independently of a Tool's outer lifecycle run."""
+        node = aliased(WorkflowNodeExecutionModel, name="wne") if statistics else WorkflowNodeExecutionModel
+        binding = aliased(WorkflowAgentNodeBinding, name="wanb") if statistics else WorkflowAgentNodeBinding
+        run = aliased(WorkflowRun, name="wr") if statistics else WorkflowRun
+        process_data_json = func.nullif(node.process_data, "")
+        if self._session.get_bind().dialect.name == "sqlite":
+            binding_id = func.json_extract(process_data_json, f"$.{WORKFLOW_AGENT_BINDING_ID_KEY}")
+        else:
+            binding_id = sa.cast(process_data_json, sa.JSON)[WORKFLOW_AGENT_BINDING_ID_KEY].as_string()
+
+        node_table = "wne" if statistics else WorkflowNodeExecutionModel.__tablename__
+        source_version = (
+            select(Workflow.version)
+            .where(
+                Workflow.id == sa.literal_column(f"{node_table}.workflow_id"),
+                Workflow.app_id == sa.literal_column(f"{node_table}.app_id"),
+                Workflow.tenant_id == sa.literal_column(f"{node_table}.tenant_id"),
+            )
+            .scalar_subquery()
+        )
+        old_binding = aliased(WorkflowAgentNodeBinding, name="old_binding")
+        single_binding_version = (
+            select(func.min(old_binding.workflow_version))
+            .where(
+                old_binding.tenant_id == sa.literal_column(f"{node_table}.tenant_id"),
+                old_binding.app_id == sa.literal_column(f"{node_table}.app_id"),
+                old_binding.workflow_id == sa.literal_column(f"{node_table}.workflow_id"),
+                old_binding.node_id == sa.literal_column(f"{node_table}.node_id"),
+            )
+            .having(func.count() == 1)
+            .scalar_subquery()
+        )
+        is_tool = node.triggered_from == WorkflowNodeExecutionTriggeredFrom.WORKFLOW_TOOL
+        # Old Tool rows may outlive their source Workflow. A sole source binding
+        # is still unambiguous; never choose among versions using the caller run.
+        old_version = sa.case((is_tool, func.coalesce(source_version, single_binding_version)), else_=run.version)
+        return and_(
+            node.tenant_id == run.tenant_id,
+            binding.tenant_id == node.tenant_id,
+            binding.app_id == node.app_id,
+            binding.workflow_id == node.workflow_id,
+            binding.node_id == node.node_id,
+            or_(is_tool, and_(node.app_id == run.app_id, node.workflow_id == run.workflow_id)),
+            or_(
+                sa.cast(binding.id, sa.String) == binding_id,
+                and_(binding_id.is_(None), binding.workflow_version == old_version),
+            ),
+        )
+
+    def _statistics_workflow_node_binding_join_sql(self) -> str:
+        # Reuse the log/detail owner predicate in both raw-SQL statistics queries.
+        return str(
+            self._workflow_node_binding_condition(statistics=True).compile(
+                dialect=self._session.get_bind().dialect,
+                compile_kwargs={"literal_binds": True},
+            )
+        )
+
     def _list_workflow_sources(self, *, app: App, agent_id: str) -> list[dict[str, Any]]:
         workflow_app = aliased(App)
         stmt = (
@@ -947,6 +1009,7 @@ def _load_workflow_daily_statistics(
             ("agent_log", "agent_backend", "usage", "completion_tokens"), "BIGINT"
         )
         binding_filters = self._statistics_workflow_binding_filters_sql(source_filter)
+        binding_join = self._statistics_workflow_node_binding_join_sql()
         run_date_filters = ""
         args: dict[str, Any] = {
             "tz": params.timezone,
@@ -981,17 +1044,14 @@ def _load_workflow_daily_statistics(
         COALESCE(SUM(COALESCE(wne.elapsed_time, 0)), 0) AS latency,
         COALESCE(SUM(COALESCE({completion_tokens}, 0)), 0) AS answer_tokens
     FROM workflow_runs wr
+    JOIN workflow_node_executions wne
+        ON wne.workflow_run_id = wr.id
     JOIN workflow_agent_node_bindings wanb
-        ON wanb.tenant_id = :tenant_id
+        ON {binding_join}
         AND wanb.agent_id = :agent_id
-        AND wanb.app_id = wr.app_id
-        AND wanb.workflow_id = wr.workflow_id
-        AND wanb.workflow_version = wr.version
         {binding_filters}
-    JOIN workflow_node_executions wne
-        ON wne.workflow_run_id = wr.id
-        AND wne.node_id = wanb.node_id
-    WHERE wr.type != :chat_workflow_type{run_date_filters}
+    JOIN apps source_app ON source_app.id = wanb.app_id AND source_app.tenant_id = wr.tenant_id
+    WHERE wr.tenant_id = :tenant_id AND wr.type != :chat_workflow_type{run_date_filters}
     GROUP BY wr.id, wr.created_by_role, wr.created_by, wr.created_at
 )
 SELECT
@@ -1085,24 +1145,23 @@ def _statistics_workflow_binding_filters_sql(source_filter: AgentSourceFilter) -
             workflow_binding_filters.append("wanb.node_id = :node_id")
         return f"AND {' AND '.join(workflow_binding_filters)}" if workflow_binding_filters else ""
 
-    @classmethod
-    def _statistics_workflow_message_scope_sql(cls, source_filter: AgentSourceFilter) -> str:
-        binding_filters = cls._statistics_workflow_binding_filters_sql(source_filter)
+    def _statistics_workflow_message_scope_sql(self, source_filter: AgentSourceFilter) -> str:
+        binding_filters = self._statistics_workflow_binding_filters_sql(source_filter)
+        binding_join = self._statistics_workflow_node_binding_join_sql()
         return f"""m.workflow_run_id IS NOT NULL
         AND EXISTS (
             SELECT 1
             FROM workflow_runs wr
+            JOIN workflow_node_executions wne
+                ON wne.workflow_run_id = wr.id
             JOIN workflow_agent_node_bindings wanb
-                ON wanb.tenant_id = :tenant_id
+                ON {binding_join}
                 AND wanb.agent_id = :agent_id
-                AND wanb.app_id = wr.app_id
-                AND wanb.workflow_id = wr.workflow_id
-                AND wanb.workflow_version = wr.version
                 {binding_filters}
-            JOIN workflow_node_executions wne
-                ON wne.workflow_run_id = wr.id
-                AND wne.node_id = wanb.node_id
+            JOIN apps source_app ON source_app.id = wanb.app_id AND source_app.tenant_id = wr.tenant_id
             WHERE wr.id = m.workflow_run_id
+                AND wr.app_id = m.app_id
+                AND wr.tenant_id = :tenant_id
                 AND wr.type = :chat_workflow_type
         )"""
 

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

@@ -20,6 +20,7 @@
 )
 from models.model import App, AppMode, Conversation, Message, MessageFeedback
 from models.workflow import (
+    Workflow,
     WorkflowExecutionStatus,
     WorkflowNodeExecutionModel,
     WorkflowNodeExecutionTriggeredFrom,
@@ -272,15 +273,6 @@ def test_statistics_workflow_app_source_covers_all_versions_and_nodes() -> None:
     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:
     apply_config_overrides(monkeypatch, DB_TYPE="postgresql")
 
@@ -489,6 +481,97 @@ def test_list_workflow_logs_uses_node_executions_without_messages(sqlite_session
     assert rows[0]["source"]["app_name"] == "Marketing Department"
 
 
+@pytest.mark.parametrize("source_identity", ["persisted-binding", "source-version", "unique-legacy-binding"])
+def test_nested_agent_logs_and_messages_keep_source_identity(sqlite_session: Session, source_identity: str) -> None:
+    source_app = _app(app_id="workflow-app-1", name="Source workflow", mode=AppMode.WORKFLOW)
+    source = Workflow(
+        id="workflow-1",
+        tenant_id="tenant-1",
+        app_id=source_app.id,
+        type=WorkflowType.WORKFLOW,
+        version="v1",
+        graph="{}",
+        _features="{}",
+        created_by="account-1",
+    )
+    run = _workflow_run()
+    run.app_id, run.workflow_id, run.version = "caller-app", "caller-workflow", "caller-version"
+    execution = _node_execution()
+    execution.triggered_from = WorkflowNodeExecutionTriggeredFrom.WORKFLOW_TOOL
+    binding = _workflow_binding()
+    if source_identity == "persisted-binding":
+        execution.process_data = json.dumps({"workflow_agent_binding_id": binding.id})
+    unused_binding = _workflow_binding(binding_id="unused-binding")
+    unused_binding.workflow_version = "v2"
+    sqlite_session.add_all([source_app, run, execution, binding])
+    if source_identity != "unique-legacy-binding":
+        sqlite_session.add_all([source, unused_binding])
+    sqlite_session.commit()
+    service = AgentObservabilityService(sqlite_session)
+    params = AgentLogQueryParams(sources=("workflow:workflow-app-1",))
+
+    logs = service.list_logs(app=_app(app_id="agent-app"), agent_id="agent-1", params=params)
+    messages = service.list_log_messages(
+        app=_app(app_id="agent-app"), agent_id="agent-1", conversation_id=execution.id, params=params
+    )
+
+    assert logs["total"] == messages["total"] == 1
+    assert logs["data"][0]["id"] == messages["data"][0]["id"] == execution.id
+    assert logs["data"][0]["source"]["id"] == "workflow:workflow-app-1:workflow-1:v1:node-1"
+    assert messages["data"][0]["total_tokens"] == 454_064
+    for source_filter in ("workflow:caller-app", "workflow:workflow-app-1:workflow-1:v2:node-1"):
+        params = AgentLogQueryParams(sources=(source_filter,))
+        assert service.list_logs(app=_app(app_id="agent-app"), agent_id="agent-1", params=params)["total"] == 0
+        assert (
+            service.list_log_messages(
+                app=_app(app_id="agent-app"), agent_id="agent-1", conversation_id=execution.id, params=params
+            )["total"]
+            == 0
+        )
+
+    if source_identity == "unique-legacy-binding":
+        # Without a captured binding or source Workflow, another Agent/version
+        # makes attribution ambiguous, even when filtering for only agent-1.
+        unused_binding.agent_id = "another-agent"
+        sqlite_session.add(unused_binding)
+        sqlite_session.commit()
+        params = AgentLogQueryParams(sources=("workflow",))
+        assert service.list_logs(app=_app(app_id="agent-app"), agent_id="agent-1", params=params)["total"] == 0
+        assert (
+            service.list_log_messages(
+                app=_app(app_id="agent-app"), agent_id="agent-1", conversation_id=execution.id, params=params
+            )["total"]
+            == 0
+        )
+
+
+@pytest.mark.parametrize("mismatch", ["run-tenant", "source-tenant", "binding-id", "unrelated-run"])
+def test_workflow_agent_observability_rejects_mismatched_ownership(sqlite_session: Session, mismatch: str) -> None:
+    source_app = _app(app_id="workflow-app-1", mode=AppMode.WORKFLOW)
+    run, execution, binding = _workflow_run(), _node_execution(), _workflow_binding()
+    execution.process_data = json.dumps({"workflow_agent_binding_id": binding.id})
+    if mismatch == "run-tenant":
+        run.tenant_id = "other-tenant"
+    elif mismatch == "source-tenant":
+        source_app.tenant_id = "other-tenant"
+    elif mismatch == "binding-id":
+        execution.process_data = json.dumps({"workflow_agent_binding_id": "missing-binding"})
+    else:
+        run.app_id, run.workflow_id = "unrelated-app", "unrelated-workflow"
+    sqlite_session.add_all([source_app, run, execution, binding])
+    sqlite_session.commit()
+    service = AgentObservabilityService(sqlite_session)
+    params = AgentLogQueryParams(sources=("workflow",))
+
+    assert service.list_logs(app=_app(app_id="agent-app"), agent_id="agent-1", params=params)["total"] == 0
+    assert (
+        service.list_log_messages(
+            app=_app(app_id="agent-app"), agent_id="agent-1", conversation_id=execution.id, params=params
+        )["total"]
+        == 0
+    )
+
+
 def test_list_workflow_messages_uses_node_execution_identity(
     monkeypatch: pytest.MonkeyPatch, sqlite_session: Session
 ) -> None: