[eric] free-trial: backend dispatch injection, hardware fingerprint, mint/clear, exhaustion upsell

backend/apps/agents/agent_manager.py

@@ -30,6 +30,7 @@ from backend.apps.agents.core.error_classify import (
     _NON_TRANSIENT_PATTERNS,
     _TRANSIENT_CAPACITY_PATTERNS,
     _is_auth_error,
+    _is_free_trial_exhausted,
     _is_long_context_error,
     _is_transient_capacity_error,
 )
@@ -1252,7 +1253,7 @@ class AgentManager:
                 and _primary_is_claude
                 and anthropic_web_search_is_reliable(
                     uses_direct_anthropic_api=_uses_direct_anthropic_api,
-                    is_pro=(getattr(global_settings, "connection_mode", "own_key") == "openswarm-pro"),
+                    is_pro=(getattr(global_settings, "connection_mode", "own_key") in ("openswarm-pro", "free-trial")),
                 )
             )
 
@@ -1592,19 +1593,21 @@ class AgentManager:
                 env["ENABLE_TOOL_SEARCH"] = "auto"
                 options_kwargs["env"] = env
                 logger.info(f"[MCP-DEBUG] Using OpenRouter for {session.model}")
-            elif api_type == "anthropic" and not resolved_is_9router and getattr(global_settings, "connection_mode", "own_key") == "openswarm-pro":
-                proxy_url = getattr(global_settings, "openswarm_proxy_url", None) or "https://api.openswarm.com"
-                bearer = getattr(global_settings, "openswarm_bearer_token", "") or ""
+            elif api_type == "anthropic" and not resolved_is_9router and getattr(global_settings, "connection_mode", "own_key") in ("openswarm-pro", "free-trial"):
+                from backend.apps.settings.credentials import proxy_auth
+                bearer, proxy_url = proxy_auth(global_settings)
+                bearer = bearer or ""
                 options_kwargs["env"] = {
                     "ANTHROPIC_AUTH_TOKEN": bearer,
                     "ANTHROPIC_BASE_URL": proxy_url,
                     # Pin subagent ids; CLI default 'claude-haiku-4-5-20251001'
                     # gets rejected by Pro's surface as "No credentials for provider: anthropic".
+                    # (Free-trial down-routes all of these to Haiku server-side.)
                     "CLAUDE_CODE_SUBAGENT_MODEL": "claude-sonnet-4-6",
                     "ANTHROPIC_SMALL_FAST_MODEL": "claude-haiku-4-5-20251001",
                     "ANTHROPIC_DEFAULT_HAIKU_MODEL": "claude-haiku-4-5-20251001",
                 }
-                logger.info(f"[MCP-DEBUG] Using OpenSwarm Pro proxy at {proxy_url}")
+                logger.info(f"[MCP-DEBUG] Using OpenSwarm cloud proxy at {proxy_url}")
             elif api_type == "anthropic" and not resolved_is_9router and global_settings.anthropic_api_key:
                 options_kwargs["env"] = {"ANTHROPIC_API_KEY": global_settings.anthropic_api_key}
                 logger.info("[MCP-DEBUG] Using direct Anthropic API key")
@@ -2700,12 +2703,12 @@ class AgentManager:
                                     # showing the SDK's Anthropic-rate
                                     # estimate, which is meaningless here.
                                     _free_route = True
-                            if (
-                                api_type == "anthropic"
-                                and getattr(global_settings, "connection_mode", "own_key") == "openswarm-pro"
-                                and getattr(global_settings, "openswarm_bearer_token", None)
-                            ):
-                                _free_route = True
+                            if api_type == "anthropic":
+                                from backend.apps.settings.credentials import proxy_auth as _proxy_auth
+                                _pa_tok, _ = _proxy_auth(global_settings)
+                                # Pro and free-trial both run server-funded, so per-token cost to the user is 0.
+                                if _pa_tok:
+                                    _free_route = True
 
                             if _free_route:
                                 cost = 0.0
@@ -2943,6 +2946,29 @@ class AgentManager:
                     })
                 except Exception:
                     logger.debug("submit_diagnostic for context_overflow failed", exc_info=True)
