feat(schedules): add persona and project settings to scheduled tasks

A scheduled task could pin the model and skill of the workstream each
firing creates; it can now also pin its persona and project, so a
schedule can run under, e.g., the researcher persona attached to a
specific project's memory bucket.

The two values live on scheduled_tasks (migration 066, Text NOT NULL
default '') and are passed verbatim to create_workstream at dispatch,
where the node resolves the persona for the workstream kind and gates
the project attach. Empty means "kind-default persona / no project",
resolved late at each firing (mirrors how empty model/skill already
behave) -- existing schedules keep byte-identical dispatch behaviour,
so there is no backfill.

Also fixes a latent bug this feature depends on: admin_create_schedule
read created_by from request.state.user_id, which AuthMiddleware never
sets, so every scheduled task stored created_by=''. It now reads
auth_result.user_id like every other console endpoint. This is now
load-bearing -- the scheduler dispatches under created_by and the node
gates the project attach against it. admin_update_schedule adopts the
editing admin as owner when a project is assigned to a pre-fix orphaned
('') schedule, and re-validates persona/project only when they change
so a since-disabled persona or lost membership does not block unrelated
edits (the node re-checks at dispatch either way).

Wired through: schema + migration (up/down + parity tested), both
storage backends, API schemas, SDK create_workstream and console
create_schedule/update_schedule, scheduler dispatch, and the admin
schedule shelf (persona + project pickers, current value preserved so
an edit cannot silently clear a filtered-out selection).

