mirror of
https://github.com/langgenius/dify.git
synced 2026-07-21 02:28:30 +08:00
fix(agent): aggregate monitor sources by workflow app (#39021)
Co-authored-by: yyh <92089059+lyzno1@users.noreply.github.com>
This commit is contained in:
zyssyz123
GitHub
parent
4571c4f276
commit
79662d49af
@ -177,7 +177,7 @@ class AgentLogsQuery(BaseModel):
|
||||
default_factory=list,
|
||||
description=(
|
||||
"Filter by one or more source IDs, e.g. webapp:<app_id> "
|
||||
"or workflow:<app_id>:<workflow_id>:<version>:<node_id>"
|
||||
"or workflow:<app_id>. Exact workflow:<app_id>:<workflow_id>:<version>:<node_id> 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:<app_id> or workflow:<app_id> 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)")
|
||||
|
||||
@ -1151,7 +1151,7 @@ Commit an uploaded file into the Agent App drive under files/<name>
|
||||
| sort_by | query | Sort by created_at or updated_at | No | string, <br>**Default:** updated_at |
|
||||
| sort_order | query | Sort order: asc or desc | No | string, <br>**Default:** desc |
|
||||
| source | query | Deprecated single source filter | No | string |
|
||||
| sources | query | Filter by one or more source IDs, e.g. webapp:<app_id> or workflow:<app_id>:<workflow_id>:<version>:<node_id> | No | [ string ] |
|
||||
| sources | query | Filter by one or more source IDs, e.g. webapp:<app_id> or workflow:<app_id>. Exact workflow:<app_id>:<workflow_id>:<version>:<node_id> 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/<name>
|
||||
| sort_by | query | Sort by created_at or updated_at | No | string, <br>**Default:** updated_at |
|
||||
| sort_order | query | Sort order: asc or desc | No | string, <br>**Default:** desc |
|
||||
| source | query | Deprecated single source filter | No | string |
|
||||
| sources | query | Filter by one or more source IDs, e.g. webapp:<app_id> or workflow:<app_id>:<workflow_id>:<version>:<node_id> | No | [ string ] |
|
||||
| sources | query | Filter by one or more source IDs, e.g. webapp:<app_id> or workflow:<app_id>. Exact workflow:<app_id>:<workflow_id>:<version>:<node_id> 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:<app_id> or workflow:<app_id> 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, <br>**Default:** updated_at | Sort by created_at or updated_at | No |
|
||||
| sort_order | string, <br>**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:<app_id> or workflow:<app_id>:<workflow_id>:<version>:<node_id> | No |
|
||||
| sources | [ string ] | Filter by one or more source IDs, e.g. webapp:<app_id> or workflow:<app_id>. Exact workflow:<app_id>:<workflow_id>:<version>:<node_id> 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:<app_id> or workflow:<app_id> 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
|
||||
|
||||
@ -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(
|
||||
*,
|
||||
|
||||
@ -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)
|
||||
|
||||
Loading…
Reference in New Issue
Block a user