"""Opt-in live wake check against a real provider (deepseek by default).
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/job_wake_live.py
Boots the real daemon, submits one prompt that starts an unawaited background
job, then watches the event stream: the wake turn must arrive on its own while
nobody polls the python kernel, and the model must be able to read the job's
output from the notice it is given.
"""
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"]
PROMPT = (
"use the python tool to run exactly this cell and nothing else: "
"import sys; job = run(sys.executable, '-c', 'import time; time.sleep(25); print(\"WAKE-DONE-42\")'); print(job.id). "
"then end your turn with a one-sentence confirmation. do not await the job, "
"do not poll it, do not run anything else."
)
def main():
with tempfile.TemporaryDirectory(prefix="albedo-live-wake-") as directory:
home = Path(directory) / "home"
workspace = Path(directory) / "workspace"
home.mkdir(mode=0o700)
workspace.mkdir()
(home / "config.json").write_text(
json.dumps(
{
"active": "deepseek",
"providers": {
"deepseek": {
"baseUrl": "https://api.deepseek.com/v1",
"apiKey": KEY,
"model": "deepseek-chat",
"protocol": "chat_completions",
}
},
}
)
)
env = dict(
os.environ,
HOME=str(Path(directory) / "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)
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_id = created["session"]
print("session", session_id)
with api(f"/sessions/{session_id}/events", {"content": PROMPT}) as response:
response.read()
print("submitted; waiting for the first run to end")
deadline = time.monotonic() + 180
while time.monotonic() < deadline:
with api(f"/sessions/{session_id}/status") as response:
status = json.load(response)
if not status["running"]:
break
time.sleep(1)
else:
raise AssertionError("first run never settled: " + json.dumps(status))
print(
"idle at",
round(time.monotonic() % 100, 1),
"s; job still running; now only listening",
)
# Hold the stream open: no status polling, no python calls, nothing that
# could carry the completion back except the wake itself.
with api(f"/sessions/{session_id}/stream") as response:
response.fp.raw._sock.settimeout(150)
seen = []
wake_at = None
while True:
line = response.readline()
if not line:
break
if not line.startswith(b"data: "):
continue
page = json.loads(line[6:])
seen.extend(page["events"])
users = [
e
for e in page["events"]
if e.get("type") == "user" and "job finished" in e.get("text", "")
]
if users and wake_at is None:
wake_at = time.monotonic()
print("WAKE TURN ARRIVED:")
print(" ", users[0]["source"], "|", users[0]["text"][:220])
messages = [e for e in page["events"] if e.get("type") == "message"]
if wake_at is not None and messages:
print("MODEL REPLIED TO THE WAKE:")
print(" ", messages[0]["text"][:600])
break
if (
time.monotonic() > wake_at + 120
if wake_at
else time.monotonic() > deadline + 150
):
break
assert wake_at is not None, "no wake arrived; events: " + json.dumps(
[e.get("type") for e in seen]
)
with contextlib.suppress(Exception):
api("/shutdown", {}).close()
print("PASS: the session woke itself and the model answered the wake")
if __name__ == "__main__":
main()