(cherry picked from commit c328bebecd)
This commit is contained in:
Patrick Buckley
2026-07-08 00:48:01 -07:00
parent 702ac43d0e
commit d48902fd01
19 changed files with 670 additions and 7 deletions
+103
View File
@@ -0,0 +1,103 @@
"""Tests for alembic migration 066 (persona + project on scheduled_tasks).
Drives ``command.upgrade``/``downgrade`` against an isolated SQLite database per
test (the 060/062/063/065 harness pattern), then asserts:
* upgrade adds the ``persona`` and ``project_id`` columns to ``scheduled_tasks``;
* a pre-066 scheduled task migrates cleanly, gaining ``""`` for both new columns
— the empty default that means "kind default persona" / "no project" and
preserves byte-identical dispatch behaviour to pre-066;
* downgrade removes both columns, returning ``scheduled_tasks`` to its exact
pre-066 shape — pinning the clean-rollback guarantee;
* up -> down -> up lands cleanly with no leftover-column conflict.
"""
from __future__ import annotations
from pathlib import Path
import sqlalchemy as sa
from alembic import command
from alembic.config import Config
_MIGRATIONS_DIR = str(
Path(__file__).resolve().parent.parent / "turnstone" / "core" / "storage" / "migrations"
)
def _alembic_cfg(db_path: Path) -> Config:
cfg = Config()
cfg.set_main_option("script_location", _MIGRATIONS_DIR)
cfg.set_main_option("sqlalchemy.url", f"sqlite:///{db_path}")
return cfg
def _insert_pre066_task(engine: sa.Engine) -> None:
with engine.begin() as conn:
conn.execute(
sa.text(
"INSERT INTO scheduled_tasks "
"(task_id, name, schedule_type, initial_message, created, updated) "
"VALUES ('t1', 'Nightly', 'cron', 'run', "
"'2026-01-01T00:00:00', '2026-01-01T00:00:00')"
)
)
class TestMigration066:
def test_upgrade_adds_persona_and_project_columns(self, tmp_path: Path) -> None:
db_path = tmp_path / "066-up.db"
command.upgrade(_alembic_cfg(db_path), "066")
engine = sa.create_engine(f"sqlite:///{db_path}")
try:
cols = {c["name"] for c in sa.inspect(engine).get_columns("scheduled_tasks")}
assert {"persona", "project_id"} <= cols
finally:
engine.dispose()
def test_preexisting_row_migrates_with_empty_default(self, tmp_path: Path) -> None:
db_path = tmp_path / "066-default.db"
cfg = _alembic_cfg(db_path)
# Stop at 065, insert a pre-066 scheduled task, THEN upgrade to 066.
command.upgrade(cfg, "065")
engine = sa.create_engine(f"sqlite:///{db_path}")
try:
_insert_pre066_task(engine)
command.upgrade(cfg, "066")
with engine.connect() as conn:
row = conn.execute(
sa.text("SELECT persona, project_id FROM scheduled_tasks WHERE task_id = 't1'")
).fetchone()
assert row is not None
assert row[0] == "" and row[1] == ""
finally:
engine.dispose()
def test_downgrade_removes_persona_and_project_columns(self, tmp_path: Path) -> None:
db_path = tmp_path / "066-down.db"
cfg = _alembic_cfg(db_path)
command.upgrade(cfg, "066")
command.downgrade(cfg, "065")
engine = sa.create_engine(f"sqlite:///{db_path}")
try:
cols = {c["name"] for c in sa.inspect(engine).get_columns("scheduled_tasks")}
assert "persona" not in cols and "project_id" not in cols
finally:
engine.dispose()
def test_downgrade_then_upgrade_round_trip(self, tmp_path: Path) -> None:
"""up -> down -> up must land cleanly (no leftover column conflict)."""
db_path = tmp_path / "066-roundtrip.db"
cfg = _alembic_cfg(db_path)
command.upgrade(cfg, "066")
command.downgrade(cfg, "065")
command.upgrade(cfg, "066")
engine = sa.create_engine(f"sqlite:///{db_path}")
try:
cols = {c["name"] for c in sa.inspect(engine).get_columns("scheduled_tasks")}
assert {"persona", "project_id"} <= cols
finally:
engine.dispose()
+200
View File
@@ -176,6 +176,206 @@ class TestScheduleAPI:
assert resp.status_code == 400
assert "future" in resp.json()["error"].lower()
@staticmethod
def _seed_persona(storage, name="researcher", kinds=None):
storage.create_persona(
{
"persona_id": f"id-{name}",
"name": name,
"display_name": name.title(),
"description": "",
"base_prompt": "You are a test persona.",
"applies_to_kinds": kinds or ["interactive"],
}
)
def test_create_with_persona_and_project(self, client, storage):
self._seed_persona(storage)
# Owned by the authenticated admin (created_by) → attachable.
storage.create_project("proj_1", "My Project", "test-admin")
resp = client.post(
"/v1/api/admin/schedules",
json=_cron_payload(persona="researcher", project_id="proj_1"),
)
assert resp.status_code == 200, resp.text
data = resp.json()
assert data["persona"] == "researcher"
assert data["project_id"] == "proj_1"
def test_create_defaults_persona_project_empty(self, client):
resp = client.post("/v1/api/admin/schedules", json=_cron_payload())
assert resp.status_code == 200
data = resp.json()
assert data["persona"] == ""
assert data["project_id"] == ""
def test_create_unknown_persona_rejected(self, client):
resp = client.post(
"/v1/api/admin/schedules",
json=_cron_payload(persona="ghost"),
)
assert resp.status_code == 400
assert "persona" in resp.json()["error"].lower()
def test_create_persona_wrong_kind_rejected(self, client, storage):
# A coordinator-only persona is refused — schedules only ever dispatch
# interactive workstreams, so the picker/validation are kind-scoped.
self._seed_persona(storage, name="orchestrator", kinds=["coordinator"])
resp = client.post(
"/v1/api/admin/schedules",
json=_cron_payload(persona="orchestrator"),
)
assert resp.status_code == 400
def test_create_unattachable_project_rejected(self, client, storage):
# A private project owned by someone else — the admin isn't a member.
storage.create_project("proj_x", "Theirs", "someone-else", visibility="private")
resp = client.post(
"/v1/api/admin/schedules",
json=_cron_payload(project_id="proj_x"),
)
assert resp.status_code == 403
def test_update_persona_and_project(self, client, storage):
self._seed_persona(storage, name="scribe")
storage.create_project("proj_2", "Proj Two", "test-admin")
task_id = client.post("/v1/api/admin/schedules", json=_cron_payload()).json()["task_id"]
resp = client.put(
f"/v1/api/admin/schedules/{task_id}",
json={"persona": "scribe", "project_id": "proj_2"},
)
assert resp.status_code == 200, resp.text
data = client.get(f"/v1/api/admin/schedules/{task_id}").json()
assert data["persona"] == "scribe"
assert data["project_id"] == "proj_2"
@staticmethod
def _legacy_task(storage, task_id="legacy"):
"""A schedule from before the created_by fix — created_by is ''."""
storage.create_scheduled_task(
task_id=task_id,
name="Legacy",
description="",
schedule_type="cron",
cron_expr="0 9 * * *",
at_time="",
target_mode="auto",
model="",
initial_message="go",
auto_approve=False,
auto_approve_tools=[],
created_by="",
next_run="2099-01-01T09:00:00",
)
def test_update_assign_project_heals_empty_created_by(self, client, storage):
# Assigning a project to an orphaned schedule adopts the editing admin
# as owner so the attach — and every future dispatch — has an identity.
self._legacy_task(storage)
storage.create_project("proj_heal", "Heal", "test-admin")
resp = client.put(
"/v1/api/admin/schedules/legacy",
json={"project_id": "proj_heal"},
)
assert resp.status_code == 200, resp.text
row = storage.get_scheduled_task("legacy")
assert row["project_id"] == "proj_heal"
assert row["created_by"] == "test-admin"
def test_update_denied_project_does_not_heal_created_by(self, client, storage):
# Healing must not become an attach bypass: a project the editing admin
# can't reach is still 403, and created_by/project stay untouched.
self._legacy_task(storage, task_id="legacy2")
storage.create_project("proj_other", "Other", "someone-else", visibility="private")
resp = client.put(
"/v1/api/admin/schedules/legacy2",
json={"project_id": "proj_other"},
)
assert resp.status_code == 403
row = storage.get_scheduled_task("legacy2")
assert row["created_by"] == ""
assert row["project_id"] == ""
def test_update_project_keeps_existing_owner(self, client, storage):
# A schedule that already has a real owner is NOT re-owned by an editing
# admin — created_by is only adopted for the orphaned "" case.
self._seed_persona(storage, name="researcher")
storage.create_scheduled_task(
task_id="owned",
name="Owned",
description="",
schedule_type="cron",
cron_expr="0 9 * * *",
at_time="",
target_mode="auto",
model="",
initial_message="go",
auto_approve=False,
auto_approve_tools=[],
created_by="original-owner",
next_run="2099-01-01T09:00:00",
)
# A public project the original owner (and anyone) can attach to.
storage.create_project("proj_pub", "Pub", "someone-else", visibility="public")
resp = client.put(
"/v1/api/admin/schedules/owned",
json={"project_id": "proj_pub"},
)
assert resp.status_code == 200, resp.text
row = storage.get_scheduled_task("owned")
assert row["project_id"] == "proj_pub"
assert row["created_by"] == "original-owner"
def test_update_unchanged_persona_skips_revalidation(self, client, storage):
# A persona disabled after creation must not block editing other fields
# when the shelf resends the unchanged slug (it still fails at dispatch).
self._seed_persona(storage, name="researcher")
task_id = client.post(
"/v1/api/admin/schedules", json=_cron_payload(persona="researcher")
).json()["task_id"]
storage.update_persona("id-researcher", enabled=False)
resp = client.put(
f"/v1/api/admin/schedules/{task_id}",
json={"name": "Renamed", "persona": "researcher"},
)
assert resp.status_code == 200, resp.text
assert resp.json()["name"] == "Renamed"
assert resp.json()["persona"] == "researcher"
def test_update_unchanged_project_skips_regate(self, client, storage):
# Project attach isn't re-gated when unchanged, so a project deleted (or
# membership lost) out from under the schedule doesn't block edits.
storage.create_project("proj_keep", "Keep", "test-admin")
task_id = client.post(
"/v1/api/admin/schedules", json=_cron_payload(project_id="proj_keep")
).json()["task_id"]
storage.delete_project("proj_keep") # a re-gate would now 400
resp = client.put(
f"/v1/api/admin/schedules/{task_id}",
json={"name": "Renamed", "project_id": "proj_keep"},
)
assert resp.status_code == 200, resp.text
assert resp.json()["project_id"] == "proj_keep"
def test_update_ignores_created_by_in_body(self, client, storage):
# created_by is never sourced from the request body — a spoofed value
# in the PUT payload is ignored (only the heal path from auth writes it).
task_id = client.post("/v1/api/admin/schedules", json=_cron_payload()).json()["task_id"]
client.put(
f"/v1/api/admin/schedules/{task_id}",
json={"name": "X", "created_by": "attacker"},
)
row = storage.get_scheduled_task(task_id)
assert row["created_by"] == "test-admin"
def test_update_unknown_persona_rejected(self, client):
task_id = client.post("/v1/api/admin/schedules", json=_cron_payload()).json()["task_id"]
resp = client.put(
f"/v1/api/admin/schedules/{task_id}",
json={"persona": "ghost"},
)
assert resp.status_code == 400
def test_get_schedule(self, client):
create_resp = client.post("/v1/api/admin/schedules", json=_cron_payload())
task_id = create_resp.json()["task_id"]
+35
View File
@@ -47,9 +47,44 @@ class TestScheduledTaskCRUD:
assert result["enabled"] == 1
assert result["created_by"] == "u_admin"
assert result["next_run"] == "2099-01-01T09:00:00"
# persona/project default to "" — empty means "kind default" / "no
# project", resolved late at dispatch (mirrors empty model/skill).
assert result["persona"] == ""
assert result["project_id"] == ""
assert "created" in result
assert "updated" in result
def test_create_with_persona_and_project(self, db):
db.create_scheduled_task(**_make_task_kwargs(persona="researcher", project_id="proj_42"))
result = db.get_scheduled_task("task_001")
assert result is not None
assert result["persona"] == "researcher"
assert result["project_id"] == "proj_42"
def test_update_persona_and_project(self, db):
db.create_scheduled_task(**_make_task_kwargs())
assert db.update_scheduled_task("task_001", persona="scribe", project_id="proj_9")
updated = db.get_scheduled_task("task_001")
assert updated is not None
assert updated["persona"] == "scribe"
assert updated["project_id"] == "proj_9"
# Clearing back to defaults is a first-class update, not a no-op.
assert db.update_scheduled_task("task_001", persona="", project_id="")
cleared = db.get_scheduled_task("task_001")
assert cleared is not None
assert cleared["persona"] == ""
assert cleared["project_id"] == ""
def test_update_created_by(self, db):
# created_by is allow-listed for update so the API can adopt an orphaned
# ("") schedule's owner. Exercised here so the Postgres backend covers
# the write too (the API test is SQLite-pinned).
db.create_scheduled_task(**_make_task_kwargs(created_by=""))
assert db.update_scheduled_task("task_001", created_by="adopted")
row = db.get_scheduled_task("task_001")
assert row is not None
assert row["created_by"] == "adopted"
def test_get_nonexistent(self, db):
assert db.get_scheduled_task("no_such_task") is None
+46
View File
@@ -156,6 +156,52 @@ class TestSchedulerTick:
assert run_kwargs["status"] == "dispatched"
assert run_kwargs["ws_id"] == "ws_abc123"
def test_dispatch_passes_persona_and_project(self, mocks):
"""persona + project_id ride to create_workstream; created_by becomes
the user_id the node gates the project attach against."""
collector, storage = mocks
task = _make_task(persona="researcher", project_id="proj_42")
storage.list_due_tasks.return_value = [task]
collector.get_nodes.return_value = ([_make_node()], 1)
collector.get_node_detail.return_value = {"server_url": "http://node-001:8080"}
scheduler = TaskScheduler(collector, storage)
with patch(
"turnstone.console.scheduler.TurnstoneServer.create_workstream",
return_value=_mock_create_response(),
) as mock_create:
scheduler._tick()
mock_create.assert_called_once()
call_kwargs = mock_create.call_args[1]
assert call_kwargs["persona"] == "researcher"
assert call_kwargs["project_id"] == "proj_42"
assert call_kwargs["user_id"] == "u_admin"
def test_dispatch_defaults_persona_project_empty(self, mocks):
"""A task row without persona/project keys dispatches with empty
strings — the node then resolves the current kind default / no attach."""
collector, storage = mocks
task = _make_task()
task.pop("persona", None)
task.pop("project_id", None)
storage.list_due_tasks.return_value = [task]
collector.get_nodes.return_value = ([_make_node()], 1)
collector.get_node_detail.return_value = {"server_url": "http://node-001:8080"}
scheduler = TaskScheduler(collector, storage)
with patch(
"turnstone.console.scheduler.TurnstoneServer.create_workstream",
return_value=_mock_create_response(),
) as mock_create:
scheduler._tick()
call_kwargs = mock_create.call_args[1]
assert call_kwargs["persona"] == ""
assert call_kwargs["project_id"] == ""
def test_dispatch_pool_mode(self, mocks):
collector, storage = mocks
+4
View File
@@ -337,6 +337,8 @@ async def test_create_schedule():
schedule_type="cron",
initial_message="Run nightly checks",
cron_expr="0 2 * * *",
persona="researcher",
project_id="proj_1",
)
assert resp.task_id == "t1"
body = captured_body[0]
@@ -344,6 +346,8 @@ async def test_create_schedule():
assert body["schedule_type"] == "cron"
assert body["cron_expr"] == "0 2 * * *"
assert body["initial_message"] == "Run nightly checks"
assert body["persona"] == "researcher"
assert body["project_id"] == "proj_1"
# Optional fields with defaults should not appear when not set
assert "description" not in body
assert "model" not in body
+6
View File
@@ -380,12 +380,16 @@ async def test_create_workstream_extended_params():
auto_approve_tools="read_file,write_file",
user_id="u42",
ws_id="ws_custom",
persona="researcher",
project_id="proj_9",
)
assert captured_body["name"] == "ext"
assert captured_body["initial_message"] == "hi"
assert captured_body["auto_approve_tools"] == "read_file,write_file"
assert captured_body["user_id"] == "u42"
assert captured_body["ws_id"] == "ws_custom"
assert captured_body["persona"] == "researcher"
assert captured_body["project_id"] == "proj_9"
@pytest.mark.anyio
@@ -406,3 +410,5 @@ async def test_create_workstream_omits_empty_params():
assert "auto_approve_tools" not in captured_body
assert "user_id" not in captured_body
assert "ws_id" not in captured_body
assert "persona" not in captured_body
assert "project_id" not in captured_body
+4
View File
@@ -151,6 +151,10 @@ class ConsoleCreateWsRequest(BaseModel):
default="",
description="Persona slug; resolved and snapshotted at creation, empty = kind default",
)
project_id: str = Field(
default="",
description="Project to attach the workstream to (validated against membership, empty = none)",
)
resume_ws: str = Field(
default="", description="Workstream ID to resume (loads previous conversation)"
)
+6
View File
@@ -195,6 +195,8 @@ class CreateScheduleRequest(BaseModel):
auto_approve: bool = Field(default=False)
auto_approve_tools: list[str] = Field(default_factory=list)
skill: str = Field(default="", description="Skill name (replaces default skills)")
persona: str = Field(default="", description="Persona slug (empty = kind default)")
project_id: str = Field(default="", description="Project to attach the workstream to")
notify_targets: list[dict[str, str]] = Field(
default_factory=list,
description="Notification targets on completion (channel_type + channel_id/user_id)",
@@ -216,6 +218,8 @@ class UpdateScheduleRequest(BaseModel):
auto_approve: bool | None = None
auto_approve_tools: list[str] | None = None
skill: str | None = None
persona: str | None = None
project_id: str | None = None
notify_targets: list[dict[str, str]] | None = None
enabled: bool | None = None
@@ -235,6 +239,8 @@ class ScheduleInfo(BaseModel):
auto_approve: bool = False
auto_approve_tools: list[str] = Field(default_factory=list)
skill: str = ""
persona: str = ""
project_id: str = ""
notify_targets: list[dict[str, str]] = Field(default_factory=list)
enabled: bool = True
created_by: str = ""
+2
View File
@@ -318,6 +318,8 @@ class TaskScheduler:
auto_approve_tools=",".join(self._parse_tools(task)),
user_id=task.get("created_by", ""),
skill=task.get("skill", ""),
persona=task.get("persona", ""),
project_id=task.get("project_id", ""),
notify_targets=task.get("notify_targets", "[]"),
# Mark the resulting ChatSession as non-interactive-for-
# consent so OAuth-MCP errors get persisted to
+97 -1
View File
@@ -6095,6 +6095,43 @@ def _validate_schedule_fields(schedule_type: str, cron_expr: str, at_time: str)
return None
def _resolve_schedule_persona(storage: Any, persona: str) -> tuple[str, str | None]:
"""Validate a schedule's persona slug against the interactive kind.
Schedules dispatch interactive workstreams, so the persona is resolved
against that kind the same eligibility rule the create handler applies,
surfaced here so a bad slug fails at edit time rather than silently at the
next firing. Returns ``(canonical_slug, None)`` on success (the resolved
row's name, never the raw input, per the persona contract) or
``("", error)`` on failure. Empty persona = kind default, always valid.
"""
if not persona:
return "", None
from turnstone.core.personas import resolve_persona_for_kind
row, err = resolve_persona_for_kind(storage, persona, "interactive")
if err:
return "", err
return (str(row["name"]) if row else persona), None
def _validate_schedule_project(
storage: Any, user_id: str, project_id: str
) -> tuple[int, str] | None:
"""Gate attaching a schedule's dispatched workstream to *project_id*.
Checked against *user_id* the schedule's ``created_by``, the identity the
scheduler dispatches under so the same owner/member rule the node enforces
at dispatch is applied up front. Returns ``None`` when allowed, else the
``(status, message)`` to surface. Empty project_id = no attach, allowed.
"""
if not project_id:
return None
from turnstone.core.auth import ensure_project_attachable
return ensure_project_attachable(user_id, project_id, storage=storage)
async def admin_preview_schedule(request: Request) -> JSONResponse:
"""POST /v1/api/admin/schedules/preview — validate timing, return next runs.
@@ -6188,7 +6225,17 @@ async def admin_create_schedule(request: Request) -> JSONResponse:
raw_tools = body.get("auto_approve_tools", [])
auto_approve_tools = raw_tools if isinstance(raw_tools, list) else []
skill_name = str(body.get("skill", "")).strip()[:256]
persona = str(body.get("persona", "")).strip()[:64]
project_id = str(body.get("project_id", "")).strip()[:64]
enabled = bool(body.get("enabled", True))
# created_by is the authenticated admin — read via auth_result like every
# other console endpoint (AuthMiddleware never sets request.state.user_id,
# so the previous ``state.user_id`` read silently stored ""). It is now
# load-bearing: the scheduler dispatches under this identity and the node
# gates the project attach against it, so an empty value would make every
# project-scoped schedule fail the attach.
auth_result = getattr(getattr(request, "state", None), "auth_result", None)
created_by = getattr(auth_result, "user_id", "") or ""
# Validate notify_targets
from turnstone.server import _validate_notify_targets
@@ -6208,6 +6255,13 @@ async def admin_create_schedule(request: Request) -> JSONResponse:
return JSONResponse({"error": "initial_message is required"}, status_code=400)
if skill_name and not storage.get_prompt_template_by_name(skill_name):
return JSONResponse({"error": f"Skill not found: {skill_name}"}, status_code=400)
persona, persona_err = _resolve_schedule_persona(storage, persona)
if persona_err:
return JSONResponse({"error": persona_err}, status_code=400)
project_denied = _validate_schedule_project(storage, created_by, project_id)
if project_denied is not None:
status_code, message = project_denied
return JSONResponse({"error": message}, status_code=status_code)
validation_err = _validate_schedule_fields(schedule_type, cron_expr, at_time)
if validation_err:
@@ -6226,7 +6280,6 @@ async def admin_create_schedule(request: Request) -> JSONResponse:
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,
@@ -6244,6 +6297,8 @@ async def admin_create_schedule(request: Request) -> JSONResponse:
next_run=next_run if enabled else "",
skill=skill_name,
notify_targets=notify_targets,
persona=persona,
project_id=project_id,
)
if not enabled:
@@ -6323,6 +6378,47 @@ async def admin_update_schedule(request: Request) -> JSONResponse:
if skill_val and not storage.get_prompt_template_by_name(skill_val):
return JSONResponse({"error": f"Skill not found: {skill_val}"}, status_code=400)
updates["skill"] = skill_val
if "persona" in body:
persona_val = str(body["persona"]).strip()[:64]
# Re-validate only when the persona actually changes: the edit shelf
# always resends the current slug, and a schedule whose persona was
# since disabled (or the picker couldn't show) must stay editable for
# its other fields. A stale persona still fails loudly at dispatch,
# where the node re-resolves it.
if persona_val != (existing.get("persona") or ""):
persona_val, persona_err = _resolve_schedule_persona(storage, persona_val)
if persona_err:
return JSONResponse({"error": persona_err}, status_code=400)
updates["persona"] = persona_val
if "project_id" in body:
project_val = str(body["project_id"]).strip()[:64]
# Re-gate only when the project actually changes: the edit shelf
# resends the current value, and membership churn (or a project the
# editing admin can't see) must not block unrelated edits — the node
# re-gates against the owner at dispatch. On an actual change, the
# schedule dispatches under created_by and the node gates the attach
# against it, so a schedule created before the created_by fix ("")
# could never attach. When a project is assigned to such an orphaned
# schedule, adopt the editing admin as its owner (never overriding a
# real created_by) and persist it so the create-time check and the
# dispatch identity agree.
if project_val != (existing.get("project_id") or ""):
editing_admin = (
getattr(
getattr(getattr(request, "state", None), "auth_result", None),
"user_id",
"",
)
or ""
)
owner = existing.get("created_by", "") or editing_admin
project_denied = _validate_schedule_project(storage, owner, project_val)
if project_denied is not None:
status_code, message = project_denied
return JSONResponse({"error": message}, status_code=status_code)
if project_val and not existing.get("created_by", ""):
updates["created_by"] = owner
updates["project_id"] = project_val
if "enabled" in body:
updates["enabled"] = bool(body["enabled"])
if "notify_targets" in body:
+60 -5
View File
@@ -1329,8 +1329,10 @@ function _populateScheduleSelect(selectId, url, labelKey, valueKey, opts) {
.then(function (data) {
const temp = sel.querySelector("[data-temporary]");
if (temp) temp.remove();
const items = opts && opts.listKey ? data[opts.listKey] : data;
let items = opts && opts.listKey ? data[opts.listKey] : data;
if (!Array.isArray(items)) return;
if (opts && typeof opts.filter === "function")
items = items.filter(opts.filter);
items.forEach(function (item) {
const opt = document.createElement("option");
opt.value = item[valueKey];
@@ -1338,7 +1340,22 @@ function _populateScheduleSelect(selectId, url, labelKey, valueKey, opts) {
opts && opts.display ? opts.display(item) : item[labelKey];
sel.appendChild(opt);
});
if (opts && opts.selected) sel.value = opts.selected;
if (opts && opts.selected) {
sel.value = opts.selected;
if (sel.value !== opts.selected) {
// The current value isn't in this list — it was filtered out
// (a disabled/wrong-kind persona) or is outside the caller-scoped
// feed (a private/archived project the editing admin can't see).
// Re-add it as a "(current)" option so it round-trips; without this
// the select falls back to the placeholder and saving an unrelated
// field would silently CLEAR the setting.
const keep = document.createElement("option");
keep.value = opts.selected;
keep.textContent = opts.selected + " (current)";
sel.appendChild(keep);
sel.value = opts.selected;
}
}
// Caller hook for placeholder annotation / other post-load tweaks.
// Used by the schedule modals to rewrite the bare "Default model"
// placeholder with the resolved alias so the label matches the
@@ -1923,7 +1940,12 @@ function _schResetForm() {
_schSetMode("daily");
}
function _schPopulateSelects(selectedModel, selectedSkill) {
function _schPopulateSelects(
selectedModel,
selectedSkill,
selectedPersona,
selectedProject,
) {
_populateScheduleSelect("sch-model", "/v1/api/models", "alias", "alias", {
listKey: "models",
selected: selectedModel || "",
@@ -1945,6 +1967,32 @@ function _schPopulateSelects(selectedModel, selectedSkill) {
},
},
);
// Schedules dispatch interactive workstreams, so only offer personas
// eligible for that kind; the label matches the home/create picker
// (display name, falling back to the slug).
_populateScheduleSelect("sch-persona", "/v1/api/personas", "name", "name", {
listKey: "personas",
selected: selectedPersona || "",
filter: function (p) {
return (p.applies_to_kinds || []).indexOf("interactive") !== -1;
},
display: function (p) {
return p.display_name || p.name;
},
});
_populateScheduleSelect(
"sch-project",
"/v1/api/projects",
"name",
"project_id",
{
listKey: "projects",
selected: selectedProject || "",
display: function (p) {
return p.name;
},
},
);
}
function _schOpen(title, tag, kind, submitLabel) {
@@ -1962,7 +2010,7 @@ function showCreateScheduleModal() {
_schWire();
_schResetForm();
document.getElementById("sch-enabled-row").hidden = true;
_schPopulateSelects("", "");
_schPopulateSelects("", "", "", "");
_schOpen("New schedule", "SCH-NEW", "create", "Create");
}
@@ -1991,7 +2039,12 @@ function showEditScheduleModal(taskId) {
? s.target_mode
: "";
document.getElementById("sch-node-group").hidden = !isSpecificNode;
_schPopulateSelects(s.model || "", s.skill || "");
_schPopulateSelects(
s.model || "",
s.skill || "",
s.persona || "",
s.project_id || "",
);
document.getElementById("sch-message").value = s.initial_message || "";
document.getElementById("sch-autoapprove").checked = !!s.auto_approve;
document.getElementById("sch-enabled").checked = !!s.enabled;
@@ -2036,6 +2089,8 @@ function _submitScheduleShelf() {
target_mode: targetMode,
model: (document.getElementById("sch-model").value || "").trim(),
skill: (document.getElementById("sch-template").value || "").trim(),
persona: (document.getElementById("sch-persona").value || "").trim(),
project_id: (document.getElementById("sch-project").value || "").trim(),
initial_message: message,
auto_approve: document.getElementById("sch-autoapprove").checked,
notify_targets: _collectNotifyTargets("sch"),
+12
View File
@@ -1533,6 +1533,18 @@
<select id="sch-template">
<option value="">None</option>
</select>
<label for="sch-persona"
>Persona <span class="label-hint">optional</span></label
>
<select id="sch-persona">
<option value="">Default persona</option>
</select>
<label for="sch-project"
>Project <span class="label-hint">optional</span></label
>
<select id="sch-project">
<option value="">No project</option>
</select>
<label for="sch-message">Initial message</label>
<textarea
id="sch-message"
+10
View File
@@ -1800,6 +1800,8 @@ class PostgreSQLBackend:
next_run: str,
skill: str = "",
notify_targets: str = "[]",
persona: str = "",
project_id: str = "",
) -> None:
from sqlalchemy.dialects import postgresql
@@ -1820,6 +1822,8 @@ class PostgreSQLBackend:
auto_approve=1 if auto_approve else 0,
auto_approve_tools=",".join(auto_approve_tools),
skill=skill,
persona=persona,
project_id=project_id,
notify_targets=notify_targets,
enabled=1,
created_by=created_by,
@@ -1862,8 +1866,14 @@ class PostgreSQLBackend:
"auto_approve",
"auto_approve_tools",
"skill",
"persona",
"project_id",
"notify_targets",
"enabled",
# created_by is only ever set by the update handler adopting an
# orphaned (pre-fix "") schedule's owner from auth_result — never
# sourced from the request body, so this is not a spoofing surface.
"created_by",
"last_run",
"next_run",
"updated",
+7 -1
View File
@@ -1058,8 +1058,14 @@ class StorageBackend(Protocol):
next_run: str,
skill: str = "",
notify_targets: str = "[]",
persona: str = "",
project_id: str = "",
) -> None:
"""Create a scheduled task. No-op if task_id already exists."""
"""Create a scheduled task. No-op if task_id already exists.
``persona`` (slug) and ``project_id`` are stamped onto the workstream
each firing creates; empty = kind-default persona / no project.
"""
...
def get_scheduled_task(self, task_id: str) -> dict[str, Any] | None:
+8
View File
@@ -258,6 +258,14 @@ scheduled_tasks = sa.Table(
sa.Column("auto_approve", sa.Integer, nullable=False, server_default="0"),
sa.Column("auto_approve_tools", sa.Text, nullable=False, server_default=""),
sa.Column("skill", sa.Text, nullable=False, server_default=""),
# persona/project_id: slug + project stamped onto the workstream each firing
# creates. Empty = "kind default persona" / "no project", mirroring how an
# empty model/skill means "use the default". Passed verbatim to
# create_workstream at dispatch, where the node resolves the persona and
# gates the project attach (scheduler.py::_dispatch_to_node). Added in
# migration 066.
sa.Column("persona", sa.Text, nullable=False, server_default=""),
sa.Column("project_id", sa.Text, nullable=False, server_default=""),
sa.Column("notify_targets", 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=""),
+10
View File
@@ -1945,6 +1945,8 @@ class SQLiteBackend:
next_run: str,
skill: str = "",
notify_targets: str = "[]",
persona: str = "",
project_id: str = "",
) -> None:
now = datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%S")
@@ -1964,6 +1966,8 @@ class SQLiteBackend:
"auto_approve": 1 if auto_approve else 0,
"auto_approve_tools": ",".join(auto_approve_tools),
"skill": skill,
"persona": persona,
"project_id": project_id,
"notify_targets": notify_targets,
"enabled": 1,
"created_by": created_by,
@@ -2005,8 +2009,14 @@ class SQLiteBackend:
"auto_approve",
"auto_approve_tools",
"skill",
"persona",
"project_id",
"notify_targets",
"enabled",
# created_by is only ever set by the update handler adopting an
# orphaned (pre-fix "") schedule's owner from auth_result — never
# sourced from the request body, so this is not a spoofing surface.
"created_by",
"last_run",
"next_run",
"updated",
@@ -0,0 +1,43 @@
"""Add persona + project settings to scheduled tasks.
A scheduled task dispatches a fresh workstream every firing. Until now it
could pin the model and skill of that workstream but not its **persona** or
**project** the two levers a manually-created workstream already carries
(``workstreams.persona`` from migration 063, ``workstreams.project_id`` from
062). These two columns close that gap so a schedule can run under, say, the
``researcher`` persona attached to a specific project's memory bucket.
Both are ``Text NOT NULL DEFAULT ''`` following the ``scheduled_tasks``
convention (``model``/``skill`` use the same shape): empty = "kind default
persona" / "no project", exactly as an empty model means "default model". The
values are stamped verbatim onto ``create_workstream`` at dispatch, where the
node resolves the persona for the workstream kind and gates the project attach
(``console/scheduler.py::_dispatch_to_node`` ``/v1/api/workstreams/new``);
nothing is resolved or enforced at migration time. Existing rows migrate
cleanly to the empty default byte-identical dispatch behaviour to pre-066.
Additive and reversible.
Revision ID: 066
Revises: 065
Create Date: 2026-07-08
"""
import sqlalchemy as sa
from alembic import op
revision = "066"
down_revision = "065"
branch_labels = None
depends_on = None
def upgrade() -> None:
with op.batch_alter_table("scheduled_tasks") as batch_op:
batch_op.add_column(sa.Column("persona", sa.Text, nullable=False, server_default=""))
batch_op.add_column(sa.Column("project_id", sa.Text, nullable=False, server_default=""))
def downgrade() -> None:
with op.batch_alter_table("scheduled_tasks") as batch_op:
batch_op.drop_column("project_id")
batch_op.drop_column("persona")
+10
View File
@@ -557,6 +557,8 @@ class AsyncTurnstoneConsole(_BaseClient):
model: str = "",
auto_approve: bool = False,
auto_approve_tools: list[str] | None = None,
persona: str = "",
project_id: str = "",
enabled: bool = True,
) -> ScheduleInfo:
body: dict[str, Any] = {
@@ -577,6 +579,10 @@ class AsyncTurnstoneConsole(_BaseClient):
body["model"] = model
if auto_approve_tools:
body["auto_approve_tools"] = auto_approve_tools
if persona:
body["persona"] = persona
if project_id:
body["project_id"] = project_id
return await self._request(
"POST", "/v1/api/admin/schedules", json_body=body, response_model=ScheduleInfo
)
@@ -600,6 +606,8 @@ class AsyncTurnstoneConsole(_BaseClient):
initial_message: Any = _UNSET,
auto_approve: Any = _UNSET,
auto_approve_tools: Any = _UNSET,
persona: Any = _UNSET,
project_id: Any = _UNSET,
enabled: Any = _UNSET,
) -> ScheduleInfo:
body: dict[str, Any] = {}
@@ -614,6 +622,8 @@ class AsyncTurnstoneConsole(_BaseClient):
("initial_message", initial_message),
("auto_approve", auto_approve),
("auto_approve_tools", auto_approve_tools),
("persona", persona),
("project_id", project_id),
("enabled", enabled),
]:
if val is not _UNSET:
+7
View File
@@ -112,6 +112,7 @@ class AsyncTurnstoneServer(_BaseClient):
ws_id: str = "",
client_type: str = "",
notify_targets: str = "",
project_id: str = "",
attachments: list[AttachmentUpload] | None = None,
) -> CreateWorkstreamResponse:
"""Create a new workstream.
@@ -127,6 +128,8 @@ class AsyncTurnstoneServer(_BaseClient):
*persona* selects the persona the workstream is created with
(resolved and snapshotted server-side; empty = the kind default).
*project_id* attaches the workstream to a project (validated
server-side against the *user_id*'s membership; empty = none).
"""
body: dict[str, Any] = {}
if name:
@@ -141,6 +144,8 @@ class AsyncTurnstoneServer(_BaseClient):
body["skill"] = skill
if persona:
body["persona"] = persona
if project_id:
body["project_id"] = project_id
if initial_message:
body["initial_message"] = initial_message
if auto_approve_tools:
@@ -634,6 +639,7 @@ class TurnstoneServer:
ws_id: str = "",
client_type: str = "",
notify_targets: str = "",
project_id: str = "",
attachments: list[AttachmentUpload] | None = None,
) -> CreateWorkstreamResponse:
return self._runner.run(
@@ -650,6 +656,7 @@ class TurnstoneServer:
ws_id=ws_id,
client_type=client_type,
notify_targets=notify_targets,
project_id=project_id,
attachments=attachments,
)
)