+            elif _is_free_trial_exhausted(e, extra_text=_stderr_tail):
+                # Free runs spent. Flip back to own_key and show a friendly
+                # "connect a model" upsell instead of a raw 402.
+                try:
+                    from backend.apps.subscription.free_trial import clear_free_trial
+                    await clear_free_trial(load_settings())
+                except Exception:
+                    logger.debug("clear_free_trial after exhaustion failed", exc_info=True)
+                friendly_msg = (
+                    "You've used your free runs. Connect a model to keep going: "
+                    "your own API key, an AI subscription you already pay for, or "
+                    "OpenSwarm Pro."
+                )
+                error_msg = Message(role="system", content=friendly_msg, branch_id=session.active_branch_id)
+                session.messages.append(error_msg)
+                await ws_manager.send_to_session(session_id, "agent:free_trial_exhausted", {
+                    "session_id": session_id,
+                    "message": friendly_msg,
+                })
+                await ws_manager.send_to_session(session_id, "agent:message", {
+                    "session_id": session_id,
+                    "message": error_msg.model_dump(mode="json"),
+                })
             elif _is_auth_error(e, extra_text=_stderr_tail):
                 # Three sub-cases the user can hit, with distinct fixes:
                 #   1. "No credentials for provider: claude", user picked a

backend/apps/agents/core/error_classify.py

@@ -31,6 +31,7 @@ _NON_TRANSIENT_PATTERNS = re.compile(
     r"|missing\s+bearer\s+token"
     r"|extra\s+usage\s+is\s+required\s+for\s+long\s+context"
     r"|long\s+context\s+(?:requests?\s+)?(?:requires?|not\s+(?:available|enabled))"
+    r"|free_trial_exhausted|used\s+your\s+free"
     r"|401|403)",
     re.IGNORECASE,
 )
@@ -53,6 +54,21 @@ def _is_long_context_error(exc: BaseException, extra_text: str = "") -> bool:
     ))
 
 
