diff --git a/pyproject.toml b/pyproject.toml index 4ebd21df..5fc684ab 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -43,10 +43,10 @@ Repository = "https://github.com/turnstonelabs/turnstone" Issues = "https://github.com/turnstonelabs/turnstone/issues" [project.optional-dependencies] -test = ["pytest>=9.0", "pytest-cov>=6.0"] +test = ["pytest>=9.0", "pytest-cov>=6.0", "croniter>=3.0"] dev = ["ruff>=0.9", "mypy>=1.14", "types-redis>=4.6"] mq = ["redis>=7.2"] -console = ["redis>=7.2"] +console = ["redis>=7.2", "croniter>=3.0"] sim = ["redis>=7.2"] anthropic = ["anthropic>=0.39"] postgres = ["psycopg[binary]>=3.2"] @@ -150,6 +150,10 @@ ignore_missing_imports = true module = ["discord", "discord.*"] ignore_missing_imports = true +[[tool.mypy.overrides]] +module = ["croniter", "croniter.*"] +ignore_missing_imports = true + [[tool.mypy.overrides]] module = ["turnstone.channels.discord.*"] disallow_subclassing_any = false diff --git a/tests/test_schedule_api.py b/tests/test_schedule_api.py new file mode 100644 index 00000000..27f93807 --- /dev/null +++ b/tests/test_schedule_api.py @@ -0,0 +1,264 @@ +"""Tests for scheduled task admin API endpoints.""" + +from __future__ import annotations + +import pytest +from starlette.applications import Starlette +from starlette.routing import Mount, Route +from starlette.testclient import TestClient + +from turnstone.console.server import ( + admin_create_schedule, + admin_delete_schedule, + admin_get_schedule, + admin_list_schedule_runs, + admin_list_schedules, + admin_update_schedule, +) +from turnstone.core.storage._sqlite import SQLiteBackend + + +@pytest.fixture +def storage(tmp_path): + """Fresh SQLite backend for each test.""" + return SQLiteBackend(str(tmp_path / "test.db")) + + +@pytest.fixture +def client(storage): + """TestClient with storage and auth bypassed.""" + app = Starlette( + routes=[ + Mount( + "/v1", + routes=[ + Route("/api/admin/schedules", admin_list_schedules), + Route("/api/admin/schedules", admin_create_schedule, methods=["POST"]), + Route("/api/admin/schedules/{task_id}", admin_get_schedule), + Route( + "/api/admin/schedules/{task_id}", + admin_update_schedule, + methods=["PUT"], + ), + Route( + "/api/admin/schedules/{task_id}", + admin_delete_schedule, + methods=["DELETE"], + ), + Route( + "/api/admin/schedules/{task_id}/runs", + admin_list_schedule_runs, + ), + ], + ), + ], + ) + app.state.auth_storage = storage + return TestClient(app) + + +def _cron_payload(**overrides): + """Build default cron schedule creation payload.""" + defaults = { + "name": "Daily report", + "description": "Generate the summary", + "schedule_type": "cron", + "cron_expr": "0 9 * * *", + "target_mode": "auto", + "model": "gpt-5", + "initial_message": "Generate the daily report", + } + defaults.update(overrides) + return defaults + + +def _at_payload(**overrides): + """Build default at-time schedule creation payload.""" + defaults = { + "name": "One-shot task", + "description": "Run once", + "schedule_type": "at", + "at_time": "2099-01-01T00:00:00+00:00", + "target_mode": "auto", + "model": "gpt-5", + "initial_message": "Do the thing", + } + defaults.update(overrides) + return defaults + + +class TestScheduleAPI: + """Tests for the 6 admin schedule endpoints.""" + + def test_list_empty(self, client): + resp = client.get("/v1/api/admin/schedules") + assert resp.status_code == 200 + data = resp.json() + assert data["schedules"] == [] + + def test_create_cron(self, client): + resp = client.post("/v1/api/admin/schedules", json=_cron_payload()) + assert resp.status_code == 200 + task = resp.json() + assert task["name"] == "Daily report" + assert task["schedule_type"] == "cron" + assert task["cron_expr"] == "0 9 * * *" + assert task["enabled"] is True + assert "task_id" in task + assert "created" in task + assert "next_run" in task + assert task["next_run"] != "" + + def test_create_at(self, client): + resp = client.post("/v1/api/admin/schedules", json=_at_payload()) + assert resp.status_code == 200 + task = resp.json() + assert task["schedule_type"] == "at" + assert task["at_time"] == "2099-01-01T00:00:00+00:00" + assert task["next_run"] == "2099-01-01T00:00:00+00:00" + + def test_create_missing_name(self, client): + payload = _cron_payload() + del payload["name"] + resp = client.post("/v1/api/admin/schedules", json=payload) + assert resp.status_code == 400 + assert "name" in resp.json()["error"].lower() + + def test_create_invalid_cron(self, client): + resp = client.post( + "/v1/api/admin/schedules", + json=_cron_payload(cron_expr="not a cron"), + ) + assert resp.status_code == 400 + assert "cron" in resp.json()["error"].lower() + + def test_create_naive_at_time(self, client): + """Naive timestamps (no timezone) should be rejected.""" + resp = client.post( + "/v1/api/admin/schedules", + json=_at_payload(at_time="2099-01-01T00:00:00"), + ) + assert resp.status_code == 400 + assert "timezone" in resp.json()["error"].lower() + + def test_create_past_at_time(self, client): + resp = client.post( + "/v1/api/admin/schedules", + json=_at_payload(at_time="2000-01-01T00:00:00+00:00"), + ) + assert resp.status_code == 400 + assert "future" in resp.json()["error"].lower() + + def test_get_schedule(self, client): + create_resp = client.post("/v1/api/admin/schedules", json=_cron_payload()) + task_id = create_resp.json()["task_id"] + + resp = client.get(f"/v1/api/admin/schedules/{task_id}") + assert resp.status_code == 200 + assert resp.json()["task_id"] == task_id + assert resp.json()["name"] == "Daily report" + + def test_get_nonexistent(self, client): + resp = client.get("/v1/api/admin/schedules/nonexistent_id") + assert resp.status_code == 404 + + def test_update_schedule(self, client): + create_resp = client.post("/v1/api/admin/schedules", json=_cron_payload()) + task_id = create_resp.json()["task_id"] + + resp = client.put( + f"/v1/api/admin/schedules/{task_id}", + json={"name": "Weekly report"}, + ) + assert resp.status_code == 200 + assert resp.json()["name"] == "Weekly report" + + # Verify via GET + get_resp = client.get(f"/v1/api/admin/schedules/{task_id}") + assert get_resp.json()["name"] == "Weekly report" + + def test_update_nonexistent(self, client): + resp = client.put( + "/v1/api/admin/schedules/nonexistent_id", + json={"name": "Nope"}, + ) + assert resp.status_code == 404 + + def test_delete_schedule(self, client): + create_resp = client.post("/v1/api/admin/schedules", json=_cron_payload()) + task_id = create_resp.json()["task_id"] + + resp = client.delete(f"/v1/api/admin/schedules/{task_id}") + assert resp.status_code == 200 + assert resp.json()["status"] == "ok" + + # Verify gone + get_resp = client.get(f"/v1/api/admin/schedules/{task_id}") + assert get_resp.status_code == 404 + + def test_delete_nonexistent(self, client): + resp = client.delete("/v1/api/admin/schedules/nonexistent_id") + assert resp.status_code == 404 + + def test_list_runs_empty(self, client): + create_resp = client.post("/v1/api/admin/schedules", json=_cron_payload()) + task_id = create_resp.json()["task_id"] + + resp = client.get(f"/v1/api/admin/schedules/{task_id}/runs") + assert resp.status_code == 200 + assert resp.json()["runs"] == [] + + def test_list_runs_nonexistent(self, client): + resp = client.get("/v1/api/admin/schedules/nonexistent_id/runs") + assert resp.status_code == 404 + + def test_create_specific_node_target(self, client): + payload = _cron_payload(target_mode="node-custom-001") + resp = client.post("/v1/api/admin/schedules", json=payload) + assert resp.status_code == 200 + data = resp.json() + assert data["target_mode"] == "node-custom-001" + + def test_list_runs_with_data(self, client, storage): + create_resp = client.post("/v1/api/admin/schedules", json=_cron_payload()) + task_id = create_resp.json()["task_id"] + + # Record runs directly in storage + storage.record_task_run( + run_id="run_001", + task_id=task_id, + node_id="node-1", + ws_id="ws_abc", + correlation_id="corr_001", + started="2025-06-01T09:00:00", + status="dispatched", + error="", + ) + storage.record_task_run( + run_id="run_002", + task_id=task_id, + node_id="node-2", + ws_id="", + correlation_id="corr_002", + started="2025-06-01T09:01:00", + status="failed", + error="No reachable nodes", + ) + + resp = client.get(f"/v1/api/admin/schedules/{task_id}/runs") + assert resp.status_code == 200 + runs = resp.json()["runs"] + assert len(runs) == 2 + # Most recent first + assert runs[0]["run_id"] == "run_002" + assert runs[0]["status"] == "failed" + assert runs[1]["run_id"] == "run_001" + + def test_list_runs_invalid_limit(self, client): + create_resp = client.post("/v1/api/admin/schedules", json=_cron_payload()) + task_id = create_resp.json()["task_id"] + + # Invalid limit should not crash — falls back to 50 + resp = client.get(f"/v1/api/admin/schedules/{task_id}/runs?limit=abc") + assert resp.status_code == 200 + assert resp.json()["runs"] == [] diff --git a/tests/test_scheduled_tasks_storage.py b/tests/test_scheduled_tasks_storage.py new file mode 100644 index 00000000..ae526718 --- /dev/null +++ b/tests/test_scheduled_tasks_storage.py @@ -0,0 +1,259 @@ +"""Tests for scheduled_tasks and scheduled_task_runs storage CRUD.""" + +from __future__ import annotations + +import time + +import pytest + +from turnstone.core.storage._sqlite import SQLiteBackend + + +@pytest.fixture +def db(tmp_path): + """Fresh SQLite backend for each test.""" + backend = SQLiteBackend(str(tmp_path / "test.db")) + return backend + + +def _make_task_kwargs(**overrides): + """Build default kwargs for create_scheduled_task.""" + defaults = { + "task_id": "task_001", + "name": "Daily report", + "description": "Generate the daily summary", + "schedule_type": "cron", + "cron_expr": "0 9 * * *", + "at_time": "", + "target_mode": "auto", + "model": "gpt-5", + "initial_message": "Generate the daily report", + "auto_approve": False, + "auto_approve_tools": [], + "created_by": "u_admin", + "next_run": "2099-01-01T09:00:00", + } + defaults.update(overrides) + return defaults + + +class TestScheduledTaskCRUD: + """Tests for scheduled_tasks table operations.""" + + def test_create_and_get(self, db): + db.create_scheduled_task(**_make_task_kwargs()) + result = db.get_scheduled_task("task_001") + assert result is not None + assert result["task_id"] == "task_001" + assert result["name"] == "Daily report" + assert result["description"] == "Generate the daily summary" + assert result["schedule_type"] == "cron" + assert result["cron_expr"] == "0 9 * * *" + assert result["at_time"] == "" + assert result["target_mode"] == "auto" + assert result["model"] == "gpt-5" + assert result["initial_message"] == "Generate the daily report" + assert result["auto_approve"] == 0 + assert result["auto_approve_tools"] == "" + assert result["enabled"] == 1 + assert result["created_by"] == "u_admin" + assert result["next_run"] == "2099-01-01T09:00:00" + assert "created" in result + assert "updated" in result + + def test_get_nonexistent(self, db): + assert db.get_scheduled_task("no_such_task") is None + + def test_create_duplicate_noop(self, db): + db.create_scheduled_task(**_make_task_kwargs(name="First")) + db.create_scheduled_task(**_make_task_kwargs(name="Second")) + result = db.get_scheduled_task("task_001") + assert result is not None + assert result["name"] == "First" # first write wins + + def test_list_tasks(self, db): + db.create_scheduled_task(**_make_task_kwargs(task_id="task_a", name="Alpha")) + # Ensure different created timestamps (resolution is 1 second) + time.sleep(1.1) + db.create_scheduled_task(**_make_task_kwargs(task_id="task_b", name="Beta")) + tasks = db.list_scheduled_tasks() + assert len(tasks) == 2 + # Ordered by created DESC — most recent first + assert tasks[0]["task_id"] == "task_b" + assert tasks[1]["task_id"] == "task_a" + + def test_update_task(self, db): + db.create_scheduled_task(**_make_task_kwargs()) + original = db.get_scheduled_task("task_001") + assert original is not None + original_updated = original["updated"] + + time.sleep(0.05) + result = db.update_scheduled_task("task_001", name="Weekly report") + assert result is True + + updated = db.get_scheduled_task("task_001") + assert updated is not None + assert updated["name"] == "Weekly report" + assert updated["updated"] >= original_updated + + def test_update_enable_disable(self, db): + db.create_scheduled_task(**_make_task_kwargs()) + task = db.get_scheduled_task("task_001") + assert task is not None + assert task["enabled"] == 1 + + db.update_scheduled_task("task_001", enabled=False) + task = db.get_scheduled_task("task_001") + assert task is not None + assert task["enabled"] == 0 + + db.update_scheduled_task("task_001", enabled=True) + task = db.get_scheduled_task("task_001") + assert task is not None + assert task["enabled"] == 1 + + def test_delete_task(self, db): + db.create_scheduled_task(**_make_task_kwargs()) + assert db.delete_scheduled_task("task_001") is True + assert db.get_scheduled_task("task_001") is None + # Deleting again returns False + assert db.delete_scheduled_task("task_001") is False + + def test_delete_cascades_runs(self, db): + db.create_scheduled_task(**_make_task_kwargs()) + db.record_task_run( + run_id="run_001", + task_id="task_001", + node_id="node_1", + ws_id="ws_abc", + correlation_id="corr_001", + started="2025-01-01T09:00:00", + status="dispatched", + error="", + ) + assert len(db.list_task_runs("task_001")) == 1 + + db.delete_scheduled_task("task_001") + assert db.list_task_runs("task_001") == [] + + def test_list_due_tasks(self, db): + db.create_scheduled_task( + **_make_task_kwargs(task_id="past", next_run="2020-01-01T00:00:00") + ) + db.create_scheduled_task( + **_make_task_kwargs(task_id="future", next_run="2099-12-31T23:59:59") + ) + now = "2025-06-01T12:00:00" + due = db.list_due_tasks(now) + assert len(due) == 1 + assert due[0]["task_id"] == "past" + + def test_list_due_tasks_skips_disabled(self, db): + db.create_scheduled_task( + **_make_task_kwargs(task_id="disabled_task", next_run="2020-01-01T00:00:00") + ) + db.update_scheduled_task("disabled_task", enabled=False) + due = db.list_due_tasks("2025-06-01T12:00:00") + assert len(due) == 0 + + def test_list_due_tasks_empty_next_run(self, db): + db.create_scheduled_task(**_make_task_kwargs(task_id="empty_next", next_run="")) + due = db.list_due_tasks("2099-12-31T23:59:59") + assert len(due) == 0 + + def test_at_task_fields(self, db): + db.create_scheduled_task( + **_make_task_kwargs( + task_id="at_task", + schedule_type="at", + cron_expr="", + at_time="2099-06-15T14:00:00", + next_run="2099-06-15T14:00:00", + ) + ) + result = db.get_scheduled_task("at_task") + assert result is not None + assert result["schedule_type"] == "at" + assert result["at_time"] == "2099-06-15T14:00:00" + + +class TestScheduledTaskRuns: + """Tests for scheduled_task_runs table operations.""" + + def test_record_and_list(self, db): + db.create_scheduled_task(**_make_task_kwargs()) + db.record_task_run( + run_id="run_a", + task_id="task_001", + node_id="node_1", + ws_id="ws_1", + correlation_id="corr_a", + started="2025-01-01T09:00:00", + status="dispatched", + error="", + ) + db.record_task_run( + run_id="run_b", + task_id="task_001", + node_id="node_2", + ws_id="ws_2", + correlation_id="corr_b", + started="2025-01-02T09:00:00", + status="dispatched", + error="", + ) + runs = db.list_task_runs("task_001") + assert len(runs) == 2 + # Ordered by started DESC — most recent first + assert runs[0]["run_id"] == "run_b" + assert runs[1]["run_id"] == "run_a" + + def test_list_runs_respects_limit(self, db): + db.create_scheduled_task(**_make_task_kwargs()) + for i in range(3): + db.record_task_run( + run_id=f"run_{i}", + task_id="task_001", + node_id="node_1", + ws_id="", + correlation_id=f"corr_{i}", + started=f"2025-01-0{i + 1}T09:00:00", + status="dispatched", + error="", + ) + runs = db.list_task_runs("task_001", limit=2) + assert len(runs) == 2 + + def test_list_runs_empty(self, db): + assert db.list_task_runs("no_such_task") == [] + + def test_prune_task_runs(self, db): + db.create_scheduled_task(**_make_task_kwargs()) + # Old run (should be pruned) + db.record_task_run( + run_id="old_run", + task_id="task_001", + node_id="node_1", + ws_id="", + correlation_id="c_old", + started="2020-01-01T00:00:00", + status="dispatched", + error="", + ) + # Recent run (should survive) + db.record_task_run( + run_id="new_run", + task_id="task_001", + node_id="node_1", + ws_id="", + correlation_id="c_new", + started="2099-01-01T00:00:00", + status="dispatched", + error="", + ) + pruned = db.prune_task_runs(retention_days=90) + assert pruned == 1 + runs = db.list_task_runs("task_001") + assert len(runs) == 1 + assert runs[0]["run_id"] == "new_run" diff --git a/tests/test_scheduler.py b/tests/test_scheduler.py new file mode 100644 index 00000000..455d1b45 --- /dev/null +++ b/tests/test_scheduler.py @@ -0,0 +1,272 @@ +"""Tests for turnstone.console.scheduler — TaskScheduler tick and dispatch.""" + +from __future__ import annotations + +from unittest.mock import MagicMock + +import pytest + +from turnstone.console.scheduler import TaskScheduler + + +@pytest.fixture +def mocks(): + """Broker, collector, and storage mocks for scheduler tests.""" + broker = MagicMock() + broker._redis = MagicMock() + collector = MagicMock() + storage = MagicMock() + return broker, collector, storage + + +def _make_task(**overrides): + """Build a minimal task dict matching storage row format.""" + defaults = { + "task_id": "task_001", + "name": "Test task", + "description": "", + "schedule_type": "cron", + "cron_expr": "0 9 * * *", + "at_time": "", + "target_mode": "auto", + "model": "gpt-5", + "initial_message": "Run the tests", + "auto_approve": 0, + "auto_approve_tools": "", + "enabled": 1, + "created_by": "u_admin", + "next_run": "2020-01-01T09:00:00", + "last_run": "", + "created": "2020-01-01T00:00:00", + "updated": "2020-01-01T00:00:00", + } + defaults.update(overrides) + return defaults + + +def _make_node(node_id="node-001", reachable=True, ws_total=2, max_ws=10): + """Build a minimal node dict matching collector output.""" + return { + "node_id": node_id, + "reachable": reachable, + "ws_total": ws_total, + "max_ws": max_ws, + } + + +class TestSchedulerTick: + """Tests for _tick() lock acquisition and dispatch logic.""" + + def test_tick_acquires_lock(self, mocks): + broker, collector, storage = mocks + broker._redis.set.return_value = True + storage.list_due_tasks.return_value = [] + + scheduler = TaskScheduler(broker, collector, storage) + scheduler._tick() + + broker._redis.set.assert_called_once() + storage.list_due_tasks.assert_called_once() + # Lock released via Lua eval (conditional delete) + broker._redis.eval.assert_called_once() + + def test_tick_skips_when_locked(self, mocks): + broker, collector, storage = mocks + broker._redis.set.return_value = None # lock held by another console + + scheduler = TaskScheduler(broker, collector, storage) + scheduler._tick() + + storage.list_due_tasks.assert_not_called() + + def test_dispatch_auto_mode(self, mocks): + broker, collector, storage = mocks + broker._redis.set.return_value = True + + task = _make_task(target_mode="auto") + storage.list_due_tasks.return_value = [task] + collector.get_nodes.return_value = ([_make_node()], 1) + + scheduler = TaskScheduler(broker, collector, storage) + scheduler._tick() + + broker.push_inbound.assert_called_once() + _, kwargs = broker.push_inbound.call_args + assert ( + kwargs.get("node_id") == "node-001" + or broker.push_inbound.call_args[1].get("node_id") == "node-001" + ) + storage.record_task_run.assert_called_once() + run_kwargs = storage.record_task_run.call_args[1] + assert run_kwargs["node_id"] == "node-001" + assert run_kwargs["status"] == "dispatched" + + def test_dispatch_pool_mode(self, mocks): + broker, collector, storage = mocks + broker._redis.set.return_value = True + + task = _make_task(target_mode="pool") + storage.list_due_tasks.return_value = [task] + + scheduler = TaskScheduler(broker, collector, storage) + scheduler._tick() + + broker.push_inbound.assert_called_once() + # Pool dispatch calls push_inbound without node_id kwarg + args, kwargs = broker.push_inbound.call_args + assert kwargs.get("node_id") is None or "node_id" not in kwargs + storage.record_task_run.assert_called_once() + run_kwargs = storage.record_task_run.call_args[1] + assert run_kwargs["node_id"] == "pool" + + def test_dispatch_all_mode(self, mocks): + broker, collector, storage = mocks + broker._redis.set.return_value = True + + task = _make_task(target_mode="all") + storage.list_due_tasks.return_value = [task] + collector.get_nodes.return_value = ( + [_make_node("node-001"), _make_node("node-002")], + 2, + ) + + scheduler = TaskScheduler(broker, collector, storage) + scheduler._tick() + + assert broker.push_inbound.call_count == 2 + assert storage.record_task_run.call_count == 2 + + def test_dispatch_specific_node(self, mocks): + broker, collector, storage = mocks + broker._redis.set.return_value = True + + task = _make_task(target_mode="node-001") + storage.list_due_tasks.return_value = [task] + + scheduler = TaskScheduler(broker, collector, storage) + scheduler._tick() + + broker.push_inbound.assert_called_once() + _, kwargs = broker.push_inbound.call_args + assert kwargs["node_id"] == "node-001" + + def test_at_task_disables_after_dispatch(self, mocks): + broker, collector, storage = mocks + broker._redis.set.return_value = True + + task = _make_task(schedule_type="at", cron_expr="", at_time="2099-01-01T00:00:00") + storage.list_due_tasks.return_value = [task] + collector.get_nodes.return_value = ([_make_node()], 1) + + scheduler = TaskScheduler(broker, collector, storage) + scheduler._tick() + + # At-task should be disabled after dispatch + update_calls = storage.update_scheduled_task.call_args_list + assert len(update_calls) == 1 + args, kwargs = update_calls[0] + assert args[0] == "task_001" + assert kwargs["enabled"] is False + assert kwargs["next_run"] == "" + + def test_cron_task_updates_next_run(self, mocks): + broker, collector, storage = mocks + broker._redis.set.return_value = True + + task = _make_task(schedule_type="cron", cron_expr="0 9 * * *") + storage.list_due_tasks.return_value = [task] + collector.get_nodes.return_value = ([_make_node()], 1) + + scheduler = TaskScheduler(broker, collector, storage) + scheduler._tick() + + update_calls = storage.update_scheduled_task.call_args_list + assert len(update_calls) == 1 + _, kwargs = update_calls[0] + assert kwargs["next_run"] != "" + assert "enabled" not in kwargs # cron tasks stay enabled + + def test_no_reachable_nodes_records_failure(self, mocks): + broker, collector, storage = mocks + broker._redis.set.return_value = True + + task = _make_task(target_mode="auto") + storage.list_due_tasks.return_value = [task] + # No reachable nodes + collector.get_nodes.return_value = ( + [_make_node("node-001", reachable=False)], + 1, + ) + + scheduler = TaskScheduler(broker, collector, storage) + scheduler._tick() + + broker.push_inbound.assert_not_called() + storage.record_task_run.assert_called_once() + run_kwargs = storage.record_task_run.call_args[1] + assert run_kwargs["status"] == "failed" + assert run_kwargs["error"] != "" + + def test_failure_does_not_advance_schedule(self, mocks): + """When dispatch fails, last_run/next_run should not be updated.""" + broker, collector, storage = mocks + broker._redis.set.return_value = True + + task = _make_task(target_mode="auto") + storage.list_due_tasks.return_value = [task] + collector.get_nodes.return_value = ([], 0) # no nodes at all + + scheduler = TaskScheduler(broker, collector, storage) + scheduler._tick() + + # update_scheduled_task should NOT be called (no last_run/next_run advance) + storage.update_scheduled_task.assert_not_called() + + def test_fan_out_capped(self, mocks): + """Fan-out 'all' mode should respect max_fan_out limit.""" + broker, collector, storage = mocks + broker._redis.set.return_value = True + + task = _make_task(target_mode="all") + storage.list_due_tasks.return_value = [task] + # 10 reachable nodes but max_fan_out=3 + nodes = [_make_node(f"node-{i:03d}") for i in range(10)] + collector.get_nodes.return_value = (nodes, 10) + + scheduler = TaskScheduler(broker, collector, storage, max_fan_out=3) + scheduler._tick() + + assert broker.push_inbound.call_count == 3 + assert storage.record_task_run.call_count == 3 + + def test_specific_node_target(self, mocks): + """Non-enum target_mode is treated as a specific node_id.""" + broker, collector, storage = mocks + broker._redis.set.return_value = True + + task = _make_task(target_mode="node-custom-123") + storage.list_due_tasks.return_value = [task] + + scheduler = TaskScheduler(broker, collector, storage) + scheduler._tick() + + broker.push_inbound.assert_called_once() + call_kwargs = broker.push_inbound.call_args + assert call_kwargs[1]["node_id"] == "node-custom-123" + + def test_user_id_in_dispatched_message(self, mocks): + """Dispatched message should include created_by as user_id.""" + import json + + broker, collector, storage = mocks + broker._redis.set.return_value = True + + task = _make_task(target_mode="pool", created_by="u_scheduler_admin") + storage.list_due_tasks.return_value = [task] + + scheduler = TaskScheduler(broker, collector, storage) + scheduler._tick() + + msg_json = broker.push_inbound.call_args[0][0] + msg_data = json.loads(msg_json) + assert msg_data["user_id"] == "u_scheduler_admin" diff --git a/turnstone/api/console_spec.py b/turnstone/api/console_spec.py index c01017bd..7b62d03b 100644 --- a/turnstone/api/console_spec.py +++ b/turnstone/api/console_spec.py @@ -23,13 +23,18 @@ from turnstone.api.schemas import ( AuthSetupRequest, AuthSetupResponse, AuthStatusResponse, + CreateScheduleRequest, CreateTokenRequest, CreateTokenResponse, CreateUserRequest, ErrorResponse, + ListScheduleRunsResponse, + ListSchedulesResponse, ListTokensResponse, ListUsersResponse, + ScheduleInfo, StatusResponse, + UpdateScheduleRequest, UserInfo, ) @@ -182,6 +187,61 @@ CONSOLE_ENDPOINTS: list[EndpointSpec] = [ error_codes=[404], tags=["Admin"], ), + # --- Schedules --- + EndpointSpec( + "/v1/api/admin/schedules", + "GET", + "List all scheduled tasks", + response_model=ListSchedulesResponse, + tags=["Schedules"], + ), + EndpointSpec( + "/v1/api/admin/schedules", + "POST", + "Create a scheduled task", + request_model=CreateScheduleRequest, + response_model=ScheduleInfo, + error_codes=[400], + tags=["Schedules"], + ), + EndpointSpec( + "/v1/api/admin/schedules/{task_id}", + "GET", + "Get a scheduled task", + response_model=ScheduleInfo, + error_codes=[404], + tags=["Schedules"], + ), + EndpointSpec( + "/v1/api/admin/schedules/{task_id}", + "PUT", + "Update a scheduled task", + request_model=UpdateScheduleRequest, + response_model=ScheduleInfo, + error_codes=[400, 404], + tags=["Schedules"], + ), + EndpointSpec( + "/v1/api/admin/schedules/{task_id}", + "DELETE", + "Delete a scheduled task", + response_model=StatusResponse, + error_codes=[404], + tags=["Schedules"], + ), + EndpointSpec( + "/v1/api/admin/schedules/{task_id}/runs", + "GET", + "List run history for a scheduled task", + response_model=ListScheduleRunsResponse, + query_params=[ + QueryParam( + "limit", "Max results (default 50, max 200)", schema_type="integer", default=50 + ), + ], + error_codes=[404], + tags=["Schedules"], + ), # --- Observability --- EndpointSpec( "/health", @@ -213,6 +273,11 @@ _ALL_MODELS: list[type[BaseModel]] = [ ConsoleCreateWsRequest, ConsoleCreateWsResponse, ConsoleHealthResponse, + CreateScheduleRequest, + UpdateScheduleRequest, + ScheduleInfo, + ListSchedulesResponse, + ListScheduleRunsResponse, ] diff --git a/turnstone/api/schemas.py b/turnstone/api/schemas.py index c2655e4f..eedbf260 100644 --- a/turnstone/api/schemas.py +++ b/turnstone/api/schemas.py @@ -155,3 +155,87 @@ class AuthStatusResponse(BaseModel): auth_enabled: bool has_users: bool setup_required: bool + + +# --------------------------------------------------------------------------- +# Schedules +# --------------------------------------------------------------------------- + + +class CreateScheduleRequest(BaseModel): + """POST /v1/api/admin/schedules request body.""" + + name: str = Field(description="Human-readable schedule name") + description: str = Field(default="", description="Optional description") + schedule_type: str = Field(description="'cron' or 'at'") + cron_expr: str = Field(default="", description="Cron expression (when schedule_type='cron')") + at_time: str = Field(default="", description="ISO8601 timestamp (when schedule_type='at')") + target_mode: str = Field(default="auto", description="auto, pool, all, or specific node_id") + model: str = Field(default="", description="Model alias for the workstream") + initial_message: str = Field(description="Message sent to the new workstream") + auto_approve: bool = Field(default=False) + auto_approve_tools: list[str] = Field(default_factory=list) + enabled: bool = Field(default=True) + + +class UpdateScheduleRequest(BaseModel): + """PUT /v1/api/admin/schedules/{task_id} request body (partial update).""" + + name: str | None = None + description: str | None = None + schedule_type: str | None = None + cron_expr: str | None = None + at_time: str | None = None + target_mode: str | None = None + model: str | None = None + initial_message: str | None = None + auto_approve: bool | None = None + auto_approve_tools: list[str] | None = None + enabled: bool | None = None + + +class ScheduleInfo(BaseModel): + """Scheduled task details.""" + + task_id: str + name: str + description: str = "" + schedule_type: str + cron_expr: str = "" + at_time: str = "" + target_mode: str = "auto" + model: str = "" + initial_message: str + auto_approve: bool = False + auto_approve_tools: list[str] = Field(default_factory=list) + enabled: bool = True + created_by: str = "" + last_run: str | None = None + next_run: str | None = None + created: str = "" + updated: str = "" + + +class ListSchedulesResponse(BaseModel): + """GET /v1/api/admin/schedules response.""" + + schedules: list[ScheduleInfo] + + +class ScheduleRunInfo(BaseModel): + """Single execution record for a scheduled task.""" + + run_id: str + task_id: str + node_id: str = "" + ws_id: str = "" + correlation_id: str = "" + started: str + status: str = "dispatched" + error: str = "" + + +class ListScheduleRunsResponse(BaseModel): + """GET /v1/api/admin/schedules/{task_id}/runs response.""" + + runs: list[ScheduleRunInfo] diff --git a/turnstone/console/scheduler.py b/turnstone/console/scheduler.py new file mode 100644 index 00000000..7ce49ce7 --- /dev/null +++ b/turnstone/console/scheduler.py @@ -0,0 +1,259 @@ +"""Background task scheduler for timed workstream dispatch. + +Runs as a daemon thread inside the console process. Checks for due tasks +every ``check_interval`` seconds and dispatches them as +``CreateWorkstreamMessage`` via the MQ broker. + +Uses Redis ``SET NX EX`` for distributed locking in multi-console deployments. +""" + +from __future__ import annotations + +import threading +import uuid +from datetime import UTC, datetime +from typing import TYPE_CHECKING, Any + +import structlog + +if TYPE_CHECKING: + from turnstone.console.collector import ClusterCollector + from turnstone.core.storage._protocol import StorageBackend + from turnstone.mq.broker import RedisBroker + +log = structlog.get_logger(__name__) + + +def _pick_best_node(collector: ClusterCollector) -> str: + """Select the reachable node with the most available capacity.""" + nodes, _ = collector.get_nodes(sort_by="activity", limit=1000, offset=0) + best_id = "" + best_headroom = -1 + for n in nodes: + if not n.get("reachable", False): + continue + headroom = n.get("max_ws", 10) - n.get("ws_total", 0) + if headroom > best_headroom: + best_headroom = headroom + best_id = n["node_id"] + return best_id + + +class TaskScheduler: + """Background scheduler for dispatching timed workstreams.""" + + def __init__( + self, + broker: RedisBroker, + collector: ClusterCollector, + storage: StorageBackend, + prefix: str = "turnstone", + check_interval: float = 15.0, + lock_ttl: int = 60, + max_fan_out: int = 20, + ) -> None: + self._broker = broker + self._collector = collector + self._storage = storage + self._prefix = prefix + self._check_interval = check_interval + self._lock_ttl = lock_ttl + self._max_fan_out = max_fan_out + self._stop_event = threading.Event() + self._thread: threading.Thread | None = None + self._tick_count = 0 + self._prune_every = 240 # ~1 hour at 15s intervals + + def start(self) -> None: + """Start the scheduler daemon thread.""" + self._stop_event.clear() + self._thread = threading.Thread(target=self._loop, daemon=True, name="scheduler") + self._thread.start() + log.info("scheduler.started", check_interval=self._check_interval) + + def stop(self) -> None: + """Stop the scheduler and wait for the thread to finish.""" + self._stop_event.set() + if self._thread is not None: + self._thread.join(timeout=5) + log.info("scheduler.stopped") + + def _loop(self) -> None: + """Main scheduler loop — tick then sleep.""" + while not self._stop_event.is_set(): + try: + self._tick() + except Exception: + log.exception("scheduler.tick_error") + self._stop_event.wait(self._check_interval) + + # Lua script for safe lock release — only delete if we still own the lock + _UNLOCK_SCRIPT = "if redis.call('get',KEYS[1])==ARGV[1] then return redis.call('del',KEYS[1]) else return 0 end" + + def _tick(self) -> None: + """Single scheduler iteration: acquire lock, query due tasks, dispatch.""" + now = datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%S") + + # Distributed lock with unique owner — prevents releasing another instance's lock + lock_key = f"{self._prefix}:scheduler:lock" + lock_value = uuid.uuid4().hex + acquired = self._broker._redis.set(lock_key, lock_value, nx=True, ex=self._lock_ttl) + if not acquired: + return + + try: + due_tasks = self._storage.list_due_tasks(now) + for task in due_tasks: + self._dispatch_task(task, now) + + # Periodic run history pruning (~once per hour) + self._tick_count += 1 + if self._tick_count % self._prune_every == 0: + pruned = self._storage.prune_task_runs(retention_days=90) + if pruned: + log.info("scheduler.pruned_runs", count=pruned) + finally: + # Only release our own lock (safe even if TTL expired and another took it) + self._broker._redis.eval( # type: ignore[no-untyped-call] + self._UNLOCK_SCRIPT, 1, lock_key, lock_value + ) + + def _dispatch_task(self, task: dict[str, Any], now: str) -> None: + """Dispatch a single task as one or more CreateWorkstreamMessages.""" + target_mode = task["target_mode"] + task_id = task["task_id"] + dispatched = False + + if target_mode == "all": + nodes, _ = self._collector.get_nodes(sort_by="activity", limit=1000, offset=0) + fan_count = 0 + for n in nodes: + if n.get("reachable", False): + if fan_count >= self._max_fan_out: + log.warning( + "scheduler.fan_out_capped", + task_id=task_id, + max_fan_out=self._max_fan_out, + ) + break + self._dispatch_to_node(task, n["node_id"], now) + fan_count += 1 + dispatched = True + if not dispatched: + self._record_failure(task, now, "No reachable nodes for fan-out") + elif target_mode == "pool": + self._dispatch_to_pool(task, now) + dispatched = True + elif target_mode == "auto": + node_id = _pick_best_node(self._collector) + if node_id: + self._dispatch_to_node(task, node_id, now) + dispatched = True + else: + self._record_failure(task, now, "No reachable nodes") + else: + # Specific node_id + self._dispatch_to_node(task, target_mode, now) + dispatched = True + + if not dispatched: + return # Don't advance schedule on failure + + # Update last_run and compute next_run + next_run = self._compute_next_run(task) + if task["schedule_type"] == "at": + self._storage.update_scheduled_task(task_id, last_run=now, next_run="", enabled=False) + else: + self._storage.update_scheduled_task(task_id, last_run=now, next_run=next_run) + + log_kw: dict[str, Any] = { + "task_id": task_id, + "target_mode": target_mode, + "schedule_type": task["schedule_type"], + "created_by": task.get("created_by", ""), + } + if task.get("auto_approve", 0): + log_kw["auto_approve"] = True + log_kw["auto_approve_tools"] = task.get("auto_approve_tools", "") + log.warning("scheduler.task_dispatched_auto_approve", **log_kw) + else: + log.info("scheduler.task_dispatched", **log_kw) + + @staticmethod + def _parse_tools(task: dict[str, Any]) -> list[str]: + raw = task.get("auto_approve_tools", "") + return [t.strip() for t in raw.split(",") if t.strip()] + + def _dispatch_to_node(self, task: dict[str, Any], node_id: str, now: str) -> None: + """Send a CreateWorkstreamMessage to a specific node.""" + from turnstone.mq.protocol import CreateWorkstreamMessage + + msg = CreateWorkstreamMessage( + name=task["name"], + model=task.get("model", ""), + target_node=node_id, + initial_message=task["initial_message"], + auto_approve=bool(task.get("auto_approve", 0)), + auto_approve_tools=self._parse_tools(task), + user_id=task.get("created_by", ""), + ) + self._broker.push_inbound(msg.to_json(), node_id=node_id) + + self._storage.record_task_run( + run_id=uuid.uuid4().hex, + task_id=task["task_id"], + node_id=node_id, + ws_id="", + correlation_id=msg.correlation_id, + started=now, + status="dispatched", + error="", + ) + + def _dispatch_to_pool(self, task: dict[str, Any], now: str) -> None: + """Send a CreateWorkstreamMessage to the shared pool queue.""" + from turnstone.mq.protocol import CreateWorkstreamMessage + + msg = CreateWorkstreamMessage( + name=task["name"], + model=task.get("model", ""), + initial_message=task["initial_message"], + auto_approve=bool(task.get("auto_approve", 0)), + auto_approve_tools=self._parse_tools(task), + user_id=task.get("created_by", ""), + ) + self._broker.push_inbound(msg.to_json()) + + self._storage.record_task_run( + run_id=uuid.uuid4().hex, + task_id=task["task_id"], + node_id="pool", + ws_id="", + correlation_id=msg.correlation_id, + started=now, + status="dispatched", + error="", + ) + + def _record_failure(self, task: dict[str, Any], now: str, error: str) -> None: + """Record a failed dispatch attempt.""" + self._storage.record_task_run( + run_id=uuid.uuid4().hex, + task_id=task["task_id"], + node_id="", + ws_id="", + correlation_id="", + started=now, + status="failed", + error=error, + ) + log.warning("scheduler.dispatch_failed", task_id=task["task_id"], error=error) + + @staticmethod + def _compute_next_run(task: dict[str, Any]) -> str: + """Compute the next run time. Returns empty string for one-shot tasks.""" + from turnstone.console.server import _compute_next_run + + return _compute_next_run( + task["schedule_type"], task.get("cron_expr", ""), task.get("at_time", "") + ) diff --git a/turnstone/console/server.py b/turnstone/console/server.py index e5c83820..3929ce17 100644 --- a/turnstone/console/server.py +++ b/turnstone/console/server.py @@ -620,8 +620,14 @@ async def _lifespan(app: Starlette) -> AsyncGenerator[None, None]: timeout=httpx.Timeout(connect=5, read=30, write=5, pool=5), headers=headers, ) + # Start scheduler if configured + scheduler = getattr(app.state, "scheduler", None) + if scheduler is not None: + scheduler.start() yield # Shutdown + if scheduler is not None: + scheduler.stop() await app.state.proxy_sse_client.aclose() await app.state.proxy_client.aclose() app.state.collector.stop() @@ -878,6 +884,277 @@ async def admin_delete_channel(request: Request) -> JSONResponse: return JSONResponse({"error": "Channel link not found"}, status_code=404) +# --------------------------------------------------------------------------- +# Admin API endpoints — scheduled tasks +# --------------------------------------------------------------------------- + + +def _normalize_task_dict(task: dict[str, Any]) -> dict[str, Any]: + """Convert DB row ints/csv to JSON-friendly bools/lists.""" + tools_str = task.get("auto_approve_tools", "") + task["auto_approve_tools"] = [s.strip() for s in tools_str.split(",") if s.strip()] + task["auto_approve"] = bool(task.get("auto_approve", 0)) + task["enabled"] = bool(task.get("enabled", 1)) + return task + + +def _compute_next_run(schedule_type: str, cron_expr: str, at_time: str) -> str: + """Compute the next run time for a schedule. Empty string if invalid.""" + if schedule_type == "at": + return at_time + if schedule_type == "cron" and cron_expr: + from datetime import UTC, datetime + + from croniter import croniter + + cron = croniter(cron_expr, datetime.now(UTC)) + next_dt = cron.get_next(datetime) + return str(next_dt.strftime("%Y-%m-%dT%H:%M:%S")) + return "" + + +def _validate_schedule_fields(schedule_type: str, cron_expr: str, at_time: str) -> str | None: + """Validate schedule type/expression. Returns error string or None.""" + if schedule_type not in ("cron", "at"): + return "schedule_type must be 'cron' or 'at'" + if schedule_type == "cron": + if not cron_expr: + return "cron_expr is required when schedule_type is 'cron'" + from croniter import croniter + + if not croniter.is_valid(cron_expr): + return f"Invalid cron expression: {cron_expr}" + if schedule_type == "at": + if not at_time: + return "at_time is required when schedule_type is 'at'" + from datetime import UTC, datetime + + try: + dt = datetime.fromisoformat(at_time) + if dt.tzinfo is None: + return ( + "at_time must include a timezone offset (e.g. 2024-01-01T12:00:00Z or +00:00)" + ) + if dt <= datetime.now(UTC): + return "at_time must be in the future" + except ValueError: + return "at_time must be a valid ISO8601 timestamp with timezone" + return None + + +async def admin_list_schedules(request: Request) -> JSONResponse: + """GET /v1/api/admin/schedules — list all scheduled tasks.""" + from turnstone.core.web_helpers import require_storage_or_503 + + storage, err = require_storage_or_503(request) + if err: + return err + tasks = storage.list_scheduled_tasks() + for t in tasks: + _normalize_task_dict(t) + return JSONResponse({"schedules": tasks}) + + +async def admin_create_schedule(request: Request) -> JSONResponse: + """POST /v1/api/admin/schedules — create a scheduled task.""" + import uuid + + from turnstone.core.web_helpers import read_json_or_400, require_storage_or_503 + + storage, err = require_storage_or_503(request) + if err: + return err + + body = await read_json_or_400(request) + if isinstance(body, JSONResponse): + return body + + name = str(body.get("name", "")).strip()[:256] + description = str(body.get("description", "")).strip()[:1024] + schedule_type = str(body.get("schedule_type", "")).strip() + cron_expr = str(body.get("cron_expr", "")).strip()[:256] + at_time = str(body.get("at_time", "")).strip()[:64] + target_mode = str(body.get("target_mode", "auto")).strip()[:256] + model = str(body.get("model", "")).strip()[:128] + initial_message = str(body.get("initial_message", "")).strip()[:4096] + auto_approve = bool(body.get("auto_approve", False)) + raw_tools = body.get("auto_approve_tools", []) + auto_approve_tools = raw_tools if isinstance(raw_tools, list) else [] + enabled = bool(body.get("enabled", True)) + + if not name: + return JSONResponse({"error": "name is required"}, status_code=400) + if not initial_message: + return JSONResponse({"error": "initial_message is required"}, status_code=400) + + validation_err = _validate_schedule_fields(schedule_type, cron_expr, at_time) + if validation_err: + return JSONResponse({"error": validation_err}, status_code=400) + + if not target_mode: + return JSONResponse({"error": "target_mode is required"}, status_code=400) + + # Cap total schedule count to prevent unbounded growth + max_schedules = 200 + existing = storage.list_scheduled_tasks() + if len(existing) >= max_schedules: + return JSONResponse( + {"error": f"Maximum of {max_schedules} schedules reached"}, status_code=409 + ) + + next_run = _compute_next_run(schedule_type, cron_expr, at_time) + task_id = uuid.uuid4().hex + created_by = getattr(getattr(request, "state", None), "user_id", "") + + storage.create_scheduled_task( + task_id=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, + created_by=created_by, + next_run=next_run if enabled else "", + ) + + if not enabled: + # Storage backends default enabled=1 on create; persist user's choice + storage.update_scheduled_task(task_id, enabled=False) + + task = storage.get_scheduled_task(task_id) + if task: + _normalize_task_dict(task) + return JSONResponse(task) + + +async def admin_get_schedule(request: Request) -> JSONResponse: + """GET /v1/api/admin/schedules/{task_id} — get single task.""" + from turnstone.core.web_helpers import require_storage_or_503 + + storage, err = require_storage_or_503(request) + if err: + return err + task_id = request.path_params["task_id"] + task = storage.get_scheduled_task(task_id) + if task is None: + return JSONResponse({"error": "Schedule not found"}, status_code=404) + _normalize_task_dict(task) + return JSONResponse(task) + + +async def admin_update_schedule(request: Request) -> JSONResponse: + """PUT /v1/api/admin/schedules/{task_id} — partial update.""" + from turnstone.core.web_helpers import read_json_or_400, require_storage_or_503 + + storage, err = require_storage_or_503(request) + if err: + return err + task_id = request.path_params["task_id"] + + existing = storage.get_scheduled_task(task_id) + if existing is None: + return JSONResponse({"error": "Schedule not found"}, status_code=404) + + body = await read_json_or_400(request) + if isinstance(body, JSONResponse): + return body + + updates: dict[str, Any] = {} + if "name" in body: + updates["name"] = str(body["name"]).strip()[:256] + if "description" in body: + updates["description"] = str(body["description"]).strip()[:1024] + if "schedule_type" in body: + updates["schedule_type"] = str(body["schedule_type"]).strip() + if "cron_expr" in body: + updates["cron_expr"] = str(body["cron_expr"]).strip()[:256] + if "at_time" in body: + updates["at_time"] = str(body["at_time"]).strip()[:64] + if "target_mode" in body: + updates["target_mode"] = str(body["target_mode"]).strip()[:256] + if "model" in body: + updates["model"] = str(body["model"]).strip()[:128] + if "initial_message" in body: + updates["initial_message"] = str(body["initial_message"]).strip()[:4096] + if "auto_approve" in body: + updates["auto_approve"] = bool(body["auto_approve"]) + if "auto_approve_tools" in body: + raw = body["auto_approve_tools"] + updates["auto_approve_tools"] = raw if isinstance(raw, list) else [] + if "enabled" in body: + updates["enabled"] = bool(body["enabled"]) + + # Validate schedule fields if changed + stype = updates.get("schedule_type", existing["schedule_type"]) + cexpr = updates.get("cron_expr", existing["cron_expr"]) + atime = updates.get("at_time", existing["at_time"]) + schedule_fields_changed = ( + "schedule_type" in updates or "cron_expr" in updates or "at_time" in updates + ) + if schedule_fields_changed: + validation_err = _validate_schedule_fields(stype, cexpr, atime) + if validation_err: + return JSONResponse({"error": validation_err}, status_code=400) + + # Recompute next_run if schedule changed or enabled toggled + if schedule_fields_changed or "enabled" in updates: + enabled = updates.get("enabled", bool(existing.get("enabled", 1))) + if enabled: + # Re-validate at_time when re-enabling a one-shot task + if stype == "at" and not schedule_fields_changed: + validation_err = _validate_schedule_fields(stype, cexpr, atime) + if validation_err: + return JSONResponse({"error": validation_err}, status_code=400) + updates["next_run"] = _compute_next_run(stype, cexpr, atime) + else: + updates["next_run"] = "" + + storage.update_scheduled_task(task_id, **updates) + task = storage.get_scheduled_task(task_id) + if task: + _normalize_task_dict(task) + return JSONResponse(task) + + +async def admin_delete_schedule(request: Request) -> JSONResponse: + """DELETE /v1/api/admin/schedules/{task_id} — delete task + runs.""" + from turnstone.core.web_helpers import require_storage_or_503 + + storage, err = require_storage_or_503(request) + if err: + return err + task_id = request.path_params["task_id"] + if storage.delete_scheduled_task(task_id): + return JSONResponse({"status": "ok"}) + return JSONResponse({"error": "Schedule not found"}, status_code=404) + + +async def admin_list_schedule_runs(request: Request) -> JSONResponse: + """GET /v1/api/admin/schedules/{task_id}/runs — run history.""" + from turnstone.core.web_helpers import require_storage_or_503 + + storage, err = require_storage_or_503(request) + if err: + return err + task_id = request.path_params["task_id"] + + # Verify task exists + if storage.get_scheduled_task(task_id) is None: + return JSONResponse({"error": "Schedule not found"}, status_code=404) + + try: + limit = min(int(request.query_params.get("limit", "50")), 200) + except (ValueError, TypeError): + limit = 50 + runs = storage.list_task_runs(task_id, limit=limit) + return JSONResponse({"runs": runs}) + + # --------------------------------------------------------------------------- # App factory # --------------------------------------------------------------------------- @@ -937,6 +1214,16 @@ def create_app( admin_delete_channel, methods=["DELETE"], ), + Route("/api/admin/schedules", admin_list_schedules), + Route("/api/admin/schedules", admin_create_schedule, methods=["POST"]), + Route("/api/admin/schedules/{task_id}", admin_get_schedule), + Route("/api/admin/schedules/{task_id}", admin_update_schedule, methods=["PUT"]), + Route( + "/api/admin/schedules/{task_id}", + admin_delete_schedule, + methods=["DELETE"], + ), + Route("/api/admin/schedules/{task_id}/runs", admin_list_schedule_runs), ], ), Route("/health", health), @@ -966,6 +1253,20 @@ def create_app( from turnstone.core.auth import LoginRateLimiter app.state.login_limiter = LoginRateLimiter() + + # Scheduler — start background thread if storage is available + if auth_storage is not None: + from turnstone.console.scheduler import TaskScheduler + + scheduler = TaskScheduler( + broker=broker, + collector=collector, + storage=auth_storage, + ) + app.state.scheduler = scheduler + else: + app.state.scheduler = None + return app diff --git a/turnstone/console/static/admin.js b/turnstone/console/static/admin.js index 9114d916..c7f45c20 100644 --- a/turnstone/console/static/admin.js +++ b/turnstone/console/static/admin.js @@ -46,10 +46,13 @@ function switchAdminTab(tab) { tab === "tokens" ? "" : "none"; document.getElementById("admin-channels").style.display = tab === "channels" ? "" : "none"; + document.getElementById("admin-schedules").style.display = + tab === "schedules" ? "" : "none"; if (tab === "users") loadAdminUsers(); if (tab === "tokens") _populateTokenUserSelect(); if (tab === "channels") _populateChannelUserSelect(); + if (tab === "schedules") loadAdminSchedules(); } // --------------------------------------------------------------------------- @@ -373,6 +376,517 @@ function confirmUnlinkChannel(channelType, channelUserId) { ); } +// --------------------------------------------------------------------------- +// Schedules +// --------------------------------------------------------------------------- + +var _csTrapHandler = null; +var _esTrapHandler = null; +var _srTrapHandler = null; +var _editScheduleTriggerEl = null; +var _runsScheduleTriggerEl = null; + +function loadAdminSchedules() { + authFetch("/v1/api/admin/schedules") + .then(function (r) { + if (!r.ok) throw new Error("Failed to load schedules"); + return r.json(); + }) + .then(function (data) { + _renderSchedules(data.schedules || []); + }) + .catch(function () { + document.getElementById("admin-schedules-table").innerHTML = + '
Failed to load schedules
'; + }); +} + +function _renderSchedules(schedules) { + var container = document.getElementById("admin-schedules-table"); + if (!schedules.length) { + container.innerHTML = + '
No scheduled tasks. Create one to get started.
'; + return; + } + var html = ""; + for (var i = 0; i < schedules.length; i++) { + var s = schedules[i]; + var typeLabel = s.schedule_type === "cron" ? "cron" : "at"; + var typeCls = s.schedule_type === "cron" ? "scope-write" : "scope-approve"; + var schedule = + s.schedule_type === "cron" + ? s.cron_expr + : (s.at_time || "").slice(0, 16).replace("T", " "); + var target = s.target_mode; + var nextRun = s.next_run + ? escapeHtml(s.next_run).slice(0, 16).replace("T", " ") + : "\u2014"; + var enabled = s.enabled; + var statusCls = enabled ? "sched-active" : "sched-disabled"; + var statusLabel = enabled ? "active" : "disabled"; + var statusDot = enabled ? "\u25cf " : "\u25cb "; + if (s.schedule_type === "at" && !enabled && s.last_run) { + statusCls = "sched-expired"; + statusLabel = "completed"; + statusDot = "\u25c9 "; + } + html += + '
' + + '' + + escapeHtml(s.name) + + "" + + '' + + typeLabel + + "" + + '' + + escapeHtml(schedule) + + "" + + '' + + escapeHtml(target) + + "" + + '' + + nextRun + + "" + + '' + + statusDot + + statusLabel + + "" + + '' + + '' + + '' + + '" + + '' + + "
"; + } + container.innerHTML = html; + // Bind buttons + var editBtns = container.querySelectorAll("[data-edit-sched]"); + for (var j = 0; j < editBtns.length; j++) { + editBtns[j].addEventListener("click", function () { + showEditScheduleModal(this.getAttribute("data-edit-sched")); + }); + } + var runsBtns = container.querySelectorAll("[data-runs-sched]"); + for (var k = 0; k < runsBtns.length; k++) { + runsBtns[k].addEventListener("click", function () { + showScheduleRuns(this.getAttribute("data-runs-sched")); + }); + } + var toggleBtns = container.querySelectorAll("[data-toggle-sched]"); + for (var m = 0; m < toggleBtns.length; m++) { + toggleBtns[m].addEventListener("click", function () { + toggleSchedule( + this.getAttribute("data-toggle-sched"), + this.getAttribute("data-enabled") === "1", + ); + }); + } + var delBtns = container.querySelectorAll("[data-delete-sched]"); + for (var n = 0; n < delBtns.length; n++) { + delBtns[n].addEventListener("click", function () { + confirmDeleteSchedule( + this.getAttribute("data-delete-sched"), + this.getAttribute("data-sname"), + ); + }); + } +} + +function toggleSchedule(taskId, currentlyEnabled) { + authFetch("/v1/api/admin/schedules/" + encodeURIComponent(taskId), { + method: "PUT", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ enabled: !currentlyEnabled }), + }) + .then(function (r) { + if (!r.ok) throw new Error("Toggle failed"); + showToast(currentlyEnabled ? "Schedule disabled" : "Schedule enabled"); + loadAdminSchedules(); + }) + .catch(function () { + showToast("Failed to toggle schedule"); + }); +} + +function confirmDeleteSchedule(taskId, name) { + showConfirmModal( + "Delete Schedule", + "Delete schedule \u2018" + + name + + "\u2019 and its run history? This cannot be undone.", + "Delete", + function () { + authFetch("/v1/api/admin/schedules/" + encodeURIComponent(taskId), { + method: "DELETE", + }) + .then(function (r) { + if (!r.ok) throw new Error("Delete failed"); + showToast("Schedule deleted"); + loadAdminSchedules(); + }) + .catch(function () { + showToast("Failed to delete schedule"); + }); + }, + ); +} + +// --- Create Schedule Modal --- + +function toggleScheduleTypeFields() { + var t = document.getElementById("cs-type").value; + document.getElementById("cs-cron-group").style.display = + t === "cron" ? "" : "none"; + document.getElementById("cs-at-group").style.display = + t === "at" ? "" : "none"; + if (t === "cron") document.getElementById("cs-cron").focus(); + else document.getElementById("cs-at").focus(); +} + +function toggleScheduleNodeField() { + var v = document.getElementById("cs-target").value; + document.getElementById("cs-node-group").style.display = + v === "node" ? "" : "none"; + if (v === "node") document.getElementById("cs-node").focus(); +} + +function showCreateScheduleModal() { + var overlay = document.getElementById("create-schedule-overlay"); + overlay.style.display = "flex"; + document.getElementById("create-schedule-error").style.display = "none"; + document.getElementById("cs-name").value = ""; + document.getElementById("cs-desc").value = ""; + document.getElementById("cs-type").value = "cron"; + document.getElementById("cs-cron").value = ""; + document.getElementById("cs-at").value = ""; + document.getElementById("cs-target").value = "auto"; + document.getElementById("cs-node").value = ""; + document.getElementById("cs-model").value = ""; + document.getElementById("cs-message").value = ""; + document.getElementById("cs-autoapprove").checked = false; + toggleScheduleTypeFields(); + toggleScheduleNodeField(); + document.getElementById("cs-submit").disabled = false; + document.getElementById("cs-submit").textContent = "Create"; + _csTrapHandler = _installTrap( + "create-schedule-overlay", + "create-schedule-box", + ); + setTimeout(function () { + document.getElementById("cs-name").focus(); + }, 50); +} + +function hideCreateScheduleModal() { + document.getElementById("create-schedule-overlay").style.display = "none"; + _csTrapHandler = _removeTrap(_csTrapHandler); + var trigger = document.querySelector("#admin-schedules .admin-action-btn"); + if (trigger) trigger.focus(); +} + +function submitCreateSchedule() { + var name = (document.getElementById("cs-name").value || "").trim(); + var desc = (document.getElementById("cs-desc").value || "").trim(); + var schedType = document.getElementById("cs-type").value; + var cronExpr = (document.getElementById("cs-cron").value || "").trim(); + var atTime = document.getElementById("cs-at").value || ""; + var targetMode = document.getElementById("cs-target").value; + var nodeId = (document.getElementById("cs-node").value || "").trim(); + var model = (document.getElementById("cs-model").value || "").trim(); + var message = (document.getElementById("cs-message").value || "").trim(); + var autoApprove = document.getElementById("cs-autoapprove").checked; + var errEl = document.getElementById("create-schedule-error"); + + if (!name) return _showModalError(errEl, "Name is required"); + if (!message) return _showModalError(errEl, "Initial message is required"); + if (schedType === "cron" && !cronExpr) + return _showModalError(errEl, "Cron expression is required"); + if (schedType === "at" && !atTime) + return _showModalError(errEl, "Run time is required"); + + // Normalize datetime-local to "YYYY-MM-DDTHH:MM:SS+00:00" (UTC) + if (schedType === "at" && atTime) { + if (atTime.length === 16) atTime += ":00"; + else if (atTime.length > 19) atTime = atTime.slice(0, 19); + atTime += "+00:00"; + } + + if (targetMode === "node") targetMode = nodeId; + + var btn = document.getElementById("cs-submit"); + btn.disabled = true; + btn.textContent = "Creating\u2026"; + + authFetch("/v1/api/admin/schedules", { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ + name: name, + description: desc, + schedule_type: schedType, + cron_expr: cronExpr, + at_time: atTime, + target_mode: targetMode, + model: model, + initial_message: message, + auto_approve: autoApprove, + }), + }) + .then(function (r) { + if (!r.ok) + return r.json().then(function (d) { + throw new Error(d.error || "Failed"); + }); + return r.json(); + }) + .then(function () { + hideCreateScheduleModal(); + showToast("Schedule '" + name + "' created"); + loadAdminSchedules(); + }) + .catch(function (err) { + btn.disabled = false; + btn.textContent = "Create"; + _showModalError(errEl, err.message || "Failed to create schedule"); + }); +} + +// --- Edit Schedule Modal --- + +function toggleEditScheduleTypeFields() { + var t = document.getElementById("es-type").value; + document.getElementById("es-cron-group").style.display = + t === "cron" ? "" : "none"; + document.getElementById("es-at-group").style.display = + t === "at" ? "" : "none"; + if (t === "cron") document.getElementById("es-cron").focus(); + else document.getElementById("es-at").focus(); +} + +function toggleEditScheduleNodeField() { + var v = document.getElementById("es-target").value; + document.getElementById("es-node-group").style.display = + v === "node" ? "" : "none"; + if (v === "node") document.getElementById("es-node").focus(); +} + +function showEditScheduleModal(taskId) { + _editScheduleTriggerEl = document.activeElement; + authFetch("/v1/api/admin/schedules/" + encodeURIComponent(taskId)) + .then(function (r) { + if (!r.ok) throw new Error("Not found"); + return r.json(); + }) + .then(function (s) { + document.getElementById("es-id").value = s.task_id; + document.getElementById("es-name").value = s.name || ""; + document.getElementById("es-desc").value = s.description || ""; + document.getElementById("es-type").value = s.schedule_type; + document.getElementById("es-cron").value = s.cron_expr || ""; + document.getElementById("es-at").value = (s.at_time || "").slice(0, 16); + var isSpecificNode = + s.target_mode && + s.target_mode !== "auto" && + s.target_mode !== "pool" && + s.target_mode !== "all"; + document.getElementById("es-target").value = isSpecificNode + ? "node" + : s.target_mode; + document.getElementById("es-node").value = isSpecificNode + ? s.target_mode + : ""; + document.getElementById("es-model").value = s.model || ""; + document.getElementById("es-message").value = s.initial_message || ""; + document.getElementById("es-autoapprove").checked = !!s.auto_approve; + document.getElementById("es-enabled").checked = !!s.enabled; + toggleEditScheduleTypeFields(); + toggleEditScheduleNodeField(); + document.getElementById("edit-schedule-error").style.display = "none"; + document.getElementById("es-submit").disabled = false; + document.getElementById("es-submit").textContent = "Save"; + var overlay = document.getElementById("edit-schedule-overlay"); + overlay.style.display = "flex"; + _esTrapHandler = _installTrap( + "edit-schedule-overlay", + "edit-schedule-box", + ); + setTimeout(function () { + document.getElementById("es-name").focus(); + }, 50); + }) + .catch(function () { + showToast("Failed to load schedule"); + }); +} + +function hideEditScheduleModal() { + document.getElementById("edit-schedule-overlay").style.display = "none"; + _esTrapHandler = _removeTrap(_esTrapHandler); + if (_editScheduleTriggerEl && _editScheduleTriggerEl.isConnected) { + _editScheduleTriggerEl.focus(); + } + _editScheduleTriggerEl = null; +} + +function submitEditSchedule() { + var taskId = document.getElementById("es-id").value; + var name = (document.getElementById("es-name").value || "").trim(); + var message = (document.getElementById("es-message").value || "").trim(); + var schedType = document.getElementById("es-type").value; + var cronExpr = (document.getElementById("es-cron").value || "").trim(); + var targetMode = document.getElementById("es-target").value; + if (targetMode === "node") + targetMode = (document.getElementById("es-node").value || "").trim(); + var atTime = document.getElementById("es-at").value || ""; + if (atTime) { + if (atTime.length === 16) atTime += ":00"; + else if (atTime.length > 19) atTime = atTime.slice(0, 19); + atTime += "+00:00"; + } + + var errEl = document.getElementById("edit-schedule-error"); + + if (!name) return _showModalError(errEl, "Name is required"); + if (!message) return _showModalError(errEl, "Initial message is required"); + if (schedType === "cron" && !cronExpr) + return _showModalError(errEl, "Cron expression is required"); + if (schedType === "at" && !atTime) + return _showModalError(errEl, "Run time is required"); + + var btn = document.getElementById("es-submit"); + btn.disabled = true; + btn.textContent = "Saving\u2026"; + + authFetch("/v1/api/admin/schedules/" + encodeURIComponent(taskId), { + method: "PUT", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ + name: (document.getElementById("es-name").value || "").trim(), + description: (document.getElementById("es-desc").value || "").trim(), + schedule_type: document.getElementById("es-type").value, + cron_expr: (document.getElementById("es-cron").value || "").trim(), + at_time: atTime, + target_mode: targetMode, + model: (document.getElementById("es-model").value || "").trim(), + initial_message: ( + document.getElementById("es-message").value || "" + ).trim(), + auto_approve: document.getElementById("es-autoapprove").checked, + enabled: document.getElementById("es-enabled").checked, + }), + }) + .then(function (r) { + if (!r.ok) + return r.json().then(function (d) { + throw new Error(d.error || "Failed"); + }); + return r.json(); + }) + .then(function () { + hideEditScheduleModal(); + showToast("Schedule updated"); + loadAdminSchedules(); + }) + .catch(function (err) { + btn.disabled = false; + btn.textContent = "Save"; + _showModalError(errEl, err.message || "Failed to update schedule"); + }); +} + +// --- Schedule Runs Modal --- + +function showScheduleRuns(taskId) { + _runsScheduleTriggerEl = document.activeElement; + authFetch( + "/v1/api/admin/schedules/" + encodeURIComponent(taskId) + "/runs?limit=50", + ) + .then(function (r) { + if (!r.ok) throw new Error("Not found"); + return r.json(); + }) + .then(function (data) { + var runs = data.runs || []; + var container = document.getElementById("schedule-runs-table"); + if (!runs.length) { + container.innerHTML = '
No runs yet
'; + } else { + var html = + ''; + for (var i = 0; i < runs.length; i++) { + var r = runs[i]; + var statusCls = + r.status === "dispatched" + ? "sched-active" + : r.status === "failed" + ? "sched-expired" + : ""; + html += + '
' + + '' + + escapeHtml(r.started || "") + .slice(0, 19) + .replace("T", " ") + + "" + + '' + + escapeHtml(r.node_id || "\u2014") + + "" + + '' + + escapeHtml(r.status) + + "" + + '' + + escapeHtml(r.error || "\u2014") + + "
"; + } + container.innerHTML = html; + } + var overlay = document.getElementById("schedule-runs-overlay"); + overlay.style.display = "flex"; + _srTrapHandler = _installTrap( + "schedule-runs-overlay", + "schedule-runs-box", + ); + }) + .catch(function () { + showToast("Failed to load run history"); + }); +} + +function hideScheduleRunsModal() { + document.getElementById("schedule-runs-overlay").style.display = "none"; + _srTrapHandler = _removeTrap(_srTrapHandler); + if (_runsScheduleTriggerEl && _runsScheduleTriggerEl.isConnected) { + _runsScheduleTriggerEl.focus(); + } + _runsScheduleTriggerEl = null; +} + // --------------------------------------------------------------------------- // Create Channel Link Modal // --------------------------------------------------------------------------- @@ -644,7 +1158,7 @@ function _modalFocusTrap(boxId) { var box = document.getElementById(boxId); if (!box) return; var focusable = box.querySelectorAll( - "input:not([disabled]), select:not([disabled]), button:not([disabled])", + "input:not([disabled]):not([type='hidden']), select:not([disabled]), textarea:not([disabled]), button:not([disabled])", ); var visible = []; for (var i = 0; i < focusable.length; i++) { @@ -678,6 +1192,10 @@ function _installTrap(overlayId, boxId, trapRef) { else if (overlayId === "token-created-overlay") hideTokenCreatedModal(); else if (overlayId === "create-channel-overlay") hideCreateChannelModal(); + else if (overlayId === "create-schedule-overlay") + hideCreateScheduleModal(); + else if (overlayId === "edit-schedule-overlay") hideEditScheduleModal(); + else if (overlayId === "schedule-runs-overlay") hideScheduleRunsModal(); else if (overlayId === "confirm-overlay") hideConfirmModal(); } }; @@ -721,6 +1239,24 @@ document.addEventListener("keydown", function (e) { hideCreateChannelModal(); return; } + var cso = document.getElementById("create-schedule-overlay"); + if (cso && cso.style.display !== "none") { + e.preventDefault(); + hideCreateScheduleModal(); + return; + } + var eso = document.getElementById("edit-schedule-overlay"); + if (eso && eso.style.display !== "none") { + e.preventDefault(); + hideEditScheduleModal(); + return; + } + var sro = document.getElementById("schedule-runs-overlay"); + if (sro && sro.style.display !== "none") { + e.preventDefault(); + hideScheduleRunsModal(); + return; + } var cf = document.getElementById("confirm-overlay"); if (cf && cf.style.display !== "none") { e.preventDefault(); @@ -735,7 +1271,7 @@ document.addEventListener("keydown", function (e) { if (!tablist) return; tablist.addEventListener("keydown", function (e) { if (e.key !== "ArrowLeft" && e.key !== "ArrowRight") return; - var tabOrder = ["users", "tokens", "channels"]; + var tabOrder = ["users", "tokens", "channels", "schedules"]; var idx = tabOrder.indexOf(_adminTab); if (e.key === "ArrowRight") idx = (idx + 1) % tabOrder.length; else idx = (idx - 1 + tabOrder.length) % tabOrder.length; diff --git a/turnstone/console/static/index.html b/turnstone/console/static/index.html index 7cc55f1e..550babcf 100644 --- a/turnstone/console/static/index.html +++ b/turnstone/console/static/index.html @@ -80,6 +80,7 @@ + @@ -142,6 +143,26 @@
Select a user to view channel links
+ + + @@ -284,6 +305,110 @@ window.TURNSTONE_KB_SHORTCUTS = [ + + + + + + + + + diff --git a/turnstone/console/static/style.css b/turnstone/console/static/style.css index f1f1e659..b34ac4d5 100644 --- a/turnstone/console/static/style.css +++ b/turnstone/console/static/style.css @@ -832,6 +832,58 @@ .admin-btn-danger:hover { opacity: 1; background: rgba(248, 113, 113, 0.1); } .admin-btn-danger:focus-visible { outline: 2px solid var(--red); outline-offset: 2px; } +.admin-btn-action { + background: none; + border: 1px solid var(--border-strong); + color: var(--fg-dim); + font-family: var(--font-display); + font-size: 10px; + font-weight: 500; + padding: 2px 8px; + border-radius: var(--radius-sm); + cursor: pointer; + opacity: 0.8; + transition: opacity 0.15s, background 0.15s; + margin-right: 4px; +} +.admin-btn-action:hover { opacity: 1; background: rgba(255, 255, 255, 0.05); color: var(--fg); } +.admin-btn-action:focus-visible { outline: 2px solid var(--accent); outline-offset: 2px; } + +/* Schedules grid: NAME | TYPE | SCHEDULE | TARGET | NEXT RUN | STATUS | ACTIONS */ +#admin-schedules .admin-colheaders, +#admin-schedules .admin-row { + grid-template-columns: 1.5fr 60px 1.2fr 80px 130px 70px 170px; +} + +/* Schedule runs grid: STARTED | NODE | STATUS | ERROR */ +.sched-runs-grid { grid-template-columns: 2fr 1fr 1fr 2fr; } + +/* Schedule status indicators */ +.sched-active { color: var(--green); font-weight: 500; } +.sched-disabled { color: var(--fg-dim); } +.sched-expired { color: var(--accent); } + +/* Wide modal variant for schedule forms */ +.admin-modal-wide { width: 480px; } + +/* Checkbox labels inside admin modals */ +.admin-modal label.admin-checkbox { + display: flex; + align-items: center; + gap: 8px; + font-size: 12px; + font-weight: 500; + text-transform: none; + letter-spacing: 0; + color: var(--fg); + cursor: pointer; + margin-top: 14px; +} +.admin-modal label.admin-checkbox input[type="checkbox"] { + width: auto; + margin: 0; +} + /* Admin modals (reuse new-ws-overlay pattern) */ .admin-modal { background: var(--bg-surface); @@ -871,7 +923,7 @@ margin-top: 12px; } .admin-modal label:first-of-type { margin-top: 0; } -.admin-modal input, .admin-modal select { +.admin-modal input:not([type="hidden"]), .admin-modal select, .admin-modal textarea { width: 100%; padding: 9px 12px; background: var(--bg); @@ -882,12 +934,13 @@ font-size: 13px; transition: border-color 0.15s, box-shadow 0.15s; } -.admin-modal input:focus, .admin-modal select:focus { +.admin-modal input:focus, .admin-modal select:focus, .admin-modal textarea:focus { border-color: var(--accent); outline: none; box-shadow: 0 0 0 3px var(--accent-dim); } -.admin-modal input::placeholder { color: var(--fg-dim); opacity: 0.6; } +.admin-modal input::placeholder, .admin-modal textarea::placeholder { color: var(--fg-dim); opacity: 0.6; } +.admin-modal textarea { resize: vertical; min-height: 40px; } .admin-modal [role="alert"] { color: var(--red); font-size: 12px; margin-bottom: 8px; display: none; } .modal-buttons { display: flex; gap: 10px; margin-top: 20px; } @@ -925,7 +978,8 @@ .modal-submit:focus-visible { outline: 2px solid var(--fg-bright); outline-offset: 2px; } .modal-submit:disabled { opacity: 0.4; cursor: not-allowed; filter: none; } -#create-user-overlay, #create-token-overlay, #token-created-overlay, #create-channel-overlay, #confirm-overlay { +#create-user-overlay, #create-token-overlay, #token-created-overlay, #create-channel-overlay, #confirm-overlay, +#create-schedule-overlay, #edit-schedule-overlay, #schedule-runs-overlay { position: fixed; inset: 0; background: rgba(0, 0, 0, 0.7); @@ -969,6 +1023,10 @@ grid-template-columns: 80px 1fr 80px; } #admin-channels .admin-col-created { display: none; } + #admin-schedules .admin-colheaders, #admin-schedules .admin-row { + grid-template-columns: 1fr 60px 80px 130px; + } + .admin-col-sschedule, .admin-col-starget, .admin-col-snext { display: none; } } /* ========================================================================== @@ -981,7 +1039,7 @@ .node-link, .dash-cell-node, .pagination button { transition: none; } .dash-row.has-link::after, .node-group-header::before { transition: none; } #new-ws-box select, #new-ws-box input, #new-ws-buttons button { transition: none; } - .admin-tab, .admin-row, .admin-btn-danger { transition: none; } + .admin-tab, .admin-row, .admin-btn-danger, .admin-btn-action { transition: none; } .admin-action-btn, .modal-cancel, .modal-submit { transition: none; } .admin-modal input, .admin-modal select { transition: none; } } diff --git a/turnstone/core/storage/_postgresql.py b/turnstone/core/storage/_postgresql.py index 10569067..2d43416b 100644 --- a/turnstone/core/storage/_postgresql.py +++ b/turnstone/core/storage/_postgresql.py @@ -879,6 +879,196 @@ class PostgreSQLBackend: conn.commit() return result.rowcount > 0 + # -- Scheduled tasks ------------------------------------------------------- + + def create_scheduled_task( + self, + task_id: str, + name: str, + description: str, + schedule_type: str, + cron_expr: str, + at_time: str, + target_mode: str, + model: str, + initial_message: str, + auto_approve: bool, + auto_approve_tools: list[str], + created_by: str, + next_run: str, + ) -> None: + from sqlalchemy.dialects import postgresql + + from turnstone.core.storage._schema import scheduled_tasks + + now = datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%S") + with self._engine.connect() as conn: + conn.execute( + postgresql.insert(scheduled_tasks) + .values( + task_id=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=1 if auto_approve else 0, + auto_approve_tools=",".join(auto_approve_tools), + enabled=1, + created_by=created_by, + next_run=next_run, + created=now, + updated=now, + ) + .on_conflict_do_nothing() + ) + conn.commit() + + def get_scheduled_task(self, task_id: str) -> dict[str, Any] | None: + from turnstone.core.storage._schema import scheduled_tasks + + with self._engine.connect() as conn: + row = conn.execute( + sa.select(scheduled_tasks).where(scheduled_tasks.c.task_id == task_id) + ).fetchone() + if row is None: + return None + return dict(row._mapping) + + def list_scheduled_tasks(self) -> list[dict[str, Any]]: + from turnstone.core.storage._schema import scheduled_tasks + + with self._engine.connect() as conn: + rows = conn.execute( + sa.select(scheduled_tasks).order_by(scheduled_tasks.c.created.desc()) + ).fetchall() + return [dict(r._mapping) for r in rows] + + _UPDATABLE_TASK_FIELDS = frozenset( + { + "name", + "description", + "schedule_type", + "cron_expr", + "at_time", + "target_mode", + "model", + "initial_message", + "auto_approve", + "auto_approve_tools", + "enabled", + "last_run", + "next_run", + "updated", + } + ) + + def update_scheduled_task(self, task_id: str, **fields: Any) -> bool: + from turnstone.core.storage._schema import scheduled_tasks + + fields = {k: v for k, v in fields.items() if k in self._UPDATABLE_TASK_FIELDS} + fields["updated"] = datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%S") + if "auto_approve" in fields: + fields["auto_approve"] = 1 if fields["auto_approve"] else 0 + if "auto_approve_tools" in fields and isinstance(fields["auto_approve_tools"], list): + fields["auto_approve_tools"] = ",".join(fields["auto_approve_tools"]) + if "enabled" in fields: + fields["enabled"] = 1 if fields["enabled"] else 0 + with self._engine.connect() as conn: + result = conn.execute( + sa.update(scheduled_tasks) + .where(scheduled_tasks.c.task_id == task_id) + .values(**fields) + ) + conn.commit() + return result.rowcount > 0 + + def delete_scheduled_task(self, task_id: str) -> bool: + from turnstone.core.storage._schema import scheduled_task_runs, scheduled_tasks + + with self._engine.connect() as conn: + conn.execute( + sa.delete(scheduled_task_runs).where(scheduled_task_runs.c.task_id == task_id) + ) + result = conn.execute( + sa.delete(scheduled_tasks).where(scheduled_tasks.c.task_id == task_id) + ) + conn.commit() + return result.rowcount > 0 + + def list_due_tasks(self, now: str) -> list[dict[str, Any]]: + from turnstone.core.storage._schema import scheduled_tasks + + with self._engine.connect() as conn: + rows = conn.execute( + sa.select(scheduled_tasks) + .where( + (scheduled_tasks.c.enabled == 1) + & (scheduled_tasks.c.next_run <= now) + & (scheduled_tasks.c.next_run != "") + ) + .order_by(scheduled_tasks.c.next_run) + .limit(100) + ).fetchall() + return [dict(r._mapping) for r in rows] + + def record_task_run( + self, + run_id: str, + task_id: str, + node_id: str, + ws_id: str, + correlation_id: str, + started: str, + status: str, + error: str, + ) -> None: + from turnstone.core.storage._schema import scheduled_task_runs + + with self._engine.connect() as conn: + conn.execute( + sa.insert(scheduled_task_runs), + { + "run_id": run_id, + "task_id": task_id, + "node_id": node_id, + "ws_id": ws_id, + "correlation_id": correlation_id, + "started": started, + "status": status, + "error": error, + }, + ) + conn.commit() + + def list_task_runs(self, task_id: str, limit: int = 50) -> list[dict[str, Any]]: + from turnstone.core.storage._schema import scheduled_task_runs + + with self._engine.connect() as conn: + rows = conn.execute( + sa.select(scheduled_task_runs) + .where(scheduled_task_runs.c.task_id == task_id) + .order_by(scheduled_task_runs.c.started.desc()) + .limit(limit) + ).fetchall() + return [dict(r._mapping) for r in rows] + + def prune_task_runs(self, retention_days: int = 90) -> int: + from datetime import timedelta + + from turnstone.core.storage._schema import scheduled_task_runs + + cutoff = (datetime.now(UTC) - timedelta(days=retention_days)).strftime("%Y-%m-%dT%H:%M:%S") + with self._engine.connect() as conn: + result = conn.execute( + sa.delete(scheduled_task_runs).where(scheduled_task_runs.c.started < cutoff) + ) + conn.commit() + return result.rowcount + # -- Lifecycle ------------------------------------------------------------- def close(self) -> None: diff --git a/turnstone/core/storage/_protocol.py b/turnstone/core/storage/_protocol.py index 7eff4b01..8e7256c2 100644 --- a/turnstone/core/storage/_protocol.py +++ b/turnstone/core/storage/_protocol.py @@ -249,6 +249,69 @@ class StorageBackend(Protocol): """Remove a channel route. Returns True if existed.""" ... + # -- Scheduled tasks ------------------------------------------------------- + + def create_scheduled_task( + self, + task_id: str, + name: str, + description: str, + schedule_type: str, + cron_expr: str, + at_time: str, + target_mode: str, + model: str, + initial_message: str, + auto_approve: bool, + auto_approve_tools: list[str], + created_by: str, + next_run: str, + ) -> None: + """Create a scheduled task. No-op if task_id already exists.""" + ... + + def get_scheduled_task(self, task_id: str) -> dict[str, Any] | None: + """Return scheduled task dict or None.""" + ... + + def list_scheduled_tasks(self) -> list[dict[str, Any]]: + """Return all scheduled tasks ordered by created DESC.""" + ... + + def update_scheduled_task(self, task_id: str, **fields: Any) -> bool: + """Update specified fields on a scheduled task. Returns True if found.""" + ... + + def delete_scheduled_task(self, task_id: str) -> bool: + """Delete a scheduled task and its run history. Returns True if found.""" + ... + + def list_due_tasks(self, now: str) -> list[dict[str, Any]]: + """Return enabled tasks whose next_run <= now, ordered by next_run.""" + ... + + def record_task_run( + self, + run_id: str, + task_id: str, + node_id: str, + ws_id: str, + correlation_id: str, + started: str, + status: str, + error: str, + ) -> None: + """Record a scheduled task execution.""" + ... + + def list_task_runs(self, task_id: str, limit: int = 50) -> list[dict[str, Any]]: + """List run history for a task, ordered by started DESC.""" + ... + + def prune_task_runs(self, retention_days: int = 90) -> int: + """Delete task runs older than retention_days. Returns count deleted.""" + ... + # -- Lifecycle ------------------------------------------------------------- def close(self) -> None: diff --git a/turnstone/core/storage/_schema.py b/turnstone/core/storage/_schema.py index 6293426c..870070e5 100644 --- a/turnstone/core/storage/_schema.py +++ b/turnstone/core/storage/_schema.py @@ -137,3 +137,48 @@ channel_routes = sa.Table( ) sa.Index("idx_channel_routes_ws", channel_routes.c.ws_id) + +# --------------------------------------------------------------------------- +# Scheduled task tables +# --------------------------------------------------------------------------- + +scheduled_tasks = sa.Table( + "scheduled_tasks", + metadata, + sa.Column("task_id", sa.Text, primary_key=True), + sa.Column("name", sa.Text, nullable=False), + sa.Column("description", sa.Text, nullable=False, server_default=""), + sa.Column("schedule_type", sa.Text, nullable=False), # "cron" or "at" + sa.Column("cron_expr", sa.Text, nullable=False, server_default=""), + sa.Column("at_time", sa.Text, nullable=False, server_default=""), # ISO8601 + sa.Column("target_mode", sa.Text, nullable=False, server_default="auto"), + sa.Column("model", sa.Text, nullable=False, server_default=""), + sa.Column("initial_message", sa.Text, nullable=False), + sa.Column("auto_approve", sa.Integer, nullable=False, server_default="0"), + sa.Column("auto_approve_tools", sa.Text, nullable=False, server_default=""), + sa.Column("enabled", sa.Integer, nullable=False, server_default="1"), + sa.Column("created_by", sa.Text, nullable=False, server_default=""), + sa.Column("last_run", sa.Text), + sa.Column("next_run", sa.Text), + sa.Column("created", sa.Text, nullable=False), + sa.Column("updated", sa.Text, nullable=False), +) + +sa.Index("idx_scheduled_tasks_enabled", scheduled_tasks.c.enabled) +sa.Index("idx_scheduled_tasks_next_run", scheduled_tasks.c.next_run) + +scheduled_task_runs = sa.Table( + "scheduled_task_runs", + metadata, + sa.Column("run_id", sa.Text, primary_key=True), + sa.Column("task_id", sa.Text, nullable=False), + sa.Column("node_id", sa.Text, nullable=False, server_default=""), + sa.Column("ws_id", sa.Text, nullable=False, server_default=""), + sa.Column("correlation_id", sa.Text, nullable=False, server_default=""), + sa.Column("started", sa.Text, nullable=False), + sa.Column("status", sa.Text, nullable=False, server_default="dispatched"), + sa.Column("error", sa.Text, nullable=False, server_default=""), +) + +sa.Index("idx_scheduled_task_runs_task_id", scheduled_task_runs.c.task_id) +sa.Index("idx_scheduled_task_runs_started", scheduled_task_runs.c.started) diff --git a/turnstone/core/storage/_sqlite.py b/turnstone/core/storage/_sqlite.py index 5edd4436..09f8d289 100644 --- a/turnstone/core/storage/_sqlite.py +++ b/turnstone/core/storage/_sqlite.py @@ -929,6 +929,194 @@ class SQLiteBackend: conn.commit() return result.rowcount > 0 + # -- Scheduled tasks ------------------------------------------------------- + + def create_scheduled_task( + self, + task_id: str, + name: str, + description: str, + schedule_type: str, + cron_expr: str, + at_time: str, + target_mode: str, + model: str, + initial_message: str, + auto_approve: bool, + auto_approve_tools: list[str], + created_by: str, + next_run: str, + ) -> None: + from turnstone.core.storage._schema import scheduled_tasks + + now = datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%S") + with self._engine.connect() as conn: + conn.execute( + sa.insert(scheduled_tasks).prefix_with("OR IGNORE"), + { + "task_id": 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": 1 if auto_approve else 0, + "auto_approve_tools": ",".join(auto_approve_tools), + "enabled": 1, + "created_by": created_by, + "next_run": next_run, + "created": now, + "updated": now, + }, + ) + conn.commit() + + def get_scheduled_task(self, task_id: str) -> dict[str, Any] | None: + from turnstone.core.storage._schema import scheduled_tasks + + with self._engine.connect() as conn: + row = conn.execute( + sa.select(scheduled_tasks).where(scheduled_tasks.c.task_id == task_id) + ).fetchone() + if row is None: + return None + return dict(row._mapping) + + def list_scheduled_tasks(self) -> list[dict[str, Any]]: + from turnstone.core.storage._schema import scheduled_tasks + + with self._engine.connect() as conn: + rows = conn.execute( + sa.select(scheduled_tasks).order_by(scheduled_tasks.c.created.desc()) + ).fetchall() + return [dict(r._mapping) for r in rows] + + _UPDATABLE_TASK_FIELDS = frozenset( + { + "name", + "description", + "schedule_type", + "cron_expr", + "at_time", + "target_mode", + "model", + "initial_message", + "auto_approve", + "auto_approve_tools", + "enabled", + "last_run", + "next_run", + "updated", + } + ) + + def update_scheduled_task(self, task_id: str, **fields: Any) -> bool: + from turnstone.core.storage._schema import scheduled_tasks + + fields = {k: v for k, v in fields.items() if k in self._UPDATABLE_TASK_FIELDS} + fields["updated"] = datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%S") + # Normalize boolean → int for auto_approve + if "auto_approve" in fields: + fields["auto_approve"] = 1 if fields["auto_approve"] else 0 + if "auto_approve_tools" in fields and isinstance(fields["auto_approve_tools"], list): + fields["auto_approve_tools"] = ",".join(fields["auto_approve_tools"]) + if "enabled" in fields: + fields["enabled"] = 1 if fields["enabled"] else 0 + with self._engine.connect() as conn: + result = conn.execute( + sa.update(scheduled_tasks) + .where(scheduled_tasks.c.task_id == task_id) + .values(**fields) + ) + conn.commit() + return result.rowcount > 0 + + def delete_scheduled_task(self, task_id: str) -> bool: + from turnstone.core.storage._schema import scheduled_task_runs, scheduled_tasks + + with self._engine.connect() as conn: + conn.execute( + sa.delete(scheduled_task_runs).where(scheduled_task_runs.c.task_id == task_id) + ) + result = conn.execute( + sa.delete(scheduled_tasks).where(scheduled_tasks.c.task_id == task_id) + ) + conn.commit() + return result.rowcount > 0 + + def list_due_tasks(self, now: str) -> list[dict[str, Any]]: + from turnstone.core.storage._schema import scheduled_tasks + + with self._engine.connect() as conn: + rows = conn.execute( + sa.select(scheduled_tasks) + .where( + (scheduled_tasks.c.enabled == 1) + & (scheduled_tasks.c.next_run <= now) + & (scheduled_tasks.c.next_run != "") + ) + .order_by(scheduled_tasks.c.next_run) + .limit(100) + ).fetchall() + return [dict(r._mapping) for r in rows] + + def record_task_run( + self, + run_id: str, + task_id: str, + node_id: str, + ws_id: str, + correlation_id: str, + started: str, + status: str, + error: str, + ) -> None: + from turnstone.core.storage._schema import scheduled_task_runs + + with self._engine.connect() as conn: + conn.execute( + sa.insert(scheduled_task_runs), + { + "run_id": run_id, + "task_id": task_id, + "node_id": node_id, + "ws_id": ws_id, + "correlation_id": correlation_id, + "started": started, + "status": status, + "error": error, + }, + ) + conn.commit() + + def list_task_runs(self, task_id: str, limit: int = 50) -> list[dict[str, Any]]: + from turnstone.core.storage._schema import scheduled_task_runs + + with self._engine.connect() as conn: + rows = conn.execute( + sa.select(scheduled_task_runs) + .where(scheduled_task_runs.c.task_id == task_id) + .order_by(scheduled_task_runs.c.started.desc()) + .limit(limit) + ).fetchall() + return [dict(r._mapping) for r in rows] + + def prune_task_runs(self, retention_days: int = 90) -> int: + from datetime import timedelta + + from turnstone.core.storage._schema import scheduled_task_runs + + cutoff = (datetime.now(UTC) - timedelta(days=retention_days)).strftime("%Y-%m-%dT%H:%M:%S") + with self._engine.connect() as conn: + result = conn.execute( + sa.delete(scheduled_task_runs).where(scheduled_task_runs.c.started < cutoff) + ) + conn.commit() + return result.rowcount + # -- Lifecycle ------------------------------------------------------------- def close(self) -> None: diff --git a/turnstone/core/storage/migrations/versions/004_scheduled_tasks.py b/turnstone/core/storage/migrations/versions/004_scheduled_tasks.py new file mode 100644 index 00000000..dd558361 --- /dev/null +++ b/turnstone/core/storage/migrations/versions/004_scheduled_tasks.py @@ -0,0 +1,62 @@ +"""Scheduled tasks and run history tables. + +Revision ID: 004 +Revises: 003 +Create Date: 2026-03-05 +""" + +import sqlalchemy as sa +from alembic import op + +revision = "004" +down_revision = "003" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.create_table( + "scheduled_tasks", + sa.Column("task_id", sa.Text, primary_key=True), + sa.Column("name", sa.Text, nullable=False), + sa.Column("description", sa.Text, nullable=False, server_default=""), + sa.Column("schedule_type", sa.Text, nullable=False), + sa.Column("cron_expr", sa.Text, nullable=False, server_default=""), + sa.Column("at_time", sa.Text, nullable=False, server_default=""), + sa.Column("target_mode", sa.Text, nullable=False, server_default="auto"), + sa.Column("model", sa.Text, nullable=False, server_default=""), + sa.Column("initial_message", sa.Text, nullable=False), + sa.Column("auto_approve", sa.Integer, nullable=False, server_default="0"), + sa.Column("auto_approve_tools", sa.Text, nullable=False, server_default=""), + sa.Column("enabled", sa.Integer, nullable=False, server_default="1"), + sa.Column("created_by", sa.Text, nullable=False, server_default=""), + sa.Column("last_run", sa.Text), + sa.Column("next_run", sa.Text), + sa.Column("created", sa.Text, nullable=False), + sa.Column("updated", sa.Text, nullable=False), + ) + op.create_index("idx_scheduled_tasks_enabled", "scheduled_tasks", ["enabled"]) + op.create_index("idx_scheduled_tasks_next_run", "scheduled_tasks", ["next_run"]) + + op.create_table( + "scheduled_task_runs", + sa.Column("run_id", sa.Text, primary_key=True), + sa.Column("task_id", sa.Text, nullable=False), + sa.Column("node_id", sa.Text, nullable=False, server_default=""), + sa.Column("ws_id", sa.Text, nullable=False, server_default=""), + sa.Column("correlation_id", sa.Text, nullable=False, server_default=""), + sa.Column("started", sa.Text, nullable=False), + sa.Column("status", sa.Text, nullable=False, server_default="dispatched"), + sa.Column("error", sa.Text, nullable=False, server_default=""), + ) + op.create_index("idx_scheduled_task_runs_task_id", "scheduled_task_runs", ["task_id"]) + op.create_index("idx_scheduled_task_runs_started", "scheduled_task_runs", ["started"]) + + +def downgrade() -> None: + op.drop_index("idx_scheduled_task_runs_started", "scheduled_task_runs") + op.drop_index("idx_scheduled_task_runs_task_id", "scheduled_task_runs") + op.drop_table("scheduled_task_runs") + op.drop_index("idx_scheduled_tasks_next_run", "scheduled_tasks") + op.drop_index("idx_scheduled_tasks_enabled", "scheduled_tasks") + op.drop_table("scheduled_tasks") diff --git a/turnstone/mq/bridge.py b/turnstone/mq/bridge.py index 7286ed09..1d308873 100644 --- a/turnstone/mq/bridge.py +++ b/turnstone/mq/bridge.py @@ -362,6 +362,9 @@ class Bridge: model = getattr(msg, "model", "") initial_message = getattr(msg, "initial_message", "") resume_session = getattr(msg, "resume_session", "") + user_id = getattr(msg, "user_id", "") + if user_id: + log.info("bridge.create_ws user_id=%s name=%s model=%s", user_id, name, model) ws_id, resumed = self._create_ws_on_server( name=name, auto_approve=auto_approve, diff --git a/turnstone/mq/protocol.py b/turnstone/mq/protocol.py index 93fcecc3..17c84916 100644 --- a/turnstone/mq/protocol.py +++ b/turnstone/mq/protocol.py @@ -96,6 +96,7 @@ class CreateWorkstreamMessage(InboundMessage): model: str = "" initial_message: str = "" resume_session: str = "" + user_id: str = "" @dataclass