feat: add a status view for task polling and honor limit on GET /tasks

cli/src/flowmesh_cli/commands/ssh.py

@@ -14,6 +14,7 @@
 import websockets
 from flowmesh import FlowMesh
 from flowmesh.exceptions import FlowMeshError, NotFoundError
+from flowmesh.models import TaskStatusView
 from flowmesh.models.common import TERMINAL_TASK_STATUSES, TaskStatus
 from flowmesh.params import append_param, extend_params
 from flowmesh.ssh import describe_direct_route
@@ -82,13 +83,13 @@ def _exec_ssh(
     os.execvp(ssh_bin, args)
 
 
-def _exit_for_task_status(task_id: str, task_info: Any) -> None:
-    status = task_info.status
+def _exit_for_task_status(task_id: str, view: TaskStatusView) -> None:
+    status = view.status
     if status == TaskStatus.DONE:
         logging.info(f"Task {task_id} completed successfully.")
         raise typer.Exit(code=0)
     if status == TaskStatus.FAILED:
-        logging.error(f"Task {task_id} failed: {task_info.error or 'unknown error'}")
+        logging.error(f"Task {task_id} failed: {view.error or 'unknown error'}")
         raise typer.Exit(code=1)
     if status == TaskStatus.CANCELLED:
         logging.info(f"Task {task_id} was cancelled.")
@@ -97,14 +98,14 @@ def _exit_for_task_status(task_id: str, task_info: Any) -> None:
 
 def _poll_status_and_exit(client: FlowMesh, task_id: str, interval: float) -> None:
     try:
-        task_info = client.tasks.wait(task_id, interval=interval)
+        view = client.tasks.wait_status(task_id, interval=interval)
     except KeyboardInterrupt:
         logging.warning("Interrupted.")
         raise typer.Exit(code=130)
     except FlowMeshError as exc:
         logging.error(str(exc))
         raise typer.Exit(code=1)
-    _exit_for_task_status(task_id, task_info)
+    _exit_for_task_status(task_id, view)
 
 
 def _stream_logs_and_exit(
@@ -130,9 +131,9 @@ def _stream_logs_and_exit(
             except NotFoundError:
                 pass
 
-            task_info = client.tasks.retrieve(task_id)
-            if task_info.status in TERMINAL_TASK_STATUSES:
-                _exit_for_task_status(task_id, task_info)
+            view = client.tasks.status(task_id)
+            if view.status in TERMINAL_TASK_STATUSES:
+                _exit_for_task_status(task_id, view)
             if not saw_logs:
                 time.sleep(interval)
     except KeyboardInterrupt:

cli/src/flowmesh_cli/commands/task.py

@@ -150,19 +150,19 @@ def watch(
     last_update: dict | None = None
     try:
         while True:
-            task = client.tasks.retrieve(task_id)
-            payload = task.model_dump(mode="json")
-            status_value = task.status
+            view = client.tasks.status(task_id)
+            status_value = view.status
             if status_value != last_status:
                 logging.log(f"[{task_id}] status: {status_value}")
                 last_status = status_value
-            latest_update = payload.get("latest_update")
+            latest_update = view.latest_update
             if isinstance(latest_update, dict) and latest_update != last_update:
                 logging.log(json.dumps(latest_update, indent=2))
                 _log_ssh_connection_instructions(task_id, latest_update, client)
                 last_update = latest_update
             if status_value in TERMINAL_TASK_STATUSES:
-                logging.log(json.dumps(payload, indent=2))
+                task = client.tasks.retrieve(task_id)
+                logging.log(json.dumps(task.model_dump(mode="json"), indent=2))
                 return
             time.sleep(interval)
     except KeyboardInterrupt:

cli/src/flowmesh_cli/core/task.py

@@ -12,16 +12,16 @@ def wait_for_task_completion(task_id: str, interval: float) -> tuple[str, str |
     """Poll task until it reaches a terminal state."""
     client = FlowMesh()
     try:
-        task = client.tasks.wait(task_id, interval=interval)
-        status = task.status
+        view = client.tasks.wait_status(task_id, interval=interval)
+        status = view.status
         match status:
             case TaskStatus.DONE:
                 return TaskStatus.DONE, None
             case TaskStatus.FAILED | TaskStatus.CANCELLED:
-                error = task.error or status.lower()
+                error = view.error or status.lower()
                 return status, error
             case _:
-                return status, task.error
+                return status, view.error
     except KeyboardInterrupt:
         logging.warning("Cancelled by user.")
         raise typer.Exit(code=1)

docs/API.md

@@ -25,7 +25,7 @@ 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`, cursor pagination). |
+| GET | `/api/v1/workflows` | List workflows (`workflow_id`, `owner`, `status`). |
 | GET | `/api/v1/workflows/{id}` | Workflow details + per-task summary. |
 | GET | `/api/v1/workflows/{id}/logs` | Query logs (`limit`, `before`/`after` cursors). |
 | GET | `/api/v1/workflows/{id}/logs/stream` | SSE log stream. |
@@ -35,8 +35,8 @@ self-authenticate the same way, sending `FLOWMESH_API_KEY` as the bearer.
 
 | Method | Path | Description |
 |--------|------|-------------|
-| GET | `/api/v1/tasks` | List tasks. Filters: `workflow_id`, `status`, `task_type`, `assigned_worker`. |
-| GET | `/api/v1/tasks/{id}` | Task details. |
+| GET | `/api/v1/tasks` | List tasks, oldest first. Filters: `workflow_id`, `status`, `task_type`, `assigned_worker`. `limit` keeps the newest; `view=status` omits specs and sources. |
+| GET | `/api/v1/tasks/{id}` | Task details. `view=status` omits the task spec and source. |
 | GET | `/api/v1/tasks/{id}/logs` | Query task logs. |
 | GET | `/api/v1/tasks/{id}/logs/stream` | SSE task log stream. |
 
@@ -121,6 +121,5 @@ Gated by `ENABLE_SERVER_SERVE_PROXY`.
 
 ## Cursor pagination
 
-List endpoints (`/api/v1/workflows`, `/api/v1/tasks`, log queries)
-accept `limit` and `before` / `after` cursors. The cursor is an opaque
-base64 of `(timestamp, id)`; do not parse client-side.
+Log queries accept `limit` and `before` / `after` cursors. A cursor is an opaque
+string taken from `entries[].cursor`; pass it back unchanged.

docs/ARCHITECTURE.md

@@ -244,9 +244,9 @@ scripts/dev/            compile_protos, sync_requirements, check_env_examples
   `(node_alias, alias)` and lasts until it is uncordoned, independent of any
   worker's lifecycle, so it also applies to a worker that registers under the
   key later.
-- **Cursor pagination.** List endpoints accept `limit` and `before` /
-  `after` cursors. The cursor is an opaque base64 of `(timestamp, id)`;
-  do not parse client-side.
+- **Cursor pagination.** Log queries accept `limit` and `before` /
+  `after` cursors. A cursor is an opaque string taken from
+  `entries[].cursor`; pass it back unchanged.
 - **Redis channels.** The runtime uses two Redis endpoints (separate instances in the
   stack):
   - control (`REDIS_CONTROL_URL`) — durable state and the control plane's

docs/SDK.md

@@ -43,9 +43,14 @@ 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 and pagination** — `client.workflows.list(status=...)`
-  and `client.tasks.list(workflow_id=..., status=...)`; pass raw cursor
-  params with `query_params=[("before", cursor), ("limit", "100")]`.
+- **List with filters** — `client.workflows.list(status=...)` and
+  `client.tasks.list(workflow_id=..., status=..., limit=...)`; tasks 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.
 - **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.
@@ -64,9 +69,9 @@ bundle missing a requested section raises `FlowMeshError`.
 
 ## Cursor pagination
 
-Cursor-enabled calls take `limit` and `before` / `after` params.
-Cursors are opaque base64 strings — pass them through; do not parse
-them.
+Log queries take `limit` and `before` / `after` params.
+A cursor is an opaque string taken from `entries[].cursor` — pass it back
+unchanged; do not parse it.
 
 ## Errors
 

pyproject.toml

@@ -91,7 +91,7 @@ runtime-agent = [
     "ipython>=9.5.0",
     "jinja2>=3.1.6",
     "matplotlib>=3.10.6",
-    "mcp>=1.28.1",
+    "mcp>=1.30.0",
     "openai>=2.25.0,<3",
     "openai-agents>=0.7.0,<0.8",
     "opencv-python-headless>=4.13.0.92",

sdk/src/flowmesh/models/__init__.py

@@ -74,7 +74,7 @@
     SFTResult,
     SSHResult,
 )
-from .tasks import HardwareUsage, TaskInfo, TaskUsage
+from .tasks import HardwareUsage, TaskInfo, TaskStatusView, TaskUsage
 from .traces import (
     ActiveWaitBreakdown,
     AssetSummary,
@@ -190,6 +190,7 @@
     "SSHLimits",
     "StorageInfo",
     "TaskInfo",
+    "TaskStatusView",
     "TaskStatus",
     "TaskTiming",
     "TaskType",

sdk/src/flowmesh/models/tasks.py

@@ -68,3 +68,33 @@ class TaskInfo(BaseModel):
     dependents: list[str]
     completed: bool
     failed: bool
+
+
+class TaskStatusView(BaseModel):
+    """A task's status, placement, timing, errors and latest update, without its
+    spec or source."""
+
+    task_id: str
+    workflow_id: str
+    status: TaskStatus
+    task_type: str | None = None
+    category: str | None = None
+    assigned_worker: str | None = None
+    submitted_at: str
+    submitted_ts: float
+    dispatched_ts: float | None = None
+    started_ts: float | None = None
+    finished_ts: float | None = None
+    error: str | None = None
+    last_error: str | None = None
+    attempts: int
+    max_attempts: int
+    next_retry_at: str | None = None
+    last_failed_worker: str | None = None
+    parent_task_id: str | None = None
+    merged_parent_id: str | None = None
+    local_name: str | None = None
+    graph_node_name: str | None = None
+    latest_update: dict[str, Any] | None = None
+    completed: bool
+    failed: bool