Add scheduled task system with cron/at scheduling (#26)

* Add scheduled task system with cron/at scheduling, admin API, and console UI

Console-integrated background scheduler dispatches workstreams on recurring
cron expressions or one-shot ISO8601 timestamps. Four target modes: auto
(best node by headroom), pool (shared queue), all (fan-out), or specific
node. Redis distributed lock with unique owner + Lua conditional release
prevents duplicate dispatch in multi-console deployments.

Storage: scheduled_tasks + scheduled_task_runs tables (migration 004),
9 protocol methods on both SQLite and PostgreSQL backends, field allowlist
on updates, run history auto-pruned at 90 days.

API: 6 CRUD endpoints under /v1/api/admin/schedules with croniter
validation, ISO8601 future-time checks, field length bounds, schedule
count cap (200), and OpenAPI spec entries with Pydantic models.

UI: Schedules tab in admin panel with create/edit/delete modals, run
history modal, cron/at type toggle, target mode select, status dots for
accessibility, responsive grid, keyboard navigation, and focus management.

Security: fan-out capped at 20 nodes/task/tick, auto-approve dispatches
logged at WARNING with created_by attribution, user_id propagated in
CreateWorkstreamMessage for audit trail.

46 tests across storage, scheduler engine, and API endpoints.

* Add croniter to test extras for CI compatibility

CI installs [test] extras but not [console], so croniter was missing
when schedule API tests import console/server.py validation functions.

* Address Copilot review: timezone validation, focus trap, enabled flag

- Reject naive at_time timestamps — require timezone offset (e.g. +00:00 or Z)
- UI appends +00:00 to datetime-local values for explicit UTC
- Fix datetime-local normalization: check length before appending seconds
- Add textarea to modal focus trap selector (prevents focus escape)
- Fix _normalize_task_dict not called in update response
- Persist enabled=false on create (storage defaults to enabled=1)
- Validate at_time is still in future when re-enabling a one-shot task
- broker._redis coupling acknowledged as tracked tech debt

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

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"] == []

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"

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"

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

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]

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

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
 
 

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 =
+        '<div class="dashboard-empty">Failed to load schedules</div>';
+    });
+}
+
+function _renderSchedules(schedules) {
+  var container = document.getElementById("admin-schedules-table");
+  if (!schedules.length) {
+    container.innerHTML =
+      '<div class="dashboard-empty">No scheduled tasks. Create one to get started.</div>';
+    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 +=
+      '<div class="admin-row" role="listitem">' +
+      '<span class="admin-col admin-col-sname">' +
+      escapeHtml(s.name) +
+      "</span>" +
+      '<span class="admin-col admin-col-stype"><span class="scope-badge ' +
+      typeCls +
+      '">' +
+      typeLabel +
+      "</span></span>" +
+      '<span class="admin-col admin-col-sschedule"><code>' +
+      escapeHtml(schedule) +
+      "</code></span>" +
+      '<span class="admin-col admin-col-starget">' +
+      escapeHtml(target) +
+      "</span>" +
+      '<span class="admin-col admin-col-snext">' +
+      nextRun +
+      "</span>" +
+      '<span class="admin-col admin-col-sstatus"><span class="' +
+      statusCls +
+      '">' +
+      statusDot +
+      statusLabel +
+      "</span></span>" +
+      '<span class="admin-col admin-col-actions">' +
+      '<button class="admin-btn-action" data-edit-sched="' +
+      escapeHtml(s.task_id) +
+      '" title="Edit">edit</button>' +
+      '<button class="admin-btn-action" data-runs-sched="' +
+      escapeHtml(s.task_id) +
+      '" title="Run history">runs</button>' +
+      '<button class="admin-btn-action" data-toggle-sched="' +
+      escapeHtml(s.task_id) +
+      '" data-enabled="' +
+      (enabled ? "1" : "0") +
+      '" title="' +
+      (enabled ? "Disable" : "Enable") +
+      '">' +
+      (enabled ? "disable" : "enable") +
+      "</button>" +
+      '<button class="admin-btn-danger" data-delete-sched="' +
+      escapeHtml(s.task_id) +
+      '" data-sname="' +
+      escapeHtml(s.name) +
+      '" title="Delete">delete</button>' +
+      "</span></div>";
+  }
+  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 = '<div class="dashboard-empty">No runs yet</div>';
+      } else {
+        var html =
+          '<div class="admin-colheaders sched-runs-grid" aria-hidden="true">' +
+          '<span class="admin-col">STARTED</span>' +
+          '<span class="admin-col">NODE</span>' +
+          '<span class="admin-col">STATUS</span>' +
+          '<span class="admin-col">ERROR</span></div>';
+        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 +=
+            '<div class="admin-row sched-runs-grid">' +
+            '<span class="admin-col">' +
+            escapeHtml(r.started || "")
+              .slice(0, 19)
+              .replace("T", " ") +
+            "</span>" +
+            '<span class="admin-col">' +
+            escapeHtml(r.node_id || "\u2014") +
+            "</span>" +
+            '<span class="admin-col"><span class="' +
+            statusCls +
+            '">' +
+            escapeHtml(r.status) +
+            "</span></span>" +
+            '<span class="admin-col">' +
+            escapeHtml(r.error || "\u2014") +
+            "</span></div>";
+        }
+        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;

