feat(m5.6): preferred_runner_id — pin tasks to a specific runner at creation

migrations/versions/20260708_ecf318dfdbe1_add_preferred_runner_id_to_tasks.py

@@ -0,0 +1,36 @@
+"""add preferred_runner_id to tasks
+
+Revision ID: ecf318dfdbe1
+Revises: 596be4c1bf2a
+Create Date: 2026-07-08 22:07:23.311486
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = 'ecf318dfdbe1'
+down_revision: str | None = '596be4c1bf2a'
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    with op.batch_alter_table('task', schema=None) as batch_op:
+        batch_op.add_column(sa.Column('preferred_runner_id', sa.String(), nullable=True))
+
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    with op.batch_alter_table('task', schema=None) as batch_op:
+        batch_op.drop_column('preferred_runner_id')
+
+    # ### end Alembic commands ###

src/panopticon/client.py

@@ -148,10 +148,13 @@ def create_task(
         memo: str | None = None,
         *,
         initial_prompt: str | None = None,
+        preferred_runner_id: str | None = None,
     ) -> JsonObj:
         body: JsonObj = {"repo_id": repo_id, "workflow": workflow, "memo": memo}
         if initial_prompt is not None:
             body["initial_prompt"] = initial_prompt
+        if preferred_runner_id is not None:
+            body["preferred_runner_id"] = preferred_runner_id
         return cast(JsonObj, self._json(self._http.post("/tasks", json=body)))
 
     def set_slug(self, task_id: str, slug: str) -> JsonObj:

src/panopticon/core/models.py

@@ -258,6 +258,10 @@ class Task:
     #: or have it respawned. Distinct from liveness — a claimed task whose container died is
     #: "claimed but down".
     claimed_by: str | None = None
+    #: The runner that should claim this task. ``None`` and ``"local"`` are equivalent — only the
+    #: ``"local"`` runner may claim the task. Any other value restricts claiming to that runner id.
+    #: Set at creation time (from the dashboard's runner selector); ``None`` until then.
+    preferred_runner_id: str | None = None
     #: Cumulative **cost-weighted** tokens the ``claude`` agent in this task's container has used,
     #: expressed in input-equivalent units (cache-reads ≈0.1×, output ≈5×). The container's Stop
     #: hook reports it via :meth:`TaskService.set_tokens_used`, recomputing the session total each

src/panopticon/sessionservice/spawner.py

@@ -124,6 +124,9 @@ def spawn_one(self, task: JsonObj) -> str | None:
         host daemon's per-task isolation still applies but the failure is visible, not silent."""
         if task["state"] in TERMINAL_LABELS or task.get("claimed_by"):
             return None
+        effective_runner = task.get("preferred_runner_id") or "local"
+        if effective_runner != self._runner_id:
+            return None  # task is reserved for a different runner
         try:
             self._client.claim(task["id"], self._runner_id)
         except httpx.HTTPStatusError as exc:
@@ -325,12 +328,17 @@ def _compose_image(self, workflow: str, repo: JsonObj) -> str | None:
         return self._images.build(workflow, repo["id"], layers, verbose=True)
 
 
-def spawnable_tasks(client: TaskServiceClient) -> Callable[[], list[JsonObj]]:
-    """This host's spawn candidates: unclaimed, non-terminal tasks (the runner claims-then-spawns).
+def spawnable_tasks(client: TaskServiceClient, runner_id: str) -> Callable[[], list[JsonObj]]:
+    """This runner's spawn candidates: unclaimed, non-terminal tasks whose effective preferred
+    runner matches ``runner_id``.
 
