[eric] merge eric/cloud-runs: headless backend, ephemeral runner, credential custody

backend/apps/agents/manager/permissions/build_effective_tool_lists.py

@@ -18,6 +18,7 @@ from backend.apps.tools_lib.tools_lib import (
     load_all_tools as load_all_tools,
     sanitize_server_name as sanitize_server_name,
 )
+from backend.config.headless import apply_headless_denies
 
 # Mutation/exec tools a read-only session must never reach: Edit (rewrites files), Bash (rm/mv/overwrite),
 # NotebookEdit (rewrites notebooks). Write is intentionally NOT here, the audit needs its one report.
@@ -33,6 +34,8 @@ def build_effective_tool_lists(
     browser_delegation_tools: List[str],
     invoke_agent_tools: List[str],
 ) -> Tuple[List[str], List[str]]:
+    # Same shadow the server registration takes: headless, the renderer-bound built-ins go straight onto disallowed instead of being offered and failing when called.
+    builtin_perms = apply_headless_denies(builtin_perms)
     effective_allowed = [
         t for t in session.allowed_tools
         if t in FULL_TOOLS and builtin_perms.get(t, "always_allow") == "always_allow"

backend/apps/agents/manager/permissions/workflow_approval.py

@@ -16,6 +16,7 @@ from typeguard import typechecked
 from backend.apps.agents.manager.permissions.ApprovalDecision import ApprovalDecision
 from backend.apps.agents.manager.permissions.decision import request_user_approval
 from backend.apps.agents.manager.streaming.HookContext import HookContext
+from backend.config.headless import is_headless
 
 logger = logging.getLogger(__name__)
 
@@ -90,7 +91,8 @@ async def resolve_ask(
 ) -> ApprovalDecision:
     """Resolve an 'ask' policy. On a workflow run, reuse a remembered decision
     (this step first, then the workflow-level fallback) instead of prompting, and
-    persist any fresh non-sensitive answer so later fires don't re-ask. Shared by
+    persist any fresh non-sensitive answer so later fires don't re-ask. Headless,
+    anything still unresolved is denied on the spot instead of prompting. Shared by
     both gates so they can't disagree (and so the first one's answer is reused by
     the second within the same call)."""
     mem = p_approval_memory.get(ctx.session_id)
@@ -113,6 +115,12 @@ async def resolve_ask(
         if prior == "deny":
             note_tool_used(ctx.session_id, tool_name, False)
             return ApprovalDecision(behavior="deny", message="Denied by a remembered workflow permission")
+    # Headless there is no one to ask, so the request would just be broadcast into the void and come back denied ten minutes later; deny now, but only after the remembered decisions above got their say.
+    if is_headless():
+        return ApprovalDecision(
+            behavior="deny",
+            message="This run is headless, so nobody can approve a tool that asks. Use a tool that doesn't need approval, or report what you'd need permission for.",
+        )
     timeout = mem.ask_timeout if mem is not None else 600.0
     decision = await request_user_approval(
         ctx.session, ctx.session_id, tool_name, tool_input, ctx.builtin_perms,

backend/apps/agents/manager/register_builtin_mcp_servers.py

@@ -12,6 +12,7 @@ from typeguard import typechecked
 
 from backend.apps.agents.core.models import AgentSession
 from backend.auth import get_auth_token
+from backend.config.headless import apply_headless_denies
 
 
 @typechecked
@@ -24,6 +25,8 @@ def register_builtin_mcp_servers(
 ) -> Tuple[List[str], List[str]]:
     import backend.apps.agents as p_agents_pkg
     agents_dir = os.path.dirname(p_agents_pkg.__file__)
+    # Headless has no renderer for a webview or a UI component, so we shadow the map once here and let the existing deny short-circuits skip those servers; nothing below may read the un-shadowed one.
+    builtin_perms = apply_headless_denies(builtin_perms)
     browser_delegation_tools = ["CreateBrowserAgent", "BrowserAgent", "BrowserAgents", "AppAgent"]
     browser_all_denied = all(
         builtin_perms.get(t, "always_allow") == "deny"

backend/apps/nine_router/credential_lease.py

@@ -0,0 +1,209 @@
+"""Handing custody of a rotating provider credential between this device and the cloud.
+
+Cloud runs execute a user's workflow while their laptop is off, using the user's OWN subscription.
+That means our server needs to refresh their token, and providers issue a new refresh token on every
+refresh while treating a replayed one as theft, revoking the entire grant family. So a credential
+gets exactly ONE holder that can rotate it, and the handover has to be ordered so there is never an
+instant where both sides can.
+
+The order is strip-then-upload, never the reverse:
+  - Strip first, then upload: worst case nobody can rotate for a moment, which is harmless because
+    the access token stays valid for hours. We restore on failure.
+  - Upload first, then strip: if the strip fails, BOTH sides hold a rotating token, which is the
+    exact incident this whole design exists to prevent.
+
+The one genuinely ambiguous case is an upload that times out after the server already committed.
+Restoring blindly there would recreate the two-holder state, so we ask the server who owns it
+before deciding.
+"""
+
+import logging
+from datetime import datetime
+from typing import Literal, Optional
+
+import httpx
+from pydantic import BaseModel, ConfigDict
+from typeguard import typechecked
+
+from backend.apps.nine_router import credential_store
+from backend.apps.settings.credentials import proxy_auth
+from backend.apps.settings.store import load_settings
+
+logger = logging.getLogger(__name__)
+
+P_TIMEOUT_S = 20.0
+
+LeaseStatus = Literal[
+    "leased",
+    "released",
+    "refreshed",
+    "not_signed_in",
+    "no_such_connection",
+    "not_rotatable",
+    "cloud_rejected",
+    "local_write_failed",
+    "ownership_unknown",
+]
+
+
+class LeaseOutcome(BaseModel):
+    model_config = ConfigDict(validate_assignment=True)
+
+    status: LeaseStatus
+    detail: str = ""
+
+
+@typechecked
+def p_cloud() -> Optional[tuple[str, str]]:
+    """(bearer, base_url) for the signed-in user, or None when there is nothing to talk to."""
+    token, base = proxy_auth(load_settings())
+    if not token or not base:
+        return None
+    return (token, base)
+
+
+@typechecked
+async def p_lease_is_cloud_owned(connection_id: str) -> Optional[bool]:
+    """True/False if we can read ownership, None if we cannot tell. The None case is load-bearing:
+    guessing here is how you end up with two rotators."""
+    cloud = p_cloud()
+    if cloud is None:
+        return None
+    token, base = cloud
+    try:
+        async with httpx.AsyncClient(timeout=P_TIMEOUT_S) as client:
+            r = await client.get(
+                f"{base}/api/credentials/status",
+                headers={"Authorization": f"Bearer {token}"},
+            )
+        if r.status_code != 200:
+            return None
+        for lease in r.json().get("leases") or []:
+            if lease.get("connection_id") == connection_id:
+                return lease.get("owner") == "cloud"
+        return False
+    except (httpx.HTTPError, ValueError):
+        return None
+
+
+@typechecked
+async def lease_to_cloud(connection_id: str) -> LeaseOutcome:
+    """Give the cloud sole custody so it can run this user's workflows while the laptop is off."""
+    cloud = p_cloud()
+    if cloud is None:
+        return LeaseOutcome(status="not_signed_in")
+    token, base = cloud
+
+    cred = credential_store.read_credential(connection_id)
+    if cred is None:
+        return LeaseOutcome(status="no_such_connection")
+    if not cred.refresh_token:
+        # Already stripped, or an api-key row. Either way there is no rotating secret to hand over.
+        return LeaseOutcome(status="not_rotatable")
+
+    refresh_token = cred.refresh_token
+    if not await credential_store.apply_to_connection(connection_id, changes={}, drop=["refreshToken"]):
+        return LeaseOutcome(status="local_write_failed")
+
+    payload = {
+        "connection_id": connection_id,
+        "provider": cred.provider,
+        "access_token": cred.access_token,
+        "refresh_token": refresh_token,
+        "expires_at": expires_ms(cred.expires_at),
+    }
+    try:
+        async with httpx.AsyncClient(timeout=P_TIMEOUT_S) as client:
+            r = await client.post(
+                f"{base}/api/credentials/lease",
+                headers={"Authorization": f"Bearer {token}"},
+                json=payload,
+            )
+        if r.status_code == 200:
+            return LeaseOutcome(status="leased")
+        await p_restore(connection_id, refresh_token)
+        return LeaseOutcome(status="cloud_rejected", detail=f"HTTP {r.status_code}")
+    except httpx.HTTPError as exc:
+        # The request may still have committed server-side, so ask before putting the token back.
+        owned = await p_lease_is_cloud_owned(connection_id)
+        if owned is True:
+            return LeaseOutcome(status="leased", detail="upload reported an error but the lease exists")
+        if owned is False:
+            await p_restore(connection_id, refresh_token)
+            return LeaseOutcome(status="cloud_rejected", detail=str(exc))
+        logger.error("lease upload outcome unknown for %s; leaving the token off this device", connection_id)
+        return LeaseOutcome(status="ownership_unknown", detail=str(exc))
+
+
+@typechecked
+async def p_restore(connection_id: str, refresh_token: str) -> None:
+    if not await credential_store.apply_to_connection(
+        connection_id, changes={"refreshToken": refresh_token}, drop=[]
+    ):
+        logger.error("could not restore the local refresh token for %s; the user must reconnect", connection_id)
+
+
+@typechecked
+def expires_ms(expires_at: Optional[str]) -> int:
+    """9Router stores an ISO string; the cloud wants unix ms. Unparseable reads as already expired,
+    which makes the server refresh on first use instead of trusting a bad clock."""
+    if not expires_at:
+        return 0
+    try:
+        return int(datetime.fromisoformat(expires_at.replace("Z", "+00:00")).timestamp() * 1000)
+    except ValueError:
+        return 0
+
+
+@typechecked
+async def release_to_device(connection_id: str) -> LeaseOutcome:
+    """Take custody back. The server hands the live refresh token home and drops its own copy."""
+    cloud = p_cloud()
+    if cloud is None:
+        return LeaseOutcome(status="not_signed_in")
+    token, base = cloud
+    try:
+        async with httpx.AsyncClient(timeout=P_TIMEOUT_S) as client:
+            r = await client.post(
+                f"{base}/api/credentials/release",
+                headers={"Authorization": f"Bearer {token}"},
+                json={"connection_id": connection_id},
+            )
+    except httpx.HTTPError as exc:
+        return LeaseOutcome(status="cloud_rejected", detail=str(exc))
+    if r.status_code != 200:
+        # 409 means a refresh is mid-exchange; the caller retries rather than taking a doomed token.
+        return LeaseOutcome(status="cloud_rejected", detail=f"HTTP {r.status_code}")
+    body = r.json()
+    ok = await credential_store.apply_to_connection(
+        connection_id,
+        changes={"accessToken": body["access_token"], "refreshToken": body["refresh_token"]},
+        drop=[],
+    )
+    return LeaseOutcome(status="released" if ok else "local_write_failed")
+
+
+@typechecked
+async def pull_access_token(connection_id: str) -> LeaseOutcome:
+    """Get a usable access token for a cloud-owned credential. This is what keeps LOCAL work going
+    once this device can no longer mint one itself."""
+    cloud = p_cloud()
+    if cloud is None:
+        return LeaseOutcome(status="not_signed_in")
+    token, base = cloud
+    try:
+        async with httpx.AsyncClient(timeout=P_TIMEOUT_S) as client:
+            r = await client.get(
+                f"{base}/api/credentials/access",
+                headers={"Authorization": f"Bearer {token}"},
+                params={"connection_id": connection_id},
+            )
+    except httpx.HTTPError as exc:
+        return LeaseOutcome(status="cloud_rejected", detail=str(exc))
+    if r.status_code != 200:
+        return LeaseOutcome(status="cloud_rejected", detail=f"HTTP {r.status_code}")
+    body = r.json()
+    ok = await credential_store.apply_to_connection(
+        connection_id, changes={"accessToken": body["access_token"]}, drop=[]
+    )
+    return LeaseOutcome(status="refreshed" if ok else "local_write_failed")

backend/apps/nine_router/credential_store.py

@@ -0,0 +1,167 @@
+"""Safe read/modify/write of 9Router's on-disk provider credentials.
+
+9Router owns ~/.9router/db.json and rewrites the whole file whenever it refreshes a token or a
+user edits a provider, and its HTTP API has no route that can write an OAuth connection's tokens
+(PUT /api/providers/[id] accepts name/priority/isActive/apiKey only, and apiKey only for apikey
+connections). So the only way to move an OAuth credential is to edit the file, which means we have
+to not race the router for it. Every mutation here happens with the router stopped.
+
+The reason any of this exists: providers hand back a NEW refresh token on every refresh and treat
+a replayed one as theft, revoking the whole grant family. So a credential may have exactly one
+holder that can refresh it. Removing `refreshToken` from a connection is what makes a given
+9Router instance structurally unable to rotate, because its refresh dispatcher bails on a falsy
+refreshToken before it ever calls the provider.
+"""
+
+import asyncio
+import json
+import logging
+import os
+import tempfile
+from typing import Any, Dict, List, Optional
+
+import httpx
+from pydantic import BaseModel, ConfigDict
+from typeguard import typechecked
+
+from backend.apps.nine_router import process
+
+logger = logging.getLogger(__name__)
+
+P_SHUTDOWN_TIMEOUT_S = 5.0
+P_DOWN_POLL_INTERVAL_S = 0.1
+P_DOWN_WAIT_S = 10.0
+
+
+class ProviderCredential(BaseModel):
+    """The transferable half of a 9Router provider connection."""
+
+    model_config = ConfigDict(validate_assignment=True)
+
+    connection_id: str
+    provider: str
+    access_token: str
+    refresh_token: Optional[str] = None
+    expires_at: Optional[str] = None
+
+
+@typechecked
+def db_path() -> str:
+    return os.path.join(process.nine_router_data_dir(), "db.json")
+
+
+@typechecked
+def p_load_db() -> Optional[Dict[str, Any]]:
+    try:
+        with open(db_path(), encoding="utf-8") as f:
+            db = json.load(f)
+        return db if isinstance(db, dict) else None
+    except (OSError, ValueError):
+        logger.warning("could not read 9router db.json", exc_info=True)
+        return None
+
+
+@typechecked
+def p_write_db(db: Dict[str, Any]) -> bool:
+    """Atomic replace at 0600. A half-written db.json costs the user every provider connection."""
+    path = db_path()
+    directory = os.path.dirname(path)
+    handle, temp_path = tempfile.mkstemp(dir=directory, prefix=".db.json.", suffix=".tmp")
+    try:
+        with os.fdopen(handle, "w", encoding="utf-8") as f:
+            json.dump(db, f, indent=2)
+        os.chmod(temp_path, 0o600)
+        os.replace(temp_path, path)
+        return True
+    except OSError:
+        logger.warning("could not write 9router db.json", exc_info=True)
+        try:
+            os.unlink(temp_path)
+        except OSError:
+            pass
+        return False
+
+
+@typechecked
+def read_credential(connection_id: str) -> Optional[ProviderCredential]:
+    """The tokens for one connection, readable whether or not the router is up."""
+    for c in process.read_persisted_connections():
+        if c.get("id") != connection_id:
+            continue
+        access = c.get("accessToken")
+        if not isinstance(access, str) or not access:
+            return None
+        refresh = c.get("refreshToken")
+        expires = c.get("expiresAt")
+        return ProviderCredential(
+            connection_id=connection_id,
+            provider=str(c.get("provider") or ""),
+            access_token=access,
+            refresh_token=refresh if isinstance(refresh, str) and refresh else None,
+            expires_at=expires if isinstance(expires, str) else None,
+        )
+    return None
+
+
+@typechecked
+def list_oauth_connection_ids() -> List[str]:
+    """Connections that carry a rotating credential; apikey rows have nothing to lease."""
+    return [
+        str(c.get("id"))
+        for c in process.read_persisted_connections()
+        if c.get("authType") == "oauth" and c.get("id")
+    ]
+
+
+@typechecked
+async def request_shutdown() -> None:
+    """Ask the router to exit over HTTP. Its own seam so a test can never reach a real router."""
+    try:
+        async with httpx.AsyncClient(timeout=P_SHUTDOWN_TIMEOUT_S, headers=process.cli_auth_headers()) as client:
+            await client.post(f"{process.NINE_ROUTER_API}/shutdown")
+    except (httpx.HTTPError, AttributeError):
+        pass
+
+
+@typechecked
+async def p_stop_router() -> bool:
+    """Down the router however we can reach it. `stop()` alone only kills one we spawned; an
+    adopted port-holder has no handle, so ask it to shut itself down over HTTP first."""
+    await request_shutdown()
+    process.stop()
+    waited = 0.0
+    while waited < P_DOWN_WAIT_S:
+        if not process.is_running():
+            return True
+        await asyncio.sleep(P_DOWN_POLL_INTERVAL_S)
+        waited += P_DOWN_POLL_INTERVAL_S
+    return not process.is_running()
+
+
+@typechecked
+async def apply_to_connection(connection_id: str, changes: Dict[str, Any], drop: List[str]) -> bool:
+    """Set `changes` and delete `drop` on one connection, with the router stopped throughout.
+
+    Refuses to run if the router will not go down, because a concurrent refresh would either lose
+    our edit or, far worse, resurrect a refresh token we are in the middle of handing away.
+    """
+    if not await p_stop_router():
+        logger.error("refusing to edit 9router db.json: router would not stop")
+        return False
+    try:
+        db = p_load_db()
+        if db is None:
+            return False
+        connections = db.get("providerConnections")
+        if not isinstance(connections, list):
+            return False
+        target = next((c for c in connections if isinstance(c, dict) and c.get("id") == connection_id), None)
+        if target is None:
+            logger.error("9router connection %s not found", connection_id)
+            return False
+        target.update(changes)
+        for key in drop:
+            target.pop(key, None)
+        return p_write_db(db)
+    finally:
+        await process.ensure_running()

backend/apps/nine_router/process.py

@@ -17,6 +17,7 @@ import os
 import secrets
 import shutil
 import socket
+import stat
 import subprocess
 import tempfile
 import time
@@ -98,7 +99,7 @@ def is_running() -> bool:
         return False
 
 
-def p_nine_router_data_dir() -> str:
+def nine_router_data_dir() -> str:
     """Where 9Router persists machine-id + auth/cli-secret, the two files we
     hash into the /api/* auth token on 0.4.x. Mirrors 9Router's own default
     (DATA_DIR env, else ~/.9router on unix, %APPDATA%/9router on win) so we read
@@ -115,6 +116,25 @@ def p_nine_router_data_dir() -> str:
     return os.path.join(os.path.expanduser("~"), ".9router")
 
 
+def harden_data_dir_permissions() -> None:
+    """Make the 9Router state dir owner-only. Its db.json holds live subscription access AND refresh
+    tokens in plaintext and 9Router writes it 0644, so on a shared machine any other local account
+    can read them. We tighten the DIRECTORY rather than the file because 9Router rewrites db.json on
+    every token refresh, which would drop a chmod on the file itself within the hour."""
+    if os.name == "nt":
+        return
+    data_dir = nine_router_data_dir()
+    try:
+        if not os.path.isdir(data_dir):
+            return
+        current = stat.S_IMODE(os.stat(data_dir).st_mode)
+        if current & 0o077:
+            os.chmod(data_dir, 0o700)
+            logger.info("tightened 9router data dir from %s to 0700", oct(current))
+    except OSError:
+        logger.warning("could not tighten 9router data dir permissions", exc_info=True)
+
+
 p_cli_token_cache: str | None = None
 
 
@@ -132,7 +152,7 @@ def cli_auth_token() -> str | None:
     if not is_running():
         return None
     try:
-        data_dir = p_nine_router_data_dir()
+        data_dir = nine_router_data_dir()
         try:
             with open(os.path.join(data_dir, "machine-id"), encoding="utf-8") as f:
                 machine_id = f.read().strip()
@@ -180,7 +200,6 @@ def p_find_9router_dir() -> str | None:
     p_is_packaged = os.environ.get("OPENSWARM_PACKAGED") == "1"
 
     if p_is_packaged:
-        import sys
         p_resources = os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))
         p_candidate = os.path.join(p_resources, "router")
         if os.path.isdir(p_candidate):
@@ -353,6 +372,7 @@ async def ensure_running():
         p_start_lock = asyncio.Lock()
     async with p_start_lock:
         await p_ensure_running_impl()
+    harden_data_dir_permissions()
     # Arm both healers the moment the router becomes a live dependency; users who never route through it never spawn them.
     if is_running():
         start_watchdog()
@@ -365,7 +385,7 @@ def read_persisted_connections() -> list[dict]:
     Empty list on any read problem."""
     try:
         import json as p_json
-        with open(os.path.join(p_nine_router_data_dir(), "db.json"), encoding="utf-8") as f:
+        with open(os.path.join(nine_router_data_dir(), "db.json"), encoding="utf-8") as f:
             db = p_json.load(f)
         return [c for c in (db.get("providerConnections") or []) if isinstance(c, dict)]
     except Exception:

backend/apps/workflows/models.py

@@ -91,6 +91,8 @@ class Workflow(BaseModel):
     steps: list[WorkflowStep] = Field(default_factory=list)
     actions: ActionsConfig = Field(default_factory=ActionsConfig)
     schedule: ScheduleConfig = Field(default_factory=ScheduleConfig)
+    # Where a SCHEDULED fire runs. "cloud" hands the timer to our servers outright, so this machine must never fire it nor roll next_run_at, or the same slot runs twice. Manual Run-now always stays local.
+    execution_target: Literal["device", "cloud"] = "device"
     permissions: list[PermissionTier] = Field(
         default_factory=lambda: [PermissionTier(kind="notify")]
     )

backend/apps/workflows/scheduler.py

@@ -289,12 +289,21 @@ def _disable_schedule(wf: Workflow) -> None:
     storage.save_workflow(wf)
 
 
+def p_timer_is_ours(wf: Workflow) -> bool:
+    """Whether this process owns a workflow's timer. Both the fire loop and the sleep math must
+    agree: if only one of them skipped cloud-hosted workflows, an overdue one would never roll
+    forward and the loop would spin at its 1s floor forever."""
+    return wf.execution_target != "cloud"
+
+
 async def _tick() -> None:
     now_utc = datetime.now(timezone.utc)
     if storage.get_paused():
         return
     due: list[Workflow] = []
     for wf in storage.list_workflows():
+        if not p_timer_is_ours(wf):
+            continue
         if not wf.schedule.enabled:
             continue
         if not is_schedule_configured(wf.schedule):
@@ -330,6 +339,8 @@ def seconds_to_next_fire() -> Optional[float]:
     now_utc = datetime.now(timezone.utc)
     soonest: Optional[datetime] = None
     for wf in storage.list_workflows():
+        if not p_timer_is_ours(wf):
+            continue
         if not wf.schedule.enabled:
             continue
         nra = _as_utc(wf.next_run_at)
@@ -452,6 +463,8 @@ def reconcile_on_startup() -> None:
     """
     now_utc = datetime.now(timezone.utc)
     for wf in storage.list_workflows():
+        if not p_timer_is_ours(wf):
+            continue
         if not wf.schedule.enabled:
             wf.next_run_at = None
             storage.save_workflow(wf)

backend/config/headless.py

@@ -0,0 +1,34 @@
+"""Headless mode: the backend running with no Electron renderer, no display, and no human
+(a Linux container). Single source of truth for the flag and for the tools that dead-end at a
+renderer, so they are dropped from the tool surface up front instead of hanging at call time."""
+
+import os
+from typing import Dict, FrozenSet
+
+from typeguard import typechecked
+
+# Each of these ends at the Electron renderer: browser/app delegation drives live webviews, ShowUI (the same gate AskUI rides) draws into the transcript, and AskUserQuestion waits on a person who isn't there.
+HEADLESS_DENIED_TOOLS: FrozenSet[str] = frozenset({
+    "CreateBrowserAgent",
+    "BrowserAgent",
+    "BrowserAgents",
+    "AppAgent",
+    "ShowUI",
+    "AskUserQuestion",
+})
+
+
+@typechecked
+def is_headless() -> bool:
+    """True when the backend was started with OPENSWARM_HEADLESS=1. Read per call rather than
+    frozen at import, so a launcher that sets it late still counts (and tests can flip it)."""
+    return os.environ.get("OPENSWARM_HEADLESS") == "1"
+
+
+@typechecked
+def apply_headless_denies(builtin_perms: Dict[str, str]) -> Dict[str, str]:
+    """The permission map with every renderer-bound tool forced to 'deny' when headless, and the
+    map itself untouched otherwise. Returns a copy so the mode never poisons the live snapshot."""
+    if not is_headless():
+        return builtin_perms
+    return {**builtin_perms, **{name: "deny" for name in HEADLESS_DENIED_TOOLS}}

backend/config/paths.py

@@ -6,8 +6,12 @@ import sys
 P_BACKEND_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
 
 p_is_packaged = os.environ.get("OPENSWARM_PACKAGED") == "1"
+p_data_root_override = os.environ.get("OPENSWARM_DATA_ROOT", "").strip()
 
-if p_is_packaged:
+if p_data_root_override:
+    # A container has no home to speak of, so the runner points this straight at its mounted volume.
+    DATA_ROOT = os.path.abspath(p_data_root_override)
+elif p_is_packaged:
     if sys.platform == "darwin":
         p_app_support = os.path.join(os.path.expanduser("~"), "Library", "Application Support", "OpenSwarm")
     elif sys.platform == "win32":

backend/tests/test_credential_lease.py

@@ -0,0 +1,256 @@
+"""Custody handover of a rotating credential, ordered so two holders is unrepresentable.
+
+Providers issue a new refresh token on every refresh and treat a replayed one as theft, revoking
+the whole grant family. So the invariant under test is not "the happy path works", it is: at no
+point does BOTH this device and the cloud hold a token that can rotate.
+
+Run:
+    cd backend && .venv/bin/python -m pytest tests/test_credential_lease.py -v
+"""
+
+from __future__ import annotations
+
+import json
+from typing import Any, Dict, List
+
+import httpx
+import pytest
+
+import backend.apps.nine_router.credential_lease as lease
+import backend.apps.nine_router.credential_store as store
+from backend.apps.nine_router import process
+
+P_CONNECTION = {
+    "id": "conn-1",
+    "provider": "claude",
+    "authType": "oauth",
+    "accessToken": "access-old",
+    "refreshToken": "refresh-live",
+    "expiresAt": "2026-08-01T00:00:00.000Z",
+    "isActive": True,
+}
+
+
+@pytest.fixture
+def p_device(tmp_path, monkeypatch):
+    """A stopped-on-demand router with one oauth connection, and a signed-in cloud identity."""
+    data_dir = tmp_path / "9router"
+    data_dir.mkdir()
+    (data_dir / "db.json").write_text(json.dumps({"providerConnections": [dict(P_CONNECTION)]}))
+    monkeypatch.setattr(process, "nine_router_data_dir", lambda: str(data_dir))
+
+    state = {"running": True}
+    monkeypatch.setattr(process, "stop", lambda: state.__setitem__("running", False))
+    monkeypatch.setattr(process, "is_running", lambda: state["running"])
+
+    async def p_ensure() -> None:
+        state["running"] = True
+
+    async def p_no_http() -> None:
+        return None
+
+    monkeypatch.setattr(process, "ensure_running", p_ensure)
+    monkeypatch.setattr(store, "request_shutdown", p_no_http)
+    monkeypatch.setattr(lease, "p_cloud", lambda: ("bearer-xyz", "https://api.example.test"))
+    return state
+
+
+def p_local() -> Dict[str, Any]:
+    db = json.loads(open(store.db_path(), encoding="utf-8").read())
+    return next(c for c in db["providerConnections"] if c["id"] == "conn-1")
+
+
+class FakeClient:
+    """Records calls and replays scripted responses; never touches the network."""
+
+    def __init__(self, script: List[Any], calls: List[Dict[str, Any]]):
+        self.script = script
+        self.calls = calls
+
+    async def __aenter__(self) -> "FakeClient":
+        return self
+
+    async def __aexit__(self, *exc: Any) -> None:
+        return None
+
+    def p_next(self, method: str, url: str, kwargs: Dict[str, Any]) -> Any:
+        self.calls.append({"method": method, "url": url, **kwargs})
+        outcome = self.script.pop(0)
+        if isinstance(outcome, Exception):
+            raise outcome
+        return outcome
+
+    async def post(self, url: str, **kwargs: Any) -> Any:
+        return self.p_next("POST", url, kwargs)
+
+    async def get(self, url: str, **kwargs: Any) -> Any:
+        return self.p_next("GET", url, kwargs)
+
+
+def p_response(status: int, body: Dict[str, Any] | None = None) -> Any:
+    return httpx.Response(status, json=body if body is not None else {})
+
+
+async def p_already_leased(harness: Dict[str, Any]) -> None:
+    """Put the device in the post-handover state: cloud owns the refresh token, device does not."""
+    harness["script"].append(p_response(200, {}))
+    await lease.lease_to_cloud("conn-1")
+    harness["script"].clear()
+    harness["calls"].clear()
+
+
+@pytest.fixture
+def p_cloud_calls(monkeypatch):
+    calls: List[Dict[str, Any]] = []
+    script: List[Any] = []
+
+    def p_factory(*args: Any, **kwargs: Any) -> FakeClient:
+        return FakeClient(script, calls)
+
+    monkeypatch.setattr(lease.httpx, "AsyncClient", p_factory)
+    return {"calls": calls, "script": script}
+
+
+@pytest.mark.asyncio
+async def test_lease_strips_locally_and_uploads(p_device, p_cloud_calls):
+    p_cloud_calls["script"].append(p_response(200, {"owner": "cloud"}))
+
+    result = await lease.lease_to_cloud("conn-1")
+
+    assert result.status == "leased"
+    assert "refreshToken" not in p_local(), "the device must not keep a token it could rotate"
+    assert p_local()["accessToken"] == "access-old", "the access token still has to work locally"
+    sent = p_cloud_calls["calls"][0]["json"]
+    assert sent["refresh_token"] == "refresh-live"
+    assert sent["provider"] == "claude"
+
+
+@pytest.mark.asyncio
+async def test_device_is_stripped_before_the_upload_is_attempted(p_device, p_cloud_calls, monkeypatch):
+    """The ordering IS the safety property. If the upload could run first, a failed strip would
+    leave two live rotators, which is the incident this design exists to prevent."""
+    observed: List[bool] = []
+
+    def p_factory(*args: Any, **kwargs: Any) -> FakeClient:
+        observed.append("refreshToken" in p_local())
+        return FakeClient(p_cloud_calls["script"], p_cloud_calls["calls"])
+
+    p_cloud_calls["script"].append(p_response(200, {}))
+    monkeypatch.setattr(lease.httpx, "AsyncClient", p_factory)
+
+    await lease.lease_to_cloud("conn-1")
+
+    assert observed == [False], "the local token was still present when the upload began"
+
+
+@pytest.mark.asyncio
+async def test_rejected_upload_restores_the_local_token(p_device, p_cloud_calls):
+    p_cloud_calls["script"].append(p_response(500, {}))
+
+    result = await lease.lease_to_cloud("conn-1")
+
+    assert result.status == "cloud_rejected"
+    assert p_local()["refreshToken"] == "refresh-live", "custody never moved, so it must come back"
+
+
+@pytest.mark.asyncio
+async def test_ambiguous_upload_asks_who_owns_it_before_restoring(p_device, p_cloud_calls):
+    """A timeout can mean the server committed anyway. Restoring blindly would recreate exactly the
+    two-holder state, so ownership is checked rather than assumed."""
+    p_cloud_calls["script"].append(httpx.ReadTimeout("boom"))
+    p_cloud_calls["script"].append(
+        p_response(200, {"leases": [{"connection_id": "conn-1", "owner": "cloud"}]})
+    )
+
+    result = await lease.lease_to_cloud("conn-1")
+
+    assert result.status == "leased"
+    assert "refreshToken" not in p_local(), "the cloud owns it; putting it back makes two rotators"
+
+
+@pytest.mark.asyncio
+async def test_ambiguous_upload_restores_when_the_cloud_does_not_have_it(p_device, p_cloud_calls):
+    p_cloud_calls["script"].append(httpx.ReadTimeout("boom"))
+    p_cloud_calls["script"].append(p_response(200, {"leases": []}))
+
+    result = await lease.lease_to_cloud("conn-1")
+
+    assert result.status == "cloud_rejected"
+    assert p_local()["refreshToken"] == "refresh-live"
+
+
+@pytest.mark.asyncio
+async def test_unknown_ownership_leaves_the_token_off_the_device(p_device, p_cloud_calls):
+    """Fail safe: when we cannot learn who owns it, the safe guess is 'not us'. Worst case the user
+    reconnects; the alternative risks revoking their whole grant."""
+    p_cloud_calls["script"].append(httpx.ReadTimeout("boom"))
+    p_cloud_calls["script"].append(httpx.ReadTimeout("also boom"))
+
+    result = await lease.lease_to_cloud("conn-1")
+
+    assert result.status == "ownership_unknown"
+    assert "refreshToken" not in p_local()
+
+
+@pytest.mark.asyncio
+async def test_release_brings_the_refresh_token_home(p_device, p_cloud_calls):
+    await p_already_leased(p_cloud_calls)
+    p_cloud_calls["script"].append(
+        p_response(200, {"access_token": "access-new", "refresh_token": "refresh-rotated"})
+    )
+
+    result = await lease.release_to_device("conn-1")
+
+    assert result.status == "released"
+    assert p_local()["refreshToken"] == "refresh-rotated", "must be the CURRENT token, not the old one"
+    assert p_local()["accessToken"] == "access-new"
+
+
+@pytest.mark.asyncio
+async def test_release_conflict_does_not_write_a_doomed_token(p_device, p_cloud_calls):
+    """409 means a refresh is mid-exchange. Taking that token would hand the device one the provider
+    is about to invalidate."""
+    await p_already_leased(p_cloud_calls)
+    p_cloud_calls["script"].append(p_response(409, {}))
+
+    result = await lease.release_to_device("conn-1")
+
+    assert result.status == "cloud_rejected"
+    assert "refreshToken" not in p_local()
+
+
+@pytest.mark.asyncio
+async def test_pull_access_token_updates_only_the_access_token(p_device, p_cloud_calls):
+    await p_already_leased(p_cloud_calls)
+    p_cloud_calls["script"].append(p_response(200, {"access_token": "access-fresh"}))
+
+    result = await lease.pull_access_token("conn-1")
+
+    assert result.status == "refreshed"
+    assert p_local()["accessToken"] == "access-fresh"
+    assert "refreshToken" not in p_local(), "pulling a token must never re-arm local rotation"
+
+
+@pytest.mark.asyncio
+async def test_api_key_connection_is_not_leasable(p_device, p_cloud_calls):
+    """No rotating secret means nothing to hand over, and no reason to touch the row."""
+    await p_already_leased(p_cloud_calls)
+
+    result = await lease.lease_to_cloud("conn-1")
+
+    assert result.status == "not_rotatable"
+    assert p_cloud_calls["calls"] == [], "a row with nothing to rotate must not be uploaded at all"
+
+
+@pytest.mark.asyncio
+async def test_signed_out_device_does_nothing(p_device, monkeypatch):
+    monkeypatch.setattr(lease, "p_cloud", lambda: None)
+    result = await lease.lease_to_cloud("conn-1")
+    assert result.status == "not_signed_in"
+    assert p_local()["refreshToken"] == "refresh-live"
+
+
+def test_expiry_converts_to_unix_ms():
+    assert lease.expires_ms("2026-08-01T00:00:00.000Z") == 1785542400000
+    assert lease.expires_ms("garbage") == 0, "an unreadable clock must read as expired, not valid"
+    assert lease.expires_ms(None) == 0

backend/tests/test_credential_store.py

@@ -0,0 +1,142 @@
+"""Moving a rotating credential out of 9Router's db.json without racing 9Router for the file.
+
+The stakes: providers issue a NEW refresh token on every refresh and treat a replayed one as
+theft, revoking the whole grant family. So exactly one holder may be able to refresh. Removing
+`refreshToken` from a connection is what makes an instance structurally unable to rotate. If that
+edit were lost to a concurrent router write, or if the router resurrected the token afterwards,
+we would have two rotators and a dead account.
+
+Run:
+    cd backend && .venv/bin/python -m pytest tests/test_credential_store.py -v
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import stat
+
+import pytest
+
+import backend.apps.nine_router.credential_store as store
+from backend.apps.nine_router import process
+
+P_CONNECTION = {
+    "id": "conn-1",
+    "provider": "claude",
+    "authType": "oauth",
+    "accessToken": "access-value",
+    "refreshToken": "refresh-value",
+    "expiresAt": "2026-08-01T00:00:00.000Z",
+    "isActive": True,
+}
+
+
+@pytest.fixture
+def p_router(tmp_path, monkeypatch):
+    """A stopped router with one oauth connection on disk. Restart is recorded, never real."""
+    data_dir = tmp_path / "9router"
+    data_dir.mkdir()
+    db = {"providerConnections": [dict(P_CONNECTION), {"id": "conn-2", "authType": "apikey"}]}
+    (data_dir / "db.json").write_text(json.dumps(db))
+    monkeypatch.setattr(process, "nine_router_data_dir", lambda: str(data_dir))
+
+    state = {"running": True, "restarts": 0}
+
+    def p_stop() -> None:
+        state["running"] = False
+
+    async def p_ensure() -> None:
+        state["restarts"] += 1
+        state["running"] = True
+
+    async def p_no_http() -> None:
+        state["shutdown_calls"] += 1
+
+    state["shutdown_calls"] = 0
+    monkeypatch.setattr(process, "stop", p_stop)
+    monkeypatch.setattr(process, "is_running", lambda: state["running"])
+    monkeypatch.setattr(process, "ensure_running", p_ensure)
+    # Hard-stubbed: without this the suite would POST /shutdown at whatever real router owns the port.
+    monkeypatch.setattr(store, "request_shutdown", p_no_http)
+    return state
+
+
+def p_connection(data_dir_owner) -> dict:
+    db = json.loads(open(store.db_path(), encoding="utf-8").read())
+    return next(c for c in db["providerConnections"] if c["id"] == "conn-1")
+
+
+def test_read_credential_returns_the_tokens(p_router):
+    cred = store.read_credential("conn-1")
+    assert cred is not None
+    assert cred.provider == "claude"
+    assert cred.access_token == "access-value"
+    assert cred.refresh_token == "refresh-value"
+
+
+def test_only_oauth_connections_are_listed(p_router):
+    assert store.list_oauth_connection_ids() == ["conn-1"]
+
+
+@pytest.mark.asyncio
+async def test_dropping_the_refresh_token_removes_the_key(p_router):
+    """Absent, not blank. 9Router's refresh dispatcher bails on a falsy refreshToken, so the key
+    being gone is precisely what makes this instance unable to rotate."""
+    ok = await store.apply_to_connection("conn-1", changes={}, drop=["refreshToken"])
+    assert ok
+    after = p_connection(p_router)
+    assert "refreshToken" not in after
+    assert after["accessToken"] == "access-value", "must not disturb the rest of the connection"
+
+
+@pytest.mark.asyncio
+async def test_router_is_stopped_then_restarted(p_router):
+    await store.apply_to_connection("conn-1", changes={}, drop=["refreshToken"])
+    assert p_router["shutdown_calls"] == 1, "an adopted router only goes down over HTTP"
+    assert p_router["restarts"] == 1
+    assert p_router["running"] is True
+
+
+@pytest.mark.asyncio
+async def test_refuses_to_edit_when_the_router_will_not_stop(p_router, monkeypatch):
+    """The load-bearing guard. Editing under a live router risks losing the edit, or worse, the
+    router rewriting the refresh token back after we have already handed it to the cloud."""
+    monkeypatch.setattr(process, "is_running", lambda: True)
+    monkeypatch.setattr(process, "stop", lambda: None)
+    monkeypatch.setattr(store, "P_DOWN_WAIT_S", 0.2)
+
+    ok = await store.apply_to_connection("conn-1", changes={}, drop=["refreshToken"])
+
+    assert ok is False
+    assert p_connection(p_router)["refreshToken"] == "refresh-value", "file must be untouched"
+
+
+@pytest.mark.asyncio
+async def test_restoring_a_refresh_token_round_trips(p_router):
+    await store.apply_to_connection("conn-1", changes={}, drop=["refreshToken"])
+    ok = await store.apply_to_connection("conn-1", changes={"refreshToken": "returned"}, drop=[])
+    assert ok
+    assert p_connection(p_router)["refreshToken"] == "returned"
+
+
+@pytest.mark.asyncio
+async def test_written_db_is_owner_only(p_router):
+    await store.apply_to_connection("conn-1", changes={}, drop=["refreshToken"])
+    mode = stat.S_IMODE(os.stat(store.db_path()).st_mode)
+    assert mode & 0o077 == 0
+
+
+@pytest.mark.asyncio
+async def test_unknown_connection_changes_nothing(p_router):
+    ok = await store.apply_to_connection("conn-missing", changes={"x": 1}, drop=[])
+    assert ok is False
+    assert p_connection(p_router)["refreshToken"] == "refresh-value"
+
+
+@pytest.mark.asyncio
+async def test_corrupt_db_is_not_overwritten(p_router):
+    open(store.db_path(), "w", encoding="utf-8").write("{not json")
+    ok = await store.apply_to_connection("conn-1", changes={}, drop=["refreshToken"])
+    assert ok is False
+    assert open(store.db_path(), encoding="utf-8").read() == "{not json"

backend/tests/test_headless_mode.py

@@ -0,0 +1,126 @@
+"""OPENSWARM_HEADLESS=1 gating: the tools that dead-end at an Electron renderer (browser/app
+delegation, ShowUI/AskUI, AskUserQuestion) must be gone from the effective tool surface, and an
+'ask' must deny on the spot instead of parking on the 600s approval timeout. Every case is paired
+with its headless-off twin, because a gate that can't be seen switching off proves nothing."""
+
+import pytest
+from unittest.mock import AsyncMock, patch
+
+from backend.apps.agents.core.models import AgentSession
+from backend.apps.agents.manager.permissions import workflow_approval
+from backend.apps.agents.manager.permissions.build_effective_tool_lists import build_effective_tool_lists
+from backend.apps.agents.manager.register_builtin_mcp_servers import register_builtin_mcp_servers
+from backend.apps.agents.manager.streaming.HookContext import HookContext
+from backend.config.headless import HEADLESS_DENIED_TOOLS
+
+BROWSER_DELEGATION = ("CreateBrowserAgent", "BrowserAgent", "BrowserAgents", "AppAgent")
+
+
+def p_session():
+    session = AgentSession(name="t", model="sonnet", dashboard_id="d")
+    session.allowed_tools = ["Read", "Bash", "AskUserQuestion"]
+    return session
+
+
+def p_run_the_real_pipeline():
+    """Registration then tool-list build, in the order the agent loop runs them."""
+    session = p_session()
+    mcp_servers = {}
+    browser_tools, invoke_tools = register_builtin_mcp_servers(
+        mcp_servers, session, {}, None, None)
+    allowed, disallowed = build_effective_tool_lists(
+        session, mcp_servers, {}, False, browser_tools, invoke_tools)
+    return mcp_servers, allowed, disallowed
+
+
+def p_ctx() -> HookContext:
+    session = p_session()
+    return HookContext(
+        session=session,
+        session_id=session.id,
+        prompt="hi",
+        builtin_perms={},
+        policy_defaults={},
+        sessions={},
+    )
+
+
+def test_the_denied_set_is_exactly_the_renderer_bound_tools():
+    assert HEADLESS_DENIED_TOOLS == frozenset(BROWSER_DELEGATION) | {"ShowUI", "AskUserQuestion"}
+
+
+def test_headless_drops_the_renderer_bound_servers_and_tools(monkeypatch):
+    monkeypatch.setenv("OPENSWARM_HEADLESS", "1")
+    mcp_servers, allowed, disallowed = p_run_the_real_pipeline()
+    assert "openswarm-browser-agent" not in mcp_servers
+    assert "openswarm-ui" not in mcp_servers
+    for tool in BROWSER_DELEGATION:
+        assert f"mcp__openswarm-browser-agent__{tool}" not in allowed
+    for ui_tool in ("ShowUI", "AskUI"):
+        assert f"mcp__openswarm-ui__{ui_tool}" not in allowed
+    assert "AskUserQuestion" not in allowed
+    assert "AskUserQuestion" in disallowed
+    # The rest of the surface is untouched; headless prunes the renderer, it doesn't lobotomise the agent.
+    assert "Read" in allowed and "Bash" in allowed
+    assert "openswarm-invoke-agent" in mcp_servers
+    assert "openswarm-apps" in mcp_servers
+
+
+def test_without_headless_every_one_of_them_is_offered(monkeypatch):
+    monkeypatch.delenv("OPENSWARM_HEADLESS", raising=False)
+    mcp_servers, allowed, _ = p_run_the_real_pipeline()
+    assert "openswarm-browser-agent" in mcp_servers
+    assert "openswarm-ui" in mcp_servers
+    for tool in BROWSER_DELEGATION:
+        assert f"mcp__openswarm-browser-agent__{tool}" in allowed
+    for ui_tool in ("ShowUI", "AskUI"):
+        assert f"mcp__openswarm-ui__{ui_tool}" in allowed
+
+
+def test_askuserquestion_survives_when_the_ui_server_is_absent(monkeypatch):
+    # With no openswarm-ui registered nothing else denies AskUserQuestion, so this isolates the headless gate.
+    monkeypatch.delenv("OPENSWARM_HEADLESS", raising=False)
+    allowed, disallowed = build_effective_tool_lists(p_session(), {}, {}, False, [], [])
+    assert "AskUserQuestion" in allowed and "AskUserQuestion" not in disallowed
+    monkeypatch.setenv("OPENSWARM_HEADLESS", "1")
+    allowed, disallowed = build_effective_tool_lists(p_session(), {}, {}, False, [], [])
+    assert "AskUserQuestion" not in allowed and "AskUserQuestion" in disallowed
+
+
+def test_only_the_exact_flag_value_turns_headless_on(monkeypatch):
+    monkeypatch.setenv("OPENSWARM_HEADLESS", "0")
+    _, allowed, _ = p_run_the_real_pipeline()
+    assert "mcp__openswarm-ui__ShowUI" in allowed
+
+
+@pytest.mark.asyncio
+async def test_headless_denies_an_ask_without_ever_prompting(monkeypatch):
+    monkeypatch.setenv("OPENSWARM_HEADLESS", "1")
+    ask = AsyncMock()
+    with patch.object(workflow_approval, "request_user_approval", new=ask):
+        decision = await workflow_approval.resolve_ask(p_ctx(), "Bash", {"command": "ls"}, None)
+    assert decision.behavior == "deny"
+    assert not ask.called  # never broadcast into the void, so never a 600s park
+
+
+@pytest.mark.asyncio
+async def test_without_headless_an_ask_still_prompts(monkeypatch):
+    monkeypatch.delenv("OPENSWARM_HEADLESS", raising=False)
+    ask = AsyncMock(return_value=workflow_approval.ApprovalDecision(behavior="allow"))
+    with patch.object(workflow_approval, "request_user_approval", new=ask):
+        decision = await workflow_approval.resolve_ask(p_ctx(), "Bash", {"command": "ls"}, None)
+    assert decision.behavior == "allow"
+    assert ask.called
+
+
+@pytest.mark.asyncio
+async def test_headless_still_honors_a_remembered_allow(monkeypatch):
+    monkeypatch.setenv("OPENSWARM_HEADLESS", "1")
+    ctx = p_ctx()
+    workflow_approval.set_workflow_approval_memory(
+        ctx.session_id, decisions={"Bash": "allow"}, step_usage={}, remember=None, ask_timeout=5.0)
+    try:
+        decision = await workflow_approval.resolve_ask(ctx, "Bash", {"command": "ls"}, None)
+    finally:
+        workflow_approval.clear_workflow_approval_memory(ctx.session_id)
+    assert decision.behavior == "allow"

backend/tests/test_router_data_dir_permissions.py

@@ -0,0 +1,56 @@
+"""9Router's state dir must be owner-only.
+
+Its db.json carries live subscription access AND refresh tokens in plaintext, and 9Router writes
+that file 0644. On a shared machine every other local account could read them. We tighten the
+directory rather than the file because 9Router rewrites db.json on every token refresh, so a chmod
+on the file itself would be undone within the hour.
+
+Run:
+    cd backend && .venv/bin/python -m pytest tests/test_router_data_dir_permissions.py -v
+"""
+
+import os
+import stat
+
+import pytest
+
+import backend.apps.nine_router.process as process
+
+
+@pytest.fixture
+def p_data_dir(tmp_path, monkeypatch):
+    d = tmp_path / "9router"
+    d.mkdir()
+    monkeypatch.setenv("DATA_DIR", str(d))
+    return d
+
+
+@pytest.mark.skipif(os.name == "nt", reason="POSIX mode bits only")
+def test_group_and_world_readable_dir_is_tightened(p_data_dir):
+    os.chmod(p_data_dir, 0o755)
+    process.harden_data_dir_permissions()
+    assert stat.S_IMODE(os.stat(p_data_dir).st_mode) == 0o700
+
+
+@pytest.mark.skipif(os.name == "nt", reason="POSIX mode bits only")
+def test_tokens_are_unreadable_by_others_afterwards(p_data_dir):
+    """The property that actually matters, stated as the attacker sees it: no bit outside the owner."""
+    os.chmod(p_data_dir, 0o755)
+    (p_data_dir / "db.json").write_text('{"providerConnections":[]}')
+    process.harden_data_dir_permissions()
+    assert stat.S_IMODE(os.stat(p_data_dir).st_mode) & 0o077 == 0
+
+
+@pytest.mark.skipif(os.name == "nt", reason="POSIX mode bits only")
+def test_already_tight_dir_is_left_alone(p_data_dir):
+    os.chmod(p_data_dir, 0o700)
+    before = os.stat(p_data_dir).st_mtime_ns
+    process.harden_data_dir_permissions()
+    assert stat.S_IMODE(os.stat(p_data_dir).st_mode) == 0o700
+    assert os.stat(p_data_dir).st_mtime_ns == before
+
+
+def test_missing_dir_does_not_raise(tmp_path, monkeypatch):
+    """Runs before 9Router has ever started, so the dir legitimately may not exist yet."""
+    monkeypatch.setenv("DATA_DIR", str(tmp_path / "nope"))
+    process.harden_data_dir_permissions()

backend/tests/test_schedule_execution_target.py

@@ -0,0 +1,113 @@
+"""A cloud-hosted workflow's timer belongs to the server, not to this machine.
+
+The failure these guard against is a double run: if the laptop is awake when a
+cloud-hosted slot comes due, both the laptop and the server fire it, and the
+user gets two of everything. Ownership is expressed once, on
+Workflow.execution_target, and every reader of the schedule must honour it.
+
+Run:
+    cd backend && .venv/bin/python -m pytest tests/test_schedule_execution_target.py -v
+"""
+
+from __future__ import annotations
+
+import asyncio
+from datetime import datetime, timedelta, timezone
+
+import pytest
+
+
+@pytest.fixture(autouse=True)
+def p_wf_env(isolated_workflows_data, reset_scheduler_state):
+    yield
+
+
+def test_default_target_is_this_device(make_wf):
+    """Every existing workflow predates the field, so the default has to keep them local."""
+    wf = make_wf()
+    assert wf.execution_target == "device"
+
+
+@pytest.mark.asyncio
+async def test_tick_does_not_fire_a_cloud_workflow(make_wf, monkeypatch):
+    from backend.apps.workflows import storage, scheduler
+    fired: list[str] = []
+
+    async def p_capture(wf, scheduled_for=None):
+        fired.append(wf.id)
+
+    monkeypatch.setattr(scheduler, "_fire", p_capture)
+    overdue = datetime.now(timezone.utc) - timedelta(minutes=5)
+
+    local = make_wf(execution_target="device", next_run_at=overdue)
+    cloud = make_wf(execution_target="cloud", next_run_at=overdue)
+    storage.save_workflow(local)
+    storage.save_workflow(cloud)
+
+    await scheduler._tick()
+    # _tick hands each fire to create_task, so nothing has actually run until we yield the loop.
+    await asyncio.sleep(0)
+
+    assert local.id in fired, "a device workflow must still fire; the test is vacuous otherwise"
+    assert cloud.id not in fired
+
+
+@pytest.mark.asyncio
+async def test_tick_does_not_roll_a_cloud_next_run_at(make_wf, monkeypatch):
+    """Rolling the timer forward locally is its own bug even when nothing fires: the server owns
+    that field, and a local write silently competes with it."""
+    from backend.apps.workflows import storage, scheduler
+
+    async def p_noop(wf, scheduled_for=None):
+        return None
+
+    monkeypatch.setattr(scheduler, "_fire", p_noop)
+    overdue = datetime.now(timezone.utc) - timedelta(minutes=5)
+    cloud = make_wf(execution_target="cloud", next_run_at=overdue)
+    storage.save_workflow(cloud)
+
+    await scheduler._tick()
+
+    after = storage.get_workflow(cloud.id)
+    assert after.next_run_at == cloud.next_run_at
+
+
+def test_sleep_math_ignores_cloud_workflows(make_wf):
+    """_tick and seconds_to_next_fire must agree. If only _tick skipped cloud workflows, an
+    overdue one would stay overdue forever and pin the loop at its 1s floor, burning a core."""
+    from backend.apps.workflows import storage, scheduler
+    overdue = datetime.now(timezone.utc) - timedelta(minutes=5)
+    storage.save_workflow(make_wf(execution_target="cloud", next_run_at=overdue))
+
+    assert scheduler.seconds_to_next_fire() is None
+    assert scheduler._seconds_until_next() == 60.0
+
+
+def test_reconcile_leaves_cloud_workflows_untouched(make_wf):
+    """A closed laptop did not "miss" a cloud run; the server ran it. Capturing it would offer the
+    user a review card for work that already happened, and rewrite a server-owned field."""
+    from backend.apps.workflows import storage, scheduler
+    anchor = datetime.now(timezone.utc) - timedelta(days=3)
+    # occurrences_between never enumerates fires from before the workflow existed, so it has to be old.
+    born = datetime.now(timezone.utc) - timedelta(days=10)
+    cloud = make_wf(execution_target="cloud", next_run_at=anchor, created_at=born)
+    storage.save_workflow(cloud)
+
+    scheduler.reconcile_on_startup()
+
+    assert storage.list_missed() == []
+    after = storage.get_workflow(cloud.id)
+    assert after.next_run_at == cloud.next_run_at
+
+
+def test_reconcile_still_captures_device_workflows(make_wf):
+    """The discriminating half: the same walk must keep working for local workflows."""
+    from backend.apps.workflows import storage, scheduler
+    anchor = datetime.now(timezone.utc) - timedelta(days=3)
+    born = datetime.now(timezone.utc) - timedelta(days=10)
+    local = make_wf(execution_target="device", next_run_at=anchor, created_at=born)
+    storage.save_workflow(local)
+
+    scheduler.reconcile_on_startup()
+
+    assert storage.list_missed() != []

openswarm-runner/Dockerfile

@@ -0,0 +1,92 @@
+# syntax=docker/dockerfile:1
+#
+# One OpenSwarm workflow run, then exit. Build context is the REPO ROOT, not this
+# directory, because the image needs backend/ and requirements.lock:
+#
+#   docker build --platform linux/amd64 -f openswarm-runner/Dockerfile -t openswarm-runner .
+#
+# Layout the image commits to (all three are load-bearing, backend code resolves
+# them with zero changes when OPENSWARM_PACKAGED=1):
+#   /app/backend      the FastAPI orchestrator
+#   /app/router       9router's standalone server, found by p_find_9router_dir()
+#   /app/python-env   UV_PYTHON target probed by tools_lib/mcp_config.py
+
+ARG PYTHON_VERSION=3.13
+ARG NODE_VERSION=20
+ARG ROUTER_VERSION=0.3.60
+ARG UV_VERSION=0.11.8
+
+FROM node:${NODE_VERSION}-bookworm-slim AS node
+
+# 9router 0.3.60 is pure JavaScript; --ignore-scripts skips a postinstall that only rebuilds a native addon the standalone server never loads.
+FROM node AS router
+ARG ROUTER_VERSION
+WORKDIR /stage
+RUN printf '{"name":"router-stage","version":"0.0.0","private":true}\n' > package.json \
+ && npm install "9router@${ROUTER_VERSION}" --no-save --no-audit --no-fund --silent --ignore-scripts \
+ && test -f node_modules/9router/app/server.js \
+ && test -z "$(find node_modules/9router -name '*.node' -print -quit)"
+
+FROM python:${PYTHON_VERSION}-slim-bookworm AS uv
+ARG UV_VERSION
+ARG TARGETARCH
+RUN set -eux; \
+    apt-get update && apt-get install -y --no-install-recommends curl ca-certificates; \
+    case "${TARGETARCH}" in \
+      amd64) triple=x86_64-unknown-linux-gnu; sha=56dd1b66701ecb62fe896abb919444e4b83c5e8645cca953e6ddd496ff8a0feb ;; \
+      arm64) triple=aarch64-unknown-linux-gnu; sha=eee8dd658d20e5ac85fec9c2326b6cbc9d83a1eef09ef07433e58698ac849591 ;; \
+      *) echo "unsupported TARGETARCH ${TARGETARCH}" >&2; exit 1 ;; \
+    esac; \
+    curl -fsSL -o /tmp/uv.tar.gz "https://github.com/astral-sh/uv/releases/download/${UV_VERSION}/uv-${triple}.tar.gz"; \
+    echo "${sha}  /tmp/uv.tar.gz" | sha256sum -c -; \
+    mkdir -p /stage; \
+    tar -xzf /tmp/uv.tar.gz -C /stage --strip-components=1
+
+# Wheels only: the runtime image ships no compiler, so a source build here is a build-time failure rather than a 3am surprise.
+FROM python:${PYTHON_VERSION}-slim-bookworm AS pydeps
+COPY backend/requirements.lock /tmp/requirements.lock
+RUN pip install --no-cache-dir --require-hashes --only-binary=:all: \
+      --prefix=/opt/pydeps -r /tmp/requirements.lock
+
+FROM python:${PYTHON_VERSION}-slim-bookworm
+
+RUN set -eux; \
+    apt-get update; \
+    apt-get install -y --no-install-recommends git ca-certificates; \
+    rm -rf /var/lib/apt/lists/*
+
+COPY --from=node /usr/local/bin/node /usr/local/bin/node
+COPY --from=pydeps /opt/pydeps /usr/local
+COPY --from=router /stage/node_modules/9router/app /app/router
+
+COPY backend /app/backend
+COPY openswarm-runner/runner /app/runner
+
+# After backend/, never before: mcp_config.resolve_command probes uv-bin last, and the repo's own copy is Mach-O.
+COPY --from=uv /stage/uv /app/backend/uv-bin/uv
+COPY --from=uv /stage/uvx /app/backend/uv-bin/uvx
+
+RUN set -eux; \
+    if ls /app/backend/.env* >/dev/null 2>&1; then echo "a dotenv reached the image; fix Dockerfile.dockerignore" >&2; exit 1; fi; \
+    mkdir -p /app/python-env/bin; \
+    ln -s /usr/local/bin/python3 /app/python-env/bin/python3; \
+    find /app/backend -name '__pycache__' -type d -prune -exec rm -rf {} +; \
+    useradd --create-home --uid 10001 --shell /usr/sbin/nologin runner; \
+    mkdir -p /data; \
+    chown -R runner:runner /app /data
+
+USER runner
+WORKDIR /app
+ENV HOME=/home/runner \
+    PYTHONPATH=/app \
+    PYTHONUNBUFFERED=1 \
+    PYTHONDONTWRITEBYTECODE=1 \
+    OPENSWARM_HEADLESS=1 \
+    OPENSWARM_PACKAGED=1 \
+    OPENSWARM_DATA_ROOT=/data/openswarm \
+    OPENSWARM_HOST=127.0.0.1 \
+    OPENSWARM_PORT=8324 \
+    DATA_DIR=/data/9router \
+    NODE_ENV=production
+
+ENTRYPOINT ["python3", "-m", "runner.main"]

openswarm-runner/Dockerfile.dockerignore

@@ -0,0 +1,18 @@
+*
+!backend
+!openswarm-runner/runner
+
+# A developer's real OAuth client secrets live here; baking them into an image that
+# gets pushed to a registry is how a laptop leaks credentials. The Dockerfile asserts
+# they are gone, so this list failing open fails the build instead of shipping.
+backend/.env
+backend/.env.*
+
+backend/data
+backend/.venv
+backend/uv-bin
+backend/tests
+backend/.pytest_cache
+**/__pycache__
+**/*.pyc
+**/.DS_Store

openswarm-runner/README.md

@@ -0,0 +1,68 @@
+# openswarm-runner
+
+One ephemeral Linux container that executes ONE OpenSwarm workflow run and exits.
+One Fly Firecracker machine per run, no state kept.
+
+## Build
+
+The build context is the **repo root**, not this directory (the image needs `backend/`
+and `backend/requirements.lock`):
+
+```bash
+docker build --platform linux/amd64 -f openswarm-runner/Dockerfile -t openswarm-runner .
+```
+
+## Run
+
+The container is told everything it needs by one JSON run spec in `OPENSWARM_RUN_SPEC`
+(or a path in `OPENSWARM_RUN_SPEC_FILE`). See `runner/run_spec.py` for the typed shape.
+
+```json
+{
+  "run_id": "cr_01J...",
+  "workflow": { "id": "wf_1", "title": "Daily digest", "model": "opus-5",
+                "steps": [{ "text": "summarize my inbox" }] },
+  "credentials": [
+    { "provider": "claude", "auth_type": "oauth",
+      "access_token": "<already refreshed by the control plane>",
+      "expires_at": "2026-07-31T20:00:00Z" }
+  ],
+  "callback": { "url": "https://api.openswarm.com/api/cloud-runs/cr_01J.../report",
+                "token": "<two-party runner token, not a user credential>" },
+  "max_run_seconds": 1800
+}
+```
+
+Exit codes: `0` ok, `1` runner crash, `2` bad spec, `3` credential expired on arrival,
+`4` backend never came up, `5` workflow failed, `6` wall-clock cap hit.
+
+## The credential rule
+
+**A `providerConnections[]` entry this runner writes never contains a `refreshToken`.**
+9Router's refresh dispatcher bails on `if (!b || !b.refreshToken) return null`, so
+omitting the field is what makes the container incapable of rotating the user's grant.
+If it ever rotated, the user's laptop would be left replaying a dead token and the
+provider would revoke the whole grant family.
+
+Two independent walls enforce it, and a third makes a leak require deleting the code
+that builds the entry:
+
+1. `ProviderCredential` forbids extra fields, so a spec carrying `refreshToken` fails
+   validation before the backend boots.
+2. `assert_no_refresh_token` re-reads the assembled db payload just before the write.
+3. `router_connection` assembles the entry from a fixed key list, never a passthrough.
+
+All three live in `runner/seed/router_credentials.py`.
+
+An access token that arrives expired fails the run (exit 3). The runner never refreshes.
+
+## Test
+
+```bash
+PYTHONPATH=.:openswarm-runner backend/.venv/bin/python3 -m pytest openswarm-runner/tests -q
+```
+
+## Deploy
+
+Not deployed. `fly.toml` is written but never applied; read its header first, the app
+has to be created onto its own isolated private network by hand before any deploy.

openswarm-runner/fly.toml

@@ -0,0 +1,51 @@
+# openswarm-runner: one ephemeral Firecracker machine per workflow run. Boots the
+# backend headless, runs the workflow, reports, exits. Nothing here is long-lived.
+#
+# THIS APP MUST NOT SHARE THE TRUSTED 6PN MESH with openswarm-cloud / openswarm-edge.
+# The agent inside has Bash and executes user prose, so from in here
+# `curl http://openswarm-cloud.internal:8080` must resolve to nothing. Fly decides an
+# app's private network AT CREATE TIME and fly.toml cannot express it, so the app is
+# created once, by hand, onto its own isolated network:
+#
+#   fly apps create openswarm-runner --org openswarm --network openswarm-runner-isolated
+#   fly deploy . --config openswarm-runner/fly.toml --dockerfile openswarm-runner/Dockerfile
+#
+# (deploy runs from the REPO ROOT: the image needs backend/ in its build context.)
+# Verify the isolation after the first deploy, do not assume it:
+#   fly ssh console -a openswarm-runner -C "getent hosts openswarm-cloud.internal"   # must fail
+#
+# There is deliberately no [http_service] and no [[services]]: the runner takes no
+# inbound traffic and gets no public IP. It reaches the control plane outbound over
+# the public internet with the callback token in the run spec, which is why the two
+# do not need a shared private network in the first place.
+#
+# Machines are created per run by the control plane (Machines API, auto_destroy=true,
+# run spec passed as OPENSWARM_RUN_SPEC). This file is the app-level shape they inherit.
+
+app = 'openswarm-runner'
+primary_region = 'iad'
+kill_signal = 'SIGTERM'
+kill_timeout = '30s'
+
+[build]
+  dockerfile = 'Dockerfile'
+
+[env]
+  # Hard wall-clock cap, enforced twice inside the container: the poll loop stops the
+  # run at this mark, and an independent thread kills the process 90s later. A run
+  # spec asking for more is clamped down to this, never up.
+  RUNNER_MAX_RUN_SECONDS = '1800'
+  OPENSWARM_HEADLESS = '1'
+  OPENSWARM_PACKAGED = '1'
+  OPENSWARM_DATA_ROOT = '/data/openswarm'
+  OPENSWARM_HOST = '127.0.0.1'
+  OPENSWARM_PORT = '8324'
+  DATA_DIR = '/data/9router'
+
+# No [[mounts]]: a run's state is garbage the moment it ends, and an ephemeral rootfs
+# means one run cannot leave a credential lying around for the next tenant to find.
+
+[[vm]]
+  cpu_kind = 'shared'
+  cpus = 2
+  memory_mb = 4096

openswarm-runner/runner/backend_process.py

@@ -0,0 +1,111 @@
+"""Boot the OpenSwarm backend inside the container and wait until it answers.
+
+Spawned the same way the desktop shell spawns it (`python -m uvicorn backend.main:app`
+on loopback) so the cloud path and the laptop path are the same code on the same
+socket. Loopback, not 0.0.0.0: every caller of this API lives in this container, and
+the agent running inside it has Bash, so there is no reason to publish the port onto
+the machine's private network.
+"""
+
+import os
+import subprocess
+import time
+from typing import Dict, List, Optional
+
+import httpx
+from pydantic import BaseModel, ConfigDict, InstanceOf
+from typeguard import typechecked
+
+HOST = "127.0.0.1"
+HEALTH_PATH = "/api/health/check"
+# 9Router is a Next.js standalone server: it binds `process.env.HOSTNAME || '0.0.0.0'`, and Docker sets HOSTNAME to the container id, so left alone it listens on the container's eth0 address and every probe of 127.0.0.1:20128 gets ECONNREFUSED. Set on the backend's env because the backend is what spawns node.
+ROUTER_BIND_HOSTNAME = "127.0.0.1"
+AUTH_TOKEN_FILENAME = "auth.token"
+# The backend imports the whole app graph before it binds; on a cold Fly machine that has been measured in tens of seconds, so the budget is generous rather than tight.
+BOOT_TIMEOUT_SECONDS = 120.0
+SHUTDOWN_GRACE_SECONDS = 10.0
+
+
+class BackendUnavailable(RuntimeError):
+    """The backend never came up, or died while we were using it."""
+
+
+class BackendProcess(BaseModel):
+    model_config = ConfigDict(validate_assignment=True)
+
+    process: InstanceOf[subprocess.Popen]
+    base_url: str
+    token: str
+
+    @typechecked
+    def headers(self) -> Dict[str, str]:
+        return {"Authorization": f"Bearer {self.token}"}
+
+    @typechecked
+    def is_alive(self) -> bool:
+        return self.process.poll() is None
+
+
+@typechecked
+def p_command(port: int) -> List[str]:
+    return ["python3", "-m", "uvicorn", "backend.main:app", "--host", HOST, "--port", str(port)]
+
+
+@typechecked
+def p_read_auth_token(data_root: str) -> str:
+    """The backend mints this before it binds, so by the time health passes the file exists."""
+    path = os.path.join(data_root, AUTH_TOKEN_FILENAME)
+    try:
+        with open(path, "r", encoding="utf-8") as handle:
+            return handle.read().strip()
+    except OSError as exc:
+        raise BackendUnavailable(f"backend is up but its auth token is unreadable at {path}: {exc}") from exc
+
+
+@typechecked
+def start_backend(app_root: str, data_root: str, port: int, deadline: float) -> BackendProcess:
+    """Spawn the backend and block until it answers health, or raise BackendUnavailable."""
+    environment = dict(os.environ)
+    environment["OPENSWARM_DATA_ROOT"] = data_root
+    environment["OPENSWARM_HEADLESS"] = "1"
+    environment["OPENSWARM_PORT"] = str(port)
+    environment["OPENSWARM_HOST"] = HOST
+    environment["HOSTNAME"] = ROUTER_BIND_HOSTNAME
+    environment["PYTHONPATH"] = app_root
+
+    process = subprocess.Popen(p_command(port), cwd=app_root, env=environment)
+    base_url = f"http://{HOST}:{port}"
+    budget = min(time.monotonic() + BOOT_TIMEOUT_SECONDS, deadline)
+
+    with httpx.Client(timeout=2.0) as client:
+        while time.monotonic() < budget:
+            if process.poll() is not None:
+                raise BackendUnavailable(f"backend exited during startup with code {process.returncode}")
+            try:
+                healthy = client.get(f"{base_url}{HEALTH_PATH}").status_code == 200
+            except httpx.HTTPError:
+                healthy = False
+            if healthy:
+                try:
+                    token = p_read_auth_token(data_root)
+                except BackendUnavailable:
+                    stop_backend(process)
+                    raise
+                return BackendProcess(process=process, base_url=base_url, token=token)
+            time.sleep(0.25)
+
+    stop_backend(process)
+    raise BackendUnavailable(f"backend did not answer {HEALTH_PATH} within its startup budget")
+
+
+@typechecked
+def stop_backend(process: Optional[subprocess.Popen]) -> None:
+    """SIGTERM then SIGKILL. The machine is about to die anyway; this just stops the logs mid-sentence."""
+    if process is None or process.poll() is not None:
+        return
+    process.terminate()
+    try:
+        process.wait(timeout=SHUTDOWN_GRACE_SECONDS)
+    except subprocess.TimeoutExpired:
+        process.kill()
+        process.wait(timeout=SHUTDOWN_GRACE_SECONDS)

openswarm-runner/runner/main.py

@@ -0,0 +1,199 @@
+"""One workflow run, one container, one exit code.
+
+Boots the backend headless, executes the workflow the control plane asked for,
+reports the result, and dies. Nothing here is meant to survive the run.
+"""
+
+import logging
+import os
+import subprocess
+import sys
+import threading
+import time
+from datetime import datetime, timezone
+from typing import Optional
+
+from pydantic import BaseModel, ConfigDict
+from typeguard import typechecked
+
+from runner.backend_process import BackendProcess, BackendUnavailable, start_backend, stop_backend
+from runner.report import RunReport, send_report
+from runner.run_spec import CallbackTarget, InvalidRunSpec, RunSpec, load_run_spec
+from runner.seed.data_root import seed_data_root
+from runner.seed.router_credentials import write_router_db
+from runner.workflow_run import RunOutcome, RunProgress, WorkflowRunFailed, execute_workflow
+
+EXIT_OK = 0
+EXIT_INTERNAL = 1
+EXIT_BAD_SPEC = 2
+EXIT_CREDENTIAL_EXPIRED = 3
+EXIT_BACKEND_UNAVAILABLE = 4
+EXIT_WORKFLOW_FAILED = 5
+EXIT_DEADLINE = 6
+
+DEFAULT_APP_ROOT = "/app"
+DEFAULT_DATA_ROOT = "/data/openswarm"
+DEFAULT_ROUTER_DATA_DIR = "/data/9router"
+DEFAULT_PORT = 8324
+# Slack between the soft deadline (stop the run, report it) and the hard one (kill the process).
+REPORT_GRACE_SECONDS = 90.0
+# Ceiling the control plane cannot raise. A cap a caller can override is not a cap.
+MAX_RUN_SECONDS_ENV = "RUNNER_MAX_RUN_SECONDS"
+DEFAULT_MAX_RUN_SECONDS = 1800
+
+logger = logging.getLogger("runner")
+
+
+class Heartbeat(BaseModel):
+    model_config = ConfigDict(validate_assignment=True)
+
+    run_id: str
+    interval_seconds: float
+    callback: Optional[CallbackTarget] = None
+    last_sent: float = 0.0
+
+    @typechecked
+    def maybe_send(self, progress: RunProgress) -> None:
+        now = time.monotonic()
+        if now - self.last_sent < self.interval_seconds:
+            return
+        self.last_sent = now
+        send_report(self.callback, RunReport(
+            run_id=self.run_id,
+            phase="heartbeat",
+            status=progress.status,
+            active_step_idx=progress.active_step_idx,
+            last_tool_label=progress.last_tool_label,
+        ))
+
+
+@typechecked
+def effective_max_run_seconds(spec: RunSpec) -> int:
+    """The shorter of what the job asked for and what this machine's config allows."""
+    try:
+        ceiling = int(os.environ.get(MAX_RUN_SECONDS_ENV, "") or DEFAULT_MAX_RUN_SECONDS)
+    except ValueError:
+        ceiling = DEFAULT_MAX_RUN_SECONDS
+    return max(60, min(spec.max_run_seconds, ceiling))
+
+
+@typechecked
+def arm_hard_stop(seconds: float) -> None:
+    """Independent backstop on machine-seconds; fires even if the graceful path is wedged."""
+    def p_fire() -> None:
+        time.sleep(seconds)
+        logger.error("hard wall-clock stop after %.0fs, killing the run", seconds)
+        os._exit(EXIT_DEADLINE)
+
+    threading.Thread(target=p_fire, daemon=True, name="hard-stop").start()
+
+
+@typechecked
+def p_fail(spec: Optional[RunSpec], status: str, message: str, code: int) -> int:
+    logger.error("%s: %s", status, message)
+    send_report(
+        spec.callback if spec else None,
+        RunReport(
+            run_id=spec.run_id if spec else "unknown",
+            phase="finished",
+            status=status,
+            exit_code=code,
+            error=message,
+        ),
+    )
+    return code
+
+
+@typechecked
+def p_exit_code_for(outcome: RunOutcome) -> int:
+    if outcome.status in ("success", "ran_late"):
+        return EXIT_OK
+    if outcome.status == "timed_out":
+        return EXIT_DEADLINE
+    return EXIT_WORKFLOW_FAILED
+
+
+@typechecked
+def p_run(spec: RunSpec, deadline: float) -> int:
+    now = datetime.now(timezone.utc)
+    expired = spec.expired_credentials(now)
+    if expired:
+        names = ", ".join(credential.provider for credential in expired)
+        return p_fail(
+            spec,
+            "failure",
+            f"access token for {names} is expired or about to expire; the runner never refreshes, "
+            "so the control plane must re-issue it",
+            EXIT_CREDENTIAL_EXPIRED,
+        )
+
+    app_root = os.environ.get("OPENSWARM_APP_ROOT", DEFAULT_APP_ROOT)
+    data_root = os.environ.get("OPENSWARM_DATA_ROOT", DEFAULT_DATA_ROOT)
+    router_data_dir = os.environ.get("DATA_DIR", DEFAULT_ROUTER_DATA_DIR)
+    port = int(os.environ.get("OPENSWARM_PORT", str(DEFAULT_PORT)))
+
+    write_router_db(router_data_dir, spec.credentials, now)
+    seed_data_root(data_root, spec)
+    logger.info("seeded data root %s and router db in %s", data_root, router_data_dir)
+
+    backend: Optional[BackendProcess] = None
+    process: Optional[subprocess.Popen] = None
+    try:
+        backend = start_backend(app_root, data_root, port, deadline)
+        process = backend.process
+        logger.info("backend healthy at %s", backend.base_url)
+
+        send_report(spec.callback, RunReport(run_id=spec.run_id, phase="started", status="running"))
+        heartbeat = Heartbeat(
+            run_id=spec.run_id,
+            interval_seconds=float(spec.callback.heartbeat_seconds) if spec.callback else 30.0,
+            callback=spec.callback,
+        )
+        outcome = execute_workflow(backend, spec.workflow.id, deadline, heartbeat.maybe_send)
+    except BackendUnavailable as exc:
+        return p_fail(spec, "failure", str(exc), EXIT_BACKEND_UNAVAILABLE)
+    except WorkflowRunFailed as exc:
+        return p_fail(spec, "failure", str(exc), EXIT_WORKFLOW_FAILED)
+    finally:
+        stop_backend(process)
+
+    code = p_exit_code_for(outcome)
+    logger.info("run %s finished as %s (exit %d)", spec.run_id, outcome.status, code)
+    send_report(spec.callback, RunReport(
+        run_id=spec.run_id,
+        phase="finished",
+        status=outcome.status,
+        exit_code=code,
+        error=outcome.error,
+        cost_usd=outcome.cost_usd,
+        answer=outcome.answer,
+        transcript=outcome.transcript,
+        system_notices=outcome.system_notices,
+    ))
+    return code
+
+
+@typechecked
+def main() -> int:
+    logging.basicConfig(
+        level=logging.INFO,
+        format="%(asctime)s %(levelname).1s %(name)s: %(message)s",
+        datefmt="%H:%M:%S",
+    )
+    try:
+        spec = load_run_spec()
+    except InvalidRunSpec as exc:
+        return p_fail(None, "failure", str(exc), EXIT_BAD_SPEC)
+
+    budget = effective_max_run_seconds(spec)
+    arm_hard_stop(budget + REPORT_GRACE_SECONDS)
+    deadline = time.monotonic() + budget
+    try:
+        return p_run(spec, deadline)
+    except Exception as exc:
+        logger.exception("runner crashed")
+        return p_fail(spec, "failure", f"runner crashed: {exc}", EXIT_INTERNAL)
+
+
+if __name__ == "__main__":
+    sys.exit(main())

openswarm-runner/runner/report.py

@@ -0,0 +1,69 @@
+"""Tell the control plane what happened. The terminal report is the run's only receipt."""
+
+import logging
+import time
+from typing import List, Literal, Optional
+
+import httpx
+from pydantic import BaseModel, ConfigDict, Field
+from typeguard import typechecked
+
+from runner.run_spec import CallbackTarget
+
+logger = logging.getLogger(__name__)
+
+TERMINAL_ATTEMPTS = 5
+TERMINAL_BACKOFF_SECONDS = 2.0
+REQUEST_TIMEOUT_SECONDS = 15.0
+
+
+class RunReport(BaseModel):
+    model_config = ConfigDict(validate_assignment=True)
+
+    run_id: str
+    phase: Literal["started", "heartbeat", "finished"]
+    status: str
+    exit_code: Optional[int] = None
+    error: Optional[str] = None
+    cost_usd: float = 0.0
+    active_step_idx: Optional[int] = None
+    last_tool_label: Optional[str] = None
+    answer: str = ""
+    transcript: str = ""
+    # The backend calls a run "success" even when the provider rejected the token; these are how the control plane sees that.
+    system_notices: List[str] = Field(default_factory=list)
+
+
+@typechecked
+def p_post_once(callback: CallbackTarget, report: RunReport) -> bool:
+    try:
+        with httpx.Client(timeout=REQUEST_TIMEOUT_SECONDS) as client:
+            response = client.post(
+                callback.url,
+                headers={"Authorization": f"Bearer {callback.token}"},
+                json=report.model_dump(mode="json"),
+            )
+        if response.status_code < 300:
+            return True
+        logger.warning("report %s rejected with HTTP %s", report.phase, response.status_code)
+        return False
+    except httpx.HTTPError as exc:
+        logger.warning("report %s failed to send: %s", report.phase, exc)
+        return False
+
+
+@typechecked
+def send_report(callback: Optional[CallbackTarget], report: RunReport) -> bool:
+    """Post a report. Terminal reports retry; heartbeats get one shot and are never retried."""
+    if callback is None:
+        logger.info("no callback configured; %s report kept local: %s", report.phase, report.status)
+        return True
+    if report.phase != "finished":
+        return p_post_once(callback, report)
+    for attempt in range(TERMINAL_ATTEMPTS):
+        if p_post_once(callback, report):
+            return True
+        if attempt + 1 < TERMINAL_ATTEMPTS:
+            time.sleep(TERMINAL_BACKOFF_SECONDS * (attempt + 1))
+    logger.error("terminal report for run %s never landed after %d attempts", report.run_id, TERMINAL_ATTEMPTS)
+    return False

openswarm-runner/runner/run_spec.py

@@ -0,0 +1,139 @@
+"""Typed description of the single workflow run this container exists to execute.
+
+The control plane hands the container exactly one of these (JSON in
+OPENSWARM_RUN_SPEC, or a path in OPENSWARM_RUN_SPEC_FILE) and nothing else. Every
+field is validated before the backend boots, so a malformed job dies in under a
+second instead of burning a machine-minute discovering it.
+"""
+
+import json
+import os
+from datetime import datetime, timedelta, timezone
+from typing import List, Literal, Optional
+
+from pydantic import BaseModel, ConfigDict, Field, model_validator
+from typeguard import typechecked
+
+from backend.apps.workflows.models import Workflow
+
+SPEC_ENV = "OPENSWARM_RUN_SPEC"
+SPEC_FILE_ENV = "OPENSWARM_RUN_SPEC_FILE"
+
+# Headroom the access token must still have on arrival. The control plane refreshes right before dispatch; anything thinner than this means its clock or its queue is broken, and we must not paper over that by refreshing ourselves.
+MIN_TOKEN_LIFETIME = timedelta(minutes=2)
+
+
+class InvalidRunSpec(ValueError):
+    """The control plane handed us something we refuse to run."""
+
+
+class ProviderCredential(BaseModel):
+    """One already-refreshed provider credential, spendable but not rotatable.
+
+    `extra="forbid"` is the first of two walls keeping a refresh token out of this
+    container: a payload carrying `refreshToken` fails validation here and the run
+    dies loudly. See runner.router_credentials for the second wall and the why.
+    """
+
+    model_config = ConfigDict(validate_assignment=True, extra="forbid")
+
+    provider: str = Field(min_length=1)
+    auth_type: Literal["oauth", "api_key"]
+    label: str = "OpenSwarm cloud run"
+    access_token: Optional[str] = None
+    api_key: Optional[str] = None
+    expires_at: Optional[datetime] = None
+    scope: Optional[str] = None
+
+    @model_validator(mode="after")
+    def p_require_matching_secret(self) -> "ProviderCredential":
+        if self.auth_type == "oauth":
+            if not self.access_token:
+                raise ValueError(f"credential for {self.provider!r} is oauth but carries no access_token")
+            if self.expires_at is None:
+                raise ValueError(f"credential for {self.provider!r} is oauth but carries no expires_at")
+            if self.api_key:
+                raise ValueError(f"credential for {self.provider!r} carries both an access_token and an api_key")
+        else:
+            if not self.api_key:
+                raise ValueError(f"credential for {self.provider!r} is api_key but carries no api_key")
+            if self.access_token:
+                raise ValueError(f"credential for {self.provider!r} carries both an access_token and an api_key")
+        return self
+
+    @typechecked
+    def remaining_lifetime(self, now: datetime) -> Optional[timedelta]:
+        """How long this credential is still good for; None when it cannot expire."""
+        if self.expires_at is None:
+            return None
+        return self.expires_at.astimezone(timezone.utc) - now.astimezone(timezone.utc)
+
+
+class CallbackTarget(BaseModel):
+    """Where the run reports back. The token is a dedicated two-party secret, never a user credential."""
+
+    model_config = ConfigDict(validate_assignment=True, extra="forbid")
+
+    url: str = Field(min_length=1)
+    token: str = Field(min_length=1)
+    heartbeat_seconds: int = Field(default=30, ge=5, le=300)
+
+
+class RunSpec(BaseModel):
+    model_config = ConfigDict(validate_assignment=True, extra="forbid")
+
+    run_id: str = Field(min_length=1)
+    workflow: Workflow
+    credentials: List[ProviderCredential] = Field(min_length=1)
+    callback: Optional[CallbackTarget] = None
+    # Hard wall-clock ceiling. Fly bills by machine-second, so an agent that wedges must cost a bounded amount.
+    max_run_seconds: int = Field(default=1800, ge=60, le=7200)
+
+    @typechecked
+    def expired_credentials(self, now: datetime) -> List[ProviderCredential]:
+        """Credentials too close to expiry to spend. The runner cannot refresh, so this is fatal, not a retry."""
+        stale: List[ProviderCredential] = []
+        for credential in self.credentials:
+            remaining = credential.remaining_lifetime(now)
+            if remaining is not None and remaining < MIN_TOKEN_LIFETIME:
+                stale.append(credential)
+        return stale
+
+    @typechecked
+    def workflow_for_disk(self) -> Workflow:
+        """The workflow as this container should see it: one run, never a schedule.
+
+        A cloud-executed workflow arrives with its schedule still configured. Left
+        enabled, the container's own scheduler would fire it a second time inside
+        the box, so the timer is stripped here rather than trusted to stay off.
+        """
+        copy = self.workflow.model_copy(deep=True)
+        copy.schedule.enabled = False
+        copy.deleted_at = None
+        copy.draft_steps = None
+        copy.next_run_at = None
+        return copy
+
+
+@typechecked
+def load_run_spec() -> RunSpec:
+    """Parse the run spec from the environment, or raise InvalidRunSpec with a legible reason."""
+    raw = os.environ.get(SPEC_ENV, "").strip()
+    source = SPEC_ENV
+    if not raw:
+        path = os.environ.get(SPEC_FILE_ENV, "").strip()
+        if not path:
+            raise InvalidRunSpec(f"no run spec: set {SPEC_ENV} to JSON or {SPEC_FILE_ENV} to a file path")
+        source = f"{SPEC_FILE_ENV}={path}"
+        try:
+            with open(path, "r", encoding="utf-8") as handle:
+                raw = handle.read()
+        except OSError as exc:
+            raise InvalidRunSpec(f"cannot read run spec from {source}: {exc}") from exc
+
+    try:
+        return RunSpec.model_validate_json(raw)
+    except json.JSONDecodeError as exc:
+        raise InvalidRunSpec(f"run spec from {source} is not valid JSON: {exc}") from exc
+    except ValueError as exc:
+        raise InvalidRunSpec(f"run spec from {source} is not a valid RunSpec: {exc}") from exc

openswarm-runner/runner/seed/data_root.py

@@ -0,0 +1,81 @@
+"""Lay down the backend's data dir before it boots, so the run is ready on the first tick.
+
+Everything here is written pre-boot on purpose: the workflow store and the settings
+store both load from disk once at startup, so seeding files is cheaper and more
+deterministic than replaying create/PATCH calls over HTTP (no aux LLM naming call,
+no schedule normalization, no chance of the container inventing a second workflow).
+"""
+
+import json
+import os
+import tempfile
+from typing import Any, Dict
+
+from typeguard import typechecked
+
+from backend.apps.settings.models import AppSettings
+from runner.run_spec import RunSpec
+
+# 9Router provider id -> the AppSettings field the backend reads a raw key from.
+API_KEY_SETTINGS_FIELD: Dict[str, str] = {
+    "anthropic": "anthropic_api_key",
+    "openai": "openai_api_key",
+    "gemini": "google_api_key",
+    "google": "google_api_key",
+    "openrouter": "openrouter_api_key",
+}
+
+# One synthetic id for every cloud run. Without it each ephemeral container mints a fresh uuid and analytics sees a brand-new "install" per run.
+CLOUD_RUNNER_INSTALLATION_ID = "openswarm-cloud-runner"
+
+
+@typechecked
+def p_write_json(path: str, payload: Any) -> None:
+    """Atomic, owner-only write; these files hold API keys."""
+    directory = os.path.dirname(path)
+    os.makedirs(directory, mode=0o700, exist_ok=True)
+    handle, temp_path = tempfile.mkstemp(dir=directory, prefix=".seed-", suffix=".json")
+    try:
+        with os.fdopen(handle, "w", encoding="utf-8") as stream:
+            json.dump(payload, stream, indent=2)
+        os.chmod(temp_path, 0o600)
+        os.replace(temp_path, path)
+    except BaseException:
+        if os.path.exists(temp_path):
+            os.unlink(temp_path)
+        raise
+
+
+@typechecked
+def settings_for_run(spec: RunSpec) -> AppSettings:
+    """The AppSettings a cloud run needs: this workflow's model, this run's keys, no telemetry."""
+    settings = AppSettings()
+    settings.default_model = spec.workflow.model
+    settings.connection_mode = "own_key"
+    settings.analytics_opt_in = False
+    settings.installation_id = CLOUD_RUNNER_INSTALLATION_ID
+    for credential in spec.credentials:
+        if credential.auth_type != "api_key":
+            continue
+        field = API_KEY_SETTINGS_FIELD.get(credential.provider)
+        if field is None:
+            raise ValueError(
+                f"no settings field for api_key provider {credential.provider!r}; "
+                f"supported: {', '.join(sorted(set(API_KEY_SETTINGS_FIELD)))}"
+            )
+        setattr(settings, field, credential.api_key)
+    return settings
+
+
+@typechecked
+def seed_data_root(data_root: str, spec: RunSpec) -> None:
+    """Write the workflow record and the settings file the backend will read at boot."""
+    workflow = spec.workflow_for_disk()
+    p_write_json(
+        os.path.join(data_root, "workflows", f"{workflow.id}.json"),
+        workflow.model_dump(mode="json"),
+    )
+    p_write_json(
+        os.path.join(data_root, "settings", "settings.json"),
+        settings_for_run(spec).model_dump(mode="json"),
+    )

openswarm-runner/runner/seed/router_credentials.py

@@ -0,0 +1,128 @@
+"""Write 9Router's credential db for one cloud run, without a refresh token. Ever.
+
+This container must be structurally incapable of rotating the user's OAuth grant.
+9Router's refresh dispatcher bails on `if (!b || !b.refreshToken) return null`, so a
+providerConnections entry with no such field is one it can spend and never rotate.
+If the runner did rotate, the user's laptop would be left replaying a dead token and
+the provider would revoke their entire grant family. That is the whole safety
+property of this file, not a style preference.
+
+Two independent walls hold it up:
+  1. ProviderCredential forbids extra fields, so a payload carrying `refreshToken`
+     never parses into the process at all.
+  2. assert_no_refresh_token re-reads the assembled payload just before the write
+     and refuses anything whose key names a refresh token, however it got there.
+"""
+
+import json
+import os
+import tempfile
+from datetime import datetime, timezone
+from typing import Any, Dict, List
+from uuid import uuid4
+
+from typeguard import typechecked
+
+from runner.run_spec import ProviderCredential
+
+DB_FILENAME = "db.json"
+
+# Normalized key fragment that must never appear anywhere in the db we write.
+FORBIDDEN_KEY_FRAGMENT = "refreshtoken"
+
+
+class RefreshTokenLeak(RuntimeError):
+    """A refresh token reached the credential writer. Fail the run rather than write it."""
+
+
+@typechecked
+def p_normalize_key(key: str) -> str:
+    return "".join(char for char in key.lower() if char.isalnum())
+
+
+@typechecked
+def p_iso(moment: datetime) -> str:
+    """9Router timestamps are ISO-8601 UTC with a Z suffix; match it exactly."""
+    return moment.astimezone(timezone.utc).isoformat(timespec="milliseconds").replace("+00:00", "Z")
+
+
+@typechecked
+def assert_no_refresh_token(payload: Any, path: str = "$") -> None:
+    """Raise if any key anywhere under `payload` names a refresh token."""
+    if isinstance(payload, dict):
+        for key, value in payload.items():
+            if FORBIDDEN_KEY_FRAGMENT in p_normalize_key(str(key)):
+                raise RefreshTokenLeak(
+                    f"refusing to write a 9Router db containing a refresh token at {path}.{key}"
+                )
+            assert_no_refresh_token(value, f"{path}.{key}")
+    elif isinstance(payload, list):
+        for index, value in enumerate(payload):
+            assert_no_refresh_token(value, f"{path}[{index}]")
+
+
+@typechecked
+def router_connection(credential: ProviderCredential, now: datetime) -> Dict[str, Any]:
+    """Build one providerConnections entry from an allow-list of keys, never a passthrough."""
+    if credential.auth_type != "oauth":
+        raise ValueError(f"credential for {credential.provider!r} is not an oauth connection")
+    entry: Dict[str, Any] = {
+        "id": str(uuid4()),
+        "provider": credential.provider,
+        "authType": "oauth",
+        "name": credential.label,
+        "priority": 1,
+        "isActive": True,
+        "createdAt": p_iso(now),
+        "updatedAt": p_iso(now),
+        "accessToken": credential.access_token,
+        "testStatus": "active",
+    }
+    if credential.expires_at is not None:
+        entry["expiresAt"] = p_iso(credential.expires_at)
+    if credential.scope:
+        entry["scope"] = credential.scope
+    return entry
+
+
+@typechecked
+def router_db_payload(credentials: List[ProviderCredential], now: datetime) -> Dict[str, Any]:
+    """A complete 9Router db seeded with this run's subscription connections and nothing else."""
+    return {
+        "providerConnections": [
+            router_connection(credential, now)
+            for credential in credentials
+            if credential.auth_type == "oauth"
+        ],
+        "providerNodes": [],
+        "proxyPools": [],
+        "modelAliases": {},
+        "mitmAlias": {},
+        "combos": [],
+        "apiKeys": [],
+        "customModels": [],
+        "pricing": {},
+        "settings": {},
+    }
+
+
+@typechecked
+def write_router_db(data_dir: str, credentials: List[ProviderCredential], now: datetime) -> str:
+    """Write $DATA_DIR/db.json owner-only and return its path."""
+    payload = router_db_payload(credentials, now)
+    assert_no_refresh_token(payload)
+
+    os.makedirs(data_dir, mode=0o700, exist_ok=True)
+    os.chmod(data_dir, 0o700)
+    path = os.path.join(data_dir, DB_FILENAME)
+    handle, temp_path = tempfile.mkstemp(dir=data_dir, prefix=".db-", suffix=".json")
+    try:
+        with os.fdopen(handle, "w", encoding="utf-8") as stream:
+            json.dump(payload, stream, indent=2)
+        os.chmod(temp_path, 0o600)
+        os.replace(temp_path, path)
+    except BaseException:
+        if os.path.exists(temp_path):
+            os.unlink(temp_path)
+        raise
+    return path

openswarm-runner/runner/workflow_run.py

@@ -0,0 +1,224 @@
+"""Drive one workflow through the backend's own HTTP surface and collect its result.
+
+Deliberately no shortcuts into agent_manager: the cloud run fires the same route the
+Run button fires, so the MCP gate, action filtering, provider routing and history all
+behave exactly as they do on a laptop.
+"""
+
+import json
+import time
+from typing import Any, Callable, Dict, List, Optional
+
+import httpx
+from pydantic import BaseModel, ConfigDict, Field
+from typeguard import typechecked
+
+from runner.backend_process import BackendProcess
+
+TERMINAL_STATUSES = ("success", "failure", "ran_late", "skipped")
+POLL_INTERVAL_SECONDS = 1.0
+TRANSCRIPT_MAX_CHARS = 14000
+
+
+class WorkflowRunFailed(RuntimeError):
+    """The backend refused to start the run at all."""
+
+
+class RunProgress(BaseModel):
+    model_config = ConfigDict(validate_assignment=True)
+
+    run_id: str
+    status: str
+    active_step_idx: Optional[int] = None
+    last_tool_label: Optional[str] = None
+
+
+class RunOutcome(BaseModel):
+    model_config = ConfigDict(validate_assignment=True)
+
+    run_id: str
+    status: str
+    error: Optional[str] = None
+    cost_usd: float = 0.0
+    session_id: Optional[str] = None
+    transcript: str = ""
+    answer: str = ""
+    system_notices: List[str] = Field(default_factory=list)
+
+
+@typechecked
+def p_block_text(block: Dict[str, Any]) -> str:
+    kind = block.get("type")
+    if kind == "text":
+        return str(block.get("text") or "")
+    if kind == "tool_use":
+        return f"[tool {block.get('name')}] {json.dumps(block.get('input') or {})[:300]}"
+    if kind == "tool_result":
+        inner = block.get("content")
+        return f"[result] {inner if isinstance(inner, str) else json.dumps(inner)[:300]}"
+    return ""
+
+
+@typechecked
+def p_message_text(message: Dict[str, Any]) -> str:
+    content = message.get("content")
+    if isinstance(content, str):
+        return content
+    if isinstance(content, list):
+        parts = [p_block_text(block) for block in content if isinstance(block, dict)]
+        return "\n".join(part for part in parts if part)
+    return ""
+
+
+@typechecked
+def render_transcript(messages: List[Dict[str, Any]]) -> str:
+    """Role-tagged flatten, tail-biased so the end of a long run always survives the cap."""
+    lines: List[str] = []
+    for message in messages:
+        if message.get("hidden"):
+            continue
+        text = p_message_text(message).strip()
+        if text:
+            lines.append(f"{str(message.get('role') or '?').upper()}: {text}")
+    joined = "\n\n".join(lines)
+    if len(joined) > TRANSCRIPT_MAX_CHARS:
+        return "...(earlier turns trimmed)...\n\n" + joined[-TRANSCRIPT_MAX_CHARS:]
+    return joined
+
+
+@typechecked
+def final_answer(messages: List[Dict[str, Any]]) -> str:
+    """Last visible assistant text: the thing a user actually asked the workflow for."""
+    for message in reversed(messages):
+        if message.get("hidden") or message.get("role") != "assistant":
+            continue
+        text = p_message_text(message).strip()
+        if text:
+            return text
+    return ""
+
+
+@typechecked
+def system_notices(messages: List[Dict[str, Any]]) -> List[str]:
+    """Every system-role bubble in the session.
+
+    The backend appends a system message only when something went wrong (a dead
+    provider token, a run error, a blocked tool), and it does NOT fail the run for
+    those, so a workflow whose credential was rejected still comes back "success".
+    Keyed on the typed role, not on the prose, and reported rather than judged: the
+    control plane decides what a notice means for billing and retries.
+    """
+    notices: List[str] = []
+    for message in messages:
+        if message.get("role") != "system" or message.get("hidden"):
+            continue
+        text = p_message_text(message).strip()
+        if text:
+            notices.append(text)
+    return notices
+
+
+@typechecked
+def p_get_json(client: httpx.Client, backend: BackendProcess, path: str) -> Dict[str, Any]:
+    response = client.get(f"{backend.base_url}{path}", headers=backend.headers())
+    response.raise_for_status()
+    payload = response.json()
+    return payload if isinstance(payload, dict) else {}
+
+
+@typechecked
+def trigger_run(client: httpx.Client, backend: BackendProcess, workflow_id: str) -> str:
+    response = client.post(
+        f"{backend.base_url}/api/workflows/{workflow_id}/run",
+        headers=backend.headers(),
+        json={},
+    )
+    response.raise_for_status()
+    body = response.json()
+    run_id = str(body.get("run_id") or "")
+    if not run_id:
+        raise WorkflowRunFailed(
+            f"backend accepted the trigger but never created a run for workflow {workflow_id}"
+        )
+    if body.get("status") == "failure":
+        raise WorkflowRunFailed(str(body.get("error") or "run failed immediately"))
+    return run_id
+
+
+@typechecked
+def p_find_run(client: httpx.Client, backend: BackendProcess, workflow_id: str, run_id: str) -> Dict[str, Any]:
+    body = p_get_json(client, backend, f"/api/workflows/{workflow_id}/runs?limit=50")
+    for record in body.get("runs") or []:
+        if isinstance(record, dict) and record.get("id") == run_id:
+            return record
+    return {}
+
+
+@typechecked
+def p_stop_run(client: httpx.Client, backend: BackendProcess, run_id: str) -> None:
+    try:
+        client.post(f"{backend.base_url}/api/workflows/runs/{run_id}/stop", headers=backend.headers())
+    except httpx.HTTPError:
+        pass
+
+
+@typechecked
+def p_collect_session(client: httpx.Client, backend: BackendProcess, session_id: str) -> List[Dict[str, Any]]:
+    try:
+        body = p_get_json(client, backend, f"/api/agents/sessions/{session_id}")
+    except httpx.HTTPError:
+        return []
+    messages = body.get("messages")
+    return [m for m in messages if isinstance(m, dict)] if isinstance(messages, list) else []
+
+
+@typechecked
+def execute_workflow(
+    backend: BackendProcess,
+    workflow_id: str,
+    deadline: float,
+    on_progress: Optional[Callable[[RunProgress], None]] = None,
+) -> RunOutcome:
+    """Fire the workflow, poll it to a terminal state, and pull the transcript back.
+
+    Blowing the deadline stops the run and reports `timed_out`; the caller still gets
+    whatever the agent produced before the wall came down.
+    """
+    with httpx.Client(timeout=30.0) as client:
+        run_id = trigger_run(client, backend, workflow_id)
+        record: Dict[str, Any] = {}
+        timed_out = False
+
+        while True:
+            record = p_find_run(client, backend, workflow_id, run_id) or record
+            status = str(record.get("status") or "running")
+            if on_progress is not None:
+                on_progress(RunProgress(
+                    run_id=run_id,
+                    status=status,
+                    active_step_idx=record.get("active_step_idx"),
+                    last_tool_label=record.get("last_tool_label"),
+                ))
+            if status in TERMINAL_STATUSES:
+                break
+            if not backend.is_alive():
+                raise WorkflowRunFailed("backend died while the workflow was running")
+            if time.monotonic() >= deadline:
+                timed_out = True
+                p_stop_run(client, backend, run_id)
+                record = p_find_run(client, backend, workflow_id, run_id) or record
+                break
+            time.sleep(POLL_INTERVAL_SECONDS)
+
+        session_id = record.get("session_id")
+        messages = p_collect_session(client, backend, str(session_id)) if session_id else []
+        return RunOutcome(
+            run_id=run_id,
+            status="timed_out" if timed_out else str(record.get("status") or "failure"),
+            error=("wall-clock cap reached before the workflow finished" if timed_out else record.get("error")),
+            cost_usd=float(record.get("cost_usd") or 0.0),
+            session_id=str(session_id) if session_id else None,
+            transcript=render_transcript(messages),
+            answer=final_answer(messages),
+            system_notices=system_notices(messages),
+        )

openswarm-runner/tests/test_router_credentials.py

@@ -0,0 +1,111 @@
+"""The runner must be structurally unable to rotate a user's OAuth grant.
+
+Every test here exists to make one class of bug unwritable: a refresh token reaching
+9Router's db.json. Delete either wall in runner/seed/router_credentials.py and these go red.
+"""
+
+import json
+import os
+import stat
+from datetime import datetime, timedelta, timezone
+
+import pytest
+
+from runner.run_spec import InvalidRunSpec, ProviderCredential, RunSpec, load_run_spec
+from runner.seed.router_credentials import (
+    RefreshTokenLeak,
+    assert_no_refresh_token,
+    router_db_payload,
+    write_router_db,
+)
+
+NOW = datetime(2026, 7, 31, 12, 0, 0, tzinfo=timezone.utc)
+ACCESS_TOKEN = "at-test-value-not-a-real-token"
+REFRESH_TOKEN = "rt-test-value-not-a-real-token"
+
+
+def spec_json(credential: dict) -> str:
+    return json.dumps({
+        "run_id": "run-1",
+        "workflow": {"id": "wf-1", "title": "Test", "steps": [{"text": "say hi"}]},
+        "credentials": [credential],
+    })
+
+
+def oauth_credential() -> ProviderCredential:
+    return ProviderCredential(
+        provider="claude",
+        auth_type="oauth",
+        access_token=ACCESS_TOKEN,
+        expires_at=NOW + timedelta(hours=8),
+    )
+
+
+@pytest.mark.parametrize("key", ["refreshToken", "refresh_token", "Refresh-Token", "oauthRefreshToken"])
+def test_a_spec_carrying_a_refresh_token_never_parses(key: str, tmp_path, monkeypatch) -> None:
+    payload = {
+        "provider": "claude",
+        "auth_type": "oauth",
+        "access_token": ACCESS_TOKEN,
+        "expires_at": (NOW + timedelta(hours=8)).isoformat(),
+        key: REFRESH_TOKEN,
+    }
+    monkeypatch.setenv("OPENSWARM_RUN_SPEC", spec_json(payload))
+    with pytest.raises(InvalidRunSpec) as caught:
+        load_run_spec()
+    assert key in str(caught.value)
+    assert not list(tmp_path.iterdir()), "a rejected spec must not leave anything on disk"
+
+
+@pytest.mark.parametrize("key", ["refreshToken", "refresh_token", "Refresh-Token", "oauthRefreshToken"])
+def test_the_writer_guard_rejects_a_poisoned_payload(key: str) -> None:
+    payload = {"providerConnections": [{"provider": "claude", key: REFRESH_TOKEN}]}
+    with pytest.raises(RefreshTokenLeak):
+        assert_no_refresh_token(payload)
+
+
+def test_the_guard_passes_a_clean_payload() -> None:
+    assert_no_refresh_token(router_db_payload([oauth_credential()], NOW))
+
+
+def test_written_db_carries_the_access_token_and_no_refresh_token(tmp_path) -> None:
+    path = write_router_db(str(tmp_path / "9router"), [oauth_credential()], NOW)
+    raw = open(path, "r", encoding="utf-8").read()
+
+    # Without this the "no refresh token" assertion below would also pass on an empty file.
+    assert ACCESS_TOKEN in raw
+    connection = json.loads(raw)["providerConnections"][0]
+    assert connection["provider"] == "claude"
+    assert connection["isActive"] is True
+    assert connection["expiresAt"] == "2026-07-31T20:00:00.000Z"
+
+    assert "refresh" not in raw.lower()
+    assert not any("refresh" in key.lower() for key in connection)
+
+
+def test_the_db_and_its_directory_are_owner_only(tmp_path) -> None:
+    path = write_router_db(str(tmp_path / "9router"), [oauth_credential()], NOW)
+    assert stat.S_IMODE(os.stat(path).st_mode) == 0o600
+    assert stat.S_IMODE(os.stat(os.path.dirname(path)).st_mode) == 0o700
+
+
+def test_an_api_key_credential_never_reaches_the_router_db(tmp_path) -> None:
+    credential = ProviderCredential(provider="anthropic", auth_type="api_key", api_key="sk-test-not-real")
+    path = write_router_db(str(tmp_path / "9router"), [credential], NOW)
+    assert json.loads(open(path, encoding="utf-8").read())["providerConnections"] == []
+
+
+def test_an_oauth_credential_without_an_access_token_is_rejected() -> None:
+    with pytest.raises(ValueError, match="no access_token"):
+        ProviderCredential(provider="claude", auth_type="oauth", expires_at=NOW)
+
+
+def test_an_expired_access_token_is_fatal_not_refreshable() -> None:
+    spec = RunSpec.model_validate_json(spec_json({
+        "provider": "claude",
+        "auth_type": "oauth",
+        "access_token": ACCESS_TOKEN,
+        "expires_at": (NOW + timedelta(seconds=30)).isoformat(),
+    }))
+    assert [credential.provider for credential in spec.expired_credentials(NOW)] == ["claude"]
+    assert spec.expired_credentials(NOW - timedelta(hours=1)) == []

openswarm-runner/tests/test_run_spec.py

@@ -0,0 +1,84 @@
+"""The run spec is the only thing the control plane can say to this container."""
+
+import json
+import os
+import stat
+
+import pytest
+
+from runner.run_spec import InvalidRunSpec, RunSpec, load_run_spec
+from runner.seed.data_root import seed_data_root, settings_for_run
+
+VALID_CREDENTIAL = {"provider": "anthropic", "auth_type": "api_key", "api_key": "sk-test-not-real"}
+
+
+def spec_body(**overrides) -> dict:
+    body = {
+        "run_id": "run-1",
+        "workflow": {
+            "id": "wf-1",
+            "title": "Daily digest",
+            "model": "opus-5",
+            "steps": [{"text": "summarize the inbox"}],
+            "schedule": {"enabled": True, "repeat_unit": "day", "hour": 9},
+        },
+        "credentials": [VALID_CREDENTIAL],
+    }
+    body.update(overrides)
+    return body
+
+
+def test_a_missing_spec_names_both_env_vars(monkeypatch) -> None:
+    monkeypatch.delenv("OPENSWARM_RUN_SPEC", raising=False)
+    monkeypatch.delenv("OPENSWARM_RUN_SPEC_FILE", raising=False)
+    with pytest.raises(InvalidRunSpec, match="OPENSWARM_RUN_SPEC_FILE"):
+        load_run_spec()
+
+
+def test_a_spec_file_is_accepted(tmp_path, monkeypatch) -> None:
+    path = tmp_path / "spec.json"
+    path.write_text(json.dumps(spec_body()), encoding="utf-8")
+    monkeypatch.delenv("OPENSWARM_RUN_SPEC", raising=False)
+    monkeypatch.setenv("OPENSWARM_RUN_SPEC_FILE", str(path))
+    assert load_run_spec().workflow.title == "Daily digest"
+
+
+def test_unknown_top_level_fields_are_rejected(monkeypatch) -> None:
+    monkeypatch.setenv("OPENSWARM_RUN_SPEC", json.dumps(spec_body(surprise="hello")))
+    with pytest.raises(InvalidRunSpec, match="surprise"):
+        load_run_spec()
+
+
+def test_a_run_needs_at_least_one_credential(monkeypatch) -> None:
+    monkeypatch.setenv("OPENSWARM_RUN_SPEC", json.dumps(spec_body(credentials=[])))
+    with pytest.raises(InvalidRunSpec):
+        load_run_spec()
+
+
+def test_the_container_never_inherits_the_schedule() -> None:
+    spec = RunSpec.model_validate(spec_body())
+    assert spec.workflow.schedule.enabled is True
+    assert spec.workflow_for_disk().schedule.enabled is False
+
+
+def test_seeding_writes_the_workflow_and_owner_only_settings(tmp_path) -> None:
+    spec = RunSpec.model_validate(spec_body())
+    seed_data_root(str(tmp_path), spec)
+
+    workflow_path = tmp_path / "workflows" / "wf-1.json"
+    settings_path = tmp_path / "settings" / "settings.json"
+    assert json.loads(workflow_path.read_text())["schedule"]["enabled"] is False
+    assert stat.S_IMODE(os.stat(settings_path).st_mode) == 0o600
+
+    settings = json.loads(settings_path.read_text())
+    assert settings["anthropic_api_key"] == "sk-test-not-real"
+    assert settings["default_model"] == "opus-5"
+    assert settings["analytics_opt_in"] is False
+
+
+def test_an_unmappable_api_key_provider_fails_loudly() -> None:
+    spec = RunSpec.model_validate(spec_body(
+        credentials=[{"provider": "wat", "auth_type": "api_key", "api_key": "x"}]
+    ))
+    with pytest.raises(ValueError, match="no settings field"):
+        settings_for_run(spec)

openswarm-runner/tests/test_workflow_run.py

@@ -0,0 +1,36 @@
+"""Reading a finished session correctly, including the failures the backend calls success."""
+
+from runner.workflow_run import final_answer, render_transcript, system_notices
+
+# Shape taken verbatim from a real container run whose provider token was rejected.
+REJECTED_TOKEN_SESSION = [
+    {"role": "user", "content": "Reply with exactly the word PONG and nothing else."},
+    {"role": "system", "content": "Provider authentication expired. Open Settings, Models and reconnect, then send your message again."},
+]
+
+ANSWERED_SESSION = [
+    {"role": "user", "content": "ping"},
+    {"role": "assistant", "content": [{"type": "tool_use", "name": "Bash", "input": {"command": "echo hi"}}]},
+    {"role": "assistant", "content": [{"type": "text", "text": "PONG"}]},
+    {"role": "assistant", "content": [{"type": "text", "text": "draft"}], "hidden": True},
+]
+
+
+def test_a_rejected_credential_surfaces_as_a_system_notice() -> None:
+    assert system_notices(REJECTED_TOKEN_SESSION) == [REJECTED_TOKEN_SESSION[1]["content"]]
+    assert final_answer(REJECTED_TOKEN_SESSION) == ""
+
+
+def test_a_healthy_run_raises_no_notices() -> None:
+    assert system_notices(ANSWERED_SESSION) == []
+
+
+def test_the_answer_is_the_last_visible_assistant_text() -> None:
+    assert final_answer(ANSWERED_SESSION) == "PONG"
+
+
+def test_the_transcript_keeps_tool_calls_and_drops_hidden_turns() -> None:
+    transcript = render_transcript(ANSWERED_SESSION)
+    assert "[tool Bash]" in transcript
+    assert "PONG" in transcript
+    assert "draft" not in transcript