+def _is_free_trial_exhausted(exc: BaseException, extra_text: str = "") -> bool:
+    """True when the cloud says the machine's free runs are spent (a 402 with
+    type free_trial_exhausted). The catch-all path uses this to flip back to
+    own_key and show a friendly connect-a-model upsell instead of a raw error.
+    """
+    combined = f"{exc!s}\n{extra_text}".strip()
+    if not combined:
+        return False
+    return bool(re.search(
+        r"free_trial_exhausted|used\s+your\s+free\s+(?:openswarm\s+)?runs",
+        combined,
+        re.IGNORECASE,
+    ))
+
+
 def _is_auth_error(exc: BaseException, extra_text: str = "") -> bool:
     """True when the upstream error is a 401/403 auth failure.
 

backend/apps/agents/providers/registry.py

@@ -248,7 +248,11 @@ def resolve_model_id_for_sdk(short_name: str, settings: AppSettings) -> str:
     if entry.get("route") == "openrouter":
         return entry.get("router_model_id", short_name)
     if entry.get("api") == "anthropic":
-        if getattr(settings, "connection_mode", "own_key") == "openswarm-pro":
+        # openswarm-pro AND free-trial both proxy-route, so resolve to the bare
+        # id (the proxy serves it) instead of the cc/-prefixed id that 401s when
+        # no Claude subscription is connected. This is the line that otherwise
+        # turns a free-trial user's first run into "No AI provider connected".
+        if getattr(settings, "connection_mode", "own_key") in ("openswarm-pro", "free-trial"):
             return entry.get("model_id", short_name)
         if getattr(settings, "anthropic_api_key", None):
             return entry.get("model_id", short_name)
@@ -337,9 +341,11 @@ async def resolve_aux_model(
         if "openrouter" in connected:
             return (or_aux, base_url)
 
-    if getattr(settings, "connection_mode", "own_key") == "openswarm-pro":
-        proxy_url = getattr(settings, "openswarm_proxy_url", None) or "https://api.openswarm.com"
-        return (bare, proxy_url)
+    if getattr(settings, "connection_mode", "own_key") in ("openswarm-pro", "free-trial"):
+        from backend.apps.settings.credentials import proxy_auth
+        token, base = proxy_auth(settings)
+        if token:
+            return (bare, base)
 
     if getattr(settings, "anthropic_api_key", None):
         return (bare, None)

backend/apps/nine_router/sync_custom.py

@@ -326,17 +326,17 @@ async def sync_openswarm_pro_as_claude(bearer_token: str | None, proxy_url: str
 
 
 async def sync_pro_routing(settings_obj) -> None:
-    """Mirror the settings' pro-mode state into the 9Router Claude lane.
-    Call after any flow that changes connection_mode or the bearer
-    (activate, sign-in, sign-out, disconnect). Never raises."""
+    """Mirror the settings' cloud-proxy state (openswarm-pro OR free-trial) into
+    the 9Router Claude lane. Call after any flow that changes connection_mode or
+    the bearer (activate, sign-in, sign-out, disconnect, free-trial arm/clear).
+    Never raises."""
     try:
-        pro = getattr(settings_obj, "connection_mode", None) == "openswarm-pro"
-        bearer = getattr(settings_obj, "openswarm_bearer_token", None)
-        proxy = getattr(settings_obj, "openswarm_proxy_url", None) or "https://api.openswarm.com"
-        active = bool(pro and bearer)
+        from backend.apps.settings.credentials import proxy_auth
+        bearer, base = proxy_auth(settings_obj)
+        active = bool(bearer)
         await sync_openswarm_pro_as_claude(
             bearer if active else None,
-            proxy if active else None,
+            base if active else None,
         )
     except Exception as e:
         logger.warning(f"OpenSwarm-Pro → Claude sync failed: {e}")

backend/apps/settings/credentials.py

@@ -10,6 +10,26 @@ if TYPE_CHECKING:
 
 OPENSWARM_DEFAULT_PROXY_URL = "https://api.openswarm.com"
 
+# Connection modes that route Claude traffic through our cloud proxy with a
+# bearer instead of a user-held key. Free-trial is openswarm-pro's cheaper
+# sibling: same proxy, but pointed at the /free sub-path the cloud meters and
+# forces to Haiku.
+PROXY_CONNECTION_MODES = ("openswarm-pro", "free-trial")
+
+
+def proxy_auth(settings: AppSettings) -> tuple[str | None, str | None]:
+    """(auth_token, base_url) for whichever cloud-proxy mode is active, else
+    (None, None). Consumers append /v1/messages to base_url as usual; for
+    free-trial the base carries the /free segment so the same SDK lands on the
+    metered route."""
+    mode = getattr(settings, "connection_mode", "own_key")
+    base = (getattr(settings, "openswarm_proxy_url", None) or OPENSWARM_DEFAULT_PROXY_URL).rstrip("/")
+    if mode == "openswarm-pro":
+        return (getattr(settings, "openswarm_bearer_token", None), base)
+    if mode == "free-trial":
+        return (getattr(settings, "free_trial_token", None), base + "/free")
+    return (None, None)
+
 
 def _check_9router() -> bool:
     """Check if 9Router is running locally."""
@@ -34,8 +54,9 @@ def validate_credentials(settings: AppSettings, provider: str = "anthropic") ->
         return
 
     if p == "anthropic":
-        if getattr(settings, "connection_mode", "own_key") == "openswarm-pro":
-            if not getattr(settings, "openswarm_bearer_token", None):
+        if getattr(settings, "connection_mode", "own_key") in PROXY_CONNECTION_MODES:
+            token, _ = proxy_auth(settings)
+            if not token:
                 raise ValueError("Open Swarm account not connected. Sign in via Settings -> API.")
             return
         if settings.anthropic_api_key:
@@ -72,10 +93,11 @@ def get_provider_credentials(settings: AppSettings, provider: str) -> dict[str,
     validate_credentials(settings, provider)
 
     if p in ("anthropic", "claude"):
-        if getattr(settings, "connection_mode", "own_key") == "openswarm-pro":
+        if getattr(settings, "connection_mode", "own_key") in PROXY_CONNECTION_MODES:
+            token, base = proxy_auth(settings)
             return {
-                "auth_token": getattr(settings, "openswarm_bearer_token", "") or "",
-                "base_url": getattr(settings, "openswarm_proxy_url", None) or OPENSWARM_DEFAULT_PROXY_URL,
+                "auth_token": token or "",
+                "base_url": base or OPENSWARM_DEFAULT_PROXY_URL,
             }
         return {"api_key": settings.anthropic_api_key or ""}
 
@@ -101,11 +123,11 @@ def get_anthropic_client(settings: AppSettings) -> anthropic.AsyncAnthropic:
     """Return an AsyncAnthropic client for the user's current connection mode."""
     import anthropic
 