-    For M1 (single host) that's every such task the service knows; scoping to this runner's own
-    assignments is an M5 refinement.
+    ``preferred_runner_id=None`` is treated as ``"local"`` — only the local runner claims
+    unaddressed tasks; a remote runner only picks up tasks explicitly assigned to it.
     """
-    return lambda: [
-        t for t in client.list_tasks() if not t["claimed_by"] and t["state"] not in TERMINAL_LABELS
-    ]
+    def _is_spawnable(t: JsonObj) -> bool:
+        if t["claimed_by"] or t["state"] in TERMINAL_LABELS:
+            return False
+        effective = t.get("preferred_runner_id") or "local"
+        return effective == runner_id
+
+    return lambda: [t for t in client.list_tasks() if _is_spawnable(t)]

src/panopticon/taskservice/api.py

@@ -76,6 +76,7 @@ class TaskSummaryOut(BaseModel):
     branch: str | None
     clone: str | None
     claimed_by: str | None
+    preferred_runner_id: str | None = None
     tokens_used: int | None
     token_estimate: int | None
     governor_task_id: str | None = None
@@ -103,6 +104,7 @@ class TaskOut(BaseModel):
     branch: str | None
     clone: str | None
     claimed_by: str | None  # the runner that owns this task (the spawn gate), or None
+    preferred_runner_id: str | None = None  # the runner restricted to claiming this task (None = local)
     tokens_used: int | None  # cost-weighted input-equivalent tokens used (None until reported)
     token_estimate: int | None  # the agent's forecast of total tokens (set in planning; None until then)
     governor_task_id: str | None = None  # the task that oversees this one, or None for ungoverned tasks
@@ -183,6 +185,7 @@ class CreateTaskIn(BaseModel):
     initial_prompt: str | None = None
     artifacts: dict[str, str] | None = None
     depends_on_task_ids: list[str] = []
+    preferred_runner_id: str | None = None
 
 
 class DependenciesIn(BaseModel):
@@ -471,6 +474,7 @@ async def create_task(body: CreateTaskIn) -> TaskOut:
                 initial_prompt=body.initial_prompt,
                 artifacts=body.artifacts,
                 depends_on_task_ids=body.depends_on_task_ids or None,
+                preferred_runner_id=body.preferred_runner_id,
             )
         )
 

src/panopticon/taskservice/service.py

@@ -255,6 +255,7 @@ async def create_task(
         initial_prompt: str | None = None,
         artifacts: dict[str, str] | None = None,
         depends_on_task_ids: list[str] | None = None,
+        preferred_runner_id: str | None = None,
     ) -> Task:
         repo = await self.get_repo(repo_id)  # ensure exists (raises NotFound)
         if governor_task_id is not None:
@@ -267,6 +268,7 @@ async def create_task(
         now = self._clock()
         task = wf.start_task(self._id(), repo_id, at=now, memo=memo, initial_prompt=initial_prompt)
         task.governor_task_id = governor_task_id
+        task.preferred_runner_id = preferred_runner_id
         task.updated_at = now  # creation time = first mutation
         await self._store.create_task(task)
         _log.info("task %s: created (workflow=%s, repo=%s)", task.id, workflow_name, repo_id)
@@ -563,10 +565,18 @@ async def claim(self, task_id: str, runner_id: str) -> Task:
         Compare-and-set: succeeds if the task is unclaimed (idempotent if this runner already holds
         it); raises :class:`AlreadyClaimed` if a different runner does. The store is the single
         writer, so the check-and-set is serialized.
+
+        ``preferred_runner_id`` (``None`` treated as ``"local"``) gates which runner may claim:
+        only the designated runner is allowed, so a remote runner cannot steal a local-only task.
         """
         task = await self.get_task(task_id)
         if task.claimed_by not in (None, runner_id):
             raise AlreadyClaimed(f"task {task_id!r} is already claimed by {task.claimed_by!r}")
+        effective = task.preferred_runner_id or "local"
+        if effective != runner_id:
+            raise AlreadyClaimed(
+                f"task {task_id!r} is reserved for runner {effective!r}; {runner_id!r} may not claim it"
+            )
         task.claimed_by = runner_id
         self.clear_lifecycle(task_id)  # drop any stale phase from a prior owner; this spawn re-reports
         await self._save_task(task)

src/panopticon/taskservice/store_sqlalchemy.py

@@ -124,6 +124,7 @@ class _TaskRow(_Base):
     branch: Mapped[str | None] = mapped_column(default=None)
     clone: Mapped[str | None] = mapped_column(default=None)
     claimed_by: Mapped[str | None] = mapped_column(default=None)
+    preferred_runner_id: Mapped[str | None] = mapped_column(default=None)
     tokens_used: Mapped[int | None] = mapped_column(default=None)
     token_estimate: Mapped[int | None] = mapped_column(default=None)
     governor_task_id: Mapped[str | None] = mapped_column(ForeignKey("task.id"), default=None)
@@ -151,6 +152,7 @@ def to_domain(self) -> Task:
             branch=self.branch,
             clone=self.clone,
             claimed_by=self.claimed_by,
+            preferred_runner_id=self.preferred_runner_id,
             tokens_used=self.tokens_used,
             token_estimate=self.token_estimate,
             governor_task_id=self.governor_task_id,
@@ -175,6 +177,7 @@ def from_domain(cls, task: Task) -> _TaskRow:
             branch=task.branch,
             clone=task.clone,
             claimed_by=task.claimed_by,
+            preferred_runner_id=task.preferred_runner_id,
             tokens_used=task.tokens_used,
             token_estimate=task.token_estimate,
             governor_task_id=task.governor_task_id,
@@ -381,6 +384,7 @@ async def _update_task(self, task: Task, stored: Sequence[HistoryEntry]) -> None
             row.branch = task.branch
             row.clone = task.clone
             row.claimed_by = task.claimed_by
+            row.preferred_runner_id = task.preferred_runner_id
             row.tokens_used = task.tokens_used
             row.token_estimate = task.token_estimate
             row.governor_task_id = task.governor_task_id

src/panopticon/terminal/dashboard.py

@@ -1365,11 +1365,36 @@ def create(result: "tuple[str, bool] | None") -> None:
                     stripped = memo_text.strip()
                     if _apply_memo_filter(stripped):
                         return
