path: root/docs/superpowers/plans/2026-10-23-cloud-models.md

# Cloud models (daemon) Implementation Plan

> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.

**Goal:** Let a chat use a cloud model from an OpenAI-compatible provider, picked per conversation, with only opted-in tools offered to it.

**Architecture:** All in `assistant.py`. `[[cloud]]` tables in config name providers; their keys come from `key_cmd` at startup into `PROVIDERS`. A model id `"<provider>:<model>"` routes a request to that provider (`route()`), everything else goes to llama-server as today. The conversation row stores the chosen model; `ask` may carry one; a `models` command lists local and cloud models; `--cloud-models <name>` fetches a provider's list for the settings page.

**Tech Stack:** Python 3.12 stdlib only (`http.client`, `urllib`, `sqlite3`, `shlex`, `subprocess`). Self-check: `python3 assistant.py --test` (prints `ok`).

Spec: `docs/superpowers/specs/2026-10-23-cloud-models-design.md`. The quickshell side (settings section, header dropdown) is a separate plan in that repo, written after this one lands.

Decision not in the spec: `write` (the canvas's To .md) has no conversation, so it stays on the local loaded model.

---

## File map

- Modify `assistant.py`: config (`DEFAULTS`, `merge`, new `check_cloud`), `load_tools` (cloud flag), providers (`provider_key`, `load_providers`, `route`, `cloud_models`, `send_models`), storage (`db_open`, `save_turn`), request path (`model_inputs`, `ask`, `stream`, `reply`, `answer`, `retitle`), socket (`cur`, `hello`, `start_new`, `handle`, `command`), `main`, `test`, `__main__`.
- Modify `examples/config.toml`, `examples/tools/{calc,now,search,weather}.toml`, `README.md`, `AGENTS.md`, the spec's status line.

Run every test with: `python3 assistant.py --test 2>&1 | tail -3` from the repo root. It logs a lot (expected tracebacks included); the last line is `ok` on success, an `AssertionError` traceback otherwise.

---

### Task 1: `[[cloud]]` config and the tool `cloud` flag

**Files:**
- Modify: `assistant.py` (`DEFAULTS` ~line 50, `merge` ~66, `load_tools` ~88, `test()` start ~1097)

- [ ] **Step 1: Write the failing tests**

In `test()`, right after the existing `for bad in ({"card": "X"}, ...)` loop, add:

```python
    good = {"name": "or", "url": "https://llm.example.org/v1", "key_cmd": "echo k",
            "models": ["a/b"]}
    assert merge(DEFAULTS, {"cloud": [good]})["cloud"] == [good]
    assert merge(DEFAULTS, {"cloud": [{**good, "models": []}]})   # before the first fetch
    assert DEFAULTS["cloud"] == []
    for bad in ({"cloud": good}, {"cloud": [{**good, "extra": 1}]},
                {"cloud": [{**good, "name": "a:b"}]}, {"cloud": [good, good]},
                {"cloud": [{**good, "models": "a/b"}]}, {"cloud": [{**good, "url": ""}]}):
        try:
            merge(DEFAULTS, bad)
            assert False, bad
        except SystemExit as e:
            assert "config: cloud" in str(e), (bad, e)
    with tempfile.TemporaryDirectory() as d:
        with open(os.path.join(d, "x.toml"), "w") as f:
            f.write('description = "d"\ncommand = ["date"]\ncloud = "yes"\n')
        try:
            load_tools(d)
            assert False
        except SystemExit as e:
            assert "cloud must be true or false" in str(e)
```

- [ ] **Step 2: Run, expect failure**

Run the self-check. Expected: `SystemExit: config: unknown key cloud` (the good case).

- [ ] **Step 3: Implement**

`DEFAULTS`, after `"thinking": True, ...`:

```python
    "cloud": [],                          # [[cloud]] providers, see check_cloud
```

New function above `merge`:

```python
def check_cloud(v):
    # [[cloud]] tables. A provider's name prefixes its models' ids
    # ("name:model"), so it is unique and holds no colon. models may be
    # empty: the settings page saves a provider before fetching its list.
    if not isinstance(v, list) or not all(isinstance(p, dict) for p in v):
        raise SystemExit("config: cloud must be [[cloud]] tables")
    seen = set()
    for p in v:
        name = p.get("name")
        if set(p) != {"name", "url", "key_cmd", "models"}:
            raise SystemExit(f"config: cloud {name!r} needs exactly name, url, key_cmd, models")
        if (not all(isinstance(p[k], str) and p[k] for k in ("name", "url", "key_cmd"))
                or not isinstance(p["models"], list)
                or not all(isinstance(m, str) for m in p["models"])):
            raise SystemExit(f"config: cloud {name!r}: name, url and key_cmd are strings, "
                             "models a list of strings")
        if ":" in name or name in seen:
            raise SystemExit(f"config: cloud name {name!r} is repeated or has a colon")
        seen.add(name)
    return v
```

In `merge`, just before `out[k] = v`:

```python
        if k == "cloud":
            v = check_cloud(v)
```

In `load_tools`, after the `needs description and command` check:

```python
        if not isinstance(t.get("cloud", False), bool):
            raise SystemExit(f"{path}: cloud must be true or false")
```

- [ ] **Step 4: Run, expect `ok`**

- [ ] **Step 5: Commit**

```bash
git add assistant.py
git commit -m "config: [[cloud]] providers and the tool cloud flag

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>"
```

---

### Task 2: Providers, keys, routing, `--cloud-models`

**Files:**
- Modify: `assistant.py` (imports, new functions right before `model_inputs` ~line 696, `main`, `__main__`, module docstring, `test()` fake server)

- [ ] **Step 1: Write the failing tests**

In `test()`'s `Fake` handler, first line of `do_GET` and first line of `do_POST`:

```python
            script["auth"] = self.headers.get("Authorization")
```

Right after `CFG["llm"] = f"http://127.0.0.1:{srv.server_address[1]}/v1"`, add:

```python
    # Cloud providers: the key from key_cmd, models fetched with it. The
    # fake answers /cloud/v1 like /v1, so the path tells them apart.
    base = f"http://127.0.0.1:{srv.server_address[1]}"
    CFG["cloud"] = [{"name": "fake", "url": base + "/cloud/v1/", "key_cmd": "echo sk-test",
                     "models": ["vendor/x:free"]},
                    {"name": "broken", "url": base + "/v1", "key_cmd": "false", "models": ["y"]}]
    PROVIDERS.update(load_providers(CFG["cloud"]))
    assert set(PROVIDERS) == {"fake"} and PROVIDERS["fake"]["key"] == "sk-test"
    assert route("fake:vendor/x:free") == (base + "/cloud/v1", "sk-test", "vendor/x:free", "fake")
    assert route("Gemma:q4") == (CFG["llm"], None, "Gemma:q4", None)   # no such provider
    try:
        route("broken:y")
        assert False
    except Refused as e:
        assert "no key" in str(e)
    assert cloud_models("fake") == ["m"] and script["auth"] == "Bearer sk-test"
    for name, why in (("broken", "key_cmd failed"), ("nope", "no cloud provider")):
        try:
            cloud_models(name)
            assert False, name
        except SystemExit as e:
            assert why in str(e), (name, e)
```

- [ ] **Step 2: Run, expect failure**

Expected: `NameError: name 'PROVIDERS' is not defined`.

- [ ] **Step 3: Implement**

Imports, alphabetical with the rest: `import shlex` after `import re`; `import urllib.error` before `import urllib.parse`.

New block right before `def model_inputs`:

```python
# Cloud providers, named in [[cloud]]. A key is key_cmd's stdout, read once
# at startup; a provider whose key_cmd fails is left out, so its models are
# not offered and a chat still on one is refused.
PROVIDERS = {}                            # name -> {"url", "key", "models"}


def provider_key(p):
    # Only the exit status is reported: a secret tool's stderr is not worth
    # the risk of it holding the secret.
    try:
        r = subprocess.run(shlex.split(p["key_cmd"]), capture_output=True, text=True,
                           stdin=subprocess.DEVNULL, timeout=10)
    except (OSError, ValueError, subprocess.TimeoutExpired) as e:
        raise RuntimeError(f"cloud {p['name']}: key_cmd failed ({type(e).__name__})") from None
    if r.returncode or not r.stdout.strip():
        raise RuntimeError(f"cloud {p['name']}: key_cmd failed (exit {r.returncode})")
    return r.stdout.strip()


def load_providers(cloud):
    out = {}
    for p in cloud:
        try:
            out[p["name"]] = {"url": p["url"].rstrip("/"), "key": provider_key(p),
                              "models": p["models"]}
        except RuntimeError as e:
            log(e)
    return out


def route(model):
    # (base url, key, the id the server knows, provider or None). A prefix
    # is a provider only when one by that name is configured, so a local id
    # with a colon stays local.
    name, sep, rest = (model or "").partition(":")
    if sep and name in PROVIDERS:
        p = PROVIDERS[name]
        return p["url"], p["key"], rest, name
    if sep and any(p["name"] == name for p in CFG["cloud"]):
        raise Refused(f"cloud {name}: no key (key_cmd failed at startup)")
    return CFG["llm"], None, model, None


def cloud_models(name):
    # For the settings page (--cloud-models): the provider's model ids,
    # fetched here so the key never reaches it.
    p = next((p for p in CFG["cloud"] if p["name"] == name), None)
    if p is None:
        raise SystemExit(f"no cloud provider named {name}")
    try:
        key = provider_key(p)
    except RuntimeError as e:
        raise SystemExit(str(e)) from None
    req = urllib.request.Request(p["url"].rstrip("/") + "/models",
                                 headers={"Authorization": f"Bearer {key}"})
    try:
        with urllib.request.urlopen(req, timeout=20) as r:
            return sorted(m["id"] for m in json.load(r)["data"])
    except urllib.error.HTTPError as e:
        raise SystemExit(f"cloud {name} {e.code}: "
                         f"{e.read().decode(errors='replace')[:300]}") from None
    except (OSError, KeyError, TypeError, ValueError) as e:
        raise SystemExit(f"cloud {name}: {e}") from None
```

`main()`, right after `DB = db_open(DB_PATH)`:

```python
        PROVIDERS.update(load_providers(CFG["cloud"]))
```

`__main__`, before the final `else`:

```python
    elif len(sys.argv) == 3 and sys.argv[1] == "--cloud-models":
        print(json.dumps(cloud_models(sys.argv[2])))
```

Module docstring, the Run paragraph becomes:

```
Run: python3 assistant.py        (--test runs the self-check, --defaults
prints the built-in config as JSON, for a settings UI, --cloud-models NAME
prints that [[cloud]] provider's model ids as JSON)
```

- [ ] **Step 4: Run, expect `ok`**

- [ ] **Step 5: Commit**

```bash
git add assistant.py
git commit -m "cloud: provider keys from key_cmd, routing by model prefix, --cloud-models

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>"
```

---

### Task 3: The conversation's model in the database

**Files:**
- Modify: `assistant.py` (`db_open` ~145, `save_turn` ~158, `test()` storage section)

- [ ] **Step 1: Write the failing tests**

In `test()`, after `assert load_conv(db, 999) is None`:

```python
    m = save_turn(db, None, "q", "r", [], [], model="fake:x")
    pick = lambda: db.execute("SELECT model FROM conversations WHERE id = ?", (m,)).fetchone()[0]
    assert pick() == "fake:x"
    save_turn(db, m, "q", "r", [], [], model="local")
    assert pick() == "local"
    save_turn(db, m, "q", "r", [], [])
    assert pick() is None                            # an ask with no choice: the loaded one
    delete_conv(db, m)
```

In the old-database check (`# An old database gains the files column.`), after its assert:

```python
        assert "model" in {r[1] for r in db_open(os.path.join(d, "h.db"))
                           .execute("PRAGMA table_info(conversations)")}
```

The last assert relies on the default: the chat's stored model is what its turns were asked with. Task 4 always passes the chat's current choice, so a NULL write only happens for a chat that never had one.

- [ ] **Step 2: Run, expect failure**

Expected: `TypeError: save_turn() got an unexpected keyword argument 'model'`.

- [ ] **Step 3: Implement**

`db_open`, after the `files` column migration:

```python
    # Added with cloud models: the model a conversation is asked with, NULL
    # for whatever the router has loaded.
    if "model" not in {r[1] for r in conn.execute("PRAGMA table_info(conversations)")}:
        conn.execute("ALTER TABLE conversations ADD COLUMN model TEXT")
```

`save_turn`: signature `def save_turn(conn, conv, user, reply, tools, messages, files=(), model=None):`, and the two conversation statements become:

```python
        if conv is None:
            conv = conn.execute("INSERT INTO conversations(title, created, updated, model) "
                                "VALUES (?, ?, ?, ?)", (title, now, now, model)).lastrowid
        else:
            conn.execute("UPDATE conversations SET updated = ?, model = ? WHERE id = ?",
                         (now, model, conv))
```

- [ ] **Step 4: Run, expect `ok`**

- [ ] **Step 5: Commit**

```bash
git add assistant.py
git commit -m "history: keep the model a conversation is asked with

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>"
```

---

### Task 4: Requests go where the chat's model lives

**Files:**
- Modify: `assistant.py` (`TOOL_SCHEMA` line 111, `cur` ~416, `hello`, `start_new`, `handle`, `command`, `model_inputs`, `ask`, `stream`, `reply`, `answer`, `retitle`, `test()`)

- [ ] **Step 1: Write the failing tests**

At the top of `test()`, next to the other imports:

```python
    import contextlib
    import io
```

`global DB, TOOLS, TOOL_SCHEMA` becomes `global DB, TOOLS`, and delete the line `TOOL_SCHEMA = schema(TOOLS)`.

In `Fake.do_POST`, right after the title branch's `return` (before `script["body"] = body`):

```python
            script.get("paths", []).append(self.path)
```

First hello check: after `assert h["pending"] is None and "ctx_max" in h` add:

```python
        assert h["chat_model"] is None
```

The two places comparing a fresh conversation event change to carry the model:

```python
        assert until("conversation")[-1] == {"type": "conversation", "id": None, "turns": [],
                                             "model": None}
```

```python
        assert {"type": "conversation", "id": None, "turns": [], "model": None} in seen
```

Right before `# llama-server down: an error, not a dead thread.`:

```python
        # Cloud: the chat's model picks the provider, and only cloud tools
        # go along. A call to another one, copied from history, gets no
        # output. The key reaches the provider and nothing else.
        saved = TOOLS
        TOOLS = {"now": {"description": "d", "command": ["sh", "-c", "echo noon"], "cloud": True},
                 "mail": {"description": "m", "command": ["sh", "-c", "echo secret subject"]}}
        logged = io.StringIO()
        with contextlib.redirect_stdout(logged):
            send(type="new")
            until("conversation")
            script["bodies"], script["paths"] = [], []
            script["replies"] = [
                [{"reasoning": "hm"},
                 {"tool_calls": [{"index": 0, "id": "k1", "function": {"name": "mail",
                                                                       "arguments": "{}"}}]}],
                [{"content": "No mail tool."}]]
            send(type="ask", text="any mail?", model="fake:vendor/x:free")
            seen = until("done")
            first, second = script["bodies"]
            assert script["paths"] == ["/cloud/v1/chat/completions"] * 2
            assert script["auth"] == "Bearer sk-test"
            assert first["model"] == "vendor/x:free" and "chat_template_kwargs" not in first
            assert [t["function"]["name"] for t in first["tools"]] == ["now"]
            assert second["messages"][-1] == {"role": "tool", "tool_call_id": "k1",
                                              "content": "error: no tool named mail"}
            assert ("think", "hm") in [(e["type"], e.get("delta")) for e in seen]
            meters = [e for e in seen if e["type"] == "ctx"]
            assert meters[-1]["model"] == "fake:vendor/x:free" and meters[-1]["max"] == 0
            assert "sk-test" not in json.dumps(seen)
            conv = cur["id"]
            assert DB.execute("SELECT model FROM conversations WHERE id = ?",
                              (conv,)).fetchone()[0] == "fake:vendor/x:free"
            # The next ask with no model stays on the chat's.
            script["replies"] = [[{"content": "Still cloud."}]]
            send(type="ask", text="again")
            until("done")
            assert script["paths"][-1] == "/cloud/v1/chat/completions"
            # Reopened, the chat reports its model.
            send(type="new")
            assert until("conversation")[-1]["model"] is None
            send(type="open", id=conv)
            assert until("conversation")[-1]["model"] == "fake:vendor/x:free"
            # A local pick goes to the router with every tool and the
            # thinking switch.
            script["replies"] = [[{"content": "Local."}]]
            send(type="ask", text="local now", model="m")
            until("done")
            assert script["paths"][-1] == "/v1/chat/completions"
            assert script["body"]["model"] == "m" and "chat_template_kwargs" in script["body"]
            assert [t["function"]["name"] for t in script["body"]["tools"]] == ["now", "mail"]
            # The dropdown's list: the router's models, then the cloud ones
            # whose key loaded.
            send(type="models")
            items = until("models")[-1]["items"]
            assert {"id": "m", "status": "loaded", "cloud": False} in items
            assert {"id": "fake:vendor/x:free", "provider": "fake", "cloud": True} in items
            assert not [i for i in items if i["id"].startswith("broken:")]
            send(type="ask", text="x", model="")
            assert until("error")[-1]["message"] == "bad model"
            send(type="ask", text="x", model="broken:y")
            e = until("error")[-1]
            assert "no key" in e["message"] and e["turn"] is True
            send(type="new")
            until("conversation")
        TOOLS = saved
        assert "sk-test" not in logged.getvalue(), logged.getvalue()
```

- [ ] **Step 2: Run, expect failure**

Expected: `KeyError: 'chat_model'` at the first hello.

- [ ] **Step 3: Implement the socket side**

Delete `TOOL_SCHEMA = schema(TOOLS) or None` (line 111): `answer` builds the schema per chat now.

`cur`:

```python
cur = {"id": None, "history": [], "turns": [], "pending": None, "model": None}
```

`hello()` gains the chat's choice (its `model` stays the one answering, from `ctx`):

```python
    return {"type": "hello", **now_state, "model": ctx["model"], "ctx_used": ctx["used"],
            "ctx_max": ctx["max"], "id": cur["id"], "turns": cur["turns"],
            "pending": cur["pending"], "chat_model": cur["model"]}
```

`start_new()`:

```python
    cur.update(id=None, history=[], turns=[], model=None)
```

`handle()`, in the `bad = (...)` chain, after the `"bad audio"` line:

```python
           else "bad model" if kind == "ask" and "model" in cmd and not (
               isinstance(cmd["model"], str) and cmd["model"])
```

`command()`:

```python
    elif kind == "ask":
        text, files, audio = cmd["text"].strip(), cmd.get("files", []), cmd.get("audio")
        if text or files or audio:
            interrupt()
            # A pick applies from this ask on and stays with the chat.
            cur["model"] = cmd.get("model", cur["model"])
            threading.Thread(target=reply, args=(turn, text, files, audio, cur["model"]),
                             daemon=True).start()
```

`new`: `emit({"type": "conversation", "id": None, "turns": [], "model": None})`.

`open`, replacing its last two lines:

```python
        interrupt()
        model = DB.execute("SELECT model FROM conversations WHERE id = ?",
                           (cmd["id"],)).fetchone()[0]
        cur.update(id=cmd["id"], history=got[0], turns=got[1], model=model)
        emit({"type": "conversation", "id": cur["id"], "turns": cur["turns"], "model": model})
```

`delete` of the current one: `emit({"type": "conversation", "id": None, "turns": [], "model": None})`.

New branch before the final `else`:

```python
    elif kind == "models":
        # Off the lock: the router can take seconds to answer.
        threading.Thread(target=send_models, daemon=True).start()
```

New function after `cloud_models`:

```python
def send_models():
    # The canvas's model dropdown: the router's models, then every cloud
    # model whose provider's key loaded.
    try:
        with urllib.request.urlopen(CFG["llm"] + "/models", timeout=10) as r:
            local = [{"id": m["id"], "status": m["status"]["value"], "cloud": False}
                     for m in json.load(r)["data"]]
    except (OSError, KeyError, TypeError, ValueError):
        local = []                        # router down: the cloud ones still show
    emit({"type": "models", "items": local + [
        {"id": f"{n}:{m}", "provider": n, "cloud": True}
        for n, p in PROVIDERS.items() for m in p["models"]]})
```

- [ ] **Step 4: Implement the request side**

`model_inputs`:

```python
def model_inputs(model=None):
    # (id, inputs) of the model a chat uses: its pick, else whatever the
    # router has resident (asking for another would swap it out under other
    # clients). inputs is None when not reported, as for every cloud model.
    if model and route(model)[3]:
        return model, None
    with urllib.request.urlopen(CFG["llm"] + "/models", timeout=10) as r:
        models = json.load(r)["data"]
    m = (next((m for m in models if m["id"] == model), None)
         or next((m for m in models if m["status"]["value"] != "unloaded"), models[0]))
    if model and m["id"] != model:
        log(f"no local model {model}; using {m['id']}")
    return m["id"], (m.get("architecture") or {}).get("input_modalities")
```

`ask`, signature and the lines up to the connection:

```python
def ask(messages, calls, tools=None, mine=None, think=None, meter=True, model=None):
    # Yields ("think", text) and ("token", text) as they stream; tool-call
    # fragments go into calls. mine is the reply's turn: a stop shuts the
    # connection from the other thread (interrupt), which ends the reading.
    # model is the id model_inputs gave, None for the router's loaded one.
    url, key, mid, cloud = route(model or llm_model())
    if cloud:
        ctx.update(model=model, max=0)    # no /props: the meter shows tokens used only
    else:
        context_size(mid)
    body = {"model": mid, "messages": messages, "stream": True,
            "stream_options": {"include_usage": True}}
    if not cloud:
        # llama.cpp's own field; strict providers refuse unknown ones.
        body["chat_template_kwargs"] = {
            "enable_thinking": CFG["thinking"] if think is None else think}
    if tools:
        body["tools"] = tools
    if CFG["temperature"] is not None:
        body["temperature"] = CFG["temperature"]
    # http.client rather than urllib: it exposes the socket a stop must shut.
    u = urllib.parse.urlsplit(url + "/chat/completions")
```

and its `yield from` line:

```python
        yield from stream(conn, u.path, body, calls, meter, key, cloud or "llama-server")
```

`stream`:

```python
def stream(conn, path, body, calls, meter=True, key=None, who="llama-server"):
    headers = {"Content-Type": "application/json"}
    if key:
        headers["Authorization"] = f"Bearer {key}"
    conn.request("POST", path, json.dumps(body).encode(), headers)
    r = conn.getresponse()
    if r.status >= 400:
        # The body says why (a full context, a bad request, a spent quota);
        # the status alone does not.
        raise RuntimeError(f"{who} {r.status}: {r.read().decode(errors='replace')[:2000]}")
```

In its loop, the reasoning lines become:

```python
            # llama-server sends reasoning_content, OpenRouter reasoning.
            think = delta.get("reasoning_content") or delta.get("reasoning")
            if think:
                yield "think", think
```

and the last line: `raise RuntimeError(other.decode(errors="replace")[:300] or f"{who} sent nothing")`.

`reply`:

```python
def reply(mine, text, files=(), audio=None, model=None):
    try:
        if audio:
            # Always the local model: the audio never leaves, only the words.
            text = transcribe(mine, audio)
            if text is None:
                return
        answer(mine, text, files, model)
```

`answer`, signature `def answer(mine, text, files=(), chosen=None):`; replace

```python
    model, inputs = model_inputs()
    images = inputs is None or "image" in inputs
```

with

```python
    model, inputs = model_inputs(chosen)
    images = inputs is None or "image" in inputs
    # A cloud chat is offered only the tools marked cloud = true: the others
    # would send mail, money or calendar to the provider. One of them called
    # anyway (copied from history) answers "no tool named".
    offered = ({n: t for n, t in TOOLS.items() if t.get("cloud") is True}
               if route(model)[3] else TOOLS)
    offer = schema(offered) or None
```

In the round loop:

```python
        for kind, piece in ask(sent, calls, offer if rnd < CFG["tool_rounds"] else None,
                               mine, model=model):
```

```python
            result = run_tool(offered, c["name"], c["arguments"])
```

Save and title:

```python
        cur["id"] = save_turn(DB, cur["id"], text, said, used, entry, chips, chosen)
```

```python
    if fresh:
        retitle(mine, conv, text, said, model)
```

`retitle`: signature `def retitle(mine, conv, user, said, model=None):`, and its ask call gets `model=model`:

```python
        text = "".join(p for k, p in ask(msgs, {}, None, mine, think=False, meter=False,
                                         model=model)
                       if k == "token")
```

- [ ] **Step 5: Run, expect `ok`**

Also `grep -n TOOL_SCHEMA assistant.py` must print nothing.

- [ ] **Step 6: Commit**

```bash
git add assistant.py
git commit -m "cloud: a chat's model routes its requests, cloud chats get cloud tools only

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>"
```

---

### Task 5: Examples and docs

**Files:**
- Modify: `examples/config.toml`, `examples/tools/{calc,now,search,weather}.toml`, `README.md`, `AGENTS.md`, `docs/superpowers/specs/2026-10-23-cloud-models-design.md`

- [ ] **Step 1: Tool flags**

```bash
cd "$(git rev-parse --show-toplevel)"
for t in calc now search weather; do sed -i '/^command = /a cloud = true' examples/tools/$t.toml; done
grep -c '^cloud = true' examples/tools/*.toml | grep -v ':0'
```

Expected: exactly `calc`, `now`, `search`, `weather`, each `:1`.

- [ ] **Step 2: `examples/config.toml`**, append:

```toml

# Cloud models, picked per chat in the canvas header, never a fallback.
# One table per provider with an OpenAI-compatible API. key_cmd prints the
# API key and runs without a shell, once at startup: the key is never
# stored here. models are the ids the canvas offers; the settings page
# fetches the list with `assistant.py --cloud-models <name>`. A cloud chat
# is offered only the tools whose file says cloud = true.
# [[cloud]]
# name = "openrouter"
# url = "https://openrouter.ai/api/v1"
# key_cmd = "pass show llm/openrouter"
# models = ["mistralai/mistral-large"]
```

Then verify it still parses: `python3 -c "import tomllib; tomllib.load(open('examples/config.toml','rb'))"` (no output).

- [ ] **Step 3: `README.md`**

Opening paragraph, last sentence (`The model and the history stay on the machine; only tools you enable (weather, web search) reach the network.`) becomes:

```
The history stays on the machine. So does the model, unless a chat is
switched to a cloud model you configured; only tools you enable (weather,
web search) reach the network otherwise.
```

In the tool TOML example, after `timeout = 10`:

```toml
cloud = false                        # true: also offered to cloud models
```

After the `config.toml` bullet, a new bullet:

```
- `[[cloud]]` in `config.toml`: providers with an OpenAI-compatible API,
  each with a `key_cmd` that prints its key (see the example). A chat
  picks a model in the canvas header; a cloud chat is sent its history,
  attachments and the results of the tools marked `cloud = true`, and
  nothing from the others. Voice is always transcribed locally.
```

- [ ] **Step 4: `AGENTS.md`**

Line 5-6, `Nothing leaves the machine except what an enabled tool fetches.` becomes:

```
Nothing leaves the machine except what an enabled tool fetches, and a
chat the user switched to a cloud model (2026-10-23).
```

After the `Tools (2026-09-26)` paragraph, a new one:

```
Cloud models (2026-10-23): `[[cloud]]` providers in `config.toml`, keys
from `key_cmd`, ids `provider:model`, chosen per chat and stored on the
conversation. Only tools with `cloud = true` go to a cloud chat. Spec:
`docs/superpowers/specs/2026-10-23-cloud-models-design.md`.
```

Agent rule `- Everything stays local: no cloud STT/TTS/LLM fallbacks.` becomes:

```
- No silent cloud fallback, ever. A cloud model is used only when the user
  picks it for a chat; tools reach a cloud chat only with `cloud = true`;
  audio is always transcribed locally; no cloud STT/TTS.
```

- [ ] **Step 5: Spec status**

`Date: 2026-10-23. Status: approved, not implemented.` becomes `Date: 2026-10-23. Status: daemon implemented (2026-10-23); quickshell side pending.`

- [ ] **Step 6: Self-check and commit**

Run the self-check, expect `ok`. Then:

```bash
git add examples README.md AGENTS.md docs/superpowers/specs/2026-10-23-cloud-models-design.md
git commit -m "docs: cloud models, example provider and cloud tools

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>"
```

---

### Task 6: Live check (with the user)

Not automatable: it needs a real provider and key. Ask the user before each outward step.

- [ ] **Step 1:** Ask the user which provider, its `key_cmd`, and one or two model ids. Ask before touching `~/.config/desktop-assistant/` (their live config): add the `[[cloud]]` table, and add `cloud = true` to their installed `calc`, `now`, `search`, `weather` tool files.
- [ ] **Step 2:** `python3 assistant.py --cloud-models <name> | head -c 300`: a JSON array of ids.
- [ ] **Step 3:** Run a second daemon on a scratch socket and data dir, as the clone tool check did (AGENTS.md, "Clone tool, live check"), so the real history is untouched. Short path: a Unix socket path is limited to 108 bytes.
- [ ] **Step 4:** Through it, with `model` set to the cloud id: a weather question (tool `weather`, answered), "how many unread mails do I have?" (no mail tool offered: must say it cannot tell), a voice clip from `/data/voice-models/samples/stt/` (`user` event from the local model, answer from the cloud), then the same chat with a local `model` and back. Record first-token times, `ctx` used, and anything the provider refused (unknown fields such as `stream_options`).
- [ ] **Step 5:** Write the findings section into AGENTS.md (`### Cloud models, live check (2026-10-23)`), ask before committing.