-    if getattr(settings, "connection_mode", "own_key") == "openswarm-pro":
-        proxy_url = getattr(settings, "openswarm_proxy_url", None) or OPENSWARM_DEFAULT_PROXY_URL
+    if getattr(settings, "connection_mode", "own_key") in PROXY_CONNECTION_MODES:
+        token, base = proxy_auth(settings)
         return anthropic.AsyncAnthropic(
-            auth_token=getattr(settings, "openswarm_bearer_token", None),
-            base_url=proxy_url,
+            auth_token=token,
+            base_url=base or OPENSWARM_DEFAULT_PROXY_URL,
         )
 
     # Prefer the user's own API key when present.

backend/apps/settings/models.py

@@ -69,6 +69,13 @@ class AppSettings(BaseModel):
     connection_mode: str = "own_key"
     openswarm_bearer_token: Optional[str] = None
     openswarm_proxy_url: Optional[str] = None
+    # Zero-config free trial: server-funded runs for a brand-new user with no
+    # key and no subscription. connection_mode flips to "free-trial" while armed;
+    # the token + remaining count are server-owned (minted by the cloud, sticky
+    # per machine). remaining is cached for the onboarding "runs low" nudge.
+    free_trial_token: Optional[str] = None
+    free_trial_remaining: Optional[int] = None
+    free_trial_runs_limit: Optional[int] = None
     openswarm_subscription_plan: Optional[str] = None
     openswarm_subscription_expires: Optional[str] = None
     openswarm_usage_cached: Optional[dict] = None

backend/apps/settings/settings.py

@@ -45,7 +45,7 @@ async def settings_lifespan():
                 getattr(s, "google_api_key", None),
                 getattr(s, "openai_api_key", None),
                 getattr(s, "openrouter_api_key", None),
-                getattr(s, "connection_mode", None) == "openswarm-pro",
+                getattr(s, "connection_mode", None) in ("openswarm-pro", "free-trial"),
                 bool(getattr(s, "custom_providers", None) or []),
             ])
             if needs_router:
@@ -59,11 +59,11 @@ async def settings_lifespan():
                 await sync_openai_api_key(s.openai_api_key)
             if getattr(s, "openrouter_api_key", None):
                 await sync_openrouter_api_key(s.openrouter_api_key)
-            if getattr(s, "connection_mode", None) == "openswarm-pro":
-                bearer = getattr(s, "openswarm_bearer_token", None)
-                proxy = getattr(s, "openswarm_proxy_url", None) or "https://api.openswarm.com"
+            if getattr(s, "connection_mode", None) in ("openswarm-pro", "free-trial"):
+                from backend.apps.settings.credentials import proxy_auth
+                bearer, base = proxy_auth(s)
                 if bearer:
-                    await sync_openswarm_pro_as_claude(bearer, proxy)
+                    await sync_openswarm_pro_as_claude(bearer, base)
             await sync_custom_providers(getattr(s, "custom_providers", None) or [])
 
         _asyncio.create_task(_boot_router_then_sync())
@@ -123,6 +123,9 @@ SERVER_OWNED_FIELDS = (
     "openswarm_subscription_plan",
     "openswarm_subscription_expires",
     "openswarm_usage_cached",
+    "free_trial_token",
+    "free_trial_remaining",
+    "free_trial_runs_limit",
     "user_id",
     "signin_method",
     "installation_id",
@@ -142,7 +145,7 @@ async def update_settings(body: AppSettings):
 
     secret_keys = {"anthropic_api_key", "openai_api_key", "google_api_key", "openrouter_api_key",
                    "claude_subscription_token", "openai_subscription_token", "gemini_subscription_token",
-                   "openswarm_bearer_token", "installation_id"}
+                   "openswarm_bearer_token", "free_trial_token", "installation_id"}
     safe = {k: v for k, v in body.model_dump().items() if k not in secret_keys}
     _sync(safe)
 

