feat: add a workflow status view and honor limit on GET /workflows

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:

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)

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. |

docs/CLI.md

@@ -37,10 +37,10 @@ flowmesh workflow logs stream <wfl-id>        # SSE log stream
 flowmesh workflow logs download <wfl-id> -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 <wfl-id> --status FAILED
 ```
 

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.

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",

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}

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)