From 6fca162e4f8ce88c3a0dfd9e1de85b4576d854d3 Mon Sep 17 00:00:00 2001
From: dawn <90008@klbr.net>
Date: Tue, 22 Sep 2026 15:26:19 +0300
Subject: [PATCH] background bash jobs wake idle sessions, local and remote
---
docs/extensions.md | 4 +
docs/files.md | 2 +-
priv/python/albedo_api.py | 4 +
priv/python/albedo_kernel.py | 19 +-
priv/python/albedo_plugins/bash.py | 85 ++++++-
priv/python/albedo_plugins/remote.py | 113 ++++++++-
src/albedo/daemon/albedo_wakes.erl | 62 +++++
src/albedo/daemon/reaper.gleam | 14 +-
src/albedo/daemon/server.gleam | 1 +
src/albedo/daemon/session.gleam | 43 +++-
src/albedo/harness/bash.gleam | 71 +++++-
src/albedo/harness/python/albedo_python.erl | 20 +-
src/albedo/harness/python/kernel.gleam | 5 +
src/albedo/harness/remote.gleam | 5 +-
src/albedo/harness/runtime.gleam | 7 +
test.sh | 2 +
test/daemon/job_wake_integration.py | 262 ++++++++++++++++++++
test/daemon/reaper_test.gleam | 25 +-
test/harness/job_wake_route_test.gleam | 91 +++++++
test/harness/job_wake_test.py | 214 ++++++++++++++++
test/manual/job_wake_live.py | 122 +++++++++
21 files changed, 1131 insertions(+), 40 deletions(-)
create mode 100644 src/albedo/daemon/albedo_wakes.erl
create mode 100644 test/daemon/job_wake_integration.py
create mode 100644 test/harness/job_wake_route_test.gleam
create mode 100644 test/harness/job_wake_test.py
create mode 100644 test/manual/job_wake_live.py
diff --git a/docs/extensions.md b/docs/extensions.md
index 4768923..ae6ba68 100644
--- a/docs/extensions.md
+++ b/docs/extensions.md
@@ -61,6 +61,10 @@ packaged short names resolve to `albedo_plugins.*`; dotted names select installe
interfaces: [`PythonApi`](../priv/python/albedo_api.py), [`work`](../src/albedo/harness/work.gleam), [`compaction strategy`](../src/albedo/harness/compaction.gleam).
+## bash
+
+`bash(command)` starts a supervised job that owns its process group; output stays on the handle (`job.tail()`, `output.read(job.id)`) and finished handles stay addressable in `jobs`. a job that finishes with its result unread wakes the session: the kernel reports the completion through the `jobs` host route, and the session submits a user turn naming the job, its exit, and how to read it, so the model never polls or awaits a handle just to learn it finished. awaiting the job, reading its result, or stopping it retires the wake. a busy session is retried, not queued in gleam: the kernel retries the notice until the run ends or the result is read. live jobs pin the kernel against the idle sweep — a detached session is not released while background work still runs, local or remote, because releasing it would kill the job and the wake it owes. remote `bash` jobs wake the same way — the remote kernel's host calls relay over the ssh connection into this session's route, and the notice names the host — and a mirror read (`rem_job.tail()`) tells the remote job it was read through one `poll` round trip.
+
## files
see [the files extension](files.md): bounded reads, guarded exact edits, and ripgrep search that runs as a supervised `bash` job instead of a raw subprocess.
diff --git a/docs/files.md b/docs/files.md
index 4b717f5..8f85aa4 100644
--- a/docs/files.md
+++ b/docs/files.md
@@ -2,7 +2,7 @@
`files` gives the model file vocabulary inside the Python session: bounded reads, one directory listing, exact edits, and search. It is enabled by default and depends on `python` and `bash`.
-Nothing here spawns a process directly. When ripgrep is installed, search runs as a supervised `bash` job, so every child belongs to a job's process group and is ended with it. Without ripgrep the same calls fall back to pure Python.
+Nothing here spawns a process directly. When ripgrep is installed, search runs as a supervised `bash` job, so every child belongs to a job's process group and is ended with it. Without ripgrep the same calls fall back to pure Python. An awaited search behaves like any awaited job; one abandoned mid-run finishes under the same wake contract as [bash](extensions.md#bash) — its result read, no wake; unread, the session is woken when it lands.
```python
files.read("src/app.py", start_line=40, end_line=80)
diff --git a/priv/python/albedo_api.py b/priv/python/albedo_api.py
index 0390b7e..5cf49e3 100644
--- a/priv/python/albedo_api.py
+++ b/priv/python/albedo_api.py
@@ -38,6 +38,10 @@ class PythonApi:
# The module list this kernel was started with, so a plugin can learn the
# session's tool selection (remote boots the same set on another machine).
modules: Sequence[str] = ()
+ # Registers one callback for a retained output channel, fired when the
+ # model reads it, so background work can withdraw a completion wake the
+ # read already satisfied. Absent when the host pre-dates the hook.
+ watch_output: Callable[[str, Callable[[], None]], None] | None = None
class Execute(TypedDict):
diff --git a/priv/python/albedo_kernel.py b/priv/python/albedo_kernel.py
index 8331923..7862031 100644
--- a/priv/python/albedo_kernel.py
+++ b/priv/python/albedo_kernel.py
@@ -234,10 +234,25 @@ def retained(id: str) -> Capture:
return capture
+READ_WATCHERS: dict[str, Callable[[], None]] = {}
+
+
+def watch_output(id: str, notify: Callable[[], None]) -> None:
+ """Register one read callback for a retained output channel (a background job)."""
+ READ_WATCHERS[id] = notify
+
+
class Output:
def read(self, id: str, *, offset: int = 0, limit: int = 4000) -> str:
"""Read retained output by cell or job id. At most 1 MiB each, 16 recent cells."""
- return retained(id).read(offset, limit)
+ capture = retained(id)
+ notify = READ_WATCHERS.pop(id, None) # one read satisfies the watcher
+ if notify is not None:
+ try:
+ notify()
+ except Exception:
+ pass # a plugin's bookkeeping must not fail the read
+ return capture.read(offset, limit)
def list(self):
"""Every retained channel: cells, background jobs, and 'native'.
@@ -847,7 +862,7 @@ def main():
modules = cast(list[str], json.loads(sys.argv[1]))
api = albedo_api.PythonApi(version=2, loop=LOOP, host=host, HostError=WorkError,
capture=background_capture, preview=PREVIEW, send=send, on_shutdown=CLEANUP.append,
- background_handle=HANDLES.append, modules=modules)
+ background_handle=HANDLES.append, modules=modules, watch_output=watch_output)
NAMESPACE.update(cells=Cells(), output=Output())
try:
albedo_api.load_plugins(modules, api, NAMESPACE)
diff --git a/priv/python/albedo_plugins/bash.py b/priv/python/albedo_plugins/bash.py
index c7c4365..0b325d2 100644
--- a/priv/python/albedo_plugins/bash.py
+++ b/priv/python/albedo_plugins/bash.py
@@ -4,14 +4,21 @@ A job runs in its own session, so its shell leads a process group and one signal
reaches the command plus every descendant that stayed in that group. The group
is ended when the command ends, when its deadline passes, and at kernel
shutdown. What survives is reported, and unfinished work keeps its running slot.
+
+A job that finishes with its result unread wakes the session: the kernel tells
+the host, the host submits a user turn naming the job, and the model never has
+to poll or await. Reading the result (tail, poll, await, output.read) or
+stopping the job withdraws the wake, and a busy session is retried until it
+goes idle, so a notice can never overtake the read that satisfies it.
"""
from __future__ import annotations
from collections import OrderedDict
-from collections.abc import Callable, Generator
+from collections.abc import Awaitable, Callable, Generator
from albedo_api import PythonApi, OutputCapture, Send
import albedo_proc
import asyncio
+import os
import uuid
loop: asyncio.AbstractEventLoop
@@ -21,6 +28,8 @@ jobs: dict[str, Job] = {} # every handle the session can
active: dict[str, Job] = {} # unfinished work: the bounded resource
retained: OrderedDict[str, Job] = OrderedDict() # finished handles, completion order
send: Send
+host: Callable[[dict[str, object]], Awaitable[object]] | None = None
+watch_output: Callable[[str, Callable[[], None]], None] | None = None
ACTIVE_LIMIT = 64 # jobs owning running processes at once
RETAINED_LIMIT = 64 # finished handles still addressable through `jobs`
@@ -28,6 +37,8 @@ COMPLETION_GRACE = 0.1 # seconds to keep reading after the command exits
EXIT_INTERVAL = 0.01 # seconds between exit checks while the command runs
SHUTDOWN_TERM = 0.25 # shared by every live group at shutdown
SHUTDOWN_KILL = 1.0
+NOTICE_RETRY = 2.0 # seconds between wake attempts while the session runs
+NOTICE_COMMAND_CAP = 200 # command characters a wake notice carries
class Job:
@@ -45,6 +56,11 @@ class Job:
self.termination: albedo_proc.Termination | None = None
self.capture: OutputCapture = capture_factory(self.id)
self.ending: asyncio.Task[albedo_proc.Termination] | None = None
+ self._awaited = False # someone awaited this job; its result reached them
+ self._read = False # the finished result was read; no wake is owed
+ self._remote = os.environ.get("ALBEDO_REMOTE_TARGET") or None
+ if watch_output is not None:
+ watch_output(self.id, self._mark_read)
self.spawning = loop.create_task(asyncio.create_subprocess_shell(
self.command, stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.STDOUT, start_new_session=True))
@@ -63,7 +79,7 @@ class Job:
def _cancelled(self, task: asyncio.Task[Job]) -> None:
if task.cancelled():
# Cancellation before _run starts skips its finally block.
- loop.create_task(self.stop())
+ loop.create_task(self._stop())
async def _spawn(self) -> asyncio.subprocess.Process:
process = await asyncio.shield(self.spawning)
@@ -82,7 +98,7 @@ class Job:
except Exception as error:
self.capture.write(f"{type(error).__name__}: {error}\n")
finally:
- ending = await self.stop()
+ ending = await self._stop()
if self.exit_code is None and self.process is not None:
self.exit_code = await self._status(self.process)
self.duration = loop.time() - self.started
@@ -93,6 +109,8 @@ class Job:
send({"type": "job", "id": self.id, "exit_code": self.exit_code,
"timed_out": self.timed_out, "cleanup": ending.as_json()})
release(self)
+ if host is not None and not self._awaited:
+ _ = loop.create_task(self._announce())
return self
async def _drain(self, process: asyncio.subprocess.Process) -> int:
@@ -129,22 +147,74 @@ class Job:
self.capture.write(chunk.decode("utf-8", errors="replace"))
def __await__(self) -> Generator[object, None, Job]:
+ self._awaited = True
return asyncio.shield(self.task).__await__()
def poll(self):
+ if self.exit_code is not None:
+ self._read = True
return self.exit_code
@property
def returncode(self) -> int | None:
"""`exit_code` under subprocess's name; both spellings answer."""
- return self.exit_code
+ return self.poll()
def tail(self, n: int = 4000) -> str:
+ if self.exit_code is not None:
+ self._read = True
data = self.capture.tail_data
return bytes(data[-max(1, min(n, preview_limit)):]).decode("utf-8", errors="replace")
+ def _mark_read(self) -> None:
+ """output.read reached this job's channel; a finished result is read."""
+ if self.exit_code is not None:
+ self._read = True
+
+ async def _announce(self) -> None:
+ """Wake the session for this job's result, retrying while it runs.
+
+ A busy session answers with a refusal rather than queueing, so the
+ notice retries until the run ends; a read or await in the meantime
+ retires it, which is also the creating-cell barrier: a cell that will
+ await its own job suppresses the wake before any turn can start.
+ """
+ while not self._read and not self._awaited:
+ try:
+ await host("jobs.completed", self._notice())
+ return
+ except Exception as error:
+ if getattr(error, "code", "") != "busy":
+ self.capture.write(f"\n[completion notice not delivered: {error}\n")
+ return
+ await asyncio.sleep(NOTICE_RETRY)
+
+ def _notice(self) -> dict[str, object]:
+ """The wake turn's display text, model text, and the facts behind both."""
+ command = self.command[:NOTICE_COMMAND_CAP] + (
+ "..." if len(self.command) > NOTICE_COMMAND_CAP else "")
+ where = f" on {self._remote}" if self._remote else ""
+ outcome = ("timed out" if self.timed_out else f"exit_code={self.exit_code}"
+ if self.exit_code is not None else "exit status unknown")
+ seconds = f"{self.duration:.1f}s" if self.duration is not None else "unknown duration"
+ display = f"bash job finished{where} ({outcome}, {seconds}): {command}"
+ text = (
+ "a background bash job finished with its result unread"
+ f"{where}: {outcome}, {seconds}, command: {command}."
+ f" its handle is jobs[{self.id!r}] in python; jobs[{self.id!r}].tail() or"
+ f" output.read({self.id!r}) reads its output."
+ " no user sent this message; use the result if the session's work needs"
+ " it, otherwise acknowledge briefly and stay idle.")
+ return {"display": display, "text": text, "id": self.id,
+ "exit_code": self.exit_code, "timed_out": self.timed_out,
+ "duration": self.duration, "host": self._remote}
+
async def stop(self) -> albedo_proc.Termination:
"""End the job's process group. A failed attempt stays retryable."""
+ self._read = True # an explicit stop is its own report; no wake is owed
+ return await self._stop()
+
+ async def _stop(self) -> albedo_proc.Termination:
if self.termination is not None and self.termination.gone:
return self.termination
process = self.process
@@ -200,6 +270,11 @@ def bash(command: object, *, timeout: float = 300) -> Job:
for the retained whole. Jobs survive cell completion and each owns its process
group. Only unfinished work counts against the concurrent quota; finished
handles stay usable in `jobs` until newer ones displace them.
+
+ A job that finishes with its result unread wakes the session by itself, so
+ polling or trailing a handle is optional: leave it in a variable, end the
+ cell, and the wake names the job when it lands. Awaiting the job, reading
+ its result, or stopping it first means no wake.
"""
if not isinstance(command, str) or not 0 < timeout <= 3600:
raise ValueError("command must be text; 0 < timeout <= 3600 required")
@@ -226,7 +301,9 @@ async def close() -> None:
def setup(api: PythonApi) -> dict[str, object]:
global loop, capture_factory, preview_limit, jobs, active, retained, send
+ global host, watch_output
loop, capture_factory, preview_limit, send = api.loop, api.capture, api.preview, api.send
+ host, watch_output = api.host, api.watch_output
jobs, active, retained = {}, {}, OrderedDict()
api.background_handle(Job)
api.on_shutdown(close)
diff --git a/priv/python/albedo_plugins/remote.py b/priv/python/albedo_plugins/remote.py
index 857be42..b9dfd98 100644
--- a/priv/python/albedo_plugins/remote.py
+++ b/priv/python/albedo_plugins/remote.py
@@ -43,7 +43,22 @@ loop: asyncio.AbstractEventLoop
host_call: Any
HostErrorType: type[Exception]
capture_factory: Any
+send_frame: Any
session_modules: list[str] = []
+_fallback_live: set[str] = set() # degraded-mode jobs still running over ssh
+
+
+def _report_live() -> None:
+ """Tell the local supervisor how many remote jobs are still live.
+
+ Releasing the kernel would kill every ssh client and with them the remote
+ kernels, so the idle sweep keeps a kernel alive while remote work runs.
+ """
+ try:
+ live = len(_fallback_live) + sum(len(c._live) for c in connections)
+ send_frame({"type": "jobs", "live": live})
+ except Exception:
+ pass # no channel or shutdown; the supervisor keeps the last count
configured: dict[str, str | None] = {}
connections: list["RemoteConnection"] = []
@@ -255,6 +270,7 @@ class RemoteRef:
self._call_id = call_id
self._label = label
self._settled: Any = _UNSET
+ self._withdrawn: bool = False
# --- mirrored job state: sync, like the local handle ---
@@ -300,7 +316,17 @@ class RemoteRef:
def _mirror(self) -> dict[str, Any] | None:
if self._handle is None:
return None
- return self._connection._mirrors.get(self._handle)
+ state = self._connection._mirrors.get(self._handle)
+ # A finished job read through its mirror is consumed on the remote side
+ # too, or its kernel would keep retrying a wake this read satisfies.
+ if (state is not None and state.get("exit_code") is not None
+ and not self._withdrawn):
+ self._withdrawn = True
+ try:
+ _ = loop.create_task(self._connection.mark_read(self._handle))
+ except RuntimeError:
+ pass # the loop is closing; the notice dies with the kernel
+ return state
# --- the value surface: identical output to the local tool ---
@@ -431,6 +457,7 @@ class RemoteConnection:
self._refs: dict[str, RemoteRef] = {}
self._mirrors: dict[str, dict[str, Any]] = {}
self._handshake: asyncio.Future[dict[str, Any]] | None = None
+ self._live: set[str] = set()
self._stderr_tail = bytearray()
self._reader: asyncio.Task[None] | None = None
self._drain: asyncio.Task[None] | None = None
@@ -449,7 +476,10 @@ class RemoteConnection:
if code != 0:
await stage(self.host)
python = self.target.get("python") or "python3"
- script = f"exec {shlex.quote(python)} -u {shlex.quote(script_path)}" \
+ # The stamp reaches the remote kernel's plugins, so a finished remote
+ # job's wake notice names the machine it ran on.
+ script = f"ALBEDO_REMOTE_TARGET={shlex.quote(self.host)} exec " \
+ + f"{shlex.quote(python)} -u {shlex.quote(script_path)}" \
+ " " + shlex.quote(json.dumps(modules if modules is not None
else default_modules()))
if self.target.get("remote_cwd"):
@@ -511,8 +541,14 @@ class RemoteConnection:
_ = loop.create_task(self._relay(frame))
elif kind == "mirror" and isinstance(frame.get("handle"), str):
self._mirrors[frame["handle"]] = frame
- # trace, done, job, job_start, cleanup describe remote cells and jobs;
- # those are addressed through references, not events on this side.
+ elif kind == "job_start" and isinstance(frame.get("id"), str):
+ self._live.add(frame["id"])
+ _report_live()
+ elif kind == "job" and isinstance(frame.get("id"), str):
+ self._live.discard(frame["id"])
+ _report_live()
+ # trace, done, cleanup describe remote cells; those are addressed
+ # through references, not events on this side.
async def _relay(self, frame: dict[str, Any]) -> None:
"""Answer one remote host-route call against this session's daemon."""
@@ -645,6 +681,18 @@ class RemoteConnection:
self._refs.pop(handle, None)
await self._send({"type": "release", "handle": handle})
+ async def mark_read(self, handle: str) -> None:
+ """Tell the remote object its result was read, retiring its wake.
+
+ A local mirror read never crosses the channel on its own, so this one
+ round trip (a poll, which the job counts as a read) is the only way a
+ finished remote job learns the model already has its output.
+ """
+ try:
+ await self.invoke(handle=handle, name="poll", timeout=30)
+ except Exception:
+ pass # the channel is gone; the notice fails its own way
+
async def tools(self) -> dict[str, Any]:
"""What the remote namespace holds, and the live references it kept."""
if self.degraded is not None:
@@ -736,6 +784,8 @@ class RemoteConnection:
except asyncio.TimeoutError:
self._process.kill()
self.closed = True
+ self._live.clear()
+ _report_live()
for future in self._pending.values():
if not future.done():
future.set_exception(RemoteLost(f"the connection to {self.host} closed"))
@@ -812,10 +862,14 @@ class FallbackJob:
self.started: float = loop.time()
self.duration: float | None = None
self.capture = capture_factory(self.id)
+ self._awaited = False
+ self._read = False
+ self._connection = connection
+ _fallback_live.add(self.id)
+ _report_live()
script = f"bash -c {shlex.quote(command)}"
if connection.target.get("remote_cwd"):
script = f"cd {shlex.quote(connection.target['remote_cwd'])} && {script}"
- self._connection = connection
self._task = loop.create_task(self._run(connection.host, script, timeout))
async def _run(self, host: str, script: str, timeout: float) -> "FallbackJob":
@@ -836,29 +890,72 @@ class FallbackJob:
finally:
copying.cancel()
self.duration = loop.time() - self.started
+ _fallback_live.discard(self.id)
+ _report_live()
+ if host_call is not None and not self._awaited:
+ _ = loop.create_task(self._announce())
return self
+ async def _announce(self) -> None:
+ """Wake the session for this command, retrying while it runs."""
+ while not self._read and not self._awaited:
+ try:
+ await host_call("jobs.completed", self._notice())
+ return
+ except HostErrorType as error:
+ if getattr(error, "code", "") != "busy":
+ self.capture.write(f"\n[completion notice not delivered: {error}\n")
+ return
+ except Exception:
+ return # the kernel is going away; the result stays on the handle
+ await asyncio.sleep(2.0)
+
+ def _notice(self) -> dict[str, object]:
+ """The wake turn's display text, model text, and the facts behind both."""
+ command = self.command[:200] + ("..." if len(self.command) > 200 else "")
+ outcome = ("timed out" if self.timed_out else f"exit_code={self.exit_code}"
+ if self.exit_code is not None else "exit status unknown")
+ seconds = f"{self.duration:.1f}s" if self.duration is not None else "unknown duration"
+ display = f"bash job finished on {self._connection.host} ({outcome}, {seconds}): {command}"
+ text = (
+ "a background bash command finished with its result unread on"
+ f" {self._connection.host} (ssh command mode): {outcome}, {seconds},"
+ f" command: {command}."
+ f" its handle is jobs[{self.id!r}] in python; jobs[{self.id!r}].tail() or"
+ f" output.read({self.id!r}) reads its output."
+ " no user sent this message; use the result if the session's work needs"
+ " it, otherwise acknowledge briefly and stay idle.")
+ return {"display": display, "text": text, "id": self.id,
+ "exit_code": self.exit_code, "timed_out": self.timed_out,
+ "duration": self.duration, "host": self._connection.host}
+
async def _copy(self, stream: Any) -> None:
while chunk := await stream.read(8192):
self.capture.write(chunk.decode("utf-8", errors="replace"))
def __await__(self):
+ self._awaited = True
return asyncio.shield(self._task).__await__()
def poll(self) -> int | None:
+ if self.exit_code is not None:
+ self._read = True
return self.exit_code
@property
def returncode(self) -> int | None:
"""`exit_code` under subprocess's name; both spellings answer."""
- return self.exit_code
+ return self.poll()
def tail(self, n: int = 4000) -> str:
+ if self.exit_code is not None:
+ self._read = True
data = self.capture.tail_data
return bytes(data[-max(1, min(n, 65536)):]).decode("utf-8", errors="replace")
async def stop(self) -> None:
"""End the ssh client; the remote command's fate is reported, not assumed."""
+ self._read = True # an explicit stop is its own report; no wake is owed
process = getattr(self, "_process", None)
if process is not None and process.returncode is None:
process.terminate()
@@ -947,9 +1044,11 @@ remote = Remote()
def setup(api: PythonApi) -> dict[str, object]:
- global loop, host_call, HostErrorType, capture_factory, session_modules
+ global loop, host_call, HostErrorType, capture_factory, send_frame
+ global session_modules
loop, host_call = api.loop, api.host
HostErrorType, capture_factory = api.HostError, api.capture
+ send_frame = api.send
session_modules = list(api.modules)
api.on_shutdown(Remote.close_all)
return {"remote": remote, "RemoteError": RemoteError}
diff --git a/src/albedo/daemon/albedo_wakes.erl b/src/albedo/daemon/albedo_wakes.erl
new file mode 100644
index 0000000..4993684
--- /dev/null
+++ b/src/albedo/daemon/albedo_wakes.erl
@@ -0,0 +1,62 @@
+%% Session wake registry: session id -> the submit closure its actor registered.
+%%
+%% The kernel reports finished background jobs through a host route, which runs
+%% outside any session actor, so the notice needs a lookup from session id to
+%% the actor that can submit a turn. The table is owned by a process nothing
+%% else depends on, so it outlives every individual session and dies with the
+%% daemon. Registered closures answer "" when the wake was submitted and a
+%% refusal message otherwise; "session is busy" is the one the kernel retries.
+-module(albedo_wakes).
+-export([register/2, forget/1, deliver/3]).
+
+-define(TABLE, albedo_wakes).
+
+register(Id, Fun) ->
+ ensure_table(),
+ true = ets:insert(?TABLE, {Id, Fun}),
+ nil.
+
+forget(Id) ->
+ case ets:whereis(?TABLE) of
+ undefined -> nil;
+ _ -> ets:delete(?TABLE, Id), nil
+ end.
+
+%% The refusal the caller should relay, or "" when the wake was submitted.
+deliver(Id, Display, Text) ->
+ case ets:whereis(?TABLE) of
+ undefined -> <<"session unavailable">>;
+ _ ->
+ case ets:lookup(?TABLE, Id) of
+ [{_, Fun}] when is_function(Fun, 2) ->
+ try Fun(Display, Text)
+ catch _:_ -> <<"session unavailable">>
+ end;
+ _ -> <<"session unavailable">>
+ end
+ end.
+
+ensure_table() ->
+ case ets:whereis(?TABLE) of
+ undefined ->
+ _ = spawn(fun table_owner/0),
+ wait_table(200);
+ _ -> ok
+ end.
+
+%% The owner needs a moment to create the table; bounded, because a table that
+%% never appears should fail loudly in register, not hang here.
+wait_table(0) -> ok;
+wait_table(N) ->
+ case ets:whereis(?TABLE) of
+ undefined -> receive after 1 -> ok end, wait_table(N - 1);
+ _ -> ok
+ end.
+
+table_owner() ->
+ try ets:new(?TABLE, [public, named_table, {read_concurrency, true}])
+ catch _:_ -> ok %% a concurrent creator won the name; ours is theirs
+ end,
+ receive stop -> ok
+ after infinity -> ok
+ end.
diff --git a/src/albedo/daemon/reaper.gleam b/src/albedo/daemon/reaper.gleam
index 08f9e06..23b45a0 100644
--- a/src/albedo/daemon/reaper.gleam
+++ b/src/albedo/daemon/reaper.gleam
@@ -1,14 +1,16 @@
//// Which idle kernels to release. Pure policy; the caller does the releasing.
////
-//// A session that is attached or running is never a victim, though the memory
-//// it holds still counts against the pool: the point is to reclaim what nobody
-//// is waiting on, not to bound how many sessions a person may keep open.
+//// A session that is attached, running, or holding live background jobs is
+//// never a victim, though the memory it holds still counts against the pool:
+//// the point is to reclaim what nobody is waiting on, not to bound how many
+//// sessions a person may keep open. Live jobs pin the kernel because releasing
+//// it kills the process groups those jobs still own and the wake they owe.
import gleam/int
import gleam/list
pub type Candidate {
- Candidate(pid: Int, idle_ms: Int, running: Bool, kilobytes: Int)
+ Candidate(pid: Int, idle_ms: Int, running: Bool, jobs: Int, kilobytes: Int)
}
pub type Limits {
@@ -21,7 +23,9 @@ pub fn victims(candidates: List(Candidate), limits: Limits) -> List(Candidate) {
// A limit shorter than the detachment grace must still be honourable.
let grace = int.min(limits.detached_ms, limits.idle_ms)
let idle =
- list.filter(candidates, fn(one) { !one.running && one.idle_ms >= grace })
+ list.filter(candidates, fn(one) {
+ !one.running && one.jobs == 0 && one.idle_ms >= grace
+ })
let #(expired, resident) =
list.partition(idle, fn(one) { one.idle_ms >= limits.idle_ms })
let total = list.fold(candidates, 0, fn(sum, one) { sum + one.kilobytes })
diff --git a/src/albedo/daemon/server.gleam b/src/albedo/daemon/server.gleam
index 35bafc3..ef0b919 100644
--- a/src/albedo/daemon/server.gleam
+++ b/src/albedo/daemon/server.gleam
@@ -347,6 +347,7 @@ fn reap(workers: List(session.Session), config: Config) -> Nil {
pid,
report.idle_ms,
report.running,
+ report.jobs,
list.key_find(usage, pid) |> result.unwrap(0),
)
})
diff --git a/src/albedo/daemon/session.gleam b/src/albedo/daemon/session.gleam
index 88d2204..56bc33a 100644
--- a/src/albedo/daemon/session.gleam
+++ b/src/albedo/daemon/session.gleam
@@ -44,7 +44,13 @@ pub type Page {
/// What a sweep needs to decide whether this session's kernel can be released.
pub type Report {
- Report(running: Bool, kernel: Option(Int), history_loaded: Bool, idle_ms: Int)
+ Report(
+ running: Bool,
+ kernel: Option(Int),
+ history_loaded: Bool,
+ idle_ms: Int,
+ jobs: Int,
+ )
}
pub type ModelSelection {
@@ -70,6 +76,7 @@ pub type Message {
String,
String,
String,
+ String,
Option(types.Image),
Subject(Result(Nil, SubmissionError)),
)
@@ -164,6 +171,17 @@ pub fn start(
unprepared(),
now_ms(),
)
+ // Background jobs wake this session through the kernel's jobs route; the
+ // registered closure lands a completion notice as an ordinary submit, so
+ // the wake reuses the whole turn pipeline and busy answers itself.
+ wakes_register(info.id, fn(display, text) {
+ case
+ actor.call(self, 10_000, Submit(display, text, "bash", "bash", None, _))
+ {
+ Ok(Nil) -> ""
+ Error(error) -> submission_error(error)
+ }
+ })
Ok(
actor.initialised(state)
|> actor.returning(self)
@@ -188,7 +206,7 @@ pub fn submit(
client_id: String,
image: Option(types.Image),
) -> Result(Nil, SubmissionError) {
- actor.call(session, 10_000, Submit(text, text, client_id, image, _))
+ actor.call(session, 10_000, Submit(text, text, client_id, "chat", image, _))
}
pub fn skills(
@@ -359,7 +377,7 @@ fn handle(state: State, message: Message) {
}
_ -> actor.continue(state)
}
- Submit(display, text, client_id, image, reply) ->
+ Submit(display, text, client_id, source, image, reply) ->
case state.run {
Some(_) -> {
process.send(reply, Error(Rejected("session is busy")))
@@ -440,7 +458,7 @@ fn handle(state: State, message: Message) {
Some(image) ->
view.user_image(
display,
- "chat",
+ source,
Some(client_id),
Some(timestamp),
image,
@@ -448,7 +466,7 @@ fn handle(state: State, message: Message) {
None ->
view.user(
display,
- "chat",
+ source,
Some(client_id),
Some(timestamp),
)
@@ -545,6 +563,7 @@ fn handle(state: State, message: Message) {
display,
skill_catalog.activation_prompt(activation),
client_id,
+ "chat",
None,
reply,
),
@@ -956,6 +975,12 @@ fn handle(state: State, message: Message) {
Some(kernel) -> option.from_result(runtime.kernel_pid(kernel))
None -> None
}
+ // Live background jobs keep their kernel: releasing it would kill work
+ // the session still owes a wake for, so the reaper counts them.
+ let jobs = case state.kernel {
+ Some(kernel) -> runtime.job_count(kernel)
+ None -> 0
+ }
process.send(
reply,
Report(
@@ -963,6 +988,7 @@ fn handle(state: State, message: Message) {
kernel,
state.history != None,
now_ms() - state.last_touch,
+ jobs,
),
)
actor.continue(state)
@@ -1013,6 +1039,7 @@ fn handle(state: State, message: Message) {
None, None -> Nil
}
runtime.reset_session(state.host, state.info.id)
+ wakes_forget(state.info.id)
process.send(reply, Nil)
actor.stop()
}
@@ -1028,6 +1055,12 @@ fn kill(pid: process.Pid) -> Nil
@external(erlang, "albedo_session", "now_ms")
fn now_ms() -> Int
+@external(erlang, "albedo_wakes", "register")
+fn wakes_register(session: String, submit: fn(String, String) -> String) -> Nil
+
+@external(erlang, "albedo_wakes", "forget")
+fn wakes_forget(session: String) -> Nil
+
@external(erlang, "albedo_session", "discard")
fn discard(path: String) -> Nil
diff --git a/src/albedo/harness/bash.gleam b/src/albedo/harness/bash.gleam
index e49d924..c7e1d28 100644
--- a/src/albedo/harness/bash.gleam
+++ b/src/albedo/harness/bash.gleam
@@ -1,15 +1,80 @@
+//// Run supervised shell commands from Python.
+
+import albedo/daemon/store
import albedo/harness/extension as harness_extension
+import gleam/dynamic/decode
+import gleam/json
+
+const instructions = "bash(command) starts a background job. Await its handle for completion, or keep it in a variable. jobs contains session-owned handles. A job's output lives on its handle (job.tail()) and stays readable in full with output.read(job.id). job.exit_code, job.duration, and job.timed_out report the exit status, wall seconds, and deadline state once the command ends. A job that finishes with its result unread wakes the session by itself, so polling a handle is optional: leave it in a variable, end the cell, and a notice naming the job arrives when it lands. Awaiting the job, reading its result, or stopping it first means no wake."
pub fn extension() -> harness_extension.Extension {
- harness_extension.python_module(
+ harness_extension.Extension(
"bash",
"Run supervised shell commands from Python.",
- "bash",
- "bash(command) starts a background job. Await its handle for completion, or keep it in a variable. jobs contains session-owned handles. A job's output lives on its handle (job.tail()) and stays readable in full with output.read(job.id). job.exit_code, job.duration, and job.timed_out report the exit status, wall seconds, and deadline state once the command ends.",
["python"],
+ [
+ harness_extension.ToolPlugin(instructions, [], ["bash"], [
+ #("jobs", route),
+ ]),
+ ],
+ fn(_) { Ok(Nil) },
)
}
pub fn plugin() -> harness_extension.Extension {
extension()
}
+
+/// The kernel's wake route: a finished background job reports itself here, and
+/// the submit closure its session registered turns the notice into an ordinary
+/// user turn, so the model never polls for completion. The reply's code is the
+/// kernel's retry signal: "busy" retries, anything else gives up.
+pub fn route(_store: store.Store, session: String, request: String) -> String {
+ case json.parse(request, request_decoder()) {
+ Ok(#("jobs.completed", args)) ->
+ case decode.run(args, notice_decoder()) {
+ Ok(notice) ->
+ case deliver(session, notice.display, notice.text) {
+ "" ->
+ json.to_string(
+ json.object([
+ #("ok", json.bool(True)),
+ #("value", json.string("delivered")),
+ ]),
+ )
+ "session is busy" -> refused("busy", "session is busy")
+ message -> refused("unavailable", message)
+ }
+ Error(_) -> refused("invalid", "invalid jobs notice")
+ }
+ _ -> refused("invalid", "unknown jobs operation")
+ }
+}
+
+fn request_decoder() -> decode.Decoder(#(String, decode.Dynamic)) {
+ use method <- decode.field("method", decode.string)
+ use args <- decode.field("args", decode.dynamic)
+ decode.success(#(method, args))
+}
+
+type Notice {
+ Notice(display: String, text: String)
+}
+
+fn notice_decoder() -> decode.Decoder(Notice) {
+ use display <- decode.field("display", decode.string)
+ use text <- decode.field("text", decode.string)
+ decode.success(Notice(display, text))
+}
+
+@external(erlang, "albedo_wakes", "deliver")
+fn deliver(session: String, display: String, text: String) -> String
+
+fn refused(code: String, message: String) -> String {
+ json.object([
+ #("ok", json.bool(False)),
+ #("code", json.string(code)),
+ #("message", json.string(message)),
+ ])
+ |> json.to_string
+}
diff --git a/src/albedo/harness/python/albedo_python.erl b/src/albedo/harness/python/albedo_python.erl
index e7c6417..abf4b4b 100644
--- a/src/albedo/harness/python/albedo_python.erl
+++ b/src/albedo/harness/python/albedo_python.erl
@@ -7,7 +7,7 @@
%% the kernel is wedged, and it returns a structured verdict instead of
%% kill(1) exit statuses the supervisor would have to guess at.
-module(albedo_python).
--export([start/6, execute/3, interrupt/1, stop/1, events/1, alive/1, os_pid/1, local_paths/0]).
+-export([start/6, execute/3, interrupt/1, stop/1, events/1, alive/1, os_pid/1, job_count/1, local_paths/0]).
-define(STARTUP_TIMEOUT, 5000).
-define(SHUTDOWN_GRACE, 2000). %% must exceed the kernel's own cleanup deadline
@@ -34,7 +34,8 @@ start(Owner, Python, Script, Cwd, Host, Modules) ->
when is_integer(KernelPid), KernelPid > 1 ->
Parent ! {Ref, {ok, self()}},
loop(#{port => Port, host => Host, active => none,
- events => [], groups => #{}, target => target_of(Ready)});
+ events => [], groups => #{}, external => 0,
+ target => target_of(Ready)});
#{<<"type">> := <<"startup_error">>, <<"message">> := Message} when is_binary(Message) ->
reap_start(Port), Parent ! {Ref, {error, {unavailable, Message}}};
_ -> reap_start(Port), Parent ! {Ref, {error, {unavailable, <<"invalid kernel handshake">>}}}
@@ -51,6 +52,15 @@ start(Owner, Python, Script, Cwd, Host, Modules) ->
end.
execute(Pid, Data, Timeout) -> call(Pid, {execute, Data, Timeout}).
+
+%% Background jobs whose groups are still owned: local job groups plus the
+%% remote jobs the remote plugin reported through "jobs" frames. A released
+%% kernel would kill them, so the idle sweep keeps such kernels alive.
+job_count(Pid) ->
+ case call(Pid, job_count) of
+ {ok, Count} when is_integer(Count), Count >= 0 -> Count;
+ _ -> 0
+ end.
events(Pid) -> case call(Pid, events) of {ok, Events} -> Events; _ -> [] end.
interrupt(Pid) -> Pid ! interrupt, nil.
@@ -94,6 +104,9 @@ loop(S = #{port := Port, active := Active}) ->
{call, From, Ref, os_pid} ->
#{target := #{pid := KernelPid}} = S,
From ! {Ref, {ok, KernelPid}}, loop(S);
+ {call, From, Ref, job_count} ->
+ Count = maps:size(maps:get(groups, S)) + maps:get(external, S, 0),
+ From ! {Ref, {ok, Count}}, loop(S);
{call, From, Ref, events} ->
From ! {Ref, {ok, lists:reverse(maps:get(events, S))}}, loop(S#{events => []});
{call, From, Ref, stop} ->
@@ -154,6 +167,9 @@ handle(#{<<"type">> := <<"job">>, <<"id">> := Id} = Message, Data, S) ->
Cleanup -> log({job_cleanup_failed, Id, Cleanup}), maps:get(groups, S)
end,
loop(S#{events => lists:sublist([Data | maps:get(events, S)], 100), groups => Groups});
+handle(#{<<"type">> := <<"jobs">>, <<"live">> := Live}, _, S)
+ when is_integer(Live), Live >= 0 ->
+ loop(S#{external => Live});
handle(#{<<"type">> := <<"trace">>}, Data, S) ->
loop(S#{events => lists:sublist([Data | maps:get(events, S)],100)});
handle(#{<<"type">> := <<"cleanup">>, <<"failures">> := Failures}, _, S) ->
diff --git a/src/albedo/harness/python/kernel.gleam b/src/albedo/harness/python/kernel.gleam
index 0360ab8..e71ad24 100644
--- a/src/albedo/harness/python/kernel.gleam
+++ b/src/albedo/harness/python/kernel.gleam
@@ -188,6 +188,11 @@ fn saved_decoder() -> decode.Decoder(Result(Saved, Error)) {
@external(erlang, "albedo_python", "os_pid")
pub fn os_pid(kernel: Kernel) -> Result(Int, Nil)
+/// Live background jobs the kernel still supervises, local groups plus remote
+/// jobs its remote plugin reported. Zero when the kernel cannot answer.
+@external(erlang, "albedo_python", "job_count")
+pub fn job_count(kernel: Kernel) -> Int
+
@external(erlang, "albedo_python", "interrupt")
pub fn interrupt(kernel: Kernel) -> Nil
diff --git a/src/albedo/harness/remote.gleam b/src/albedo/harness/remote.gleam
index 0f651bd..542200c 100644
--- a/src/albedo/harness/remote.gleam
+++ b/src/albedo/harness/remote.gleam
@@ -18,7 +18,10 @@ pub fn extension() -> harness_extension.Extension {
<> " immediately and synchronously, like local bash: the handle exists right away,"
<> " job.tail(), job.id, job.exit_code, job.duration, and job.timed_out answer"
<> " synchronously from the mirrored output stream,"
- <> " await job waits for completion, and await job.stop() stops it. Every other call"
+ <> " await job waits for completion, and await job.stop() stops it. A remote"
+ <> " job that finishes with its result unread wakes the session by itself,"
+ <> " naming the host it ran on; awaiting it, reading its result, or stopping"
+ <> " it first means no wake. Every other call"
<> " returns a reference that settles on its first await: await rem.files.find(...)"
<> " resolves to its value, await rem.work.list() to the daemon's answer. Results"
<> " cross as real objects, so remote output looks exactly like local output; a"
diff --git a/src/albedo/harness/runtime.gleam b/src/albedo/harness/runtime.gleam
index ef32f1d..c4dc686 100644
--- a/src/albedo/harness/runtime.gleam
+++ b/src/albedo/harness/runtime.gleam
@@ -193,6 +193,13 @@ pub fn kernel_pid(session: Session) -> Result(Int, Nil) {
python.os_pid(session.kernel)
}
+/// Background jobs whose groups the kernel still owns, local or remote. A
+/// released kernel would kill them, so the idle sweep keeps kernels with
+/// live jobs alive.
+pub fn job_count(session: Session) -> Int {
+ python.job_count(session.kernel)
+}
+
pub fn save_state(
session: Session,
path: String,
diff --git a/test.sh b/test.sh
index dba53f0..6e6adda 100755
--- a/test.sh
+++ b/test.sh
@@ -12,6 +12,7 @@ gleam format --check src test
gleam test
python3 test/harness/python_plugin_test.py
python3 test/harness/files_plugin_test.py
+python3 test/harness/job_wake_test.py
python3 test/harness/remote_kernel_test.py
python3 test/harness/remote_plugin_test.py
@@ -25,3 +26,4 @@ python3 test/daemon/mcp_integration.py
python3 test/daemon/context_integration.py
python3 test/daemon/kernel_reset_integration.py
python3 test/daemon/idle_reap_integration.py
+python3 test/daemon/job_wake_integration.py
diff --git a/test/daemon/job_wake_integration.py b/test/daemon/job_wake_integration.py
new file mode 100644
index 0000000..56dd147
--- /dev/null
+++ b/test/daemon/job_wake_integration.py
@@ -0,0 +1,262 @@
+"""A background job wakes an idle session end-to-end, through the real daemon.
+
+The fake provider answers the first turn with a python tool call that starts a
+job and returns, answers the second with plain text so the run ends, and then
+must be asked a third time: the job finishing with its result unread submits
+the wake turn through the kernel's jobs route, the registry, and the session
+actor's registered submit closure.
+"""
+import contextlib
+import glob
+import http.server
+import json
+import os
+from pathlib import Path
+import subprocess
+import tempfile
+import threading
+import time
+import urllib.request
+
+ROOT = Path(__file__).resolve().parents[2]
+
+
+def toolchain_path():
+ """gleam and erlang when they only exist as nix store results."""
+ entries = ["/nix/store/*gleam*/bin", "/nix/store/*erlang*/bin"]
+ found = [path for pattern in entries for path in glob.glob(pattern)]
+ return ":".join(found)
+
+
+TOOLCHAIN = toolchain_path()
+
+def content_text(item):
+ content = item.get("content", "")
+ if isinstance(content, str):
+ return content
+ if isinstance(content, list):
+ return " ".join(part["text"] for part in content
+ if isinstance(part, dict) and isinstance(part.get("text"), str))
+ return ""
+
+
+class Provider(http.server.BaseHTTPRequestHandler):
+ requests = []
+ lock = threading.Lock()
+
+ def log_message(self, *_):
+ pass
+
+ def do_POST(self):
+ request = json.loads(self.rfile.read(int(self.headers["Content-Length"])))
+ chat = self.path.endswith("chat/completions")
+ inputs = request["messages" if chat else "input"]
+ tool_turns = sum(item.get("role") == "tool" or item.get("type") == "function_call_output"
+ for item in inputs)
+ tool_turns = sum(item.get("role") == "tool" or item.get("type") == "function_call_output"
+ for item in inputs)
+ latest_user = next((content_text(item) for item in reversed(inputs)
+ if item.get("role") == "user"), "")
+ with self.lock:
+ self.requests.append({"request": request, "latest_user": latest_user})
+ # A fresh user request starts a job; the tool turn after it and the wake
+ # turn (whose user message is the notice) both end the run.
+ last = inputs[-1] if inputs else {}
+ starts = last.get("role") == "user" and "start a slow job" in latest_user
+ done = not starts
+ sleep = "20" if "detached" in latest_user else "1.2"
+ code = f'job = bash("sleep {sleep}; echo wake-done")\njob.id'
+ call_id = f"call-{tool_turns + 1}"
+ arguments = json.dumps({"code": code, "timeout_ms": 60000})
+ self.send_response(200)
+ self.send_header("Content-Type", "text/event-stream")
+ self.end_headers()
+
+ def event(value):
+ self.wfile.write(("data: " + json.dumps(value) + "\n\n").encode())
+ self.wfile.flush()
+ try:
+ if chat:
+ if not done:
+ for i in range(0, len(arguments), 20):
+ function = {"arguments": arguments[i:i+20]}
+ call = {"index": 0, "function": function}
+ if i == 0:
+ function["name"] = "python"
+ call.update(id=call_id, type="function")
+ event({"id": "r1", "choices": [{"index": 0, "delta": {"tool_calls": [call]},
+ "finish_reason": None}]})
+ event({"id": "r1", "choices": [{"index": 0, "delta": {},
+ "finish_reason": "tool_calls"}]})
+ else:
+ event({"id": "r2", "choices": [{"index": 0, "delta": {"content": "finished"},
+ "finish_reason": None}]})
+ event({"id": "r2", "choices": [{"index": 0, "delta": {},
+ "finish_reason": "stop"}]})
+ self.wfile.write(b"data: [DONE]\n\n")
+ self.wfile.flush()
+ else:
+ event({"type": "response.created", "response": {"id": "r2" if done else "r1"}})
+ if not done:
+ for i in range(0, len(arguments), 20):
+ event({"type": "response.function_call_arguments.delta",
+ "output_index": 0, "delta": arguments[i:i+20]})
+ output = [{"id": "fc1", "type": "function_call", "call_id": call_id,
+ "name": "python", "arguments": arguments, "status": "completed"}]
+ else:
+ event({"type": "response.output_text.delta", "output_index": 0,
+ "content_index": 0, "delta": "finished"})
+ output = [{"id": "m1", "type": "message", "role": "assistant",
+ "status": "completed",
+ "content": [{"type": "output_text", "text": "finished",
+ "annotations": []}]}]
+ event({"type": "response.completed",
+ "response": {"id": "r2" if done else "r1", "status": "completed",
+ "output": output,
+ "usage": {"input_tokens": 10, "output_tokens": 20}}})
+ except (BrokenPipeError, ConnectionResetError):
+ pass
+
+
+def run(protocol, endpoint):
+ with tempfile.TemporaryDirectory(prefix="albedo-wake-e2e-") as directory:
+ home = Path(directory) / "home"
+ workspace = Path(directory) / "workspace"
+ home.mkdir(mode=0o700)
+ workspace.mkdir()
+ (home / "extensions.json").write_text(json.dumps({"models": {"refreshHours": 0}}))
+ (home / "config.json").write_text(json.dumps({
+ "active": "alpha",
+ "providers": {"alpha": {"baseUrl": endpoint + "/alpha/v1", "apiKey": "key",
+ "model": "fixture-alpha", "protocol": protocol}},
+ }))
+ env = dict(os.environ, HOME=str(Path(directory) / "user-home"), ALBEDO_HOME=str(home),
+ ALBEDO_IDLE_SECONDS="10")
+ import shutil
+ if TOOLCHAIN and shutil.which("gleam") is None:
+ env["PATH"] = TOOLCHAIN + ":" + env.get("PATH", "")
+ for obsolete in ("ALBEDO_API_KEY", "ALBEDO_BASE_URL", "ALBEDO_MODEL", "ALBEDO_PROTOCOL"):
+ env.pop(obsolete, None)
+ connection = None
+ requests_at_start = len(Provider.requests)
+ try:
+ first = subprocess.run(["node", "cli/bin/albedo.mjs", "sessions"], cwd=ROOT,
+ env=env, text=True, capture_output=True, timeout=45)
+ 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):
+ req = urllib.request.Request(base + path, headers=headers,
+ data=None if body is None else json.dumps(body).encode())
+ return urllib.request.urlopen(req, timeout=15)
+
+ def seen():
+ with Provider.lock:
+ return Provider.requests[requests_at_start:]
+
+ def status(session_id):
+ with api(f"/sessions/{session_id}/status") as response:
+ return json.load(response)
+
+ def settle(session_id, timeout=30):
+ deadline = time.monotonic() + timeout
+ value = None
+ while time.monotonic() < deadline:
+ value = status(session_id)
+ if not value["running"]:
+ return value
+ time.sleep(0.05)
+ raise AssertionError(f"session never settled: {value}")
+
+ created = json.loads(subprocess.run(
+ ["node", "cli/bin/albedo.mjs", "new", str(workspace)], cwd=ROOT, env=env,
+ text=True, capture_output=True, timeout=45, check=True).stdout)
+ session_id = created["session"]
+ with api(f"/sessions/{session_id}/events", {"content": "start a slow job"}):
+ pass
+ settle(session_id)
+ assert len(seen()) == 2, [r["latest_user"] for r in seen()]
+
+ def stream(query):
+ with api(f"/sessions/{session_id}/stream{query}") as response:
+ while True:
+ line = response.readline()
+ if line.startswith(b"data: "):
+ return json.loads(line[6:])
+
+ # A cursor before the wake; the live event keeps its source.
+ before = stream("?after_seq=-1")
+ # The job outlives the run; its completion must submit a wake turn.
+ deadline = time.monotonic() + 30
+ while time.monotonic() < deadline:
+ if len(seen()) >= 3:
+ break
+ time.sleep(0.1)
+ settle(session_id)
+ assert len(seen()) == 3, [r["latest_user"] for r in seen()]
+ wake = seen()[2]["latest_user"]
+ assert "background bash job finished" in wake, wake
+ assert "jobs[" in wake and "output.read" in wake, wake
+
+ live = stream(f"?after_seq={before['cursor']}")["events"]
+ live_wake = [e for e in live
+ if e.get("type") == "user" and "bash job finished" in e.get("text", "")]
+ assert len(live_wake) == 1, [e.get("type") for e in live]
+ assert live_wake[0]["source"] == "bash", live_wake[0]
+ assert live_wake[0]["clientId"] == "bash", live_wake[0]
+ assert "exit_code=0" in live_wake[0]["text"], live_wake[0]
+
+ # The durable snapshot renders user turns uniformly, so the wake is
+ # identified by its text there; the provider request already proved it ran.
+ durable = [e for e in stream("?after_seq=-1")["events"]
+ if e.get("type") == "user" and "bash job finished" in e.get("text", "")]
+ assert len(durable) == 1, "the wake turn did not commit durably"
+
+ # No further wake: the notice was delivered once.
+ time.sleep(2.5)
+ assert len(seen()) == 3, [r["latest_user"] for r in seen()]
+
+ # A detached idle session keeps its kernel while a job runs: without
+ # the live-job pin, the 10-second idle sweep would release the kernel
+ # and kill the 20-second job before it could wake anyone.
+ with api(f"/sessions/{session_id}/events",
+ {"content": "start a slow job, detached"}) as response:
+ response.read()
+ settle(session_id)
+ assert len(seen()) == 5, [r["latest_user"] for r in seen()]
+ deadline = time.monotonic() + 50
+ while time.monotonic() < deadline:
+ if len(seen()) >= 6:
+ break
+ time.sleep(0.2) # no daemon contact: the session stays detached
+ settle(session_id)
+ assert len(seen()) == 6, [r["latest_user"] for r in seen()]
+ assert "background bash job finished" in seen()[5]["latest_user"], seen()[5]
+ print(protocol + ": background job wake delivered a turn, once, and survived an idle detach")
+ finally:
+ if connection:
+ with contextlib.suppress(Exception):
+ api("/shutdown", {}).close()
+ deadline = time.monotonic() + 10
+ while time.monotonic() < deadline:
+ try:
+ os.kill(connection["pid"], 0)
+ except ProcessLookupError:
+ break
+ time.sleep(0.05)
+ else:
+ os.kill(connection["pid"], 9)
+
+
+if __name__ == "__main__":
+ server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Provider)
+ thread = threading.Thread(target=server.serve_forever, daemon=True)
+ thread.start()
+ try:
+ for protocol in ("responses", "chat_completions"):
+ run(protocol, f"http://127.0.0.1:{server.server_port}/v1")
+ finally:
+ server.shutdown()
diff --git a/test/daemon/reaper_test.gleam b/test/daemon/reaper_test.gleam
index 7bc3991..0abb7f4 100644
--- a/test/daemon/reaper_test.gleam
+++ b/test/daemon/reaper_test.gleam
@@ -8,19 +8,24 @@ fn pids(victims: List(Candidate)) -> List(Int) {
list.map(victims, fn(victim: Candidate) { victim.pid })
}
-pub fn attached_and_running_kernels_are_never_released_test() {
+pub fn attached_running_and_job_holding_kernels_are_never_released_test() {
[
// attached moments ago, however much it holds
- Candidate(1, 0, False, 100_000),
+ Candidate(1, 0, False, 0, 100_000),
// detached but working
- Candidate(2, 600_000, True, 100_000),
+ Candidate(2, 600_000, True, 0, 100_000),
+ // detached, idle past every limit, but a background job still runs
+ Candidate(3, 600_000, False, 1, 100_000),
]
|> reaper.victims(limits)
|> should.equal([])
}
pub fn kernels_idle_past_the_limit_are_released_test() {
- [Candidate(1, 61_000, False, 10), Candidate(2, 40_000, False, 10)]
+ [
+ Candidate(1, 61_000, False, 0, 10),
+ Candidate(2, 40_000, False, 0, 10),
+ ]
|> reaper.victims(limits)
|> pids
|> should.equal([1])
@@ -29,9 +34,9 @@ pub fn kernels_idle_past_the_limit_are_released_test() {
pub fn a_pool_over_budget_loses_the_longest_unattended_first_test() {
let victims =
[
- Candidate(1, 31_000, False, 600),
- Candidate(2, 45_000, False, 600),
- Candidate(3, 35_000, False, 600),
+ Candidate(1, 31_000, False, 0, 600),
+ Candidate(2, 45_000, False, 0, 600),
+ Candidate(3, 35_000, False, 0, 600),
]
|> reaper.victims(limits)
// 1800 held against a 1000 budget: the two quietest go, the freshest stays.
@@ -40,7 +45,7 @@ pub fn a_pool_over_budget_loses_the_longest_unattended_first_test() {
pub fn memory_freed_by_the_idle_rule_counts_against_the_budget_test() {
let victims =
- [Candidate(1, 120_000, False, 900), Candidate(2, 31_000, False, 600)]
+ [Candidate(1, 120_000, False, 0, 900), Candidate(2, 31_000, False, 0, 600)]
|> reaper.victims(limits)
// Releasing the expired kernel already brings the pool under budget.
victims |> pids |> should.equal([1])
@@ -48,14 +53,14 @@ pub fn memory_freed_by_the_idle_rule_counts_against_the_budget_test() {
pub fn a_short_limit_still_releases_test() {
let brief = Limits(idle_ms: 5000, budget_kb: 1000, detached_ms: 30_000)
- [Candidate(1, 6000, False, 10)]
+ [Candidate(1, 6000, False, 0, 10)]
|> reaper.victims(brief)
|> pids
|> should.equal([1])
}
pub fn a_session_attached_moments_ago_is_spared_even_over_budget_test() {
- [Candidate(1, 1000, False, 5000)]
+ [Candidate(1, 1000, False, 0, 5000)]
|> reaper.victims(limits)
|> should.equal([])
}
diff --git a/test/harness/job_wake_route_test.gleam b/test/harness/job_wake_route_test.gleam
new file mode 100644
index 0000000..b0e55d6
--- /dev/null
+++ b/test/harness/job_wake_route_test.gleam
@@ -0,0 +1,91 @@
+//// The jobs route: a kernel wake notice lands as a submit through the registry.
+
+import albedo/harness/bash
+import albedo/harness/runtime
+import gleam/dynamic/decode
+import gleam/erlang/process
+import gleam/json
+import gleam/option.{None, Some}
+import gleeunit/should
+
+@external(erlang, "albedo_wakes", "register")
+fn register(session: String, submit: fn(String, String) -> String) -> Nil
+
+@external(erlang, "albedo_wakes", "forget")
+fn forget(session: String) -> Nil
+
+type Reply {
+ Reply(ok: Bool, code: option.Option(String))
+}
+
+fn notice(method: String) -> String {
+ json.object([
+ #("method", json.string(method)),
+ #(
+ "args",
+ json.object([
+ #("display", json.string("bash job finished (exit_code=0)")),
+ #("text", json.string("wake")),
+ ]),
+ ),
+ ])
+ |> json.to_string
+}
+
+fn reply_of(answer: String) -> Reply {
+ let decoder = {
+ use ok <- decode.field("ok", decode.bool)
+ use code <- decode.optional_field(
+ "code",
+ None,
+ decode.optional(decode.string),
+ )
+ decode.success(Reply(ok, code))
+ }
+ let assert Ok(reply) = json.parse(answer, decoder)
+ reply
+}
+
+fn store() -> runtime.Runtime {
+ let assert Ok(host) = runtime.start(":memory:")
+ host
+}
+
+pub fn delivered_wake_reaches_the_registered_submit_test() {
+ let host = store()
+ let seen = process.new_subject()
+ register("route-delivered", fn(display, text) {
+ process.send(seen, #(display, text))
+ ""
+ })
+ bash.route(runtime.ledger(host), "route-delivered", notice("jobs.completed"))
+ |> reply_of
+ |> should.equal(Reply(True, None))
+ let assert Ok(#("bash job finished (exit_code=0)", text)) =
+ process.receive(seen, 1000)
+ text |> should.equal("wake")
+ forget("route-delivered")
+}
+
+pub fn busy_wake_answers_with_the_code_the_kernel_retries_test() {
+ let host = store()
+ register("route-busy", fn(_display, _text) { "session is busy" })
+ bash.route(runtime.ledger(host), "route-busy", notice("jobs.completed"))
+ |> reply_of
+ |> should.equal(Reply(False, Some("busy")))
+ forget("route-busy")
+}
+
+pub fn an_unregistered_session_refuses_without_the_retry_code_test() {
+ let host = store()
+ bash.route(runtime.ledger(host), "route-nobody", notice("jobs.completed"))
+ |> reply_of
+ |> should.equal(Reply(False, Some("unavailable")))
+}
+
+pub fn an_unknown_jobs_operation_is_refused_test() {
+ let host = store()
+ bash.route(runtime.ledger(host), "route-unknown", notice("jobs.consumed"))
+ |> reply_of
+ |> should.equal(Reply(False, Some("invalid")))
+}
diff --git a/test/harness/job_wake_test.py b/test/harness/job_wake_test.py
new file mode 100644
index 0000000..e864fd2
--- /dev/null
+++ b/test/harness/job_wake_test.py
@@ -0,0 +1,214 @@
+"""The wake protocol: unfinished jobs tell the host, reads retire the notice.
+
+The owner here is a fake daemon: it answers the kernel's host calls, so the
+bash plugin's notice machinery can be exercised against the real kernel without
+gleam. The replies cover the whole contract: accepted, refused with busy
+(retried), refused otherwise (given up), and no call at all when the job was
+awaited or its result was already read.
+"""
+import json
+import os
+import select
+import struct
+import subprocess
+import sys
+import tempfile
+import time
+import unittest
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[2]
+KERNEL = ROOT / "priv" / "python" / "albedo_kernel.py"
+MAX_FRAME = 8 * 1024 * 1024
+RETRY_WINDOW = 3.2 # NOTICE_RETRY plus slack; must not mistake a retry for silence
+
+
+class Owner:
+ """A minimal owner that answers host calls and buffers every other frame."""
+
+ def __init__(self, modules, env=None):
+ self.workspace = tempfile.mkdtemp(prefix="albedo-wake-")
+ environment = dict(os.environ)
+ environment.update(env or {})
+ self.process = subprocess.Popen(
+ [sys.executable, "-u", str(KERNEL), json.dumps(modules)],
+ stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
+ cwd=self.workspace, env=environment)
+ self.buffered = []
+ self.calls = [] # host calls the owner has seen, in order
+ self.answers = [] # canned answers, popped per call; default ok
+
+ def close(self):
+ self.send({"type": "shutdown"})
+ self.process.wait(timeout=10)
+ for stream in (self.process.stdin, self.process.stdout, self.process.stderr):
+ stream.close()
+ self.buffered.clear()
+
+ def send(self, message):
+ data = json.dumps(message).encode()
+ self.process.stdin.write(struct.pack(">I", len(data)) + data)
+ self.process.stdin.flush()
+
+ def recv(self, timeout=0.05):
+ """One frame, or None when the kernel stays quiet for the window."""
+ ready, _, _ = select.select([self.process.stdout], [], [], timeout)
+ if not ready:
+ return None
+ header = self.process.stdout.read(4)
+ size = struct.unpack(">I", header)[0]
+ assert size <= MAX_FRAME, "control frame exceeds the ceiling"
+ return json.loads(self.process.stdout.read(size))
+
+ def wait_for(self, predicate, timeout=10.0):
+ deadline = time.monotonic() + timeout
+ while time.monotonic() < deadline:
+ for index, frame in enumerate(self.buffered):
+ if predicate(frame):
+ return self.buffered.pop(index)
+ frame = self.recv(timeout=max(0.05, deadline - time.monotonic()))
+ if frame is not None:
+ self.buffered.append(frame)
+ raise AssertionError(f"no matching frame; buffered: "
+ f"{[f.get('type') for f in self.buffered]}")
+
+ def expect_no_call(self, method, window=RETRY_WINDOW):
+ deadline = time.monotonic() + window
+ while time.monotonic() < deadline:
+ try:
+ frame = self.wait_for(
+ lambda f: f.get("type") == "call" and f.get("method") == method,
+ timeout=max(0.05, deadline - time.monotonic()))
+ except AssertionError:
+ return
+ raise AssertionError(f"unexpected {method} call: {frame}")
+
+ def serve_calls(self):
+ """Answer every host call currently waiting; returns the wake payloads."""
+ notices = []
+ while True:
+ try:
+ frame = self.wait_for(lambda f: f.get("type") == "call", timeout=0.4)
+ except AssertionError:
+ return notices
+ self.calls.append(frame)
+ notices.append(frame)
+ answer = self.answers.pop(0) if self.answers else {"ok": True, "value": "delivered"}
+ self.send({"type": "reply", "id": frame["id"], "value": answer})
+
+ def invoke(self, call_id, **frame):
+ self.send({"type": "invoke", "id": call_id, **frame})
+ while True:
+ reply = self.wait_for(lambda f: f.get("id") == call_id and f["type"] == "invoked")
+ if reply.get("ok") is True or "error" in reply:
+ return reply
+
+
+class WakeProtocolTest(unittest.TestCase):
+ def setUp(self):
+ self.owner = Owner(["bash"])
+ self.addCleanup(self.owner.close)
+ self.assertEqual(self.owner.wait_for(lambda f: f.get("type") == "ready")["type"], "ready")
+
+ def start_job(self, command):
+ reply = self.owner.invoke("j1", name="bash", args=[command])
+ self.assertTrue(reply["ok"])
+ return reply["handle"]
+
+ def wait_job_done(self):
+ """Wait on the raw completion frame; awaiting the job would retire the wake."""
+ start = self.owner.wait_for(lambda f: f.get("type") == "job_start")
+ done = self.owner.wait_for(
+ lambda f: f.get("type") == "job" and f.get("id") == start["id"])
+ return done
+
+ def read_notice(self, frame):
+ args = frame["args"]
+ self.assertIn("display", args)
+ self.assertIn("text", args)
+ self.assertIn("", args["text"])
+ return args
+
+ def test_an_unread_job_completion_reports_to_the_host(self):
+ handle = self.start_job("echo wake-echo")
+ self.wait_job_done()
+ notices = self.owner.serve_calls()
+ wakes = [f for f in notices if f["method"] == "jobs.completed"]
+ self.assertEqual(len(wakes), 1)
+ args = self.read_notice(wakes[0])
+ self.assertEqual(args["exit_code"], 0)
+ self.assertIn("wake-echo", args["text"])
+ self.assertIsNone(args["host"])
+ self.owner.expect_no_call("jobs.completed")
+
+ def test_a_busy_session_is_retried_until_accepted(self):
+ handle = self.start_job("echo retry-echo")
+ self.wait_job_done()
+ # First attempt: busy. The plugin must try again rather than give up.
+ frame = self.owner.wait_for(lambda f: f.get("type") == "call")
+ self.owner.send({"type": "reply", "id": frame["id"],
+ "value": {"ok": False, "code": "busy", "message": "session is busy"}})
+ retry = self.owner.wait_for(
+ lambda f: f.get("type") == "call" and f.get("method") == "jobs.completed",
+ timeout=RETRY_WINDOW)
+ self.owner.send({"type": "reply", "id": retry["id"],
+ "value": {"ok": True, "value": "delivered"}})
+ self.owner.expect_no_call("jobs.completed")
+
+ def test_a_permanent_refusal_gives_up_without_retrying(self):
+ handle = self.start_job("echo refused-echo")
+ self.wait_job_done()
+ frame = self.owner.wait_for(lambda f: f.get("type") == "call")
+ self.owner.send({"type": "reply", "id": frame["id"],
+ "value": {"ok": False, "code": "unavailable", "message": "no session"}})
+ self.owner.expect_no_call("jobs.completed")
+
+ def test_an_awaited_job_never_reports(self):
+ handle = self.start_job("echo awaited-echo")
+ self.owner.invoke("w1", target={"handle": handle}, **{"await": True})
+ self.owner.expect_no_call("jobs.completed")
+
+ def test_a_read_result_retires_a_pending_notice(self):
+ handle = self.start_job("echo read-echo")
+ self.wait_job_done()
+ # The notice is in flight (busy) when the result is read; the retry must stop.
+ frame = self.owner.wait_for(lambda f: f.get("type") == "call")
+ self.owner.send({"type": "reply", "id": frame["id"],
+ "value": {"ok": False, "code": "busy", "message": "session is busy"}})
+ tail = self.owner.invoke("t1", target={"handle": handle}, name="tail")
+ self.assertIn("read-echo", tail["value"])
+ self.owner.expect_no_call("jobs.completed")
+
+ def test_output_read_retires_a_pending_notice(self):
+ handle = self.start_job("echo output-echo")
+ job_id = self.wait_job_done()["id"]
+ frame = self.owner.wait_for(lambda f: f.get("type") == "call")
+ self.owner.send({"type": "reply", "id": frame["id"],
+ "value": {"ok": False, "code": "busy", "message": "session is busy"}})
+ read = self.owner.invoke("o1", name="output.read", args=[job_id])
+ self.assertIn("output-echo", read["value"])
+ self.owner.expect_no_call("jobs.completed")
+
+ def test_a_stopped_job_wakes_no_one(self):
+ handle = self.start_job("sleep 5")
+ stopped = self.owner.invoke("s1", target={"handle": handle}, name="stop")
+ self.assertTrue(stopped["ok"])
+ self.owner.expect_no_call("jobs.completed", window=1.5)
+
+ def test_the_remote_stamp_names_the_host(self):
+ owner = Owner(["bash"], env={"ALBEDO_REMOTE_TARGET": "trimounts"})
+ self.addCleanup(owner.close)
+ self.assertEqual(owner.wait_for(lambda f: f.get("type") == "ready")["type"], "ready")
+ reply = owner.invoke("j1", name="bash", args=["echo remote-echo"])
+ start = owner.wait_for(lambda f: f.get("type") == "job_start")
+ owner.wait_for(lambda f: f.get("type") == "job" and f.get("id") == start["id"])
+ frame = owner.wait_for(lambda f: f.get("type") == "call")
+ args = frame["args"]
+ self.assertEqual(args["host"], "trimounts")
+ self.assertIn(" on trimounts", args["display"])
+ owner.send({"type": "reply", "id": frame["id"],
+ "value": {"ok": True, "value": "delivered"}})
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/test/manual/job_wake_live.py b/test/manual/job_wake_live.py
new file mode 100644
index 0000000..4e0aa75
--- /dev/null
+++ b/test/manual/job_wake_live.py
@@ -0,0 +1,122 @@
+"""Opt-in live wake check against a real provider (deepseek by default).
+
+Needs a real key: reads it from prime-agent's auth.json (or $ALBEDO_TEST_AUTH).
+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]
+AUTH_PATH = os.environ.get("ALBEDO_TEST_AUTH",
+ os.path.expanduser("~/.prime/agent/auth.json"))
+AUTH = json.load(open(AUTH_PATH))
+KEY = AUTH["deepseek"]["key"]
+PROMPT = (
+ "use the python tool to run exactly this cell and nothing else: "
+ "job = bash(\"sleep 25; echo 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(["node", "cli/bin/albedo.mjs", "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(
+ ["node", "cli/bin/albedo.mjs", "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?after_seq=-1") 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 "bash 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()
--
2.54.0