turnstone/console/static/index.html

@@ -80,6 +80,7 @@
       <button id="tab-users" class="admin-tab active" data-tab="users" role="tab" aria-selected="true" aria-controls="admin-users" tabindex="0" onclick="switchAdminTab('users')">Users</button>
       <button id="tab-tokens" class="admin-tab" data-tab="tokens" role="tab" aria-selected="false" aria-controls="admin-tokens" tabindex="-1" onclick="switchAdminTab('tokens')">Tokens</button>
       <button id="tab-channels" class="admin-tab" data-tab="channels" role="tab" aria-selected="false" aria-controls="admin-channels" tabindex="-1" onclick="switchAdminTab('channels')">Channels</button>
+      <button id="tab-schedules" class="admin-tab" data-tab="schedules" role="tab" aria-selected="false" aria-controls="admin-schedules" tabindex="-1" onclick="switchAdminTab('schedules')">Schedules</button>
     </div>
 
     <!-- Users Tab -->
@@ -142,6 +143,26 @@
         <div class="dashboard-empty">Select a user to view channel links</div>
       </div>
     </div>
+
+    <!-- Schedules Tab -->
+    <div id="admin-schedules" class="admin-panel" role="tabpanel" aria-labelledby="tab-schedules" style="display:none">
+      <div class="admin-toolbar">
+        <span class="section-header" style="margin:0">SCHEDULED TASKS</span>
+        <button class="admin-action-btn" onclick="showCreateScheduleModal()">+ New schedule</button>
+      </div>
+      <div class="admin-colheaders" aria-hidden="true">
+        <span class="admin-col admin-col-sname">NAME</span>
+        <span class="admin-col admin-col-stype">TYPE</span>
+        <span class="admin-col admin-col-sschedule">SCHEDULE</span>
+        <span class="admin-col admin-col-starget">TARGET</span>
+        <span class="admin-col admin-col-snext">NEXT RUN</span>
+        <span class="admin-col admin-col-sstatus">STATUS</span>
+        <span class="admin-col admin-col-actions">ACTIONS</span>
+      </div>
+      <div id="admin-schedules-table" role="list" aria-label="Scheduled tasks" aria-live="polite">
+        <div class="dashboard-empty">Loading schedules...</div>
+      </div>
+    </div>
   </div>
 </div>
 
