diff --git a/tests/test_migration_066.py b/tests/test_migration_066.py
new file mode 100644
index 00000000..cc19a9bc
--- /dev/null
+++ b/tests/test_migration_066.py
@@ -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()
diff --git a/tests/test_schedule_api.py b/tests/test_schedule_api.py
index ad9b82de..d2b2ade9 100644
--- a/tests/test_schedule_api.py
+++ b/tests/test_schedule_api.py
@@ -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"]
diff --git a/tests/test_scheduled_tasks_storage.py b/tests/test_scheduled_tasks_storage.py
index 8f285eed..898601f3 100644
--- a/tests/test_scheduled_tasks_storage.py
+++ b/tests/test_scheduled_tasks_storage.py
@@ -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
 
diff --git a/tests/test_scheduler.py b/tests/test_scheduler.py
index 5f277672..ad1e50fb 100644
--- a/tests/test_scheduler.py
+++ b/tests/test_scheduler.py
@@ -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
 
diff --git a/tests/test_sdk_console.py b/tests/test_sdk_console.py
index b3054959..61aa523c 100644
--- a/tests/test_sdk_console.py
+++ b/tests/test_sdk_console.py
@@ -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
diff --git a/tests/test_sdk_server.py b/tests/test_sdk_server.py
index edc23855..17d6c0f6 100644
--- a/tests/test_sdk_server.py
+++ b/tests/test_sdk_server.py
@@ -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
diff --git a/turnstone/api/console_schemas.py b/turnstone/api/console_schemas.py
index d8cdf301..ce5a1634 100644
--- a/turnstone/api/console_schemas.py
+++ b/turnstone/api/console_schemas.py
@@ -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)"
     )
diff --git a/turnstone/api/schemas.py b/turnstone/api/schemas.py
index 8ece8b16..f57e20ce 100644
--- a/turnstone/api/schemas.py
+++ b/turnstone/api/schemas.py
@@ -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 = ""
diff --git a/turnstone/console/scheduler.py b/turnstone/console/scheduler.py
index 7767dacb..f66d60ba 100644
--- a/turnstone/console/scheduler.py
+++ b/turnstone/console/scheduler.py
@@ -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
diff --git a/turnstone/console/server.py b/turnstone/console/server.py
index e8d1797d..c051e2d3 100644
--- a/turnstone/console/server.py
+++ b/turnstone/console/server.py
@@ -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:
diff --git a/turnstone/console/static/admin.js b/turnstone/console/static/admin.js
index 2db846c5..3352211c 100644
--- a/turnstone/console/static/admin.js
+++ b/turnstone/console/static/admin.js
@@ -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"),
diff --git a/turnstone/console/static/index.html b/turnstone/console/static/index.html
index e0ad0101..9b2fc693 100644
--- a/turnstone/console/static/index.html
+++ b/turnstone/console/static/index.html
@@ -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"
diff --git a/turnstone/core/storage/_postgresql.py b/turnstone/core/storage/_postgresql.py
index b9313132..37f081dd 100644
--- a/turnstone/core/storage/_postgresql.py
+++ b/turnstone/core/storage/_postgresql.py
@@ -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",
diff --git a/turnstone/core/storage/_protocol.py b/turnstone/core/storage/_protocol.py
index eb49cfe2..86c2aa94 100644
--- a/turnstone/core/storage/_protocol.py
+++ b/turnstone/core/storage/_protocol.py
@@ -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:
diff --git a/turnstone/core/storage/_schema.py b/turnstone/core/storage/_schema.py
index 412707ce..80096031 100644
--- a/turnstone/core/storage/_schema.py
+++ b/turnstone/core/storage/_schema.py
@@ -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=""),
diff --git a/turnstone/core/storage/_sqlite.py b/turnstone/core/storage/_sqlite.py
index e0c6bd8a..976989f6 100644
--- a/turnstone/core/storage/_sqlite.py
+++ b/turnstone/core/storage/_sqlite.py
@@ -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",
diff --git a/turnstone/core/storage/migrations/versions/066_scheduled_task_persona_project.py b/turnstone/core/storage/migrations/versions/066_scheduled_task_persona_project.py
new file mode 100644
index 00000000..f168fb58
--- /dev/null
+++ b/turnstone/core/storage/migrations/versions/066_scheduled_task_persona_project.py
@@ -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")
diff --git a/turnstone/sdk/console.py b/turnstone/sdk/console.py
index aa9d0e20..93a41dda 100644
--- a/turnstone/sdk/console.py
+++ b/turnstone/sdk/console.py
@@ -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:
diff --git a/turnstone/sdk/server.py b/turnstone/sdk/server.py
index cb41f40e..b9de6063 100644
--- a/turnstone/sdk/server.py
+++ b/turnstone/sdk/server.py
@@ -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,
             )
         )