From 79662d49af8b66ef4aabb2a3ad22df451a0e048f Mon Sep 17 00:00:00 2001 From: zyssyz123 <916125788@qq.com> Date: Wed, 15 Jul 2026 22:08:57 +0800 Subject: [PATCH] fix(agent): aggregate monitor sources by workflow app (#39021) Co-authored-by: yyh <92089059+lyzno1@users.noreply.github.com> --- api/controllers/console/agent/roster.py | 7 +- api/openapi/markdown/console-openapi.md | 10 +-- api/services/agent/observability_service.py | 46 +++++++----- .../agent/test_agent_observability_service.py | 74 ++++++++++++++++++- 4 files changed, 111 insertions(+), 26 deletions(-) diff --git a/api/controllers/console/agent/roster.py b/api/controllers/console/agent/roster.py index c4acb501cc3..744378e2383 100644 --- a/api/controllers/console/agent/roster.py +++ b/api/controllers/console/agent/roster.py @@ -177,7 +177,7 @@ class AgentLogsQuery(BaseModel): default_factory=list, description=( "Filter by one or more source IDs, e.g. webapp: " - "or workflow::::" + "or workflow:. Exact workflow:::: IDs remain supported." ), ) sort_by: str = Field(default="updated_at", description="Sort by created_at or updated_at") @@ -221,7 +221,10 @@ class AgentLogsQuery(BaseModel): class AgentStatisticsQuery(BaseModel): source: str | None = Field( default=None, - description="Filter by all, console/explore, api/service-api, web-app, debugger, openapi, or trigger", + description=( + "Filter by a structured webapp: or workflow: source ID. " + "Legacy invoke sources and exact workflow version/node source IDs remain supported." + ), ) start: str | None = Field(default=None, description="Start date (YYYY-MM-DD HH:MM)") end: str | None = Field(default=None, description="End date (YYYY-MM-DD HH:MM)") diff --git a/api/openapi/markdown/console-openapi.md b/api/openapi/markdown/console-openapi.md index 578a5b0be07..ca151d20a91 100644 --- a/api/openapi/markdown/console-openapi.md +++ b/api/openapi/markdown/console-openapi.md @@ -1151,7 +1151,7 @@ Commit an uploaded file into the Agent App drive under files/ | sort_by | query | Sort by created_at or updated_at | No | string,
**Default:** updated_at | | sort_order | query | Sort order: asc or desc | No | string,
**Default:** desc | | source | query | Deprecated single source filter | No | string | -| sources | query | Filter by one or more source IDs, e.g. webapp: or workflow:::: | No | [ string ] | +| sources | query | Filter by one or more source IDs, e.g. webapp: or workflow:. Exact workflow:::: IDs remain supported. | No | [ string ] | | start | query | Start date (YYYY-MM-DD HH:MM) | No | string | | status | query | Deprecated single status filter | No | string | | statuses | query | Filter by one or more of success, failed, paused | No | [ string ] | @@ -1175,7 +1175,7 @@ Commit an uploaded file into the Agent App drive under files/ | sort_by | query | Sort by created_at or updated_at | No | string,
**Default:** updated_at | | sort_order | query | Sort order: asc or desc | No | string,
**Default:** desc | | source | query | Deprecated single source filter | No | string | -| sources | query | Filter by one or more source IDs, e.g. webapp: or workflow:::: | No | [ string ] | +| sources | query | Filter by one or more source IDs, e.g. webapp: or workflow:. Exact workflow:::: IDs remain supported. | No | [ string ] | | start | query | Start date (YYYY-MM-DD HH:MM) | No | string | | status | query | Deprecated single status filter | No | string | | statuses | query | Filter by one or more of success, failed, paused | No | [ string ] | @@ -1372,7 +1372,7 @@ Infer CLI tool + ENV suggestions from a standardized Agent App skill | Name | Located in | Description | Required | Schema | | ---- | ---------- | ----------- | -------- | ------ | | end | query | End date (YYYY-MM-DD HH:MM) | No | string | -| source | query | Filter by all, console/explore, api/service-api, web-app, debugger, openapi, or trigger | No | string | +| source | query | Filter by a structured webapp: or workflow: source ID. Legacy invoke sources and exact workflow version/node source IDs remain supported. | No | string | | start | query | Start date (YYYY-MM-DD HH:MM) | No | string | | agent_id | path | | Yes | string (uuid) | @@ -14386,7 +14386,7 @@ section may be empty, which is how callers express "no knowledge layer". | sort_by | string,
**Default:** updated_at | Sort by created_at or updated_at | No | | sort_order | string,
**Default:** desc | Sort order: asc or desc | No | | source | string | Deprecated single source filter | No | -| sources | [ string ] | Filter by one or more source IDs, e.g. webapp: or workflow:::: | No | +| sources | [ string ] | Filter by one or more source IDs, e.g. webapp: or workflow:. Exact workflow:::: IDs remain supported. | No | | start | string | Start date (YYYY-MM-DD HH:MM) | No | | status | string | Deprecated single status filter | No | | statuses | [ string ] | Filter by one or more of success, failed, paused | No | @@ -14835,7 +14835,7 @@ Origin that created or imported the Agent. | Name | Type | Description | Required | | ---- | ---- | ----------- | -------- | | end | string | End date (YYYY-MM-DD HH:MM) | No | -| source | string | Filter by all, console/explore, api/service-api, web-app, debugger, openapi, or trigger | No | +| source | string | Filter by a structured webapp: or workflow: source ID. Legacy invoke sources and exact workflow version/node source IDs remain supported. | No | | start | string | Start date (YYYY-MM-DD HH:MM) | No | #### AgentStatus diff --git a/api/services/agent/observability_service.py b/api/services/agent/observability_service.py index a209b8de0f0..c542422d9b7 100644 --- a/api/services/agent/observability_service.py +++ b/api/services/agent/observability_service.py @@ -94,15 +94,17 @@ class AgentObservabilityService: if lowered == "workflow": return AgentSourceFilter(kind="workflow") if lowered.startswith("workflow:"): - parts = normalized.split(":", 4) - if len(parts) != 5 or not all(parts[1:]): + parts = normalized.split(":") + if len(parts) == 2 and parts[1]: + return AgentSourceFilter(kind="workflow", app_id=parts[1]) + if len(parts) < 5 or not all(parts[1:]): raise ValueError(f"Unsupported source: {source}") return AgentSourceFilter( kind="workflow", app_id=parts[1], workflow_id=parts[2], - workflow_version=parts[3], - node_id=parts[4], + workflow_version=":".join(parts[3:-1]), + node_id=parts[-1], ) return AgentSourceFilter(kind="webapp", invoke_from=cls.resolve_source(source)) @@ -390,26 +392,17 @@ class AgentObservabilityService: def _list_workflow_sources(self, *, app: App, agent_id: str) -> list[dict[str, Any]]: workflow_app = aliased(App) stmt = ( - select( - workflow_app, - WorkflowAgentNodeBinding.workflow_id, - WorkflowAgentNodeBinding.workflow_version, - WorkflowAgentNodeBinding.node_id, - ) + select(workflow_app) + .select_from(WorkflowAgentNodeBinding) .join(workflow_app, workflow_app.id == WorkflowAgentNodeBinding.app_id) .where(WorkflowAgentNodeBinding.tenant_id == app.tenant_id, WorkflowAgentNodeBinding.agent_id == agent_id) - .order_by(workflow_app.name.asc(), WorkflowAgentNodeBinding.node_id.asc()) + .order_by(workflow_app.name.asc(), workflow_app.id.asc()) ) rows = self._session.execute(stmt).all() deduped: dict[str, dict[str, Any]] = {} for row in rows: - source = self._serialize_workflow_source( - app=row[0], - workflow_id=row.workflow_id, - workflow_version=row.workflow_version, - node_id=row.node_id, - ) - deduped[source["id"]] = source + source_app = row[0] + deduped[source_app.id] = self._serialize_workflow_app_source(app=source_app) return list(deduped.values()) @classmethod @@ -535,6 +528,23 @@ class AgentObservabilityService: "node_id": None, } + @staticmethod + def _serialize_workflow_app_source(*, app: App) -> dict[str, Any]: + """Serialize the app-level source used by log and monitoring filters.""" + icon_type = app.icon_type.value if app.icon_type else None + return { + "id": f"workflow:{app.id}", + "type": "workflow", + "app_id": app.id, + "app_name": app.name, + "app_icon_type": icon_type, + "app_icon": app.icon, + "app_icon_background": app.icon_background, + "workflow_id": None, + "workflow_version": None, + "node_id": None, + } + @staticmethod def _serialize_workflow_source( *, 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 1b735a18a62..82801156686 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 @@ -25,6 +25,11 @@ def test_resolve_source_filter_accepts_structured_sources() -> None: 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" @@ -32,12 +37,20 @@ def test_resolve_source_filter_accepts_structured_sources() -> None: 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:broken") + 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: @@ -67,6 +80,17 @@ def test_statistics_explicit_source_filters_invoke_from() -> None: 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_message_scope_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_apply_status_filter_accepts_multiple_statuses() -> None: class FakeStmt: def __init__(self): @@ -115,6 +139,7 @@ def test_source_serializers_return_structured_frontend_shape() -> None: ) 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", @@ -134,11 +159,58 @@ def test_source_serializers_return_structured_frontend_shape() -> 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)