"""Opt-in live e2e for the session command surface against a real provider
(deepseek by default): a real model drives the typed `commands` bindings
through the real tool loop.
Needs a real key: reads the deepseek profile's key from albedo's creds.json
(or the file $ALBEDO_TEST_CREDS names).
Not part of test.sh; run manually:
python3 test/manual/commands_live.py
Proves, with no fixtures in the path: the minted typed methods and their
help/signature, the model-mode read of /model, the model-mode refusal to
switch, the skills command with exact argument round-trip, the served command
catalog, and a user-mode model switch that the next real request actually
uses.
"""
import contextlib
import glob
import json
import os
from pathlib import Path
import subprocess
import tempfile
import time
import urllib.request
ROOT = Path(__file__).resolve().parents[2]
CREDS_PATH = os.environ.get("ALBEDO_TEST_CREDS") or os.path.join(
os.environ.get("ALBEDO_HOME") or os.path.expanduser("~/.albedo"), "creds.json"
)
KEY = json.load(open(CREDS_PATH))["providers"]["deepseek"]["apiKey"]
MODEL = "deepseek-chat"
SKILL_BODY = "LIVE_SKILL_BODY_7391"
CELL_ONE = """
import inspect
print("NAMES", sorted(n for n in vars(type(commands)) if not n.startswith("_")))
print("SIG", inspect.signature(commands.model))
print("DOC", commands.model.__doc__.replace(chr(10), " | "))
print("SEL", await commands.model())
print("CAT", [(c["name"], c["method"], c.get("usage")) for c in await commands.catalog()])
act = await commands.probe("one two")
print("ACT", repr(act["arguments"]), "LIVE_SKILL_BODY_7391" in act["instructions"])
print("INVOKED", (await commands.invoke("/probe", "raw text"))["arguments"])
try:
await commands.model("unreachable-model")
print("SWITCH-BUG")
except CommandsError as e:
print("REFUSED", e)
"""
PROMPT_ONE = (
"use the python tool to run exactly this one cell and nothing else, then "
"end your turn with the single word done:\n" + CELL_ONE
)
PROMPT_TWO = (
"use the python tool to run exactly this one cell and nothing else, then "
"end your turn with the single word done:\n"
"print('NOW', (await commands.model())['model'])"
)
def main():
with tempfile.TemporaryDirectory(prefix="albedo-live-commands-") as directory:
root = Path(directory)
home, workspace, user_home = root / "home", root / "workspace", root / "user"
for path in (home, workspace, user_home):
path.mkdir(mode=0o700)
skill = workspace / ".albedo" / "skills" / "probe" / "SKILL.md"
skill.parent.mkdir(parents=True)
skill.write_text(
"---\nname: probe\ndescription: live probe skill\n---\n" + SKILL_BODY + "\n"
)
(home / "config.json").write_text(
json.dumps(
{
"active": "deepseek",
"providers": {
"deepseek": {
"baseUrl": "https://api.deepseek.com/v1",
"apiKey": KEY,
"model": MODEL,
"protocol": "chat_completions",
}
},
}
)
)
(home / "extensions.json").write_text(
json.dumps({"models": {"refreshHours": 0}})
)
env = dict(
os.environ,
HOME=str(user_home),
ALBEDO_HOME=str(home),
ALBEDO_IDLE_SECONDS="600",
)
toolchain = ":".join(
glob.glob("/nix/store/*gleam*/bin") + glob.glob("/nix/store/*erlang*/bin")
)
if toolchain:
env["PATH"] = toolchain + ":" + env["PATH"]
first = subprocess.run(
[str(ROOT / "cli/bin/albedo"), "sessions"],
cwd=ROOT,
env=env,
text=True,
capture_output=True,
timeout=120,
)
assert first.returncode == 0, first.stdout + first.stderr
connection = json.loads((home / "daemon.json").read_text())
base = f"http://127.0.0.1:{connection['port']}"
headers = {
"Authorization": "Bearer " + connection["token"],
"Content-Type": "application/json",
}
def api(path, body=None):
request = urllib.request.Request(
base + path,
headers=headers,
data=None if body is None else json.dumps(body).encode(),
)
return urllib.request.urlopen(request, timeout=60)
def ready(session, timeout=180):
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
with api(f"/sessions/{session}/status") as response:
state = json.load(response)
if not state["running"]:
return
time.sleep(1)
raise AssertionError(f"session never settled: {state}")
def outputs(session):
"""The first stream page after reset is the full history snapshot."""
with api(f"/sessions/{session}/stream") as response:
response.fp.raw._sock.settimeout(60)
while True:
line = response.readline()
if not line:
return []
if line.startswith(b"data: "):
return json.loads(line[6:])["events"]
created = json.loads(
subprocess.run(
[str(ROOT / "cli/bin/albedo"), "new", str(workspace)],
cwd=ROOT,
env=env,
text=True,
capture_output=True,
timeout=120,
check=True,
).stdout
)
session = created["session"]
print("session", session)
with api(f"/sessions/{session}/commands") as response:
catalog = json.load(response)
by_name = {c["name"]: c for c in catalog}
assert "/model" in by_name and "/context" in by_name and "/probe" in by_name, (
by_name.keys()
)
assert by_name["/model"]["usage"] == "/model [model] [provider]", by_name[
"/model"
]
assert (
by_name["/probe"]["method"] == "probe" and by_name["/probe"]["userTurn"]
), by_name["/probe"]
print(
"PASS catalog serves /model /context /probe with usage and mintable methods"
)
print("turn 1 submitted", flush=True)
with api(f"/sessions/{session}/events", {"content": PROMPT_ONE}) as response:
response.read()
ready(session)
print("turn 1 settled, reading stream", flush=True)
tools = [e for e in outputs(session) if e.get("type") == "tool"]
text = json.dumps([e.get("result", "") for e in tools])
print("turn 1 tool outputs:", len(tools), flush=True)
for marker in (
"NAMES",
"SIG",
"DOC",
"SEL",
"CAT",
"ACT",
"INVOKED",
"REFUSED",
):
assert marker in text, (marker, text[-2000:])
assert "['context', 'model', 'probe']" in text, text[-2000:]
assert "INVOKED raw text" in text, text[-2000:]
assert "(model=None, provider=None)" in text, text[-2000:]
assert "Usage: /model [model] [provider]" in text, text[-2000:]
assert f"'{MODEL}'" in text, text[-2000:]
assert "'/model', 'model', '/model [model] [provider]'" in text, text[-2000:]
assert "'one two' True" in text, (
"exact arguments + skill body round-trip",
text[-2000:],
)
assert "SWITCH-BUG" not in text and "user action between turns" in text, text[
-2000:
]
print("PASS real model used the typed bindings: help/signature, read, catalog,")
print(" skill with exact args, and the model-mode switch refusal")
switch_to = MODEL
try:
with urllib.request.urlopen(
urllib.request.Request(
"https://api.deepseek.com/models",
headers={"Authorization": "Bearer " + KEY},
),
timeout=15,
) as response:
listed = [m["id"] for m in json.load(response)["data"]]
candidates = [m for m in listed if m != MODEL and "reasoner" not in m] or [
m for m in listed if m != MODEL
]
if candidates:
switch_to = candidates[0]
except Exception as probe_error:
print("model list unavailable:", probe_error)
if switch_to == MODEL:
print("SKIP switch: no second deepseek model to switch to")
else:
with api(
f"/sessions/{session}/commands",
{"name": "/model", "args": {"model": switch_to}},
) as response:
selection = json.load(response)
assert selection["result"]["model"] == switch_to, selection
assert selection["result"]["provider"] == "deepseek", selection
with api(
f"/sessions/{session}/events", {"content": PROMPT_TWO}
) as response:
response.read()
ready(session)
tools = [e for e in outputs(session) if e.get("type") == "tool"]
assert f"NOW {switch_to}" in json.dumps(tools), json.dumps(tools)[-2000:]
print(f"PASS user switch landed: the next real request ran on {switch_to}")
with contextlib.suppress(Exception):
api("/shutdown", {}).close()
print("PASS: live command surface e2e")
if __name__ == "__main__":
main()