diff --git a/cli/src/flowmesh_cli/commands/task.py b/cli/src/flowmesh_cli/commands/task.py index 1706da8f0..88809bb5f 100644 --- a/cli/src/flowmesh_cli/commands/task.py +++ b/cli/src/flowmesh_cli/commands/task.py @@ -79,6 +79,9 @@ def list_tasks( "--failed/--not-failed", help="Filter by failure state", ), + limit: int | None = typer.Option( + None, "--limit", min=1, max=10_000, help="Keep only the newest N tasks" + ), query: list[str] | None = typer.Option( None, "--query", "-q", help="Filter tasks by key=value pairs" ), @@ -97,6 +100,7 @@ def list_tasks( graph_node_name=graph_node_name, completed=completed, failed=failed, + limit=limit, query_params=query_params, ) except FlowMeshError as exc: diff --git a/cli/src/flowmesh_cli/commands/workflow.py b/cli/src/flowmesh_cli/commands/workflow.py index d986a5490..dea775a4f 100644 --- a/cli/src/flowmesh_cli/commands/workflow.py +++ b/cli/src/flowmesh_cli/commands/workflow.py @@ -5,7 +5,7 @@ import typer from flowmesh import FlowMesh from flowmesh.exceptions import FlowMeshError -from flowmesh.models.common import TERMINAL_WORKFLOW_STATUSES, WorkflowStatus +from flowmesh.models.common import TERMINAL_WORKFLOW_STATUSES from flowmesh.resources.workflows import WorkflowFormat from ..core import logging @@ -100,6 +100,9 @@ def list_workflows( "--task-id", help="Filter by task id in workflow (repeatable)", ), + limit: int | None = typer.Option( + None, "--limit", min=1, max=10_000, help="Keep only the newest N workflows" + ), query: list[str] | None = typer.Option( None, "--query", "-q", help="Filter workflows by key=value pairs" ), @@ -112,6 +115,7 @@ def list_workflows( workflow_id=workflow_id, status=status or None, task_ids=task_id or None, + limit=limit, query_params=query_params, ) except FlowMeshError as exc: @@ -130,12 +134,13 @@ def watch( last_status: str | None = None try: while True: - wf = client.workflows.retrieve(workflow_id) - status_value = WorkflowStatus(wf.status) + view = client.workflows.status(workflow_id) + status_value = view.status if status_value != last_status: logging.log(f"[{workflow_id}] status: {status_value}") last_status = status_value if status_value in TERMINAL_WORKFLOW_STATUSES: + wf = client.workflows.retrieve(workflow_id) logging.log(wf.model_dump_json(indent=2)) return time.sleep(interval) diff --git a/docs/API.md b/docs/API.md index b1bff4ac9..3e9e2278d 100644 --- a/docs/API.md +++ b/docs/API.md @@ -25,8 +25,8 @@ self-authenticate the same way, sending `FLOWMESH_API_KEY` as the bearer. |--------|------|-------------| | POST | `/api/v1/workflows` | Submit a workflow. Body is YAML (`text/plain`) or JSON; set `Workflow-Format: n8n` for n8n graphs. | | POST | `/api/v1/workflows/validate` | Parse without executing. | -| GET | `/api/v1/workflows` | List workflows (`workflow_id`, `owner`, `status`). | -| GET | `/api/v1/workflows/{id}` | Workflow details + per-task summary. | +| GET | `/api/v1/workflows` | List workflows, oldest first. Filters: `workflow_id`, `status`, `task_ids`. `limit` keeps the newest; `view=status` returns only each workflow's status and progress. | +| GET | `/api/v1/workflows/{id}` | Workflow details + per-task summary. `view=status` returns only the workflow's status and progress. | | GET | `/api/v1/workflows/{id}/logs` | Query logs (`limit`, `before`/`after` cursors). | | GET | `/api/v1/workflows/{id}/logs/stream` | SSE log stream. | | POST | `/api/v1/workflows/{id}/cancel` | Cancel a workflow and all in-flight tasks. | diff --git a/docs/CLI.md b/docs/CLI.md index 18d1bbc00..6b334500e 100644 --- a/docs/CLI.md +++ b/docs/CLI.md @@ -37,10 +37,10 @@ flowmesh workflow logs stream # SSE log stream flowmesh workflow logs download -o logs/ ``` -List, filter, paginate: +List and filter; `--limit` keeps the newest matches: ```bash -flowmesh workflow list --status DONE +flowmesh workflow list --status DONE --limit 20 flowmesh task list --workflow-id --status FAILED ``` diff --git a/docs/SDK.md b/docs/SDK.md index 25e4ca616..acd5e272e 100644 --- a/docs/SDK.md +++ b/docs/SDK.md @@ -43,14 +43,19 @@ async with AsyncFlowMesh(base_url="...", api_key="...") as client: - **Submit YAML / JSON / n8n** — `client.workflows.submit(text_or_mapping)`; pass `workflow_format="n8n"` for n8n graphs. - **Validate without executing** — `client.workflows.validate(text_or_mapping)`. -- **List with filters** — `client.workflows.list(status=...)` and - `client.tasks.list(workflow_id=..., status=..., limit=...)`; tasks come oldest +- **List with filters** — `client.workflows.list(status=..., limit=...)` and + `client.tasks.list(workflow_id=..., status=..., limit=...)`; both come oldest submission first, and `limit` keeps the newest. Pass further filters with `query_params`. - **Poll task status** — `client.tasks.status(task_id)` and `client.tasks.list_statuses(...)` return status views without the task spec or source. `client.tasks.wait_status(task_id)` polls until the task finishes and returns its status view; `client.tasks.wait(task_id)` returns the full task. +- **Poll workflow status** — `client.workflows.status(wf_id)` and + `client.workflows.list_statuses(...)` return status views with per-state task + counts instead of task ID lists. `client.workflows.wait_status(wf_id)` polls + until the workflow is `DONE`, `FAILED` or `CANCELLED` and returns its status + view; `client.workflows.wait(wf_id)` returns the full workflow. - **Stream logs** — `client.workflows.stream_logs(wf_id)` and `client.tasks.stream_logs(task_id)` yield server-sent events; the iterator stops when the source closes. diff --git a/sdk/src/flowmesh/models/__init__.py b/sdk/src/flowmesh/models/__init__.py index 5411c4bcb..01e3972d1 100644 --- a/sdk/src/flowmesh/models/__init__.py +++ b/sdk/src/flowmesh/models/__init__.py @@ -103,6 +103,7 @@ ) from .workflows import ( Workflow, + WorkflowStatusView, WorkflowSubmitResponse, WorkflowSubmitTaskEntry, WorkflowValidateResponse, @@ -206,6 +207,7 @@ "WorkerStatus", "Workflow", "WorkflowStatus", + "WorkflowStatusView", "WorkflowSubmitResponse", "WorkflowSubmitTaskEntry", "WorkflowValidateResponse", diff --git a/sdk/src/flowmesh/models/workflows.py b/sdk/src/flowmesh/models/workflows.py index 005e462aa..38e874605 100644 --- a/sdk/src/flowmesh/models/workflows.py +++ b/sdk/src/flowmesh/models/workflows.py @@ -1,6 +1,8 @@ """Workflow-related models.""" -from pydantic import BaseModel +from typing import Any + +from pydantic import BaseModel, model_validator from .common import TaskStatus, WorkflowStatus @@ -46,3 +48,41 @@ class Workflow(BaseModel): completed_tasks: list[str] failed_tasks: list[str] cancelled_tasks: list[str] + + +_COUNTED_LISTS = { + "task_count": "task_ids", + "dispatched_count": "dispatched_tasks", + "completed_count": "completed_tasks", + "failed_count": "failed_tasks", + "cancelled_count": "cancelled_tasks", +} + + +class WorkflowStatusView(BaseModel): + """A workflow's status, timestamps and per-state task counts, without its task + ID lists.""" + + workflow_id: str + status: WorkflowStatus + submitted_at: str + updated_at: str + task_count: int + dispatched_count: int + completed_count: int + failed_count: int + cancelled_count: int + + @model_validator(mode="before") + @classmethod + def _count_task_lists(cls, data: Any) -> Any: + # A server that ignores `view=status` returns the full workflow; count its + # task ID lists instead. + if not isinstance(data, dict): + return data + counts = { + count: len(data[ids]) + for count, ids in _COUNTED_LISTS.items() + if count not in data and isinstance(data.get(ids), list) + } + return {**data, **counts} diff --git a/sdk/src/flowmesh/resources/workflows.py b/sdk/src/flowmesh/resources/workflows.py index 004d84bfc..d79689e9d 100644 --- a/sdk/src/flowmesh/resources/workflows.py +++ b/sdk/src/flowmesh/resources/workflows.py @@ -1,6 +1,7 @@ """Workflow resource operations.""" import asyncio +import builtins import json import time from collections.abc import AsyncIterator, Iterator @@ -17,6 +18,7 @@ ) from ..models.workflows import ( Workflow, + WorkflowStatusView, WorkflowSubmitResponse, WorkflowValidateResponse, ) @@ -40,6 +42,23 @@ def _workflow_content_type(workflow_format: WorkflowFormat) -> str: return "application/json" if workflow_format == "n8n" else "text/plain" +def _list_params( + workflow_id: str | None = None, + status: str | list[str] | None = None, + task_ids: str | list[str] | None = None, + limit: int | None = None, + query_params: list[tuple[str, str]] | None = None, +) -> list[tuple[str, str]]: + params: list[tuple[str, str]] = [] + append_param(params, "workflow_id", workflow_id) + extend_params(params, "status", status) + extend_params(params, "task_ids", task_ids) + append_param(params, "limit", limit) + if query_params: + params.extend(query_params) + return params + + class Workflows(SyncResource): """Synchronous workflow operations.""" @@ -85,23 +104,56 @@ def retrieve(self, workflow_id: str) -> Workflow: data = self._client._request("GET", f"/workflows/{workflow_id}") return Workflow.model_validate(data) + def status(self, workflow_id: str) -> WorkflowStatusView: + """Retrieve a workflow's status view, with task counts instead of task IDs. + + Prefer this over ``retrieve`` for polling.""" + data = self._client._request( + "GET", f"/workflows/{workflow_id}", params=[("view", "status")] + ) + return WorkflowStatusView.model_validate(data) + def list( self, workflow_id: str | None = None, status: str | list[str] | None = None, task_ids: str | list[str] | None = None, + limit: int | None = None, query_params: list[tuple[str, str]] | None = None, ) -> list[Workflow]: - """List workflows with optional filters.""" - params: list[tuple[str, str]] = [] - append_param(params, "workflow_id", workflow_id) - extend_params(params, "status", status) - extend_params(params, "task_ids", task_ids) - if query_params: - params.extend(query_params) + """List workflows with optional filters, oldest submission first. + + ``limit`` keeps only the newest matching workflows.""" + params = _list_params( + workflow_id=workflow_id, + status=status, + task_ids=task_ids, + limit=limit, + query_params=query_params, + ) data = self._client._request("GET", "/workflows", params=params or None) return [Workflow.model_validate(w) for w in data] + def list_statuses( + self, + workflow_id: str | None = None, + status: str | builtins.list[str] | None = None, + task_ids: str | builtins.list[str] | None = None, + limit: int | None = None, + query_params: builtins.list[tuple[str, str]] | None = None, + ) -> builtins.list[WorkflowStatusView]: + """List the status views of workflows; takes the filters of ``list``.""" + params = _list_params( + workflow_id=workflow_id, + status=status, + task_ids=task_ids, + limit=limit, + query_params=query_params, + ) + params.append(("view", "status")) + data = self._client._request("GET", "/workflows", params=params) + return [WorkflowStatusView.model_validate(w) for w in data] + def cancel(self, workflow_id: str) -> Workflow: """Cancel a running workflow.""" data = self._client._request("POST", f"/workflows/{workflow_id}/cancel") @@ -151,11 +203,21 @@ def wait( workflow_id: str, interval: float = 2.0, ) -> Workflow: - """Poll a workflow until it reaches a terminal state.""" + """Poll a workflow until it reaches a terminal state, then retrieve it.""" + self.wait_status(workflow_id, interval=interval) + return self.retrieve(workflow_id) + + def wait_status( + self, + workflow_id: str, + interval: float = 2.0, + ) -> WorkflowStatusView: + """Poll a workflow until it reaches a terminal state and return its status + view.""" while True: - workflow = self.retrieve(workflow_id) - if workflow.status in TERMINAL_WORKFLOW_STATUSES: - return workflow + view = self.status(workflow_id) + if view.status in TERMINAL_WORKFLOW_STATUSES: + return view time.sleep(interval) @@ -199,23 +261,56 @@ async def retrieve(self, workflow_id: str) -> Workflow: data = await self._client._request("GET", f"/workflows/{workflow_id}") return Workflow.model_validate(data) + async def status(self, workflow_id: str) -> WorkflowStatusView: + """Retrieve a workflow's status view, with task counts instead of task IDs. + + Prefer this over ``retrieve`` for polling.""" + data = await self._client._request( + "GET", f"/workflows/{workflow_id}", params=[("view", "status")] + ) + return WorkflowStatusView.model_validate(data) + async def list( self, workflow_id: str | None = None, status: str | list[str] | None = None, task_ids: str | list[str] | None = None, + limit: int | None = None, query_params: list[tuple[str, str]] | None = None, ) -> list[Workflow]: - """List workflows with optional filters.""" - params: list[tuple[str, str]] = [] - append_param(params, "workflow_id", workflow_id) - extend_params(params, "status", status) - extend_params(params, "task_ids", task_ids) - if query_params: - params.extend(query_params) + """List workflows with optional filters, oldest submission first. + + ``limit`` keeps only the newest matching workflows.""" + params = _list_params( + workflow_id=workflow_id, + status=status, + task_ids=task_ids, + limit=limit, + query_params=query_params, + ) data = await self._client._request("GET", "/workflows", params=params or None) return [Workflow.model_validate(w) for w in data] + async def list_statuses( + self, + workflow_id: str | None = None, + status: str | builtins.list[str] | None = None, + task_ids: str | builtins.list[str] | None = None, + limit: int | None = None, + query_params: builtins.list[tuple[str, str]] | None = None, + ) -> builtins.list[WorkflowStatusView]: + """List the status views of workflows; takes the filters of ``list``.""" + params = _list_params( + workflow_id=workflow_id, + status=status, + task_ids=task_ids, + limit=limit, + query_params=query_params, + ) + params.append(("view", "status")) + data = await self._client._request("GET", "/workflows", params=params) + return [WorkflowStatusView.model_validate(w) for w in data] + async def cancel(self, workflow_id: str) -> Workflow: """Cancel a running workflow.""" data = await self._client._request("POST", f"/workflows/{workflow_id}/cancel") @@ -263,9 +358,17 @@ async def download_logs( yield out_path async def wait(self, workflow_id: str, interval: float = 2.0) -> Workflow: - """Poll a workflow until it reaches a terminal state.""" + """Poll a workflow until it reaches a terminal state, then retrieve it.""" + await self.wait_status(workflow_id, interval=interval) + return await self.retrieve(workflow_id) + + async def wait_status( + self, workflow_id: str, interval: float = 2.0 + ) -> WorkflowStatusView: + """Poll a workflow until it reaches a terminal state and return its status + view.""" while True: - workflow = await self.retrieve(workflow_id) - if workflow.status in TERMINAL_WORKFLOW_STATUSES: - return workflow + view = await self.status(workflow_id) + if view.status in TERMINAL_WORKFLOW_STATUSES: + return view await asyncio.sleep(interval) diff --git a/src/server/registries/workflow.py b/src/server/registries/workflow.py index cbd6ee956..ea3bdb042 100644 --- a/src/server/registries/workflow.py +++ b/src/server/registries/workflow.py @@ -14,6 +14,8 @@ model_serializer, ) +from shared.utils.time import parse_iso_datetime + from ..clients.redis import ( WORKFLOWS_SET_KEY, RedisClient, @@ -106,6 +108,15 @@ class Workflow(BaseModel): cancelled_tasks: list[str] = Field(description="Cancelled task identifiers.") +def workflow_order(workflow: Workflow) -> tuple[float, str]: + """Order by submission time, then id; an unparsable time sorts oldest.""" + try: + submitted = parse_iso_datetime(workflow.submitted_at) + except ValueError: + submitted = None + return (submitted.timestamp() if submitted else 0.0), workflow.workflow_id + + def _create_workflow_record( workflow_id: str, tasks: list[TaskRecord] ) -> tuple[WorkflowRecord, list[str], list[str]]: diff --git a/src/server/routers/v1/workflows.py b/src/server/routers/v1/workflows.py index 714ccf618..72f3571a5 100644 --- a/src/server/routers/v1/workflows.py +++ b/src/server/routers/v1/workflows.py @@ -1,7 +1,7 @@ import asyncio import json import logging -from typing import Any +from typing import Any, Literal from fastapi import ( APIRouter, @@ -13,6 +13,7 @@ status, ) from fastapi.responses import StreamingResponse +from starlette.datastructures import QueryParams from shared.schemas.event import TaskEvent @@ -36,9 +37,10 @@ workflow_log_stream_key, ) from ...hooks import SUBMISSION_GUARDS, ResourceAction, ResourceKind -from ...registries.workflow import Workflow, WorkflowRegistry +from ...registries.workflow import Workflow, WorkflowRegistry, workflow_order from ...schemas.logs import LogEntry, LogEvent, LogQueryResponse from ...schemas.workflow import ( + WorkflowStatusView, WorkflowSubmitResponse, WorkflowSubmitTaskEntry, WorkflowValidateResponse, @@ -70,6 +72,16 @@ } router = APIRouter(prefix="/workflows", tags=["Workflows"]) +# Query keys the list endpoint consumes itself; they are not model-field filters. +_CONTROL_PARAMS = ("view", "limit") + +type WorkflowView = Literal["full", "status"] +_VIEW_QUERY = Query( + "full", + description="`status` returns only status, timestamps and per-state task " + "counts (no task ID lists).", +) + def _parse_submission_body(raw_body: bytes, content_type: str) -> str: if "application/json" in content_type: @@ -268,13 +280,15 @@ async def validate_workflow( "DONE (all tasks completed). " ), response_description="Workflow details", + response_model=Workflow | WorkflowStatusView, ) async def get_workflow( workflow_id: str, + view: WorkflowView = _VIEW_QUERY, principal: PrincipalContext = Depends(authenticate_connection), registry: WorkflowRegistry = Depends(get_workflow_registry), logger: logging.Logger = Depends(get_logger), -) -> Workflow: +) -> Workflow | WorkflowStatusView: await require_permission( principal, ResourceKind.WORKFLOW, workflow_id, ResourceAction.READ, logger ) @@ -284,6 +298,8 @@ async def get_workflow( status_code=status.HTTP_404_NOT_FOUND, detail=f"Workflow '{workflow_id}' not found", ) + if view == "status": + return WorkflowStatusView.from_workflow(workflow) return workflow @@ -514,15 +530,20 @@ async def cancel_workflow( @router.get( "", summary="List workflows", - description="List submitted workflows.", + description="List submitted workflows ordered by submission, oldest first.", response_description="List of workflows", + response_model=list[Workflow] | list[WorkflowStatusView], ) async def list_workflows( request: Request, + view: WorkflowView = _VIEW_QUERY, + limit: int | None = Query( + None, ge=1, le=10_000, description="Keep only the newest N matching workflows." + ), principal: PrincipalContext = Depends(authenticate_connection), registry: WorkflowRegistry = Depends(get_workflow_registry), logger: logging.Logger = Depends(get_logger), -) -> list[Workflow]: +) -> list[Workflow] | list[WorkflowStatusView]: workflow_ids = await registry.get_workflow_ids_async() allowed = await resolve_accessible_ids( principal, ResourceKind.WORKFLOW, ResourceAction.READ, logger @@ -530,7 +551,20 @@ async def list_workflows( if allowed is not None: workflow_ids = workflow_ids & allowed workflows = await registry.get_workflows_async(sorted(workflow_ids)) - return filter_models_by_queries(workflows, request.query_params) + filters = QueryParams( + [ + (key, value) + for key, value in request.query_params.multi_items() + if key not in _CONTROL_PARAMS + ] + ) + workflows = filter_models_by_queries(workflows, filters) + workflows.sort(key=workflow_order) + if limit is not None: + workflows = workflows[-limit:] + if view == "status": + return [WorkflowStatusView.from_workflow(wf) for wf in workflows] + return workflows def _get_workflow_from_request( diff --git a/src/server/schemas/workflow.py b/src/server/schemas/workflow.py index d0c9b43b8..e6302a462 100644 --- a/src/server/schemas/workflow.py +++ b/src/server/schemas/workflow.py @@ -1,5 +1,9 @@ +from typing import Self + from pydantic import BaseModel, Field +from ..registries.workflow import Workflow, WorkflowStatus + class WorkflowSubmitTaskEntry(BaseModel): task_id: str = Field(description="Task identifier.") @@ -40,3 +44,32 @@ class WorkflowValidateResponse(BaseModel): tasks: list[WorkflowValidateTaskEntry] = Field( description="Validated task entries." ) + + +class WorkflowStatusView(BaseModel): + """A workflow's status, timestamps and per-state task counts, without its task + ID lists.""" + + workflow_id: str = Field(description="Workflow identifier.") + status: WorkflowStatus = Field(description="Workflow status.") + submitted_at: str = Field(description="Submission timestamp.") + updated_at: str = Field(description="Last update timestamp.") + task_count: int = Field(description="Number of tasks in the workflow.") + dispatched_count: int = Field(description="Number of dispatched tasks.") + completed_count: int = Field(description="Number of completed tasks.") + failed_count: int = Field(description="Number of failed tasks.") + cancelled_count: int = Field(description="Number of cancelled tasks.") + + @classmethod + def from_workflow(cls, workflow: Workflow) -> Self: + return cls( + workflow_id=workflow.workflow_id, + status=workflow.status, + submitted_at=workflow.submitted_at, + updated_at=workflow.updated_at, + task_count=len(workflow.task_ids), + dispatched_count=len(workflow.dispatched_tasks), + completed_count=len(workflow.completed_tasks), + failed_count=len(workflow.failed_tasks), + cancelled_count=len(workflow.cancelled_tasks), + ) diff --git a/tests/cli/test_cli_commands.py b/tests/cli/test_cli_commands.py index cdef40ff8..d934b4be7 100644 --- a/tests/cli/test_cli_commands.py +++ b/tests/cli/test_cli_commands.py @@ -6,6 +6,7 @@ import typer from flowmesh.models import ( TaskStatus, + WorkflowStatus, WorkflowSubmitResponse, ) from flowmesh_cli.cli import build_cli_app @@ -69,6 +70,26 @@ def test_format_flag(self, mock_hc: MagicMock, tmp_path: Path) -> None: assert call_args.kwargs["workflow_format"] == "n8n" +class TestWorkflowWatch: + @patch("flowmesh_cli.commands.workflow.FlowMesh") + def test_polls_status_then_prints_full_workflow(self, mock_hc: MagicMock) -> None: + client = _mock_client() + client.workflows.status.side_effect = [ + MagicMock(status=WorkflowStatus.DISPATCHED), + MagicMock(status=WorkflowStatus.DONE), + ] + client.workflows.retrieve.return_value.model_dump_json.return_value = ( + '{"workflow_id": "wf-1"}' + ) + mock_hc.return_value = client + + result = runner.invoke(_app(), ["workflow", "watch", "wf-1", "--interval", "0"]) + assert result.exit_code == 0 + assert client.workflows.status.call_count == 2 + client.workflows.retrieve.assert_called_once_with("wf-1") + assert '"workflow_id": "wf-1"' in result.stdout + + class TestWorkflowList: @patch("flowmesh_cli.commands.workflow.FlowMesh") def test_status_filter(self, mock_hc: MagicMock) -> None: @@ -99,12 +120,70 @@ def test_query_filter_forwarded(self, mock_hc: MagicMock) -> None: ("status", "DONE"), ] + @patch("flowmesh_cli.commands.workflow.FlowMesh") + def test_limit_forwarded(self, mock_hc: MagicMock) -> None: + client = _mock_client() + client.workflows.list.return_value = [] + mock_hc.return_value = client + + result = runner.invoke(_app(), ["workflow", "list", "--limit", "20"]) + assert result.exit_code == 0 + assert client.workflows.list.call_args.kwargs["limit"] == 20 + + @patch("flowmesh_cli.commands.workflow.FlowMesh") + def test_limit_defaults_to_none(self, mock_hc: MagicMock) -> None: + client = _mock_client() + client.workflows.list.return_value = [] + mock_hc.return_value = client + + result = runner.invoke(_app(), ["workflow", "list"]) + assert result.exit_code == 0 + assert client.workflows.list.call_args.kwargs["limit"] is None + + @patch("flowmesh_cli.commands.workflow.FlowMesh") + def test_limit_out_of_range_rejected(self, mock_hc: MagicMock) -> None: + result = runner.invoke(_app(), ["workflow", "list", "--limit", "0"]) + assert result.exit_code != 0 + mock_hc.assert_not_called() + # ------------------------------------------------------------------ # # Task commands # ------------------------------------------------------------------ # +class TestTaskList: + @patch("flowmesh_cli.commands.task.FlowMesh") + def test_limit_forwarded(self, mock_hc: MagicMock) -> None: + client = _mock_client() + client.tasks.list.return_value = [] + mock_hc.return_value = client + + result = runner.invoke( + _app(), ["task", "list", "--workflow-id", "wf-1", "--limit", "50"] + ) + assert result.exit_code == 0 + call_kwargs = client.tasks.list.call_args.kwargs + assert call_kwargs["limit"] == 50 + assert call_kwargs["workflow_id"] == "wf-1" + + @patch("flowmesh_cli.commands.task.FlowMesh") + def test_limit_defaults_to_none(self, mock_hc: MagicMock) -> None: + client = _mock_client() + client.tasks.list.return_value = [] + mock_hc.return_value = client + + result = runner.invoke(_app(), ["task", "list"]) + assert result.exit_code == 0 + assert client.tasks.list.call_args.kwargs["limit"] is None + + @patch("flowmesh_cli.commands.task.FlowMesh") + def test_limit_out_of_range_rejected(self, mock_hc: MagicMock) -> None: + result = runner.invoke(_app(), ["task", "list", "--limit", "10001"]) + assert result.exit_code != 0 + mock_hc.assert_not_called() + + class TestTaskInfo: @patch("flowmesh_cli.commands.task.FlowMesh") def test_id_forwarded(self, mock_hc: MagicMock) -> None: diff --git a/tests/sdk/test_models.py b/tests/sdk/test_models.py index 3f66382ab..3619e9428 100644 --- a/tests/sdk/test_models.py +++ b/tests/sdk/test_models.py @@ -26,6 +26,7 @@ WorkerHardware, WorkerInfo, Workflow, + WorkflowStatusView, WorkflowSubmitResponse, WorkflowValidateResponse, ) @@ -62,6 +63,7 @@ from server.schemas.node import WorkerHardware as SrvWorkerHardware from server.schemas.ssh import SSHConnectionInfo as SrvSSHConnectionInfo from server.schemas.task import TaskStatusView as SrvTaskStatusView +from server.schemas.workflow import WorkflowStatusView as SrvWorkflowStatusView from server.schemas.workflow import WorkflowSubmitResponse as SrvWorkflowSubmitResponse from server.schemas.workflow import ( WorkflowSubmitTaskEntry as SrvWorkflowSubmitTaskEntry, @@ -284,6 +286,33 @@ def test_workflow_validate(self) -> None: assert wf.workflow_id == "wf-abc123" assert wf.status == "DONE" + def test_workflow_status_view_validate(self) -> None: + payload = _dump(SrvWorkflowStatusView.from_workflow(_SRV_WORKFLOW)) + view = WorkflowStatusView.model_validate(payload) + assert view.workflow_id == "wf-abc123" + assert (view.task_count, view.completed_count, view.failed_count) == (2, 2, 0) + + def test_workflow_status_view_counts_full_workflow(self) -> None: + workflow = _SRV_WORKFLOW.model_copy( + update={ + "task_ids": [f"t-{i}" for i in range(10)], + "dispatched_tasks": ["t-0"], + "completed_tasks": ["t-1", "t-2"], + "failed_tasks": ["t-3", "t-4", "t-5"], + "cancelled_tasks": ["t-6", "t-7", "t-8", "t-9"], + } + ) + view = WorkflowStatusView.model_validate(_dump(workflow)) + assert view.status == "DONE" + counts = ( + view.task_count, + view.dispatched_count, + view.completed_count, + view.failed_count, + view.cancelled_count, + ) + assert counts == (10, 1, 2, 3, 4) + def test_workflow_submit_response(self) -> None: server = SrvWorkflowSubmitResponse( ok=True, diff --git a/tests/sdk/test_request_building.py b/tests/sdk/test_request_building.py index beb3638e2..5543913ff 100644 --- a/tests/sdk/test_request_building.py +++ b/tests/sdk/test_request_building.py @@ -11,7 +11,7 @@ import pytest import respx from flowmesh import AsyncFlowMesh, FlowMesh -from flowmesh.models import TaskStatus +from flowmesh.models import TaskStatus, WorkflowStatus from pydantic import BaseModel from server.registries.node import Node as SrvNode @@ -22,6 +22,7 @@ from server.schemas.common import OkResponse as SrvOkResponse from server.schemas.common import VersionResponse as SrvVersionResponse from server.schemas.task import TaskStatusView as SrvTaskStatusView +from server.schemas.workflow import WorkflowStatusView as SrvWorkflowStatusView from server.schemas.workflow import WorkflowSubmitResponse as SrvWorkflowSubmitResponse from server.schemas.workflow import ( WorkflowSubmitTaskEntry as SrvWorkflowSubmitTaskEntry, @@ -108,6 +109,11 @@ def _status_json( ) +def _workflow_status_json(status: SrvWorkflowStatus) -> dict[str, Any]: + view = SrvWorkflowStatusView.from_workflow(_WORKFLOW_RESPONSE) + return _json(view.model_copy(update={"status": status})) + + def _json(server_obj: BaseModel) -> dict: # type: ignore[type-arg] """Dump a server Pydantic model to a JSON-compatible dict.""" return server_obj.model_dump(mode="json") @@ -387,6 +393,100 @@ async def test_async_status_calls_request_status_view(self) -> None: assert list_route.calls[0].request.url.params["limit"] == "1" +class TestWorkflowPolling: + @respx.mock + def test_status_requests_status_view(self, mock_client: FlowMesh) -> None: + route = respx.get(route_url("get_workflow", workflow_id="wf-1")).respond( + json=_workflow_status_json(SrvWorkflowStatus.DISPATCHED) + ) + view = mock_client.workflows.status("wf-1") + assert route.calls[0].request.url.params["view"] == "status" + assert view.status == WorkflowStatus.DISPATCHED + + @respx.mock + def test_list_sends_limit(self, mock_client: FlowMesh) -> None: + route = respx.get(route_url("list_workflows")).respond(json=[]) + mock_client.workflows.list(status="DONE", limit=2) + params = route.calls[0].request.url.params + assert params["limit"] == "2" + assert params["status"] == "DONE" + assert "view" not in params + + @respx.mock + def test_list_statuses_requests_status_view(self, mock_client: FlowMesh) -> None: + route = respx.get(route_url("list_workflows")).respond( + json=[_workflow_status_json(SrvWorkflowStatus.PENDING)] + ) + views = mock_client.workflows.list_statuses(task_ids="t-1", limit=5) + params = route.calls[0].request.url.params + assert params["view"] == "status" + assert params["task_ids"] == "t-1" + assert params["limit"] == "5" + assert [v.workflow_id for v in views] == ["wf-1"] + + @respx.mock + def test_wait_polls_status_then_fetches_full_workflow( + self, mock_client: FlowMesh + ) -> None: + route = respx.get(route_url("get_workflow", workflow_id="wf-1")).mock( + side_effect=[ + httpx.Response( + 200, json=_workflow_status_json(SrvWorkflowStatus.DISPATCHED) + ), + httpx.Response(200, json=_workflow_status_json(SrvWorkflowStatus.DONE)), + httpx.Response(200, json=_json(_WORKFLOW_RESPONSE)), + ] + ) + workflow = mock_client.workflows.wait("wf-1", interval=0) + views = [call.request.url.params.get("view") for call in route.calls] + assert views == ["status", "status", None] + assert workflow.task_ids == [] + + @respx.mock + def test_wait_status_returns_terminal_view(self, mock_client: FlowMesh) -> None: + route = respx.get(route_url("get_workflow", workflow_id="wf-1")).mock( + side_effect=[ + httpx.Response( + 200, json=_workflow_status_json(SrvWorkflowStatus.PENDING) + ), + httpx.Response( + 200, json=_workflow_status_json(SrvWorkflowStatus.FAILED) + ), + ] + ) + view = mock_client.workflows.wait_status("wf-1", interval=0) + views = [call.request.url.params.get("view") for call in route.calls] + assert views == ["status", "status"] + assert view.status == WorkflowStatus.FAILED + + @pytest.mark.anyio + @respx.mock + async def test_async_polling_requests_status_view(self) -> None: + get_route = respx.get(route_url("get_workflow", workflow_id="wf-1")).mock( + side_effect=[ + httpx.Response(200, json=_workflow_status_json(SrvWorkflowStatus.DONE)), + httpx.Response(200, json=_workflow_status_json(SrvWorkflowStatus.DONE)), + httpx.Response(200, json=_json(_WORKFLOW_RESPONSE)), + ] + ) + list_route = respx.get(route_url("list_workflows")).respond( + json=[_workflow_status_json(SrvWorkflowStatus.DONE)] + ) + async with AsyncFlowMesh( + base_url=TEST_BASE_URL, api_key="flm-test-key" + ) as client: + view = await client.workflows.status("wf-1") + workflow = await client.workflows.wait("wf-1", interval=0) + views = await client.workflows.list_statuses(limit=1) + assert view.status == WorkflowStatus.DONE + assert workflow.workflow_id == "wf-1" + assert [v.workflow_id for v in views] == ["wf-1"] + get_views = [call.request.url.params.get("view") for call in get_route.calls] + assert get_views == ["status", "status", None] + assert list_route.calls[0].request.url.params["view"] == "status" + assert list_route.calls[0].request.url.params["limit"] == "1" + + class TestNodeRegister: @respx.mock def test_register_includes_version_when_given(self, mock_client: FlowMesh) -> None: diff --git a/tests/sdk/test_schema_compat.py b/tests/sdk/test_schema_compat.py index 38ed8d54c..8d4ad36e6 100644 --- a/tests/sdk/test_schema_compat.py +++ b/tests/sdk/test_schema_compat.py @@ -45,6 +45,7 @@ WorkerStatus, Workflow, WorkflowStatus, + WorkflowStatusView, WorkflowSubmitResponse, WorkflowSubmitTaskEntry, WorkflowValidateResponse, @@ -82,6 +83,7 @@ from server.schemas.task import TaskStatusView as SrvTaskStatusView from server.schemas.worker import WorkerCordon as SrvWorkerCordon from server.schemas.worker import WorkerCordonResult as SrvWorkerCordonResult +from server.schemas.workflow import WorkflowStatusView as SrvWorkflowStatusView from server.schemas.workflow import WorkflowSubmitResponse as SrvWorkflowSubmitResponse from server.schemas.workflow import ( WorkflowSubmitTaskEntry as SrvWorkflowSubmitTaskEntry, @@ -180,6 +182,7 @@ (SrvWorkflowValidateTaskEntry, WorkflowValidateTaskEntry), # Workflow registry (SrvWorkflow, Workflow), + (SrvWorkflowStatusView, WorkflowStatusView), # Task models (last_queue_ts is server-internal scheduling field) (SrvTaskUsage, TaskUsage), # Worker models diff --git a/tests/server/test_workflows_router.py b/tests/server/test_workflows_router.py new file mode 100644 index 000000000..1cc86c707 --- /dev/null +++ b/tests/server/test_workflows_router.py @@ -0,0 +1,206 @@ +"""Router tests for `GET /api/v1/workflows` status view, ordering and `limit`.""" + +import logging +from collections.abc import Iterator, Sequence +from unittest.mock import MagicMock + +import pytest +from fastapi import FastAPI +from httpx import ASGITransport, AsyncClient +from lumid_hooks import ResourceRef + +from server.app_state import get_workflow_registry +from server.auth.security import PrincipalContext, authenticate_connection +from server.hooks import PERMISSION_CHECKERS +from server.registries.workflow import Workflow, WorkflowStatus +from server.routers.v1 import workflows as workflows_router +from server.schemas.workflow import WorkflowStatusView + +PREFIX = "/api/v1" + + +class _ScopedChecker: + """Scopes workflow access to a fixed set of workflow ids.""" + + name = "scoped" + + def __init__(self, allowed: frozenset[str]) -> None: + self._allowed = allowed + + async def require( + self, + principal: PrincipalContext, + resource: ResourceRef, + action: str, + logger: logging.Logger, + ) -> None: + return None + + async def accessible_ids( + self, + principal: PrincipalContext, + kind: str, + action: str, + logger: logging.Logger, + ) -> frozenset[str] | None: + return self._allowed + + +class _StubRegistry: + def __init__(self, workflows: Sequence[Workflow]) -> None: + self._workflows = {wf.workflow_id: wf for wf in workflows} + + async def get_workflow_async(self, workflow_id: str) -> Workflow | None: + return self._workflows.get(workflow_id) + + async def get_workflow_ids_async(self) -> set[str]: + return set(self._workflows) + + async def get_workflows_async(self, workflow_ids: Sequence[str]) -> list[Workflow]: + return [self._workflows[wid] for wid in workflow_ids if wid in self._workflows] + + +def _workflow( + workflow_id: str, + submitted_at: str = "2026-10-12T10:00:00+00:00", + status: WorkflowStatus = WorkflowStatus.DISPATCHED, +) -> Workflow: + return Workflow( + workflow_id=workflow_id, + task_ids=[f"tsk-{i}" for i in range(10)], + submitted_at=submitted_at, + updated_at="2026-10-12T10:05:00+00:00", + status=status, + dispatched_tasks=["tsk-0"], + completed_tasks=["tsk-1", "tsk-2"], + failed_tasks=["tsk-3", "tsk-4", "tsk-5"], + cancelled_tasks=["tsk-6", "tsk-7", "tsk-8", "tsk-9"], + ) + + +@pytest.fixture +def no_checkers() -> Iterator[None]: + PERMISSION_CHECKERS.clear() + yield + PERMISSION_CHECKERS.clear() + + +def _client(*workflows: Workflow) -> AsyncClient: + app = FastAPI() + app.state.logger = logging.getLogger("test.workflows_router") + app.include_router(workflows_router.router, prefix=PREFIX) + registry = _StubRegistry(workflows) + app.dependency_overrides[get_workflow_registry] = lambda: registry + app.dependency_overrides[authenticate_connection] = lambda: MagicMock( + spec=PrincipalContext + ) + return AsyncClient(transport=ASGITransport(app=app), base_url="http://t") + + +def test_status_view_shared_fields_match_workflow() -> None: + for name, field in WorkflowStatusView.model_fields.items(): + if info := Workflow.model_fields.get(name): + served = (field.annotation, field.description) + assert served == (info.annotation, info.description), name + + +@pytest.mark.anyio +async def test_get_workflow_status_view_counts_tasks(no_checkers: None) -> None: + async with _client(_workflow("wfl-1")) as ac: + full = await ac.get(f"{PREFIX}/workflows/wfl-1") + slim = await ac.get(f"{PREFIX}/workflows/wfl-1", params={"view": "status"}) + + assert full.status_code == 200 and slim.status_code == 200 + f, s = full.json(), slim.json() + assert "task_ids" in f + assert set(s) == set(WorkflowStatusView.model_fields) + shared = set(s) & set(f) + assert shared == {"workflow_id", "status", "submitted_at", "updated_at"} + assert {key: f[key] for key in shared} == {key: s[key] for key in shared} + assert ( + s["task_count"], + s["dispatched_count"], + s["completed_count"], + s["failed_count"], + s["cancelled_count"], + ) == (10, 1, 2, 3, 4) + + +@pytest.mark.anyio +async def test_get_workflow_unknown_view_is_rejected(no_checkers: None) -> None: + async with _client(_workflow("wfl-1")) as ac: + resp = await ac.get(f"{PREFIX}/workflows/wfl-1", params={"view": "bogus"}) + assert resp.status_code == 422 + + +@pytest.mark.anyio +async def test_list_workflows_orders_by_submission_and_limits_to_newest( + no_checkers: None, +) -> None: + old = _workflow("wfl-b", submitted_at="2026-10-12T09:00:00+00:00") + new = _workflow("wfl-a", submitted_at="2026-10-12T11:00:00+00:00") + garbage = _workflow("wfl-c", submitted_at="not-a-timestamp") + + async with _client(new, old, garbage) as ac: + every = await ac.get(f"{PREFIX}/workflows") + newest = await ac.get(f"{PREFIX}/workflows", params={"limit": 1}) + + assert [w["workflow_id"] for w in every.json()] == ["wfl-c", "wfl-b", "wfl-a"] + assert [w["workflow_id"] for w in newest.json()] == ["wfl-a"] + + +@pytest.mark.anyio +async def test_list_workflows_limit_applies_after_filters(no_checkers: None) -> None: + done = _workflow( + "wfl-a", submitted_at="2026-10-12T09:00:00+00:00", status=WorkflowStatus.DONE + ) + running = _workflow("wfl-b", submitted_at="2026-10-12T11:00:00+00:00") + + async with _client(done, running) as ac: + resp = await ac.get( + f"{PREFIX}/workflows", + params={"status": "DONE", "limit": 1, "view": "status"}, + ) + + assert [w["workflow_id"] for w in resp.json()] == ["wfl-a"] + + +@pytest.mark.anyio +async def test_list_workflows_limit_applies_after_permissions( + no_checkers: None, +) -> None: + older = _workflow("wfl-a", submitted_at="2026-10-12T09:00:00+00:00") + newer = _workflow("wfl-b", submitted_at="2026-10-12T10:00:00+00:00") + hidden = _workflow("wfl-c", submitted_at="2026-10-12T11:00:00+00:00") + + PERMISSION_CHECKERS.append(_ScopedChecker(frozenset({"wfl-a", "wfl-b"}))) + async with _client(older, newer, hidden) as ac: + resp = await ac.get(f"{PREFIX}/workflows", params={"limit": 1}) + + assert [w["workflow_id"] for w in resp.json()] == ["wfl-b"] + + +@pytest.mark.anyio +async def test_list_workflows_status_view(no_checkers: None) -> None: + async with _client(_workflow("wfl-1"), _workflow("wfl-2")) as ac: + slim = await ac.get(f"{PREFIX}/workflows", params={"view": "status"}) + + assert slim.status_code == 200 and len(slim.json()) == 2 + assert all(set(w) == set(WorkflowStatusView.model_fields) for w in slim.json()) + + +@pytest.mark.anyio +@pytest.mark.parametrize("limit", [0, 10_001]) +async def test_list_workflows_rejects_out_of_range_limit( + no_checkers: None, limit: int +) -> None: + async with _client() as ac: + resp = await ac.get(f"{PREFIX}/workflows", params={"limit": limit}) + assert resp.status_code == 422 + + +@pytest.mark.anyio +async def test_list_workflows_unknown_view_is_rejected(no_checkers: None) -> None: + async with _client() as ac: + resp = await ac.get(f"{PREFIX}/workflows", params={"view": "bogus"}) + assert resp.status_code == 422