diff --git a/docs/architecture.md b/docs/architecture.md index 3c6a7436..925e9ba8 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -75,6 +75,7 @@ turnstone/ client.py TurnstoneClient library + TurnResult for MQ-based access console/ collector.py ClusterCollector — aggregates state from all nodes via Redis + HTTP + scheduler.py TaskScheduler — background cron/at scheduler, dispatches via MQ server.py Cluster dashboard HTTP server + SSE + CLI entry point static/ Cluster dashboard web UI (page-specific HTML, CSS, JS) channels/ diff --git a/docs/console.md b/docs/console.md index 4ccf745d..76a83c64 100644 --- a/docs/console.md +++ b/docs/console.md @@ -423,6 +423,150 @@ to create the initial admin user and receive a JWT in one step. See --- +## Scheduled Tasks + +The console includes a background **TaskScheduler** daemon that creates workstreams on a timed basis via the MQ broker. It supports cron-based recurring schedules and one-shot `at` schedules. + +### Architecture + +The scheduler runs as a daemon thread inside the console process. Every `check_interval` seconds (default 15) it: + +1. Acquires a distributed lock via Redis `SET NX EX` (prevents duplicate dispatch in multi-console deployments) +2. Queries the storage backend for tasks whose `next_run <= now` and `enabled = true` +3. Dispatches each due task as one or more `CreateWorkstreamMessage` via MQ +4. Updates `last_run` and computes the next `next_run` (or disables one-shot `at` tasks) +5. Releases the lock via Lua script (safe conditional delete) + +Run history is automatically pruned (runs older than 90 days) approximately once per hour. + +### Schedule Types + +| Type | Field | Behavior | +|------|-------|----------| +| `cron` | `cron_expr` | Recurring schedule using standard 5-field cron syntax. Requires `croniter`. | +| `at` | `at_time` | One-shot: fires once at the given ISO 8601 timestamp (must include timezone), then auto-disables. | + +### Target Modes + +| Mode | Behavior | +|------|----------| +| `auto` | Picks the reachable node with the most available capacity | +| `pool` | Pushes to the shared inbound queue (any bridge picks it up) | +| `all` | Fan-out to all reachable nodes (capped at `max_fan_out`, default 20) | +| `` | Targets a specific node by ID | + +### Configuration + +| Parameter | Default | Description | +|-----------|---------|-------------| +| `check_interval` | `15.0` | Seconds between scheduler ticks | +| `lock_ttl` | `60` | Distributed lock TTL in seconds | +| `max_fan_out` | `20` | Maximum nodes for `all` target mode | + +Dependency: `croniter` (installed with turnstone). + +### Schedule API + +All schedule endpoints require `approve` scope. Maximum 200 schedules. + +#### `GET /v1/api/admin/schedules` + +List all scheduled tasks. + +```json +{ + "schedules": [ + { + "task_id": "a1b2c3d4", + "name": "nightly-checks", + "description": "Run nightly health checks", + "schedule_type": "cron", + "cron_expr": "0 2 * * *", + "at_time": "", + "target_mode": "auto", + "model": "", + "initial_message": "Run the nightly health check suite.", + "auto_approve": false, + "auto_approve_tools": [], + "enabled": true, + "created_by": "u_admin", + "last_run": "2026-03-05T02:00:00Z", + "next_run": "2026-03-06T02:00:00Z", + "created": "2026-03-01T12:00:00Z", + "updated": "2026-03-05T02:00:01Z" + } + ] +} +``` + +#### `POST /v1/api/admin/schedules` + +Create a scheduled task. + +Request: + +```json +{ + "name": "nightly-checks", + "description": "Run nightly health checks", + "schedule_type": "cron", + "cron_expr": "0 2 * * *", + "target_mode": "auto", + "initial_message": "Run the nightly health check suite.", + "auto_approve": false, + "enabled": true +} +``` + +Required fields: `name`, `schedule_type`, `initial_message`. For `cron` schedules provide `cron_expr`; for `at` schedules provide `at_time` (ISO 8601 with timezone, must be in the future). + +Response: `ScheduleInfo` (same shape as list items above). Returns `400` for invalid cron syntax, naive timestamps, or past `at_time`. Returns `409` if the 200-schedule cap is reached. + +#### `GET /v1/api/admin/schedules/{task_id}` + +Get a single scheduled task. Returns `ScheduleInfo` or `404`. + +#### `PUT /v1/api/admin/schedules/{task_id}` + +Partial update — only include fields to change. If `schedule_type`, `cron_expr`, or `at_time` change, `next_run` is recomputed automatically. + +```json +{ + "enabled": false +} +``` + +Response: updated `ScheduleInfo`. Returns `400` for validation errors, `404` if not found. + +#### `DELETE /v1/api/admin/schedules/{task_id}` + +Delete a scheduled task and all its run history. Returns `{"status": "ok"}` or `404`. + +#### `GET /v1/api/admin/schedules/{task_id}/runs?limit=50` + +List execution history for a task (most recent first). `limit` defaults to 50, max 200. + +```json +{ + "runs": [ + { + "run_id": "r_abc123", + "task_id": "a1b2c3d4", + "node_id": "db-west-04", + "ws_id": "ws_xyz", + "correlation_id": "corr_789", + "started": "2026-03-05T02:00:00Z", + "status": "dispatched", + "error": "" + } + ] +} +``` + +Status is `dispatched` on success or `failed` with an `error` message (e.g. no reachable nodes). Failed runs do not advance `next_run`. + +--- + ## CLI Commands The `/cluster` command in the turnstone CLI queries the console's HTTP API. Requires `--console-url` or `[console] url` in config.toml. diff --git a/docs/sdk.md b/docs/sdk.md index 5086f99c..b0a24876 100644 --- a/docs/sdk.md +++ b/docs/sdk.md @@ -96,6 +96,12 @@ Both `TurnstoneConsole` (sync) and `AsyncTurnstoneConsole` (async) expose: | | `workstreams(*, state, node, search, sort, page, per_page)` | `ClusterWorkstreamsResponse` | | | `node_detail(node_id)` | `NodeDetailResponse` | | | `create_workstream(*, node_id, name, model, initial_message)` | `ConsoleCreateWsResponse` | +| **Schedules** | `list_schedules()` | `ListSchedulesResponse` | +| | `create_schedule(*, name, schedule_type, initial_message, ...)` | `ScheduleInfo` | +| | `get_schedule(task_id)` | `ScheduleInfo` | +| | `update_schedule(task_id, *, name=..., enabled=..., ...)` | `ScheduleInfo` | +| | `delete_schedule(task_id)` | `StatusResponse` | +| | `list_schedule_runs(task_id, *, limit=50)` | `ListScheduleRunsResponse` | | **Streaming** | `stream_cluster_events()` | `Iterator[ClusterEvent]` | | **Auth** | `login(username=..., password=...)` / `login(token="ts_xxx")` | `AuthLoginResponse` | | | `logout()` | `StatusResponse` | diff --git a/sdk/typescript/src/console.ts b/sdk/typescript/src/console.ts index a9cc6f51..bd455241 100644 --- a/sdk/typescript/src/console.ts +++ b/sdk/typescript/src/console.ts @@ -10,9 +10,14 @@ import type { ConsoleCreateWsRequest, ConsoleCreateWsResponse, ConsoleHealthResponse, + CreateScheduleRequest, + ListScheduleRunsResponse, + ListSchedulesResponse, NodeDetailResponse, NodesOptions, + ScheduleInfo, StatusResponse, + UpdateScheduleRequest, WorkstreamsOptions, } from "./types.js"; @@ -111,4 +116,40 @@ export class TurnstoneConsole extends BaseClient { async health(): Promise { return this.request("GET", "/health"); } + + // -- Schedules ------------------------------------------------------------ + + async listSchedules(): Promise { + return this.request("GET", "/v1/api/admin/schedules"); + } + + async createSchedule(opts: CreateScheduleRequest): Promise { + return this.request("POST", "/v1/api/admin/schedules", { json: opts }); + } + + async getSchedule(taskId: string): Promise { + return this.request("GET", `/v1/api/admin/schedules/${taskId}`); + } + + async updateSchedule( + taskId: string, + opts: UpdateScheduleRequest, + ): Promise { + return this.request("PUT", `/v1/api/admin/schedules/${taskId}`, { + json: opts, + }); + } + + async deleteSchedule(taskId: string): Promise { + return this.request("DELETE", `/v1/api/admin/schedules/${taskId}`); + } + + async listScheduleRuns( + taskId: string, + opts?: { limit?: number }, + ): Promise { + return this.request("GET", `/v1/api/admin/schedules/${taskId}/runs`, { + params: { limit: opts?.limit ?? 50 }, + }); + } } diff --git a/sdk/typescript/src/index.ts b/sdk/typescript/src/index.ts index 31f89ca1..6c50a90a 100644 --- a/sdk/typescript/src/index.ts +++ b/sdk/typescript/src/index.ts @@ -103,6 +103,12 @@ export type { ConsoleCreateWsRequest, ConsoleCreateWsResponse, ConsoleHealthResponse, + CreateScheduleRequest, + UpdateScheduleRequest, + ScheduleInfo, + ScheduleRunInfo, + ListSchedulesResponse, + ListScheduleRunsResponse, TurnResult, SendAndWaitOptions, NodesOptions, diff --git a/sdk/typescript/src/types.ts b/sdk/typescript/src/types.ts index 4fcbd0eb..527464da 100644 --- a/sdk/typescript/src/types.ts +++ b/sdk/typescript/src/types.ts @@ -269,6 +269,77 @@ export interface ConsoleHealthResponse { versions: string[]; } +// --------------------------------------------------------------------------- +// Console API — Schedules +// --------------------------------------------------------------------------- + +export interface CreateScheduleRequest { + name: string; + schedule_type: string; + initial_message: string; + description?: string; + cron_expr?: string; + at_time?: string; + target_mode?: string; + model?: string; + auto_approve?: boolean; + auto_approve_tools?: string[]; + enabled?: boolean; +} + +export interface UpdateScheduleRequest { + name?: string; + description?: string; + schedule_type?: string; + cron_expr?: string; + at_time?: string; + target_mode?: string; + model?: string; + initial_message?: string; + auto_approve?: boolean; + auto_approve_tools?: string[]; + enabled?: boolean; +} + +export interface ScheduleInfo { + task_id: string; + name: string; + description: string; + schedule_type: string; + cron_expr: string; + at_time: string; + target_mode: string; + model: string; + initial_message: string; + auto_approve: boolean; + auto_approve_tools: string[]; + enabled: boolean; + created_by: string; + last_run: string | null; + next_run: string | null; + created: string; + updated: string; +} + +export interface ListSchedulesResponse { + schedules: ScheduleInfo[]; +} + +export interface ScheduleRunInfo { + run_id: string; + task_id: string; + node_id: string; + ws_id: string; + correlation_id: string; + started: string; + status: string; + error: string; +} + +export interface ListScheduleRunsResponse { + runs: ScheduleRunInfo[]; +} + // --------------------------------------------------------------------------- // SDK-specific types // --------------------------------------------------------------------------- diff --git a/tests/test_sdk_console.py b/tests/test_sdk_console.py index 35fce8b7..a5870ee9 100644 --- a/tests/test_sdk_console.py +++ b/tests/test_sdk_console.py @@ -2,6 +2,8 @@ from __future__ import annotations +import json + import httpx import pytest @@ -240,3 +242,140 @@ async def test_query_params_passed(): assert "state=running" in captured_url[0] assert "page=2" in captured_url[0] assert "per_page=25" in captured_url[0] + + +# --------------------------------------------------------------------------- +# Schedules +# --------------------------------------------------------------------------- + +_SCHEDULE_FIXTURE = { + "task_id": "t1", + "name": "nightly", + "description": "", + "schedule_type": "cron", + "cron_expr": "0 2 * * *", + "at_time": "", + "target_mode": "auto", + "model": "", + "initial_message": "Run nightly checks", + "auto_approve": False, + "auto_approve_tools": [], + "enabled": True, + "created_by": "u1", + "last_run": None, + "next_run": "2026-03-06T02:00:00Z", + "created": "2026-03-05T12:00:00Z", + "updated": "2026-03-05T12:00:00Z", +} + + +@pytest.mark.anyio +async def test_list_schedules(): + transport = _mock_transport( + {"GET /v1/api/admin/schedules": _json_response({"schedules": [_SCHEDULE_FIXTURE]})} + ) + async with httpx.AsyncClient(transport=transport, base_url="http://test") as hc: + client = AsyncTurnstoneConsole(httpx_client=hc) + resp = await client.list_schedules() + assert len(resp.schedules) == 1 + assert resp.schedules[0].task_id == "t1" + assert resp.schedules[0].name == "nightly" + + +@pytest.mark.anyio +async def test_create_schedule(): + captured_body: list[dict] = [] + + def handler(request: httpx.Request) -> httpx.Response: + captured_body.append(json.loads(request.content)) + return _json_response(_SCHEDULE_FIXTURE) + + transport = httpx.MockTransport(handler) + async with httpx.AsyncClient(transport=transport, base_url="http://test") as hc: + client = AsyncTurnstoneConsole(httpx_client=hc) + resp = await client.create_schedule( + name="nightly", + schedule_type="cron", + initial_message="Run nightly checks", + cron_expr="0 2 * * *", + ) + assert resp.task_id == "t1" + body = captured_body[0] + assert body["name"] == "nightly" + assert body["schedule_type"] == "cron" + assert body["cron_expr"] == "0 2 * * *" + assert body["initial_message"] == "Run nightly checks" + # Optional fields with defaults should not appear when not set + assert "description" not in body + assert "model" not in body + + +@pytest.mark.anyio +async def test_get_schedule(): + transport = _mock_transport( + {"GET /v1/api/admin/schedules/t1": _json_response(_SCHEDULE_FIXTURE)} + ) + async with httpx.AsyncClient(transport=transport, base_url="http://test") as hc: + client = AsyncTurnstoneConsole(httpx_client=hc) + resp = await client.get_schedule("t1") + assert resp.task_id == "t1" + assert resp.schedule_type == "cron" + + +@pytest.mark.anyio +async def test_update_schedule_partial(): + """Only explicitly-passed fields should appear in the request body.""" + captured_body: list[dict] = [] + + def handler(request: httpx.Request) -> httpx.Response: + captured_body.append(json.loads(request.content)) + return _json_response({**_SCHEDULE_FIXTURE, "enabled": False}) + + transport = httpx.MockTransport(handler) + async with httpx.AsyncClient(transport=transport, base_url="http://test") as hc: + client = AsyncTurnstoneConsole(httpx_client=hc) + resp = await client.update_schedule("t1", enabled=False) + assert resp.enabled is False + body = captured_body[0] + assert body == {"enabled": False} + + +@pytest.mark.anyio +async def test_delete_schedule(): + transport = _mock_transport( + {"DELETE /v1/api/admin/schedules/t1": _json_response({"status": "ok"})} + ) + async with httpx.AsyncClient(transport=transport, base_url="http://test") as hc: + client = AsyncTurnstoneConsole(httpx_client=hc) + resp = await client.delete_schedule("t1") + assert resp.status == "ok" + + +@pytest.mark.anyio +async def test_list_schedule_runs(): + transport = _mock_transport( + { + "GET /v1/api/admin/schedules/t1/runs": _json_response( + { + "runs": [ + { + "run_id": "r1", + "task_id": "t1", + "node_id": "n1", + "ws_id": "ws1", + "correlation_id": "c1", + "started": "2026-03-05T02:00:00Z", + "status": "dispatched", + "error": "", + } + ] + } + ) + } + ) + async with httpx.AsyncClient(transport=transport, base_url="http://test") as hc: + client = AsyncTurnstoneConsole(httpx_client=hc) + resp = await client.list_schedule_runs("t1", limit=10) + assert len(resp.runs) == 1 + assert resp.runs[0].run_id == "r1" + assert resp.runs[0].status == "dispatched" diff --git a/turnstone/sdk/console.py b/turnstone/sdk/console.py index 031588c0..224fac00 100644 --- a/turnstone/sdk/console.py +++ b/turnstone/sdk/console.py @@ -25,12 +25,17 @@ from turnstone.api.schemas import ( AuthLoginResponse, AuthSetupResponse, AuthStatusResponse, + ListScheduleRunsResponse, + ListSchedulesResponse, + ScheduleInfo, StatusResponse, ) from turnstone.sdk._base import _BaseClient from turnstone.sdk._sync import _SyncRunner from turnstone.sdk.events import ClusterEvent +_UNSET: Any = object() + if TYPE_CHECKING: from collections.abc import AsyncIterator, Iterator @@ -179,6 +184,109 @@ class AsyncTurnstoneConsole(_BaseClient): async def health(self) -> ConsoleHealthResponse: return await self._request("GET", "/health", response_model=ConsoleHealthResponse) + # -- schedules ----------------------------------------------------------- + + async def list_schedules(self) -> ListSchedulesResponse: + return await self._request( + "GET", "/v1/api/admin/schedules", response_model=ListSchedulesResponse + ) + + async def create_schedule( + self, + *, + name: str, + schedule_type: str, + initial_message: str, + description: str = "", + cron_expr: str = "", + at_time: str = "", + target_mode: str = "auto", + model: str = "", + auto_approve: bool = False, + auto_approve_tools: list[str] | None = None, + enabled: bool = True, + ) -> ScheduleInfo: + body: dict[str, Any] = { + "name": name, + "schedule_type": schedule_type, + "initial_message": initial_message, + "target_mode": target_mode, + "auto_approve": auto_approve, + "enabled": enabled, + } + if description: + body["description"] = description + if cron_expr: + body["cron_expr"] = cron_expr + if at_time: + body["at_time"] = at_time + if model: + body["model"] = model + if auto_approve_tools: + body["auto_approve_tools"] = auto_approve_tools + return await self._request( + "POST", "/v1/api/admin/schedules", json_body=body, response_model=ScheduleInfo + ) + + async def get_schedule(self, task_id: str) -> ScheduleInfo: + return await self._request( + "GET", f"/v1/api/admin/schedules/{task_id}", response_model=ScheduleInfo + ) + + async def update_schedule( + self, + task_id: str, + *, + name: Any = _UNSET, + description: Any = _UNSET, + schedule_type: Any = _UNSET, + cron_expr: Any = _UNSET, + at_time: Any = _UNSET, + target_mode: Any = _UNSET, + model: Any = _UNSET, + initial_message: Any = _UNSET, + auto_approve: Any = _UNSET, + auto_approve_tools: Any = _UNSET, + enabled: Any = _UNSET, + ) -> ScheduleInfo: + body: dict[str, Any] = {} + for key, val in [ + ("name", name), + ("description", description), + ("schedule_type", schedule_type), + ("cron_expr", cron_expr), + ("at_time", at_time), + ("target_mode", target_mode), + ("model", model), + ("initial_message", initial_message), + ("auto_approve", auto_approve), + ("auto_approve_tools", auto_approve_tools), + ("enabled", enabled), + ]: + if val is not _UNSET: + body[key] = val + return await self._request( + "PUT", + f"/v1/api/admin/schedules/{task_id}", + json_body=body, + response_model=ScheduleInfo, + ) + + async def delete_schedule(self, task_id: str) -> StatusResponse: + return await self._request( + "DELETE", f"/v1/api/admin/schedules/{task_id}", response_model=StatusResponse + ) + + async def list_schedule_runs( + self, task_id: str, *, limit: int = 50 + ) -> ListScheduleRunsResponse: + return await self._request( + "GET", + f"/v1/api/admin/schedules/{task_id}/runs", + params={"limit": limit}, + response_model=ListScheduleRunsResponse, + ) + class TurnstoneConsole: """Synchronous client for the turnstone console API. @@ -274,6 +382,84 @@ class TurnstoneConsole: def health(self) -> ConsoleHealthResponse: return self._runner.run(self._async.health()) + # -- schedules ----------------------------------------------------------- + + def list_schedules(self) -> ListSchedulesResponse: + return self._runner.run(self._async.list_schedules()) + + def create_schedule( + self, + *, + name: str, + schedule_type: str, + initial_message: str, + description: str = "", + cron_expr: str = "", + at_time: str = "", + target_mode: str = "auto", + model: str = "", + auto_approve: bool = False, + auto_approve_tools: list[str] | None = None, + enabled: bool = True, + ) -> ScheduleInfo: + return self._runner.run( + self._async.create_schedule( + name=name, + schedule_type=schedule_type, + initial_message=initial_message, + description=description, + cron_expr=cron_expr, + at_time=at_time, + target_mode=target_mode, + model=model, + auto_approve=auto_approve, + auto_approve_tools=auto_approve_tools, + enabled=enabled, + ) + ) + + def get_schedule(self, task_id: str) -> ScheduleInfo: + return self._runner.run(self._async.get_schedule(task_id)) + + def update_schedule( + self, + task_id: str, + *, + name: Any = _UNSET, + description: Any = _UNSET, + schedule_type: Any = _UNSET, + cron_expr: Any = _UNSET, + at_time: Any = _UNSET, + target_mode: Any = _UNSET, + model: Any = _UNSET, + initial_message: Any = _UNSET, + auto_approve: Any = _UNSET, + auto_approve_tools: Any = _UNSET, + enabled: Any = _UNSET, + ) -> ScheduleInfo: + return self._runner.run( + self._async.update_schedule( + task_id, + name=name, + description=description, + schedule_type=schedule_type, + cron_expr=cron_expr, + at_time=at_time, + target_mode=target_mode, + model=model, + initial_message=initial_message, + auto_approve=auto_approve, + auto_approve_tools=auto_approve_tools, + enabled=enabled, + ) + ) + + def delete_schedule(self, task_id: str) -> StatusResponse: + return self._runner.run(self._async.delete_schedule(task_id)) + + def list_schedule_runs(self, task_id: str, *, limit: int = 50) -> ListScheduleRunsResponse: + return self._runner.run(self._async.list_schedule_runs(task_id, limit=limit)) + # -- lifecycle ----------------------------------------------------------- def close(self) -> None: