From 32d3ac7271a54b619580cbdb062eb10e6c043d68 Mon Sep 17 00:00:00 2001 From: Yao Lu Date: Mon, 12 Oct 2026 21:05:06 +0800 Subject: [PATCH 01/10] feat(tasks): ?view=status polling view, honour limit on GET /tasks (#225) GET /api/v1/tasks/{id} returns the full task: spec plus its YAML source, ~230 KB each for a Lumilake inference task, so a poller reading one status field pulled ~0.5-1 MB per call. GET /api/v1/tasks ignored `limit` (unknown key -> no-op), so `limit=100` returned every task: 240+ MB on the home cluster on 2026-10-12, the day status polling under load preceded refused connections and 502s. Server: - `?view=status` on GET /tasks/{id} and GET /tasks returns TaskStatusView: status, placement, timing, errors. Default stays `full` (compatible). - GET /tasks honours `limit` (after filtering). view/limit are not treated as model-field filters. SDK: - tasks.status(id) -> TaskStatusView (sync + async). Works against older servers too: they ignore ?view and the slim model ignores extra fields. - tasks.wait() polls status() and fetches the full task once at the end, so its return type is unchanged. - tasks.list(limit=...). Tests: router (status view omits spec/source and matches full on shared keys; limit; bad view -> 422) and SDK/server parity for TaskStatusView. Co-Authored-By: Claude Opus 5.5 (1M context) Signed-off-by: Noppanat Wadlom --- sdk/src/flowmesh/models/__init__.py | 3 +- sdk/src/flowmesh/models/tasks.py | 32 ++++++++++++++ sdk/src/flowmesh/resources/tasks.py | 36 +++++++++++++--- src/server/routers/v1/tasks.py | 35 +++++++++++++-- src/server/task/models.py | 67 +++++++++++++++++++++++++++++ tests/sdk/test_schema_compat.py | 6 +++ tests/server/test_tasks_router.py | 47 ++++++++++++++++++++ 7 files changed, 214 insertions(+), 12 deletions(-) diff --git a/sdk/src/flowmesh/models/__init__.py b/sdk/src/flowmesh/models/__init__.py index 796016aba..5411c4bcb 100644 --- a/sdk/src/flowmesh/models/__init__.py +++ b/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", diff --git a/sdk/src/flowmesh/models/tasks.py b/sdk/src/flowmesh/models/tasks.py index ab18c3b24..f32ecfbfd 100644 --- a/sdk/src/flowmesh/models/tasks.py +++ b/sdk/src/flowmesh/models/tasks.py @@ -68,3 +68,35 @@ class TaskInfo(BaseModel): dependents: list[str] completed: bool failed: bool + + +class TaskStatusView(BaseModel): + """The polling view of a task (``GET /tasks/{id}?view=status``). + + Status, placement, timing and errors only -- no task spec or source. A server + that predates the view returns the full task, which validates here too (the + extra fields are ignored), so this works against any server version.""" + + task_id: str + workflow_id: str | None = None + status: TaskStatus + task_type: str | None = None + category: str | None = None + assigned_worker: str | None = None + submitted_at: str | None = None + submitted_ts: float | None = None + 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 | None = None + max_attempts: int | None = None + 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 + completed: bool | None = None + failed: bool | None = None diff --git a/sdk/src/flowmesh/resources/tasks.py b/sdk/src/flowmesh/resources/tasks.py index 16d63b45c..9561e3755 100644 --- a/sdk/src/flowmesh/resources/tasks.py +++ b/sdk/src/flowmesh/resources/tasks.py @@ -13,7 +13,7 @@ LogQueryResponse, OkResponse, ) -from ..models.tasks import TaskInfo +from ..models.tasks import TaskInfo, TaskStatusView from ..params import append_param, extend_params from ..ssh import task_ssh_info, wait_for_ssh_info, wait_for_ssh_info_async from ._base import AsyncResource, SyncResource @@ -27,6 +27,16 @@ def retrieve(self, task_id: str) -> TaskInfo: data = self._client._request("GET", f"/tasks/{task_id}") return TaskInfo.model_validate(data) + def status(self, task_id: str) -> TaskStatusView: + """Retrieve only a task's status, placement, timing and errors. + + Use this for polling: the full task (``retrieve``) carries the spec and + source, ~0.5-1 MB for a Lumilake inference task (FlowMesh#225).""" + data = self._client._request( + "GET", f"/tasks/{task_id}", params=[("view", "status")] + ) + return TaskStatusView.model_validate(data) + def list( self, task_id: str | None = None, @@ -38,6 +48,7 @@ def list( graph_node_name: str | None = None, completed: bool | None = None, failed: bool | None = None, + limit: int | None = None, query_params: list[tuple[str, str]] | None = None, ) -> list[TaskInfo]: """List tasks with optional filters.""" @@ -51,6 +62,7 @@ def list( append_param(params, "graph_node_name", graph_node_name) append_param(params, "completed", completed) append_param(params, "failed", failed) + append_param(params, "limit", limit) if query_params: params.extend(query_params) data = self._client._request("GET", "/tasks", params=params or None) @@ -107,9 +119,8 @@ def wait( ) -> TaskInfo: """Poll a task until it reaches a terminal state.""" while True: - task = self.retrieve(task_id) - if task.status in TERMINAL_TASK_STATUSES: - return task + if self.status(task_id).status in TERMINAL_TASK_STATUSES: + return self.retrieve(task_id) time.sleep(interval) def ssh_info(self, task_id: str) -> dict[str, Any] | None: @@ -135,6 +146,16 @@ async def retrieve(self, task_id: str) -> TaskInfo: data = await self._client._request("GET", f"/tasks/{task_id}") return TaskInfo.model_validate(data) + async def status(self, task_id: str) -> TaskStatusView: + """Retrieve only a task's status, placement, timing and errors. + + Use this for polling: the full task (``retrieve``) carries the spec and + source, ~0.5-1 MB for a Lumilake inference task (FlowMesh#225).""" + data = await self._client._request( + "GET", f"/tasks/{task_id}", params=[("view", "status")] + ) + return TaskStatusView.model_validate(data) + async def list( self, task_id: str | None = None, @@ -146,6 +167,7 @@ async def list( graph_node_name: str | None = None, completed: bool | None = None, failed: bool | None = None, + limit: int | None = None, query_params: list[tuple[str, str]] | None = None, ) -> list[TaskInfo]: """List tasks with optional filters.""" @@ -159,6 +181,7 @@ async def list( append_param(params, "graph_node_name", graph_node_name) append_param(params, "completed", completed) append_param(params, "failed", failed) + append_param(params, "limit", limit) if query_params: params.extend(query_params) data = await self._client._request("GET", "/tasks", params=params or None) @@ -217,9 +240,8 @@ async def wait( ) -> TaskInfo: """Poll a task until it reaches a terminal state.""" while True: - task = await self.retrieve(task_id) - if task.status in TERMINAL_TASK_STATUSES: - return task + if (await self.status(task_id)).status in TERMINAL_TASK_STATUSES: + return await self.retrieve(task_id) await asyncio.sleep(interval) async def ssh_info(self, task_id: str) -> dict[str, Any] | None: diff --git a/src/server/routers/v1/tasks.py b/src/server/routers/v1/tasks.py index 7dba7bf66..8fc7408da 100644 --- a/src/server/routers/v1/tasks.py +++ b/src/server/routers/v1/tasks.py @@ -1,7 +1,7 @@ import asyncio import json import logging -from typing import Any +from typing import Any, Literal from fastapi import APIRouter, Depends, Header, HTTPException from fastapi import Path as ApiPath @@ -30,12 +30,24 @@ from ...registries.worker import WorkerRegistry from ...schemas.common import OkResponse from ...schemas.logs import LogEntry, LogEvent, LogQueryResponse +from ...task.models import TaskStatusView from ...task.runtime import TaskInfo, TaskRuntime from ...utils.misc import filter_models_by_queries router = APIRouter(prefix="/tasks", tags=["Tasks"]) _TASK_LIST = TypeAdapter(list[TaskInfo]) +_TASK_STATUS_LIST = TypeAdapter(list[TaskStatusView]) + +# Query keys the list endpoint consumes itself; they are not model-field filters. +_CONTROL_PARAMS = ("view", "limit") + +TaskView = Literal["full", "status"] +_VIEW_QUERY = Query( + "full", + description="`status` returns only status, placement, timing and errors " + "(no task spec or source) -- use it for polling.", +) def _strip_private_fields(data: dict[str, Any]) -> dict[str, Any]: @@ -57,7 +69,7 @@ def _pushed_down_queries( remaining = [ (key, value) for key, value in query_params.multi_items() - if key not in ("workflow_id", "status") + if key not in ("workflow_id", "status", *_CONTROL_PARAMS) ] return workflow_ids, statuses, QueryParams(remaining) @@ -81,10 +93,14 @@ def _sanitize_latest_update(info: TaskInfo) -> None: summary="List tasks", description="List all tasks.", response_description="List of task details.", - response_model=list[TaskInfo], + response_model=list[TaskInfo] | list[TaskStatusView], ) async def list_tasks( request: Request, + view: TaskView = _VIEW_QUERY, + limit: int | None = Query( + None, ge=1, description="Return at most this many tasks (after filtering)." + ), principal: PrincipalContext = Depends(authenticate_connection), runtime: TaskRuntime = Depends(get_runtime), logger: logging.Logger = Depends(get_logger), @@ -101,6 +117,13 @@ def _list_tasks_json() -> bytes: tasks = filter_models_by_queries(tasks, remaining) if allowed is not None: tasks = [task for task in tasks if task.task_id in allowed] + if limit is not None: + # `limit` used to be ignored (an unknown filter key), so a client + # asking for 100 tasks got every task: 240+ MB on a busy cluster. + tasks = tasks[:limit] + if view == "status": + views = [TaskStatusView.from_info(task) for task in tasks] + return _TASK_STATUS_LIST.dump_json(views, by_alias=True) for task in tasks: _sanitize_latest_update(task) return _TASK_LIST.dump_json(tasks, by_alias=True) @@ -115,13 +138,15 @@ def _list_tasks_json() -> bytes: summary="Get a task", description="Get task details by ID.", response_description="Task details.", + response_model=TaskInfo | TaskStatusView, ) async def get_task( task_id: str = ApiPath(..., min_length=1), + view: TaskView = _VIEW_QUERY, principal: PrincipalContext = Depends(authenticate_connection), runtime: TaskRuntime = Depends(get_runtime), logger: logging.Logger = Depends(get_logger), -) -> TaskInfo: +) -> TaskInfo | TaskStatusView: await require_permission( principal, ResourceKind.TASK, task_id, ResourceAction.READ, logger ) @@ -130,6 +155,8 @@ async def get_task( raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="task not found" ) + if view == "status": + return TaskStatusView.from_info(info) _sanitize_latest_update(info) return info diff --git a/src/server/task/models.py b/src/server/task/models.py index f1d6dd4f3..f8439f468 100644 --- a/src/server/task/models.py +++ b/src/server/task/models.py @@ -278,3 +278,70 @@ class TaskParsingResult(BaseModel): task_id: str = Field(description="Task identifier.") graph_node_name: str | None = Field(description="Original graph node name.") depends_on: list[str] = Field(description="Dependency task IDs.") + + +# Fields of TaskStatusView, in the order they are documented. +_STATUS_VIEW_FIELDS = ( + "task_id", + "workflow_id", + "status", + "task_type", + "category", + "assigned_worker", + "submitted_at", + "submitted_ts", + "dispatched_ts", + "started_ts", + "finished_ts", + "error", + "last_error", + "attempts", + "max_attempts", + "next_retry_at", + "last_failed_worker", + "parent_task_id", + "merged_parent_id", + "local_name", + "graph_node_name", + "completed", + "failed", +) + + +class TaskStatusView(BaseModel): + """The polling view of a task (``?view=status``): status, placement, timing + and errors -- no spec, no ``source``. + + A full TaskInfo carries the task spec plus its YAML source (~230 KB each for a + Lumilake inference task), so pollers asking only "is it done yet" were + pulling ~0.5-1 MB per call (mlsys-io/FlowMesh#225).""" + + task_id: str + workflow_id: str | None = None + status: str + task_type: str | None = None + category: str | None = None + assigned_worker: str | None = None + submitted_at: str | None = None + submitted_ts: float | None = None + 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 | None = None + max_attempts: int | None = None + 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 + completed: bool | None = None + failed: bool | None = None + + @classmethod + def from_info(cls, info: "TaskInfo") -> "TaskStatusView": + return cls.model_validate( + {name: getattr(info, name, None) for name in _STATUS_VIEW_FIELDS} + ) diff --git a/tests/sdk/test_schema_compat.py b/tests/sdk/test_schema_compat.py index 35b35a400..3fd23f4d8 100644 --- a/tests/sdk/test_schema_compat.py +++ b/tests/sdk/test_schema_compat.py @@ -32,6 +32,7 @@ SSHLimits, StorageInfo, TaskInfo, + TaskStatusView, TaskType, TaskUsage, VersionResponse, @@ -91,6 +92,7 @@ WorkflowValidateTaskEntry as SrvWorkflowValidateTaskEntry, ) from server.task.models import TaskInfo as SrvTaskInfo +from server.task.models import TaskStatusView as SrvTaskStatusView from server.task.models import TaskUsage as SrvTaskUsage from shared.schemas.artifact import ArtifactContext as SrvArtifactContext from shared.schemas.artifact import ArtifactRef as SrvArtifactRef @@ -262,6 +264,10 @@ def test_task_info_fields() -> None: assert_fields_match(SrvTaskInfo, TaskInfo, skip_server_fields={"last_queue_ts"}) +def test_task_status_view_fields() -> None: + assert_fields_match(SrvTaskStatusView, TaskStatusView) + + def test_worker_register_response_fields() -> None: assert_fields_match(SrvWorkerRegisterResponse, WorkerRegisterResponse) diff --git a/tests/server/test_tasks_router.py b/tests/server/test_tasks_router.py index 7873fc612..ff094b478 100644 --- a/tests/server/test_tasks_router.py +++ b/tests/server/test_tasks_router.py @@ -301,3 +301,50 @@ async def ping() -> dict[str, bool]: assert ping_resp.status_code == 200 assert elapsed < 0.3 assert list_resp.status_code == 200 + + +# mlsys-io/FlowMesh#225: pollers asked for a ~0.5-1 MB task to read one status +# field, and `limit` on the list was silently ignored (240+ MB responses). + + +@pytest.mark.anyio +async def test_get_task_status_view_omits_spec_and_source(no_checkers: None) -> None: + runtime, _ = build_runtime() + wf = await _register(runtime, _PAYLOAD) + task_id = _task_ids(runtime, wf)[0] + + async with _client(runtime) as ac: + full = await ac.get(f"{PREFIX}/tasks/{task_id}") + slim = await ac.get(f"{PREFIX}/tasks/{task_id}", params={"view": "status"}) + + assert full.status_code == 200 and slim.status_code == 200 + f, s = full.json(), slim.json() + assert "task" in f and "source" in f # default unchanged + assert "task" not in s and "source" not in s + for key in ("task_id", "status", "attempts", "error", "assigned_worker"): + assert s[key] == f[key] + + +@pytest.mark.anyio +async def test_list_tasks_honours_limit_and_status_view(no_checkers: None) -> None: + runtime, _ = build_runtime() + wf = await _register(runtime, _PAYLOAD) + assert len(_task_ids(runtime, wf)) == 2 + + async with _client(runtime) as ac: + one = await ac.get(f"{PREFIX}/tasks", params={"workflow_id": wf, "limit": 1}) + slim = await ac.get( + f"{PREFIX}/tasks", params={"workflow_id": wf, "view": "status"} + ) + + assert one.status_code == 200 and len(one.json()) == 1 + assert slim.status_code == 200 and len(slim.json()) == 2 + assert all("task" not in t and "source" not in t for t in slim.json()) + + +@pytest.mark.anyio +async def test_unknown_view_is_rejected(no_checkers: None) -> None: + runtime, _ = build_runtime() + async with _client(runtime) as ac: + resp = await ac.get(f"{PREFIX}/tasks", params={"view": "bogus"}) + assert resp.status_code == 422 From e30a1d63ea8eb6e9d3ee42025d9f3bb60d20792b Mon Sep 17 00:00:00 2001 From: Noppanat Wadlom Date: Sat, 17 Oct 2026 04:08:12 +0000 Subject: [PATCH 02/10] refactor: tighten the task status view schema and add latest_update Signed-off-by: Noppanat Wadlom --- sdk/src/flowmesh/models/tasks.py | 21 +++++----- src/server/routers/v1/tasks.py | 10 ++--- src/server/schemas/task.py | 56 ++++++++++++++++++++++++++ src/server/task/models.py | 67 ------------------------------- tests/sdk/test_schema_compat.py | 2 +- tests/server/test_tasks_router.py | 26 +++++++++--- 6 files changed, 92 insertions(+), 90 deletions(-) create mode 100644 src/server/schemas/task.py diff --git a/sdk/src/flowmesh/models/tasks.py b/sdk/src/flowmesh/models/tasks.py index f32ecfbfd..25d7c0058 100644 --- a/sdk/src/flowmesh/models/tasks.py +++ b/sdk/src/flowmesh/models/tasks.py @@ -71,32 +71,31 @@ class TaskInfo(BaseModel): class TaskStatusView(BaseModel): - """The polling view of a task (``GET /tasks/{id}?view=status``). + """A task's status, placement, timing and errors, without its spec or source. - Status, placement, timing and errors only -- no task spec or source. A server - that predates the view returns the full task, which validates here too (the - extra fields are ignored), so this works against any server version.""" + A server without this view returns the full task, which validates here too.""" task_id: str - workflow_id: str | None = None + workflow_id: str status: TaskStatus task_type: str | None = None category: str | None = None assigned_worker: str | None = None - submitted_at: str | None = None - submitted_ts: float | 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 | None = None - max_attempts: int | 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 - completed: bool | None = None - failed: bool | None = None + latest_update: dict[str, Any] | None = None + completed: bool + failed: bool diff --git a/src/server/routers/v1/tasks.py b/src/server/routers/v1/tasks.py index 8fc7408da..76d410bf3 100644 --- a/src/server/routers/v1/tasks.py +++ b/src/server/routers/v1/tasks.py @@ -30,7 +30,7 @@ from ...registries.worker import WorkerRegistry from ...schemas.common import OkResponse from ...schemas.logs import LogEntry, LogEvent, LogQueryResponse -from ...task.models import TaskStatusView +from ...schemas.task import TaskStatusView from ...task.runtime import TaskInfo, TaskRuntime from ...utils.misc import filter_models_by_queries @@ -46,7 +46,7 @@ _VIEW_QUERY = Query( "full", description="`status` returns only status, placement, timing and errors " - "(no task spec or source) -- use it for polling.", + "(no task spec or source) — use it for polling.", ) @@ -121,11 +121,11 @@ def _list_tasks_json() -> bytes: # `limit` used to be ignored (an unknown filter key), so a client # asking for 100 tasks got every task: 240+ MB on a busy cluster. tasks = tasks[:limit] + for task in tasks: + _sanitize_latest_update(task) if view == "status": views = [TaskStatusView.from_info(task) for task in tasks] return _TASK_STATUS_LIST.dump_json(views, by_alias=True) - for task in tasks: - _sanitize_latest_update(task) return _TASK_LIST.dump_json(tasks, by_alias=True) return Response( @@ -155,9 +155,9 @@ async def get_task( raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="task not found" ) + _sanitize_latest_update(info) if view == "status": return TaskStatusView.from_info(info) - _sanitize_latest_update(info) return info diff --git a/src/server/schemas/task.py b/src/server/schemas/task.py new file mode 100644 index 000000000..d9d68f2c8 --- /dev/null +++ b/src/server/schemas/task.py @@ -0,0 +1,56 @@ +from typing import Any, Self + +from pydantic import BaseModel, Field + +from ..task.models import TaskInfo + + +class TaskStatusView(BaseModel): + """A task's status, placement, timing and errors, without its spec or source.""" + + task_id: str = Field(description="Task identifier.") + workflow_id: str = Field(description="Workflow identifier.") + status: str = Field(description="Task status.") + task_type: str | None = Field(default=None, description="Task type.") + category: str | None = Field(default=None, description="Task category.") + assigned_worker: str | None = Field( + default=None, description="Assigned worker identifier." + ) + submitted_at: str = Field(description="Submission timestamp.") + submitted_ts: float = Field(description="Submission epoch seconds.") + dispatched_ts: float | None = Field( + default=None, description="Dispatch timestamp (epoch seconds)." + ) + started_ts: float | None = Field( + default=None, description="Start timestamp (epoch seconds)." + ) + finished_ts: float | None = Field( + default=None, description="Finish timestamp (epoch seconds)." + ) + error: str | None = Field(default=None, description="Error message, if any.") + last_error: str | None = Field( + default=None, description="Most recent executor error message." + ) + attempts: int = Field(description="Attempt count.") + max_attempts: int = Field(description="Max retry count.") + next_retry_at: str | None = Field(default=None, description="Next retry timestamp.") + last_failed_worker: str | None = Field( + default=None, description="The most recent worker to have failed this task." + ) + parent_task_id: str | None = Field( + default=None, description="Parent task identifier." + ) + merged_parent_id: str | None = Field( + default=None, description="Merged parent task identifier." + ) + local_name: str | None = Field(default=None, description="Workflow stage name.") + graph_node_name: str | None = Field(default=None, description="Graph node name.") + latest_update: dict[str, Any] | None = Field( + default=None, description="Latest mid-task update payload." + ) + completed: bool = Field(description="Whether the task completed successfully.") + failed: bool = Field(description="Whether the task failed.") + + @classmethod + def from_info(cls, info: TaskInfo) -> Self: + return cls.model_validate(info, from_attributes=True) diff --git a/src/server/task/models.py b/src/server/task/models.py index f8439f468..f1d6dd4f3 100644 --- a/src/server/task/models.py +++ b/src/server/task/models.py @@ -278,70 +278,3 @@ class TaskParsingResult(BaseModel): task_id: str = Field(description="Task identifier.") graph_node_name: str | None = Field(description="Original graph node name.") depends_on: list[str] = Field(description="Dependency task IDs.") - - -# Fields of TaskStatusView, in the order they are documented. -_STATUS_VIEW_FIELDS = ( - "task_id", - "workflow_id", - "status", - "task_type", - "category", - "assigned_worker", - "submitted_at", - "submitted_ts", - "dispatched_ts", - "started_ts", - "finished_ts", - "error", - "last_error", - "attempts", - "max_attempts", - "next_retry_at", - "last_failed_worker", - "parent_task_id", - "merged_parent_id", - "local_name", - "graph_node_name", - "completed", - "failed", -) - - -class TaskStatusView(BaseModel): - """The polling view of a task (``?view=status``): status, placement, timing - and errors -- no spec, no ``source``. - - A full TaskInfo carries the task spec plus its YAML source (~230 KB each for a - Lumilake inference task), so pollers asking only "is it done yet" were - pulling ~0.5-1 MB per call (mlsys-io/FlowMesh#225).""" - - task_id: str - workflow_id: str | None = None - status: str - task_type: str | None = None - category: str | None = None - assigned_worker: str | None = None - submitted_at: str | None = None - submitted_ts: float | None = None - 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 | None = None - max_attempts: int | None = None - 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 - completed: bool | None = None - failed: bool | None = None - - @classmethod - def from_info(cls, info: "TaskInfo") -> "TaskStatusView": - return cls.model_validate( - {name: getattr(info, name, None) for name in _STATUS_VIEW_FIELDS} - ) diff --git a/tests/sdk/test_schema_compat.py b/tests/sdk/test_schema_compat.py index 3fd23f4d8..38ed8d54c 100644 --- a/tests/sdk/test_schema_compat.py +++ b/tests/sdk/test_schema_compat.py @@ -79,6 +79,7 @@ from server.schemas.node import StorageInfo as SrvStorageInfo from server.schemas.node import WorkerHardware as SrvWorkerHardware from server.schemas.node import WorkerRegisterResponse as SrvWorkerRegisterResponse +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 WorkflowSubmitResponse as SrvWorkflowSubmitResponse @@ -92,7 +93,6 @@ WorkflowValidateTaskEntry as SrvWorkflowValidateTaskEntry, ) from server.task.models import TaskInfo as SrvTaskInfo -from server.task.models import TaskStatusView as SrvTaskStatusView from server.task.models import TaskUsage as SrvTaskUsage from shared.schemas.artifact import ArtifactContext as SrvArtifactContext from shared.schemas.artifact import ArtifactRef as SrvArtifactRef diff --git a/tests/server/test_tasks_router.py b/tests/server/test_tasks_router.py index ff094b478..f17e59e84 100644 --- a/tests/server/test_tasks_router.py +++ b/tests/server/test_tasks_router.py @@ -16,6 +16,8 @@ from server.auth.security import PrincipalContext, authenticate_connection from server.hooks import PERMISSION_CHECKERS from server.routers.v1 import tasks as tasks_router +from server.schemas.task import TaskStatusView +from server.task.models import TaskInfo from server.task.runtime import TaskRuntime from tests.server.task.merge_harness import build_runtime @@ -303,8 +305,9 @@ async def ping() -> dict[str, bool]: assert list_resp.status_code == 200 -# mlsys-io/FlowMesh#225: pollers asked for a ~0.5-1 MB task to read one status -# field, and `limit` on the list was silently ignored (240+ MB responses). +def test_status_view_fields_are_task_info_fields() -> None: + served = set(TaskInfo.model_fields) | set(TaskInfo.model_computed_fields) + assert set(TaskStatusView.model_fields) <= served @pytest.mark.anyio @@ -312,6 +315,7 @@ async def test_get_task_status_view_omits_spec_and_source(no_checkers: None) -> runtime, _ = build_runtime() wf = await _register(runtime, _PAYLOAD) task_id = _task_ids(runtime, wf)[0] + runtime.tasks[task_id].latest_update = {"ssh": {"host": "h", "_secret": "s"}} async with _client(runtime) as ac: full = await ac.get(f"{PREFIX}/tasks/{task_id}") @@ -319,10 +323,20 @@ async def test_get_task_status_view_omits_spec_and_source(no_checkers: None) -> assert full.status_code == 200 and slim.status_code == 200 f, s = full.json(), slim.json() - assert "task" in f and "source" in f # default unchanged - assert "task" not in s and "source" not in s - for key in ("task_id", "status", "attempts", "error", "assigned_worker"): - assert s[key] == f[key] + assert "task" in f and "source" in f + assert set(s) == set(TaskStatusView.model_fields) + assert {key: f[key] for key in s} == s + assert s["latest_update"] == {"ssh": {"host": "h"}} + + +@pytest.mark.anyio +async def test_get_task_unknown_view_is_rejected(no_checkers: None) -> None: + runtime, _ = build_runtime() + wf = await _register(runtime, _PAYLOAD) + task_id = _task_ids(runtime, wf)[0] + async with _client(runtime) as ac: + resp = await ac.get(f"{PREFIX}/tasks/{task_id}", params={"view": "bogus"}) + assert resp.status_code == 422 @pytest.mark.anyio From 97320e35010ac6c743454ac739daa1cc2003e154 Mon Sep 17 00:00:00 2001 From: Noppanat Wadlom Date: Sat, 17 Oct 2026 04:10:07 +0000 Subject: [PATCH 03/10] fix: list tasks by submission and keep the newest under limit Signed-off-by: Noppanat Wadlom --- docs/API.md | 9 ++++---- docs/ARCHITECTURE.md | 2 +- src/server/routers/v1/tasks.py | 10 ++++----- src/server/task/models.py | 8 +++++++ tests/server/test_tasks_router.py | 36 ++++++++++++++++++++++++++----- 5 files changed, 49 insertions(+), 16 deletions(-) diff --git a/docs/API.md b/docs/API.md index 2a49e3e54..05caf8355 100644 --- a/docs/API.md +++ b/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 submission first. Filters: `workflow_id`, `status`, `task_type`, `assigned_worker`. `limit` keeps the newest matching tasks; `view=status` returns each task's status view. | +| GET | `/api/v1/tasks/{id}` | Task details. `view=status` returns only status, placement, timing, errors and the latest update, without the task spec or source; use it for polling. | | 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 +Log queries accept `limit` and `before` / `after` cursors. The cursor is an opaque base64 of `(timestamp, id)`; do not parse client-side. diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index d8aca337e..5f74e7135 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -244,7 +244,7 @@ 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` / +- **Cursor pagination.** Log queries accept `limit` and `before` / `after` cursors. The cursor is an opaque base64 of `(timestamp, id)`; do not parse client-side. - **Redis channels.** The runtime uses two Redis endpoints (separate instances in the diff --git a/src/server/routers/v1/tasks.py b/src/server/routers/v1/tasks.py index 76d410bf3..796796e91 100644 --- a/src/server/routers/v1/tasks.py +++ b/src/server/routers/v1/tasks.py @@ -31,6 +31,7 @@ from ...schemas.common import OkResponse from ...schemas.logs import LogEntry, LogEvent, LogQueryResponse from ...schemas.task import TaskStatusView +from ...task.models import task_order from ...task.runtime import TaskInfo, TaskRuntime from ...utils.misc import filter_models_by_queries @@ -91,7 +92,7 @@ def _sanitize_latest_update(info: TaskInfo) -> None: @router.get( "", summary="List tasks", - description="List all tasks.", + description="List tasks ordered by submission, oldest first.", response_description="List of task details.", response_model=list[TaskInfo] | list[TaskStatusView], ) @@ -99,7 +100,7 @@ async def list_tasks( request: Request, view: TaskView = _VIEW_QUERY, limit: int | None = Query( - None, ge=1, description="Return at most this many tasks (after filtering)." + None, ge=1, le=10_000, description="Return only the newest this many tasks." ), principal: PrincipalContext = Depends(authenticate_connection), runtime: TaskRuntime = Depends(get_runtime), @@ -117,10 +118,9 @@ def _list_tasks_json() -> bytes: tasks = filter_models_by_queries(tasks, remaining) if allowed is not None: tasks = [task for task in tasks if task.task_id in allowed] + tasks.sort(key=task_order) if limit is not None: - # `limit` used to be ignored (an unknown filter key), so a client - # asking for 100 tasks got every task: 240+ MB on a busy cluster. - tasks = tasks[:limit] + tasks = tasks[-limit:] for task in tasks: _sanitize_latest_update(task) if view == "status": diff --git a/src/server/task/models.py b/src/server/task/models.py index f1d6dd4f3..a4b6fbf13 100644 --- a/src/server/task/models.py +++ b/src/server/task/models.py @@ -256,6 +256,14 @@ def _serialize(self, handler: SerializerFunctionWrapHandler) -> dict[str, Any]: return handler(redacted) +# A task's position in a listing: its submission time, then its id. +type TaskOrder = tuple[float, str] + + +def task_order(record: TaskRecord) -> TaskOrder: + return record.submitted_ts, record.task_id + + class TaskInfo(TaskRecord): depends_on: list[str] = Field(description="Dependency task IDs.") pending_dependencies: list[str] = Field( diff --git a/tests/server/test_tasks_router.py b/tests/server/test_tasks_router.py index f17e59e84..f2bcdbfd4 100644 --- a/tests/server/test_tasks_router.py +++ b/tests/server/test_tasks_router.py @@ -340,20 +340,46 @@ async def test_get_task_unknown_view_is_rejected(no_checkers: None) -> None: @pytest.mark.anyio -async def test_list_tasks_honours_limit_and_status_view(no_checkers: None) -> None: +async def test_list_tasks_orders_by_submission_and_limits_to_newest( + no_checkers: None, +) -> None: + runtime, _ = build_runtime() + wf = await _register(runtime, _PAYLOAD) + first, second = _task_ids(runtime, wf) + runtime.tasks[first].submitted_ts = 200.0 + runtime.tasks[second].submitted_ts = 100.0 + + async with _client(runtime) as ac: + every = await ac.get(f"{PREFIX}/tasks", params={"workflow_id": wf}) + newest = await ac.get(f"{PREFIX}/tasks", params={"workflow_id": wf, "limit": 1}) + + assert [t["task_id"] for t in every.json()] == [second, first] + assert [t["task_id"] for t in newest.json()] == [first] + + +@pytest.mark.anyio +async def test_list_tasks_status_view(no_checkers: None) -> None: runtime, _ = build_runtime() wf = await _register(runtime, _PAYLOAD) - assert len(_task_ids(runtime, wf)) == 2 async with _client(runtime) as ac: - one = await ac.get(f"{PREFIX}/tasks", params={"workflow_id": wf, "limit": 1}) slim = await ac.get( f"{PREFIX}/tasks", params={"workflow_id": wf, "view": "status"} ) - assert one.status_code == 200 and len(one.json()) == 1 assert slim.status_code == 200 and len(slim.json()) == 2 - assert all("task" not in t and "source" not in t for t in slim.json()) + assert all(set(t) == set(TaskStatusView.model_fields) for t in slim.json()) + + +@pytest.mark.anyio +@pytest.mark.parametrize("limit", [0, 10_001]) +async def test_list_tasks_rejects_out_of_range_limit( + no_checkers: None, limit: int +) -> None: + runtime, _ = build_runtime() + async with _client(runtime) as ac: + resp = await ac.get(f"{PREFIX}/tasks", params={"limit": limit}) + assert resp.status_code == 422 @pytest.mark.anyio From 13dc9c8525065c07be4e8b6633174ac37ec96700 Mon Sep 17 00:00:00 2001 From: Noppanat Wadlom Date: Sat, 17 Oct 2026 04:12:15 +0000 Subject: [PATCH 04/10] feat: add tasks.list_statuses and poll SSH info through the status view Signed-off-by: Noppanat Wadlom --- docs/SDK.md | 13 ++- sdk/src/flowmesh/resources/tasks.py | 172 ++++++++++++++++++++++------ sdk/src/flowmesh/ssh.py | 8 +- tests/sdk/test_models.py | 13 +++ tests/sdk/test_request_building.py | 96 ++++++++++++++++ 5 files changed, 256 insertions(+), 46 deletions(-) diff --git a/docs/SDK.md b/docs/SDK.md index d7d7ea9d5..7535a77de 100644 --- a/docs/SDK.md +++ b/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(task_id)` polls the status view and returns the full + task once it finishes. - **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,7 +69,7 @@ bundle missing a requested section raises `FlowMeshError`. ## Cursor pagination -Cursor-enabled calls take `limit` and `before` / `after` params. +Log queries take `limit` and `before` / `after` params. Cursors are opaque base64 strings — pass them through; do not parse them. diff --git a/sdk/src/flowmesh/resources/tasks.py b/sdk/src/flowmesh/resources/tasks.py index 9561e3755..e5efeb065 100644 --- a/sdk/src/flowmesh/resources/tasks.py +++ b/sdk/src/flowmesh/resources/tasks.py @@ -1,6 +1,7 @@ """Task resource operations.""" import asyncio +import builtins import time from collections.abc import AsyncIterator, Iterator from pathlib import Path @@ -19,6 +20,35 @@ from ._base import AsyncResource, SyncResource +def _list_params( + task_id: str | None = None, + workflow_id: str | None = None, + status: str | list[str] | None = None, + category: str | None = None, + task_type: str | None = None, + assigned_worker: str | None = None, + graph_node_name: str | None = None, + completed: bool | None = None, + failed: bool | 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, "task_id", task_id) + append_param(params, "workflow_id", workflow_id) + extend_params(params, "status", status) + append_param(params, "category", category) + append_param(params, "task_type", task_type) + append_param(params, "assigned_worker", assigned_worker) + append_param(params, "graph_node_name", graph_node_name) + append_param(params, "completed", completed) + append_param(params, "failed", failed) + append_param(params, "limit", limit) + if query_params: + params.extend(query_params) + return params + + class Tasks(SyncResource): """Synchronous task operations.""" @@ -28,10 +58,9 @@ def retrieve(self, task_id: str) -> TaskInfo: return TaskInfo.model_validate(data) def status(self, task_id: str) -> TaskStatusView: - """Retrieve only a task's status, placement, timing and errors. + """Retrieve a task's status view, without its spec or source. - Use this for polling: the full task (``retrieve``) carries the spec and - source, ~0.5-1 MB for a Lumilake inference task (FlowMesh#225).""" + Prefer this over ``retrieve`` for polling.""" data = self._client._request( "GET", f"/tasks/{task_id}", params=[("view", "status")] ) @@ -51,23 +80,57 @@ def list( limit: int | None = None, query_params: list[tuple[str, str]] | None = None, ) -> list[TaskInfo]: - """List tasks with optional filters.""" - params: list[tuple[str, str]] = [] - append_param(params, "task_id", task_id) - append_param(params, "workflow_id", workflow_id) - extend_params(params, "status", status) - append_param(params, "category", category) - append_param(params, "task_type", task_type) - append_param(params, "assigned_worker", assigned_worker) - append_param(params, "graph_node_name", graph_node_name) - append_param(params, "completed", completed) - append_param(params, "failed", failed) - append_param(params, "limit", limit) - if query_params: - params.extend(query_params) + """List tasks with optional filters, oldest submission first. + + ``limit`` keeps only the newest matching tasks.""" + params = _list_params( + task_id, + workflow_id, + status, + category, + task_type, + assigned_worker, + graph_node_name, + completed, + failed, + limit, + query_params, + ) data = self._client._request("GET", "/tasks", params=params or None) return [TaskInfo.model_validate(t) for t in data] + def list_statuses( + self, + task_id: str | None = None, + workflow_id: str | None = None, + status: str | builtins.list[str] | None = None, + category: str | None = None, + task_type: str | None = None, + assigned_worker: str | None = None, + graph_node_name: str | None = None, + completed: bool | None = None, + failed: bool | None = None, + limit: int | None = None, + query_params: builtins.list[tuple[str, str]] | None = None, + ) -> builtins.list[TaskStatusView]: + """List the status views of tasks; takes the filters of ``list``.""" + params = _list_params( + task_id, + workflow_id, + status, + category, + task_type, + assigned_worker, + graph_node_name, + completed, + failed, + limit, + query_params, + ) + params.append(("view", "status")) + data = self._client._request("GET", "/tasks", params=params) + return [TaskStatusView.model_validate(t) for t in data] + def stop(self, task_id: str) -> OkResponse: """Request a running task to stop.""" data = self._client._request("POST", f"/tasks/{task_id}/stop") @@ -125,7 +188,7 @@ def wait( def ssh_info(self, task_id: str) -> dict[str, Any] | None: """Return SSH publish info for a task if present.""" - return task_ssh_info(self.retrieve(task_id)) + return task_ssh_info(self.status(task_id)) def wait_for_ssh( self, @@ -134,7 +197,7 @@ def wait_for_ssh( ) -> dict[str, Any]: """Poll a task until SSH session information is available.""" return wait_for_ssh_info( - self.retrieve, task_id, interval=interval, sleep=time.sleep + self.status, task_id, interval=interval, sleep=time.sleep ) @@ -147,10 +210,9 @@ async def retrieve(self, task_id: str) -> TaskInfo: return TaskInfo.model_validate(data) async def status(self, task_id: str) -> TaskStatusView: - """Retrieve only a task's status, placement, timing and errors. + """Retrieve a task's status view, without its spec or source. - Use this for polling: the full task (``retrieve``) carries the spec and - source, ~0.5-1 MB for a Lumilake inference task (FlowMesh#225).""" + Prefer this over ``retrieve`` for polling.""" data = await self._client._request( "GET", f"/tasks/{task_id}", params=[("view", "status")] ) @@ -170,23 +232,57 @@ async def list( limit: int | None = None, query_params: list[tuple[str, str]] | None = None, ) -> list[TaskInfo]: - """List tasks with optional filters.""" - params: list[tuple[str, str]] = [] - append_param(params, "task_id", task_id) - append_param(params, "workflow_id", workflow_id) - extend_params(params, "status", status) - append_param(params, "category", category) - append_param(params, "task_type", task_type) - append_param(params, "assigned_worker", assigned_worker) - append_param(params, "graph_node_name", graph_node_name) - append_param(params, "completed", completed) - append_param(params, "failed", failed) - append_param(params, "limit", limit) - if query_params: - params.extend(query_params) + """List tasks with optional filters, oldest submission first. + + ``limit`` keeps only the newest matching tasks.""" + params = _list_params( + task_id, + workflow_id, + status, + category, + task_type, + assigned_worker, + graph_node_name, + completed, + failed, + limit, + query_params, + ) data = await self._client._request("GET", "/tasks", params=params or None) return [TaskInfo.model_validate(t) for t in data] + async def list_statuses( + self, + task_id: str | None = None, + workflow_id: str | None = None, + status: str | builtins.list[str] | None = None, + category: str | None = None, + task_type: str | None = None, + assigned_worker: str | None = None, + graph_node_name: str | None = None, + completed: bool | None = None, + failed: bool | None = None, + limit: int | None = None, + query_params: builtins.list[tuple[str, str]] | None = None, + ) -> builtins.list[TaskStatusView]: + """List the status views of tasks; takes the filters of ``list``.""" + params = _list_params( + task_id, + workflow_id, + status, + category, + task_type, + assigned_worker, + graph_node_name, + completed, + failed, + limit, + query_params, + ) + params.append(("view", "status")) + data = await self._client._request("GET", "/tasks", params=params) + return [TaskStatusView.model_validate(t) for t in data] + async def stop(self, task_id: str) -> OkResponse: """Request a running task to stop.""" data = await self._client._request("POST", f"/tasks/{task_id}/stop") @@ -246,7 +342,7 @@ async def wait( async def ssh_info(self, task_id: str) -> dict[str, Any] | None: """Return SSH publish info for a task if present.""" - task = await self.retrieve(task_id) + task = await self.status(task_id) return task_ssh_info(task) async def wait_for_ssh( @@ -256,7 +352,7 @@ async def wait_for_ssh( ) -> dict[str, Any]: """Poll a task until SSH session information is available.""" return await wait_for_ssh_info_async( - self.retrieve, + self.status, task_id, interval=interval, sleep=asyncio.sleep, diff --git a/sdk/src/flowmesh/ssh.py b/sdk/src/flowmesh/ssh.py index 671a7c630..30406468f 100644 --- a/sdk/src/flowmesh/ssh.py +++ b/sdk/src/flowmesh/ssh.py @@ -11,10 +11,10 @@ from ._constants import API_VERSION_PREFIX from .exceptions import FlowMeshError from .models.common import TaskStatus -from .models.tasks import TaskInfo +from .models.tasks import TaskInfo, TaskStatusView -def task_ssh_info(task: TaskInfo) -> dict[str, Any] | None: +def task_ssh_info(task: TaskInfo | TaskStatusView) -> dict[str, Any] | None: """Extract SSH publish information from a task update.""" latest = task.latest_update if not isinstance(latest, dict): @@ -24,7 +24,7 @@ def task_ssh_info(task: TaskInfo) -> dict[str, Any] | None: def wait_for_ssh_info( - retrieve_task: Callable[[str], TaskInfo], + retrieve_task: Callable[[str], TaskInfo | TaskStatusView], task_id: str, interval: float, sleep: Callable[[float], Any], @@ -45,7 +45,7 @@ def wait_for_ssh_info( async def wait_for_ssh_info_async( - retrieve_task: Callable[[str], Awaitable[TaskInfo]], + retrieve_task: Callable[[str], Awaitable[TaskInfo | TaskStatusView]], task_id: str, interval: float, sleep: Any, diff --git a/tests/sdk/test_models.py b/tests/sdk/test_models.py index 72b854cb0..3f66382ab 100644 --- a/tests/sdk/test_models.py +++ b/tests/sdk/test_models.py @@ -19,6 +19,7 @@ OkResponse, ProfileSummary, TaskInfo, + TaskStatusView, TaskTiming, TaskUsage, VersionResponse, @@ -60,6 +61,7 @@ from server.schemas.node import StorageInfo as SrvStorageInfo 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 WorkflowSubmitResponse as SrvWorkflowSubmitResponse from server.schemas.workflow import ( WorkflowSubmitTaskEntry as SrvWorkflowSubmitTaskEntry, @@ -324,6 +326,17 @@ def test_task_info_accepts_source_or_raw_yaml(self, key: str) -> None: payload.pop("raw_yaml" if key == "source" else "source") assert TaskInfo.model_validate(payload).source == expected + def test_task_status_view_validate(self) -> None: + payload = _dump(SrvTaskStatusView.from_info(_SRV_TASK_INFO)) + view = TaskStatusView.model_validate(payload) + assert view.task_id == "t-abc" + assert view.status == "DONE" + + def test_task_status_view_accepts_full_task(self) -> None: + view = TaskStatusView.model_validate(_dump(_SRV_TASK_INFO)) + assert view.attempts == 1 + assert view.completed is True + def test_task_usage(self) -> None: usage = TaskUsage.model_validate(_dump(_SRV_TASK_USAGE)) assert usage.runtime_sec == 60.0 diff --git a/tests/sdk/test_request_building.py b/tests/sdk/test_request_building.py index 2491ef4f9..b877d32f0 100644 --- a/tests/sdk/test_request_building.py +++ b/tests/sdk/test_request_building.py @@ -7,6 +7,7 @@ import json from typing import Any +import httpx import pytest import respx from flowmesh import AsyncFlowMesh, FlowMesh @@ -19,10 +20,15 @@ from server.registries.workflow import WorkflowStatus as SrvWorkflowStatus 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 WorkflowSubmitResponse as SrvWorkflowSubmitResponse from server.schemas.workflow import ( WorkflowSubmitTaskEntry as SrvWorkflowSubmitTaskEntry, ) +from server.task.models import TaskInfo as SrvTaskInfo +from shared.tasks.envelope import TaskEnvelopeTemplate +from shared.tasks.specs.misc import EchoSpecTemplate +from shared.tasks.task_type import TaskType from .helpers import clear_env_and_config_file from .router_app import TEST_BASE_URL, route_url @@ -71,6 +77,32 @@ def mock_client() -> FlowMesh: ) +_TASK_RESPONSE = SrvTaskInfo( + task_id="t-1", + workflow_id="wf-1", + owner_id="usr-1", + source="apiVersion: flowmesh/v1\nkind: Task", + task=TaskEnvelopeTemplate( + apiVersion="flowmesh/v1", + kind="Task", + spec=EchoSpecTemplate(taskType=TaskType.ECHO), + ), + status="DONE", + submitted_at="2025-01-15T10:30:00Z", + submitted_ts=1705312200.0, + depends_on=[], + pending_dependencies=[], + dependents=[], + completed=True, + failed=False, +) + + +def _status_json(status: str) -> dict[str, Any]: + view = SrvTaskStatusView.from_info(_TASK_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") @@ -221,6 +253,70 @@ def test_list_workers_with_tags(self, mock_client: FlowMesh) -> None: assert "tags=a100" in url +class TestTaskPolling: + @respx.mock + def test_status_requests_status_view(self, mock_client: FlowMesh) -> None: + route = respx.get(route_url("get_task", task_id="t-1")).respond( + json=_status_json("DISPATCHED") + ) + view = mock_client.tasks.status("t-1") + assert route.calls[0].request.url.params["view"] == "status" + assert view.status == "DISPATCHED" + + @respx.mock + def test_list_sends_limit(self, mock_client: FlowMesh) -> None: + route = respx.get(route_url("list_tasks")).respond(json=[]) + mock_client.tasks.list(workflow_id="wf-1", limit=2) + params = route.calls[0].request.url.params + assert params["limit"] == "2" + 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_tasks")).respond( + json=[_status_json("PENDING")] + ) + views = mock_client.tasks.list_statuses(status="PENDING", limit=5) + params = route.calls[0].request.url.params + assert params["view"] == "status" + assert params["status"] == "PENDING" + assert params["limit"] == "5" + assert [v.task_id for v in views] == ["t-1"] + + @respx.mock + def test_wait_polls_status_then_fetches_full_task( + self, mock_client: FlowMesh + ) -> None: + route = respx.get(route_url("get_task", task_id="t-1")).mock( + side_effect=[ + httpx.Response(200, json=_status_json("DISPATCHED")), + httpx.Response(200, json=_status_json("DONE")), + httpx.Response(200, json=_json(_TASK_RESPONSE)), + ] + ) + task = mock_client.tasks.wait("t-1", interval=0) + views = [call.request.url.params.get("view") for call in route.calls] + assert views == ["status", "status", None] + assert task.source == _TASK_RESPONSE.source + + @pytest.mark.anyio + @respx.mock + async def test_async_wait_polls_status_then_fetches_full_task(self) -> None: + route = respx.get(route_url("get_task", task_id="t-1")).mock( + side_effect=[ + httpx.Response(200, json=_status_json("DONE")), + httpx.Response(200, json=_json(_TASK_RESPONSE)), + ] + ) + async with AsyncFlowMesh( + base_url=TEST_BASE_URL, api_key="flm-test-key" + ) as client: + task = await client.tasks.wait("t-1", interval=0) + views = [call.request.url.params.get("view") for call in route.calls] + assert views == ["status", None] + assert task.task_id == "t-1" + + class TestNodeRegister: @respx.mock def test_register_includes_version_when_given(self, mock_client: FlowMesh) -> None: From 8055829a6bac443eb5b1bf1d01175d29e38731b5 Mon Sep 17 00:00:00 2001 From: Noppanat Wadlom Date: Sat, 17 Oct 2026 04:12:58 +0000 Subject: [PATCH 05/10] feat: poll the task status view in CLI watch and SSH log streaming Signed-off-by: Noppanat Wadlom --- cli/src/flowmesh_cli/commands/ssh.py | 2 +- cli/src/flowmesh_cli/commands/task.py | 10 +++++----- tests/cli/test_cli_commands.py | 21 +++++++++++++++++++++ 3 files changed, 27 insertions(+), 6 deletions(-) diff --git a/cli/src/flowmesh_cli/commands/ssh.py b/cli/src/flowmesh_cli/commands/ssh.py index 1d23ece28..9c1033c45 100644 --- a/cli/src/flowmesh_cli/commands/ssh.py +++ b/cli/src/flowmesh_cli/commands/ssh.py @@ -130,7 +130,7 @@ def _stream_logs_and_exit( except NotFoundError: pass - task_info = client.tasks.retrieve(task_id) + task_info = client.tasks.status(task_id) if task_info.status in TERMINAL_TASK_STATUSES: _exit_for_task_status(task_id, task_info) if not saw_logs: diff --git a/cli/src/flowmesh_cli/commands/task.py b/cli/src/flowmesh_cli/commands/task.py index 062c85fae..1706da8f0 100644 --- a/cli/src/flowmesh_cli/commands/task.py +++ b/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: diff --git a/tests/cli/test_cli_commands.py b/tests/cli/test_cli_commands.py index 44baa43e1..cdef40ff8 100644 --- a/tests/cli/test_cli_commands.py +++ b/tests/cli/test_cli_commands.py @@ -5,6 +5,7 @@ import typer from flowmesh.models import ( + TaskStatus, WorkflowSubmitResponse, ) from flowmesh_cli.cli import build_cli_app @@ -118,6 +119,26 @@ def test_id_forwarded(self, mock_hc: MagicMock) -> None: client.tasks.retrieve.assert_called_once_with("t-123") +class TestTaskWatch: + @patch("flowmesh_cli.commands.task.FlowMesh") + def test_polls_status_then_prints_full_task(self, mock_hc: MagicMock) -> None: + client = _mock_client() + client.tasks.status.side_effect = [ + MagicMock(status=TaskStatus.DISPATCHED, latest_update=None), + MagicMock(status=TaskStatus.DONE, latest_update=None), + ] + client.tasks.retrieve.return_value.model_dump.return_value = { + "task_id": "t-123" + } + mock_hc.return_value = client + + result = runner.invoke(_app(), ["task", "watch", "t-123", "--interval", "0"]) + assert result.exit_code == 0 + assert client.tasks.status.call_count == 2 + client.tasks.retrieve.assert_called_once_with("t-123") + assert '"task_id": "t-123"' in result.stdout + + class TestVersion: def test_version_flag(self) -> None: result = runner.invoke(_app(), ["--version"]) From 89b82acde00bcb2bebb331e6dfaf1e704c507f31 Mon Sep 17 00:00:00 2001 From: Noppanat Wadlom Date: Sat, 17 Oct 2026 04:23:21 +0000 Subject: [PATCH 06/10] refactor: align task status view descriptions and pass list filters by name Signed-off-by: Noppanat Wadlom --- sdk/src/flowmesh/models/tasks.py | 3 +- sdk/src/flowmesh/resources/tasks.py | 88 ++++++++++++++--------------- src/server/routers/v1/tasks.py | 11 ++-- src/server/schemas/task.py | 3 +- src/server/task/models.py | 6 +- tests/sdk/test_request_building.py | 43 +++++++++++++- tests/server/test_tasks_router.py | 13 +++-- 7 files changed, 105 insertions(+), 62 deletions(-) diff --git a/sdk/src/flowmesh/models/tasks.py b/sdk/src/flowmesh/models/tasks.py index 25d7c0058..ef72871cf 100644 --- a/sdk/src/flowmesh/models/tasks.py +++ b/sdk/src/flowmesh/models/tasks.py @@ -71,7 +71,8 @@ class TaskInfo(BaseModel): class TaskStatusView(BaseModel): - """A task's status, placement, timing and errors, without its spec or source. + """A task's status, placement, timing, errors and latest update, without its + spec or source. A server without this view returns the full task, which validates here too.""" diff --git a/sdk/src/flowmesh/resources/tasks.py b/sdk/src/flowmesh/resources/tasks.py index e5efeb065..a9a8ba511 100644 --- a/sdk/src/flowmesh/resources/tasks.py +++ b/sdk/src/flowmesh/resources/tasks.py @@ -84,17 +84,17 @@ def list( ``limit`` keeps only the newest matching tasks.""" params = _list_params( - task_id, - workflow_id, - status, - category, - task_type, - assigned_worker, - graph_node_name, - completed, - failed, - limit, - query_params, + task_id=task_id, + workflow_id=workflow_id, + status=status, + category=category, + task_type=task_type, + assigned_worker=assigned_worker, + graph_node_name=graph_node_name, + completed=completed, + failed=failed, + limit=limit, + query_params=query_params, ) data = self._client._request("GET", "/tasks", params=params or None) return [TaskInfo.model_validate(t) for t in data] @@ -115,17 +115,17 @@ def list_statuses( ) -> builtins.list[TaskStatusView]: """List the status views of tasks; takes the filters of ``list``.""" params = _list_params( - task_id, - workflow_id, - status, - category, - task_type, - assigned_worker, - graph_node_name, - completed, - failed, - limit, - query_params, + task_id=task_id, + workflow_id=workflow_id, + status=status, + category=category, + task_type=task_type, + assigned_worker=assigned_worker, + graph_node_name=graph_node_name, + completed=completed, + failed=failed, + limit=limit, + query_params=query_params, ) params.append(("view", "status")) data = self._client._request("GET", "/tasks", params=params) @@ -236,17 +236,17 @@ async def list( ``limit`` keeps only the newest matching tasks.""" params = _list_params( - task_id, - workflow_id, - status, - category, - task_type, - assigned_worker, - graph_node_name, - completed, - failed, - limit, - query_params, + task_id=task_id, + workflow_id=workflow_id, + status=status, + category=category, + task_type=task_type, + assigned_worker=assigned_worker, + graph_node_name=graph_node_name, + completed=completed, + failed=failed, + limit=limit, + query_params=query_params, ) data = await self._client._request("GET", "/tasks", params=params or None) return [TaskInfo.model_validate(t) for t in data] @@ -267,17 +267,17 @@ async def list_statuses( ) -> builtins.list[TaskStatusView]: """List the status views of tasks; takes the filters of ``list``.""" params = _list_params( - task_id, - workflow_id, - status, - category, - task_type, - assigned_worker, - graph_node_name, - completed, - failed, - limit, - query_params, + task_id=task_id, + workflow_id=workflow_id, + status=status, + category=category, + task_type=task_type, + assigned_worker=assigned_worker, + graph_node_name=graph_node_name, + completed=completed, + failed=failed, + limit=limit, + query_params=query_params, ) params.append(("view", "status")) data = await self._client._request("GET", "/tasks", params=params) diff --git a/src/server/routers/v1/tasks.py b/src/server/routers/v1/tasks.py index 796796e91..f22c4193e 100644 --- a/src/server/routers/v1/tasks.py +++ b/src/server/routers/v1/tasks.py @@ -46,8 +46,8 @@ TaskView = Literal["full", "status"] _VIEW_QUERY = Query( "full", - description="`status` returns only status, placement, timing and errors " - "(no task spec or source) — use it for polling.", + description="`status` returns only status, placement, timing, errors and the " + "latest update (no task spec or source) — use it for polling.", ) @@ -62,8 +62,9 @@ def _pushed_down_queries( Returns (workflow_ids, statuses, remaining). ``workflow_id`` and ``status`` are matched by membership in ``TaskRuntime.list_tasks`` before any TaskInfo - is built, so a query that only filters on them never dumps a model. The - remaining params still go through ``filter_models_by_queries``. + is built, and the control params (``view``, ``limit``) are dropped, so a + query that only filters on them never dumps a model. The remaining params + still go through ``filter_models_by_queries``. """ workflow_ids = query_params.getlist("workflow_id") statuses = query_params.getlist("status") @@ -100,7 +101,7 @@ async def list_tasks( request: Request, view: TaskView = _VIEW_QUERY, limit: int | None = Query( - None, ge=1, le=10_000, description="Return only the newest this many tasks." + None, ge=1, le=10_000, description="Keep only the newest N matching tasks." ), principal: PrincipalContext = Depends(authenticate_connection), runtime: TaskRuntime = Depends(get_runtime), diff --git a/src/server/schemas/task.py b/src/server/schemas/task.py index d9d68f2c8..601e51ca6 100644 --- a/src/server/schemas/task.py +++ b/src/server/schemas/task.py @@ -6,7 +6,8 @@ class TaskStatusView(BaseModel): - """A task's status, placement, timing and errors, without its spec or source.""" + """A task's status, placement, timing, errors and latest update, without its + spec or source.""" task_id: str = Field(description="Task identifier.") workflow_id: str = Field(description="Workflow identifier.") diff --git a/src/server/task/models.py b/src/server/task/models.py index a4b6fbf13..3295cf86f 100644 --- a/src/server/task/models.py +++ b/src/server/task/models.py @@ -256,11 +256,7 @@ def _serialize(self, handler: SerializerFunctionWrapHandler) -> dict[str, Any]: return handler(redacted) -# A task's position in a listing: its submission time, then its id. -type TaskOrder = tuple[float, str] - - -def task_order(record: TaskRecord) -> TaskOrder: +def task_order(record: TaskRecord) -> tuple[float, str]: return record.submitted_ts, record.task_id diff --git a/tests/sdk/test_request_building.py b/tests/sdk/test_request_building.py index b877d32f0..ec56d1fb5 100644 --- a/tests/sdk/test_request_building.py +++ b/tests/sdk/test_request_building.py @@ -98,9 +98,13 @@ def mock_client() -> FlowMesh: ) -def _status_json(status: str) -> dict[str, Any]: +def _status_json( + status: str, latest_update: dict[str, Any] | None = None +) -> dict[str, Any]: view = SrvTaskStatusView.from_info(_TASK_RESPONSE) - return _json(view.model_copy(update={"status": status})) + return _json( + view.model_copy(update={"status": status, "latest_update": latest_update}) + ) def _json(server_obj: BaseModel) -> dict: # type: ignore[type-arg] @@ -316,6 +320,41 @@ async def test_async_wait_polls_status_then_fetches_full_task(self) -> None: assert views == ["status", None] assert task.task_id == "t-1" + @respx.mock + def test_wait_for_ssh_polls_status_view(self, mock_client: FlowMesh) -> None: + ssh = {"mode": "direct", "host": "h", "port": 22} + route = respx.get(route_url("get_task", task_id="t-1")).mock( + side_effect=[ + httpx.Response(200, json=_status_json("DISPATCHED")), + httpx.Response(200, json=_status_json("DISPATCHED", {"ssh": ssh})), + ] + ) + assert mock_client.tasks.wait_for_ssh("t-1", interval=0) == ssh + assert [c.request.url.params["view"] for c in route.calls] == ["status"] * 2 + + @pytest.mark.anyio + @respx.mock + async def test_async_status_calls_request_status_view(self) -> None: + ssh = {"mode": "direct", "host": "h", "port": 22} + task_route = respx.get(route_url("get_task", task_id="t-1")).respond( + json=_status_json("DISPATCHED", {"ssh": ssh}) + ) + list_route = respx.get(route_url("list_tasks")).respond( + json=[_status_json("DISPATCHED")] + ) + async with AsyncFlowMesh( + base_url=TEST_BASE_URL, api_key="flm-test-key" + ) as client: + view = await client.tasks.status("t-1") + info = await client.tasks.wait_for_ssh("t-1", interval=0) + views = await client.tasks.list_statuses(limit=1) + assert view.status == "DISPATCHED" + assert info == ssh + assert [v.task_id for v in views] == ["t-1"] + assert all(c.request.url.params["view"] == "status" for c in task_route.calls) + assert list_route.calls[0].request.url.params["view"] == "status" + assert list_route.calls[0].request.url.params["limit"] == "1" + class TestNodeRegister: @respx.mock diff --git a/tests/server/test_tasks_router.py b/tests/server/test_tasks_router.py index f2bcdbfd4..c36604772 100644 --- a/tests/server/test_tasks_router.py +++ b/tests/server/test_tasks_router.py @@ -305,9 +305,14 @@ async def ping() -> dict[str, bool]: assert list_resp.status_code == 200 -def test_status_view_fields_are_task_info_fields() -> None: - served = set(TaskInfo.model_fields) | set(TaskInfo.model_computed_fields) - assert set(TaskStatusView.model_fields) <= served +def test_status_view_fields_match_task_info() -> None: + for name, field in TaskStatusView.model_fields.items(): + if computed := TaskInfo.model_computed_fields.get(name): + served = (computed.return_type, computed.description) + else: + info = TaskInfo.model_fields[name] + served = (info.annotation, info.description) + assert (field.annotation, field.description) == served, name @pytest.mark.anyio @@ -383,7 +388,7 @@ async def test_list_tasks_rejects_out_of_range_limit( @pytest.mark.anyio -async def test_unknown_view_is_rejected(no_checkers: None) -> None: +async def test_list_tasks_unknown_view_is_rejected(no_checkers: None) -> None: runtime, _ = build_runtime() async with _client(runtime) as ac: resp = await ac.get(f"{PREFIX}/tasks", params={"view": "bogus"}) From e946a5a56a809a99f17fa6b55efc73e7159fcbbc Mon Sep 17 00:00:00 2001 From: Noppanat Wadlom Date: Sat, 17 Oct 2026 04:23:23 +0000 Subject: [PATCH 07/10] docs: describe log cursors as the opaque entries[].cursor strings Signed-off-by: Noppanat Wadlom --- docs/API.md | 4 ++-- docs/ARCHITECTURE.md | 4 ++-- docs/SDK.md | 4 ++-- 3 files changed, 6 insertions(+), 6 deletions(-) diff --git a/docs/API.md b/docs/API.md index 05caf8355..546e6fab6 100644 --- a/docs/API.md +++ b/docs/API.md @@ -121,5 +121,5 @@ Gated by `ENABLE_SERVER_SERVE_PROXY`. ## Cursor pagination -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. diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 5f74e7135..7a354539a 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -245,8 +245,8 @@ scripts/dev/ compile_protos, sync_requirements, check_env_examples worker's lifecycle, so it also applies to a worker that registers under the key later. - **Cursor pagination.** Log queries accept `limit` and `before` / - `after` cursors. The cursor is an opaque base64 of `(timestamp, id)`; - do not parse client-side. + `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 diff --git a/docs/SDK.md b/docs/SDK.md index 7535a77de..d67248d0d 100644 --- a/docs/SDK.md +++ b/docs/SDK.md @@ -70,8 +70,8 @@ bundle missing a requested section raises `FlowMeshError`. ## Cursor pagination Log queries take `limit` and `before` / `after` params. -Cursors are opaque base64 strings — pass them through; do not parse -them. +A cursor is an opaque string taken from `entries[].cursor` — pass it back +unchanged; do not parse it. ## Errors From 950e6d5fd0689ad15d4825e42820f657e952a41f Mon Sep 17 00:00:00 2001 From: Noppanat Wadlom Date: Sat, 17 Oct 2026 04:37:05 +0000 Subject: [PATCH 08/10] chore: bump mcp to 1.30.0 for CVE-2026-59951 Signed-off-by: Noppanat Wadlom --- pyproject.toml | 2 +- src/worker/requirements/requirements.txt | 2 +- uv.lock | 12 ++++++------ 3 files changed, 8 insertions(+), 8 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 02434d4cf..cda947972 100644 --- a/pyproject.toml +++ b/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", diff --git a/src/worker/requirements/requirements.txt b/src/worker/requirements/requirements.txt index 282844ed8..ba3a7f171 100644 --- a/src/worker/requirements/requirements.txt +++ b/src/worker/requirements/requirements.txt @@ -25,7 +25,7 @@ hydra-core==1.3.7 ipython==9.5.0 jinja2==3.1.6 matplotlib==3.10.6 -mcp==1.28.1 +mcp==1.30.0 nvidia-ml-py==13.580.82 openai-agents==0.7.0 openai==2.54.0 diff --git a/uv.lock b/uv.lock index f9d8be34a..8ca5c3a6d 100644 --- a/uv.lock +++ b/uv.lock @@ -2253,7 +2253,7 @@ ci = [ { name = "jinja2", specifier = ">=3.1.6" }, { name = "lumid-hooks", specifier = ">=0.2.0" }, { name = "matplotlib", specifier = ">=3.10.6" }, - { name = "mcp", specifier = ">=1.28.1" }, + { name = "mcp", specifier = ">=1.30.0" }, { name = "mypy", specifier = ">=1.19.1" }, { name = "nvidia-ml-py", specifier = ">=13.580.82" }, { name = "openai", specifier = ">=2.25.0,<3" }, @@ -2357,7 +2357,7 @@ runtime-agent = [ { name = "ipython", specifier = ">=9.5.0" }, { name = "jinja2", specifier = ">=3.1.6" }, { name = "matplotlib", specifier = ">=3.10.6" }, - { name = "mcp", specifier = ">=1.28.1" }, + { name = "mcp", specifier = ">=1.30.0" }, { name = "openai", specifier = ">=2.25.0,<3" }, { name = "openai-agents", specifier = ">=0.7.0,<0.8" }, { name = "opencv-python-headless", specifier = ">=4.13.0.92" }, @@ -2469,7 +2469,7 @@ runtime-worker-cpu = [ { name = "ipython", specifier = ">=9.5.0" }, { name = "jinja2", specifier = ">=3.1.6" }, { name = "matplotlib", specifier = ">=3.10.6" }, - { name = "mcp", specifier = ">=1.28.1" }, + { name = "mcp", specifier = ">=1.30.0" }, { name = "nvidia-ml-py", specifier = ">=13.580.82" }, { name = "openai", specifier = ">=2.25.0,<3" }, { name = "openai-agents", specifier = ">=0.7.0,<0.8" }, @@ -4058,7 +4058,7 @@ wheels = [ [[package]] name = "mcp" -version = "1.28.1" +version = "1.30.0" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "anyio" }, @@ -4076,9 +4076,9 @@ dependencies = [ { name = "typing-inspection" }, { name = "uvicorn", marker = "sys_platform != 'emscripten'" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/6e/77/9450b8f251a13affb6281997d0523c4615f8a8b35d0b21ff30db3a5aac9d/mcp-1.28.1.tar.gz", hash = "sha256:d51e36a5f5644faea4f85ea649bfffa6bc6c26770d42798ad6a3de3d2ba69683", size = 638501, upload-time = "2026-06-26T12:57:29.093Z" } +sdist = { url = "https://files.pythonhosted.org/packages/ba/93/0142dc84a666daf8ad51a34268f34c12fd6fda4f3810c4be2504eecc8212/mcp-1.30.0.tar.gz", hash = "sha256:445414625fce5c295faa505bb11bacece661ab6f4028d57c935db57820b7a3e4", size = 680511, upload-time = "2026-09-07T14:34:15.845Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/e2/5e/d118fce19f87a2e7d8101c35c8ae0ec289098a4df0ff244cec23e415aca0/mcp-1.28.1-py3-none-any.whl", hash = "sha256:2726bca5e7193f61c5dde8b12500a6de2d9acf6d1a1c0be9e8c2e706437991df", size = 222620, upload-time = "2026-06-26T12:57:27.218Z" }, + { url = "https://files.pythonhosted.org/packages/f5/f4/e58bc33317c92a0203664daaf00bf6f41166cc0149e5d6870a03f7cd004a/mcp-1.30.0-py3-none-any.whl", hash = "sha256:666edb5009503e1047c9d60346a756f94b261f05cc2625f23d41c728ffc484d0", size = 234581, upload-time = "2026-09-07T14:34:14.266Z" }, ] [[package]] From c428bb658d3ceb007b13a2c7d0632485dc4fdb62 Mon Sep 17 00:00:00 2001 From: Noppanat Wadlom Date: Sat, 17 Oct 2026 06:40:44 +0000 Subject: [PATCH 09/10] refactor: trim task view descriptions and alias TaskView with type Signed-off-by: Noppanat Wadlom --- docs/API.md | 4 ++-- sdk/src/flowmesh/models/tasks.py | 4 +--- src/server/routers/v1/tasks.py | 4 ++-- 3 files changed, 5 insertions(+), 7 deletions(-) diff --git a/docs/API.md b/docs/API.md index 546e6fab6..b1bff4ac9 100644 --- a/docs/API.md +++ b/docs/API.md @@ -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, oldest submission first. Filters: `workflow_id`, `status`, `task_type`, `assigned_worker`. `limit` keeps the newest matching tasks; `view=status` returns each task's status view. | -| GET | `/api/v1/tasks/{id}` | Task details. `view=status` returns only status, placement, timing, errors and the latest update, without the task spec or source; use it for polling. | +| 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. | diff --git a/sdk/src/flowmesh/models/tasks.py b/sdk/src/flowmesh/models/tasks.py index ef72871cf..bfa2114ff 100644 --- a/sdk/src/flowmesh/models/tasks.py +++ b/sdk/src/flowmesh/models/tasks.py @@ -72,9 +72,7 @@ class TaskInfo(BaseModel): class TaskStatusView(BaseModel): """A task's status, placement, timing, errors and latest update, without its - spec or source. - - A server without this view returns the full task, which validates here too.""" + spec or source.""" task_id: str workflow_id: str diff --git a/src/server/routers/v1/tasks.py b/src/server/routers/v1/tasks.py index f22c4193e..b530ea46f 100644 --- a/src/server/routers/v1/tasks.py +++ b/src/server/routers/v1/tasks.py @@ -43,11 +43,11 @@ # Query keys the list endpoint consumes itself; they are not model-field filters. _CONTROL_PARAMS = ("view", "limit") -TaskView = Literal["full", "status"] +type TaskView = Literal["full", "status"] _VIEW_QUERY = Query( "full", description="`status` returns only status, placement, timing, errors and the " - "latest update (no task spec or source) — use it for polling.", + "latest update (no task spec or source).", ) From b4418500bedd7f7c45390404f57685dc723beb46 Mon Sep 17 00:00:00 2001 From: Noppanat Wadlom Date: Sat, 17 Oct 2026 06:40:47 +0000 Subject: [PATCH 10/10] feat: add tasks.wait_status and wait on it from the CLI Signed-off-by: Noppanat Wadlom --- cli/src/flowmesh_cli/commands/ssh.py | 17 ++++++++------- cli/src/flowmesh_cli/core/task.py | 8 +++---- docs/SDK.md | 4 ++-- sdk/src/flowmesh/resources/tasks.py | 32 ++++++++++++++++++++++------ tests/sdk/test_request_building.py | 31 +++++++++++++++++++++++++++ 5 files changed, 72 insertions(+), 20 deletions(-) diff --git a/cli/src/flowmesh_cli/commands/ssh.py b/cli/src/flowmesh_cli/commands/ssh.py index 9c1033c45..adbe3a986 100644 --- a/cli/src/flowmesh_cli/commands/ssh.py +++ b/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.status(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: diff --git a/cli/src/flowmesh_cli/core/task.py b/cli/src/flowmesh_cli/core/task.py index b1504a451..652171762 100644 --- a/cli/src/flowmesh_cli/core/task.py +++ b/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) diff --git a/docs/SDK.md b/docs/SDK.md index d67248d0d..25e4ca616 100644 --- a/docs/SDK.md +++ b/docs/SDK.md @@ -49,8 +49,8 @@ async with AsyncFlowMesh(base_url="...", api_key="...") as client: `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(task_id)` polls the status view and returns the full - task once it finishes. + 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. diff --git a/sdk/src/flowmesh/resources/tasks.py b/sdk/src/flowmesh/resources/tasks.py index a9a8ba511..d6b78de07 100644 --- a/sdk/src/flowmesh/resources/tasks.py +++ b/sdk/src/flowmesh/resources/tasks.py @@ -180,10 +180,20 @@ def wait( task_id: str, interval: float = 2.0, ) -> TaskInfo: - """Poll a task until it reaches a terminal state.""" + """Poll a task until it reaches a terminal state, then retrieve it.""" + self.wait_status(task_id, interval=interval) + return self.retrieve(task_id) + + def wait_status( + self, + task_id: str, + interval: float = 2.0, + ) -> TaskStatusView: + """Poll a task until it reaches a terminal state and return its status view.""" while True: - if self.status(task_id).status in TERMINAL_TASK_STATUSES: - return self.retrieve(task_id) + view = self.status(task_id) + if view.status in TERMINAL_TASK_STATUSES: + return view time.sleep(interval) def ssh_info(self, task_id: str) -> dict[str, Any] | None: @@ -334,10 +344,20 @@ async def wait( task_id: str, interval: float = 2.0, ) -> TaskInfo: - """Poll a task until it reaches a terminal state.""" + """Poll a task until it reaches a terminal state, then retrieve it.""" + await self.wait_status(task_id, interval=interval) + return await self.retrieve(task_id) + + async def wait_status( + self, + task_id: str, + interval: float = 2.0, + ) -> TaskStatusView: + """Poll a task until it reaches a terminal state and return its status view.""" while True: - if (await self.status(task_id)).status in TERMINAL_TASK_STATUSES: - return await self.retrieve(task_id) + view = await self.status(task_id) + if view.status in TERMINAL_TASK_STATUSES: + return view await asyncio.sleep(interval) async def ssh_info(self, task_id: str) -> dict[str, Any] | None: diff --git a/tests/sdk/test_request_building.py b/tests/sdk/test_request_building.py index ec56d1fb5..beb3638e2 100644 --- a/tests/sdk/test_request_building.py +++ b/tests/sdk/test_request_building.py @@ -11,6 +11,7 @@ import pytest import respx from flowmesh import AsyncFlowMesh, FlowMesh +from flowmesh.models import TaskStatus from pydantic import BaseModel from server.registries.node import Node as SrvNode @@ -320,6 +321,36 @@ async def test_async_wait_polls_status_then_fetches_full_task(self) -> None: assert views == ["status", None] assert task.task_id == "t-1" + @respx.mock + def test_wait_status_returns_terminal_view(self, mock_client: FlowMesh) -> None: + route = respx.get(route_url("get_task", task_id="t-1")).mock( + side_effect=[ + httpx.Response(200, json=_status_json("DISPATCHED")), + httpx.Response(200, json=_status_json("DONE")), + ] + ) + view = mock_client.tasks.wait_status("t-1", interval=0) + views = [call.request.url.params.get("view") for call in route.calls] + assert views == ["status", "status"] + assert view.status == TaskStatus.DONE + + @pytest.mark.anyio + @respx.mock + async def test_async_wait_status_returns_terminal_view(self) -> None: + route = respx.get(route_url("get_task", task_id="t-1")).mock( + side_effect=[ + httpx.Response(200, json=_status_json("DISPATCHED")), + httpx.Response(200, json=_status_json("FAILED")), + ] + ) + async with AsyncFlowMesh( + base_url=TEST_BASE_URL, api_key="flm-test-key" + ) as client: + view = await client.tasks.wait_status("t-1", interval=0) + views = [call.request.url.params.get("view") for call in route.calls] + assert views == ["status", "status"] + assert view.status == TaskStatus.FAILED + @respx.mock def test_wait_for_ssh_polls_status_view(self, mock_client: FlowMesh) -> None: ssh = {"mode": "direct", "host": "h", "port": 22}