-                    if auto_submit and stripped:
-                        self._client.create_task(repo, workflow, stripped, initial_prompt=stripped)
+
+                    def _submit(runner_id: str | None) -> None:
+                        if auto_submit and stripped:
+                            self._client.create_task(
+                                repo, workflow, stripped,
+                                initial_prompt=stripped, preferred_runner_id=runner_id,
+                            )
+                        else:
+                            self._client.create_task(
+                                repo, workflow, stripped or None, preferred_runner_id=runner_id,
+                            )
+                        self.action_refresh()
+
+                    live_runners = self._client.live_runners()
+                    if len(live_runners) >= 2:
+                        # Sort: "local" first, then alphabetically by id.
+                        sorted_runners = sorted(
+                            live_runners,
+                            key=lambda r: ("" if r["id"] == "local" else r["id"]),
+                        )
+                        runner_ids = [r["id"] for r in sorted_runners]
+
+                        def pick_runner(runner_id: str | None) -> None:
+                            _submit(runner_id)
+
+                        self.push_screen(ChoiceScreen("runner", runner_ids), pick_runner)
                     else:
-                        self._client.create_task(repo, workflow, stripped or None)
-                    self.action_refresh()
+                        # 0 or 1 runners: no selector — use the single live runner, or None (local).
+                        preferred = live_runners[0]["id"] if len(live_runners) == 1 else None
+                        _submit(preferred)
 
                 self.push_screen(MemoScreen(auto_submit_default), create)
 

tests/test_api.py

@@ -259,18 +259,19 @@ def test_set_turn_and_blocked(client: TestClient) -> None:
 
 
 def test_claim_release_over_rest(client: TestClient) -> None:
-    task_id = _new_task(client)
+    resp = client.post("/tasks", json={"repo_id": "r1", "workflow": "spike", "preferred_runner_id": "host-1"})
+    task_id = resp.json()["id"]
     assert client.get(f"/tasks/{task_id}").json()["claimed_by"] is None
 
     claimed = client.put(f"/tasks/{task_id}/claim", json={"runner_id": "host-1"})
     assert claimed.status_code == 200 and claimed.json()["claimed_by"] == "host-1"
 
-    # a different runner is refused with 409 while it's held
+    # a different runner is refused with 409 (already claimed by host-1)
     assert client.put(f"/tasks/{task_id}/claim", json={"runner_id": "host-2"}).status_code == 409
 
-    # release frees it; another runner can then claim
+    # release frees it; the preferred runner can reclaim
     assert client.delete(f"/tasks/{task_id}/claim").json()["claimed_by"] is None
-    assert client.put(f"/tasks/{task_id}/claim", json={"runner_id": "host-2"}).json()["claimed_by"] == "host-2"
+    assert client.put(f"/tasks/{task_id}/claim", json={"runner_id": "host-1"}).json()["claimed_by"] == "host-1"
 
 
 # -- artifacts ----------------------------------------------------------------------

tests/test_dashboard.py

@@ -174,13 +174,17 @@ def list_workflows_for_repo(self, repo_id: str) -> list[dict[str, str]]:
     def list_operations(self, task_id: str) -> dict[str, str]:
         return self._operations
 
+    def live_runners(self) -> list[dict[str, Any]]:
+        return []  # 0 live runners by default → no runner selector shown
+
     def create_task(
         self,
         repo_id: str,
         workflow: str,
         memo: str | None = None,
         *,
         initial_prompt: str | None = None,
+        preferred_runner_id: str | None = None,
     ) -> dict[str, Any]:
         self.created.append((repo_id, workflow, memo, initial_prompt))
         return {"id": "new"}

tests/test_host.py

@@ -310,7 +310,7 @@ def test_run_host_spawns_then_provisions_end_to_end(tmp_path: Path) -> None:
     asyncio.run(service.create_repo(Repo(id="r1", name="acme/widgets", git_url="https://forge/r1.git", default_base="trunk")))
     with TestClient(create_app(service)) as http:
         client = TaskServiceClient(http)
-        task_id = client.create_task("r1", "spike")["id"]
+        task_id = client.create_task("r1", "spike", preferred_runner_id="host-1")["id"]
         runner = _FakeRunner()
 
         def one_pass() -> Callable[[], bool]:

tests/test_runner_liveness.py

@@ -110,8 +110,8 @@ def test_reclaim_releases_only_the_dead_runners_non_terminal_claims(
 ) -> None:
     service, base = served
     client = TaskServiceClient(httpx.Client(base_url=base, timeout=10.0))
-    mine = client.create_task("r1", "spike")["id"]
-    other = client.create_task("r1", "spike")["id"]
+    mine = client.create_task("r1", "spike", preferred_runner_id="host-dead")["id"]
+    other = client.create_task("r1", "spike", preferred_runner_id="host-live")["id"]
     client.claim(mine, "host-dead")
     client.claim(other, "host-live")