@@ -284,6 +305,110 @@ window.TURNSTONE_KB_SHORTCUTS = [
   </div>
 </div>
 
+<!-- Create Schedule Modal -->
+<div id="create-schedule-overlay" style="display:none" role="dialog" aria-modal="true" aria-labelledby="create-schedule-title">
+  <div id="create-schedule-box" class="admin-modal admin-modal-wide">
+    <h2 id="create-schedule-title">New Schedule</h2>
+    <div id="create-schedule-error" role="alert" aria-live="assertive"></div>
+    <label for="cs-name">Name</label>
+    <input id="cs-name" type="text" placeholder="Daily health check" autocomplete="off">
+    <label for="cs-desc">Description <span class="label-hint">optional</span></label>
+    <input id="cs-desc" type="text" placeholder="" autocomplete="off">
+    <label for="cs-type">Schedule type</label>
+    <select id="cs-type" onchange="toggleScheduleTypeFields()">
+      <option value="cron">Cron (recurring)</option>
+      <option value="at">At (one-shot)</option>
+    </select>
+    <div id="cs-cron-group">
+      <label for="cs-cron">Cron expression</label>
+      <input id="cs-cron" type="text" placeholder="0 9 * * MON-FRI" autocomplete="off" spellcheck="false" aria-describedby="cs-cron-hint">
+      <span id="cs-cron-hint" class="label-hint" style="display:block;margin-top:3px">min hour day month weekday</span>
+    </div>
+    <div id="cs-at-group" style="display:none">
+      <label for="cs-at">Run at</label>
+      <input id="cs-at" type="datetime-local">
+    </div>
+    <label for="cs-target">Target</label>
+    <select id="cs-target" onchange="toggleScheduleNodeField()">
+      <option value="auto">Auto (best available)</option>
+      <option value="pool">Pool (any bridge)</option>
+      <option value="all">All nodes</option>
+      <option value="node">Specific node...</option>
+    </select>
+    <div id="cs-node-group" style="display:none">
+      <label for="cs-node">Node ID</label>
+      <input id="cs-node" type="text" placeholder="node-001" autocomplete="off" spellcheck="false">
+    </div>
+    <label for="cs-model">Model <span class="label-hint">optional</span></label>
+    <input id="cs-model" type="text" placeholder="Default model" autocomplete="off">
+    <label for="cs-message">Initial message</label>
+    <textarea id="cs-message" rows="3" placeholder="What should the workstream do?"></textarea>
+    <label class="admin-checkbox"><input id="cs-autoapprove" type="checkbox"> Auto-approve tool calls</label>
+    <div class="modal-buttons">
+      <button class="modal-cancel" onclick="hideCreateScheduleModal()">Cancel</button>
+      <button id="cs-submit" class="modal-submit" onclick="submitCreateSchedule()">Create</button>
+    </div>
+  </div>
+</div>
+
+<!-- Edit Schedule Modal -->
+<div id="edit-schedule-overlay" style="display:none" role="dialog" aria-modal="true" aria-labelledby="edit-schedule-title">
+  <div id="edit-schedule-box" class="admin-modal admin-modal-wide">
+    <h2 id="edit-schedule-title">Edit Schedule</h2>
+    <div id="edit-schedule-error" role="alert" aria-live="assertive"></div>
+    <input id="es-id" type="hidden">
+    <label for="es-name">Name</label>
+    <input id="es-name" type="text" autocomplete="off">
+    <label for="es-desc">Description</label>
+    <input id="es-desc" type="text" autocomplete="off">
+    <label for="es-type">Schedule type</label>
+    <select id="es-type" onchange="toggleEditScheduleTypeFields()">
+      <option value="cron">Cron (recurring)</option>
+      <option value="at">At (one-shot)</option>
+    </select>
+    <div id="es-cron-group">
+      <label for="es-cron">Cron expression</label>
+      <input id="es-cron" type="text" autocomplete="off" spellcheck="false">
+    </div>
+    <div id="es-at-group" style="display:none">
+      <label for="es-at">Run at</label>
+      <input id="es-at" type="datetime-local">
+    </div>
+    <label for="es-target">Target</label>
+    <select id="es-target" onchange="toggleEditScheduleNodeField()">
+      <option value="auto">Auto (best available)</option>
+      <option value="pool">Pool (any bridge)</option>
+      <option value="all">All nodes</option>
+      <option value="node">Specific node...</option>
+    </select>
+    <div id="es-node-group" style="display:none">
+      <label for="es-node">Node ID</label>
+      <input id="es-node" type="text" autocomplete="off" spellcheck="false">
+    </div>
+    <label for="es-model">Model</label>
+    <input id="es-model" type="text" autocomplete="off">
+    <label for="es-message">Initial message</label>
+    <textarea id="es-message" rows="3"></textarea>
+    <label class="admin-checkbox"><input id="es-autoapprove" type="checkbox"> Auto-approve tool calls</label>
+    <label class="admin-checkbox"><input id="es-enabled" type="checkbox"> Enabled</label>
+    <div class="modal-buttons">
+      <button class="modal-cancel" onclick="hideEditScheduleModal()">Cancel</button>
+      <button id="es-submit" class="modal-submit" onclick="submitEditSchedule()">Save</button>
+    </div>
+  </div>
+</div>
+
+<!-- Schedule Runs Modal -->
+<div id="schedule-runs-overlay" style="display:none" role="dialog" aria-modal="true" aria-labelledby="schedule-runs-title">
+  <div id="schedule-runs-box" class="admin-modal admin-modal-wide">
+    <h2 id="schedule-runs-title">Run History</h2>
+    <div id="schedule-runs-table"></div>
+    <div class="modal-buttons">
+      <button class="modal-cancel" onclick="hideScheduleRunsModal()">Close</button>
+    </div>
+  </div>
+</div>
+
 <script src="/static/admin.js"></script>
 <script src="/static/app.js"></script>
 </body>

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; }
 }

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:

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:

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)

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:

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

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,

turnstone/mq/protocol.py

@@ -96,6 +96,7 @@ class CreateWorkstreamMessage(InboundMessage):
     model: str = ""
     initial_message: str = ""
     resume_session: str = ""
+    user_id: str = ""
 
 
 @dataclass