backend/apps/subscription/free_trial.py

@@ -0,0 +1,193 @@
+"""Zero-config free trial: arm/refresh/clear the server-funded free runs.
+
+A brand-new user with no key and no subscription gets a small number of agent
+runs funded through the cloud's shared pool, so they see the product work before
+being asked to connect anything. Identity is a hashed hardware fingerprint
+computed here in the backend (no Electron IPC), so deleting/reinstalling the app
+does not reset the count. The cloud is authoritative for the run count and the
+forced cheap model; this module only mirrors state into settings and 9Router.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import logging
+import os
+import platform
+import re
+import subprocess
+
+import httpx
+
+from backend.apps.settings.credentials import OPENSWARM_DEFAULT_PROXY_URL
+from backend.apps.settings.settings import save_settings_async
+
+logger = logging.getLogger(__name__)
+
+# Namespaces the hash so a raw hardware UUID never leaves the device. Public on
+# purpose (open-source): it only prevents transmitting the raw id, not a secret.
+_FP_SALT = "openswarm-free-trial-v1"
+
+
+def _enabled() -> bool:
+    return os.environ.get("OPENSWARM_FREE_TRIAL_ENABLED", "1") != "0"
+
+
+def _raw_hardware_id() -> str | None:
+    """A stable per-machine id that survives app reinstall / data wipe."""
+    system = platform.system()
+    try:
+        if system == "Darwin":
+            out = subprocess.run(
+                ["ioreg", "-rd1", "-c", "IOPlatformExpertDevice"],
+                capture_output=True, text=True, timeout=3,
+            ).stdout
+            m = re.search(r'"IOPlatformUUID"\s*=\s*"([^"]+)"', out)
+            return m.group(1) if m else None
+        if system == "Windows":
+            import winreg  # type: ignore
+            with winreg.OpenKey(
+                winreg.HKEY_LOCAL_MACHINE, r"SOFTWARE\Microsoft\Cryptography"
+            ) as key:
+                val, _ = winreg.QueryValueEx(key, "MachineGuid")
+                return val or None
+        for path in ("/etc/machine-id", "/var/lib/dbus/machine-id"):
+            if os.path.exists(path):
+                with open(path, "r") as f:
+                    v = f.read().strip()
+                    if v:
+                        return v
+    except Exception:
+        return None
+    return None
+
+
+def _fingerprint(settings_obj) -> str | None:
+    raw = _raw_hardware_id()
+    if not raw:
+        # Fail-soft: installation_id is less durable (regenerates on wipe) but
+        # better than nothing on a machine where the hardware id can't be read.
+        raw = getattr(settings_obj, "installation_id", None)
+    if not raw:
+        return None
+    return hashlib.sha256((_FP_SALT + raw).encode("utf-8")).hexdigest()
+
+
+def _has_own_model(s) -> bool:
+    """True if the user already has any real model path; never shadow it."""
+    if any(getattr(s, k, None) for k in (
+        "anthropic_api_key", "openai_api_key", "google_api_key", "openrouter_api_key",
+        "claude_subscription_token", "openai_subscription_token", "gemini_subscription_token",
+    )):
+        return True
+    if getattr(s, "connection_mode", "own_key") == "openswarm-pro" and getattr(s, "openswarm_bearer_token", None):
+        return True
+    for cp in (getattr(s, "custom_providers", None) or []):
+        name = cp.get("name") if isinstance(cp, dict) else getattr(cp, "name", None)
+        base = cp.get("base_url") if isinstance(cp, dict) else getattr(cp, "base_url", None)
+        if (name or "").strip() and (base or "").strip():
+            return True
+    return False
+
+
+def _proxy_base(settings_obj) -> str:
+    return (getattr(settings_obj, "openswarm_proxy_url", None) or OPENSWARM_DEFAULT_PROXY_URL).rstrip("/")
+
+
+async def _sync_routing(settings_obj) -> None:
+    try:
+        from backend.apps.nine_router import sync_pro_routing
+        await sync_pro_routing(settings_obj)
+    except Exception as e:
+        logger.debug("free-trial routing sync skipped: %s", e)
+
+
+async def clear_free_trial(settings_obj) -> None:
+    """Drop the trial token and revert to own_key. Keeps free_trial_remaining
+    (so the UI knows it's spent) and never touches a real paid mode."""
+    if getattr(settings_obj, "connection_mode", "own_key") == "free-trial":
+        settings_obj.connection_mode = "own_key"
+    settings_obj.free_trial_token = None
+    await save_settings_async(settings_obj)
+    await _sync_routing(settings_obj)
+
+
+async def arm_free_trial(settings_obj) -> dict:
+    """Mint (or re-fetch) the machine's grant and, if runs remain, flip into
+    free-trial mode. Guarded: never arms over a real key/subscription."""
+    if not _enabled():
+        return {"armed": False, "reason": "disabled"}
+    if getattr(settings_obj, "connection_mode", "own_key") not in ("own_key", "free-trial"):
+        return {"armed": False, "reason": "other_mode"}
+    if _has_own_model(settings_obj):
+        return {"armed": False, "reason": "has_model"}
+
+    fp = _fingerprint(settings_obj)
+    if not fp:
+        return {"armed": False, "reason": "no_fingerprint"}
+
+    base = _proxy_base(settings_obj)
+    payload: dict = {"fingerprint_hash": fp}
+    if getattr(settings_obj, "installation_id", None):
+        payload["install_id"] = settings_obj.installation_id
+    if getattr(settings_obj, "user_id", None):
+        payload["user_id"] = settings_obj.user_id
+
+    try:
+        async with httpx.AsyncClient(timeout=10.0) as client:
+            r = await client.post(f"{base}/api/free-trial/mint", json=payload)
+    except httpx.HTTPError as e:
+        logger.debug("free-trial mint network error: %s", e)
+        return {"armed": False, "reason": "network"}
+    if r.status_code != 200:
+        return {"armed": False, "reason": "upstream", "code": r.status_code}
+
+    data = r.json()
+    remaining = int(data.get("runs_remaining") or 0)
+    settings_obj.free_trial_remaining = remaining
+    settings_obj.free_trial_runs_limit = int(data.get("runs_limit") or 0) or None
+
+    if remaining > 0:
+        settings_obj.connection_mode = "free-trial"
+        settings_obj.free_trial_token = data.get("trial_token")
+        settings_obj.openswarm_proxy_url = base
+        await save_settings_async(settings_obj)
+        await _sync_routing(settings_obj)
+        return {"armed": True, "runs_remaining": remaining, "runs_limit": settings_obj.free_trial_runs_limit}
+
+    # Already spent on this machine: record it but don't arm.
+    await clear_free_trial(settings_obj)
+    return {"armed": False, "reason": "exhausted", "runs_remaining": 0}
+
+
+async def refresh_free_trial(settings_obj) -> dict:
+    """Re-read remaining runs from the cloud. Called after a session ends so the
+    onboarding 'runs low' nudge stays honest. Clears the trial when spent."""
+    token = getattr(settings_obj, "free_trial_token", None)
+    if getattr(settings_obj, "connection_mode", "own_key") != "free-trial" or not token:
+        return {"connected": False, "runs_remaining": getattr(settings_obj, "free_trial_remaining", None)}
+
+    base = _proxy_base(settings_obj)
+    try:
+        async with httpx.AsyncClient(timeout=5.0) as client:
+            r = await client.post(
+                f"{base}/api/free-trial/status",
+                headers={"Authorization": f"Bearer {token}"},
+            )
+    except httpx.HTTPError:
+        return {"connected": True, "runs_remaining": getattr(settings_obj, "free_trial_remaining", None)}
+
+    if r.status_code == 401:
+        await clear_free_trial(settings_obj)
+        return {"connected": False, "runs_remaining": getattr(settings_obj, "free_trial_remaining", None)}
+    if r.status_code != 200:
+        return {"connected": True, "runs_remaining": getattr(settings_obj, "free_trial_remaining", None)}
+
+    data = r.json()
+    remaining = int(data.get("runs_remaining") or 0)
+    settings_obj.free_trial_remaining = remaining
+    if remaining <= 0:
+        await clear_free_trial(settings_obj)
+        return {"connected": False, "runs_remaining": 0}
+    await save_settings_async(settings_obj)
+    return {"connected": True, "runs_remaining": remaining, "runs_limit": getattr(settings_obj, "free_trial_runs_limit", None)}

backend/apps/subscription/router.py

@@ -343,6 +343,26 @@ async def portal():
     return {"url": data.get("url")}
 
 
+# ---------------------------------------------------------------------------
+# Free zero-config trial
+# ---------------------------------------------------------------------------
+
+@subscription.router.post("/free-trial/mint")
+async def free_trial_mint():
+    """Renderer calls this on launch when no model is connected. Mints (or
+    re-fetches) the machine's grant and arms free-trial mode if runs remain.
+    Best-effort: never raises, returns {armed: false, reason} on any miss."""
+    from backend.apps.subscription.free_trial import arm_free_trial
+    return await arm_free_trial(load_settings())
+
+
+@subscription.router.post("/free-trial/status")
+async def free_trial_status():
+    """Refresh remaining free runs from the cloud (drives the onboarding nudge)."""
+    from backend.apps.subscription.free_trial import refresh_free_trial
+    return await refresh_free_trial(load_settings())
+
+
 # ---------------------------------------------------------------------------
 # POST /api/subscription/disconnect
 # ---------------------------------------------------------------------------

backend/tests/test_free_trial.py

@@ -0,0 +1,57 @@
+"""Free-trial dispatch injection: the pure-logic pieces that decide routing."""
+
+import backend  # noqa: F401  (path sanity asserted below)
+
+from backend.apps.settings.models import AppSettings
+from backend.apps.settings.credentials import proxy_auth
+from backend.apps.agents.core.error_classify import (
+    _is_free_trial_exhausted,
+    _is_transient_capacity_error,
+)
+from backend.apps.agents.providers.registry import resolve_model_id_for_sdk
+from backend.apps.subscription.free_trial import _has_own_model
+
+
+def test_proxy_auth_for_each_mode():
+    assert proxy_auth(AppSettings()) == (None, None)
+
+    pro = AppSettings(
+        connection_mode="openswarm-pro",
+        openswarm_bearer_token="bear",
+        openswarm_proxy_url="https://api.openswarm.com",
+    )
+    assert proxy_auth(pro) == ("bear", "https://api.openswarm.com")
+
+    free = AppSettings(
+        connection_mode="free-trial",
+        free_trial_token="ftk",
+        openswarm_proxy_url="https://api.openswarm.com",
+    )
+    # Free-trial carries the /free segment so the same SDK lands on the metered route.
+    assert proxy_auth(free) == ("ftk", "https://api.openswarm.com/free")
+
+
+def test_free_trial_resolves_to_a_bare_anthropic_id():
+    s = AppSettings(connection_mode="free-trial", free_trial_token="ftk")
+    mid = resolve_model_id_for_sdk("sonnet", s)
+    # The bug this fixes: without the free-trial branch this returns a cc/-prefixed
+    # id that 401s when no Claude subscription is connected.
+    assert "cc/" not in mid
+    assert mid.startswith("claude-")
+
+
+def test_exhaustion_is_classified_and_not_retried():
+    assert _is_free_trial_exhausted(Exception("error type free_trial_exhausted"))
+    assert _is_free_trial_exhausted(Exception("You've used your free OpenSwarm runs"))
+    assert not _is_free_trial_exhausted(Exception("overloaded, try again"))
+    # Must NOT look transient, or the agent loop would retry a spent trial forever.
+    assert not _is_transient_capacity_error(Exception("free_trial_exhausted"))
+
+
+def test_has_own_model_never_shadows_a_real_provider():
+    assert not _has_own_model(AppSettings(connection_mode="free-trial", free_trial_token="x"))
+    assert not _has_own_model(AppSettings())
+    assert _has_own_model(AppSettings(anthropic_api_key="sk-ant-x"))
+    assert _has_own_model(
+        AppSettings(connection_mode="openswarm-pro", openswarm_bearer_token="b")
+    )