kernel: run the python kernel detached behind a bridge, with a durable seq/ack session layer so it survives dropped connections and daemon restarts
Signed-off-by: dawn <90008@klbr.net>
78 unmodified lines798081828383848586875 unmodified lines78 unmodified lines- **`native/render/`** — optional Rust renderer behind the `view` extension, which shows the model its changes as highlighted images for the final review pass. Install with `native/render/install.sh`; needs cargo.
### robot-docs/
`robot-docs/` is one model-facing doc per subsystem (agents, auth, settings, kernel-state, migrations, cache-ttl, cache-warming, commands, compaction, context, extensions, files, mcp, models, paperclips, provider-requests, proxy, quota, skills, usage-feed, webhooks, workspaces). This file is the map; those are the details. When you change a subsystem's behavior, update its doc in the same change; when you need to understand one, read it there first.`robot-docs/` is one model-facing doc per subsystem (agents, auth, settings, kernel-state, migrations, cache-ttl, cache-warming, commands, compaction, context, extensions, files, kernel, mcp, models, paperclips, provider-requests, proxy, quota, skills, usage-feed, webhooks, workspaces). This file is the map; those are the details. When you change a subsystem's behavior, update its doc in the same change; when you need to understand one, read it there first.
Read the page for a subsystem before you edit it, and name the module that owns the behavior before you write it. Ownership rules live in those docs (an extension owns its own migrations and tables; the daemon only collects them), and the mistakes that cost the most were ones a doc already ruled out. Use the `design-preflight` skill for anything beyond a small change.
### Other directories5 unmodified lines9 unmodified lines1011121314151617181948 unmodified lines9 unmodified lines# an earlier benchmark run.set -euo pipefailcd "$(dirname "$0")"export ALBEDO_NO_BROWSER=1# a detached kernel a suite left behind ends soon after, not in an hourexport ALBEDO_KERNEL_GRACE_SECONDS=20
ruff checkruff format --check priv/python test cli/internal/storage/maintenance.pyty check48 unmodified lines131 unmodified linesExpand all13213313413513613613713813914080 unmodified lines131 unmodified linesExpand allTop-level traces emit before `done`. Nested `cells.run` traces emit before waiting for the host's `cells.finish` acknowledgement and remain separate from the parent trace. The daemon persists these events when the enclosing top-level execution completes, after which `cells.trace` can read the results independently of kernel buffers. This lifecycle applies to newly started kernels and does not change output retention, image delivery, or background task execution.
interfaces: [`PythonApi`](../priv/python/albedo_api.py), [`work`](../src/albedo/harness/work.gleam), [`compaction strategy`](../src/albedo/harness/compaction.gleam).
Kernel control messages are checked by `priv/python/albedo_protocol.py` before delivery to the event loop. Invalid JSON, oversized frames, unknown message types, and invalid fields enter the kernel shutdown path, including plugin cleanup and child-process cleanup. Host replies are checked before their pending futures are removed or completed. Extra fields on known messages are accepted. Missing methods, expired references, and tool exceptions remain call errors.The kernel runs detached behind a bridge, with a durable session layer under these messages; see [kernel](kernel.md). Kernel control messages are checked by `priv/python/albedo_protocol.py` before delivery to the event loop. Invalid JSON, oversized frames, unknown message types, and invalid fields enter the kernel shutdown path, including plugin cleanup and child-process cleanup. Host replies are checked before their pending futures are removed or completed. Extra fields on known messages are accepted. Missing methods, expired references, and tool exceptions remain call errors.
## run
Output read watchers use weak references to bound job methods. They remain registered across early reads, so a read after completion still retires the notice. Forgetting or evicting an output channel removes its watcher. Job handles and output channels keep their separate retention order and existing limits. A retained output watcher does not keep an evicted job alive, and an explicitly held job keeps its own output after archive eviction.80 unmodified lines123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122# the detached kernel
a session's python kernel (`priv/python/albedo_kernel.py`) is its own process,detached from the daemon: it outlives a dropped connection and a daemonrestart, keeping its namespace and its background jobs. the daemon reaches itthrough a bridge, and every message between the two travels through a smallsession layer that neither loses nor repeats a message across a reconnect.
## processes
- **kernel**: started in its own session and process group, stdio not tied to anyone. it listens on `kernel.sock` in its run directory, `$ALBEDO_HOME/run/<kernel id>/` (0700; the socket is bound relative to the directory so a long temp home stays under the ~104-byte `sun_path` limit). `kernel.log` there catches what it writes before its own output capture starts.- **bridge** (`priv/python/albedo_bridge.py`): what the daemon's erlang port runs now, with the same `{packet, 4}` stdio framing the kernel used to have. `start <run_dir> <modules-json>` starts the kernel beside it (the token goes over the kernel's stdin, never argv) and waits for its socket; `attach <run_dir>` connects to a running one. it announces its own bundle hash (`{"bridge": {"bundle"}}`), then copies bytes both ways until either side closes. exit status 3 means no kernel is there. killing it never touches the kernel. #56 runs the same argv over ssh.- **port owner** (`albedo_python.erl`): one erlang process per kernel, as before. it owns the job groups the kernel reports and the `albedo_signal.py` termination ladder, aimed at the pid/pgid/leader the kernel declares in its hello.
a kernel run with plain `albedo_kernel.py <modules>` (no `--run`) keeps theold raw stdio protocol and dies with its pipe: the remote plugin's targets usethat.
## handshake
the daemon writes `{"attach": {"kernel", "token", "ack", "grace"}}` first. thekernel checks the token (kept in the daemon's sqlite), bumps its epoch, cutsoff any older connection (newest attach wins), and answers`{"hello": {protocol, bundle, epoch, ack, pid, pgid, leader, ready, jobs,external, slots, dropped}}`, or `{"refused": reason}`. `protocol` (1) and thehello/snapshot/shutdown frames never change shape. `bundle` is`albedo_bundle.digest()`, the same content hash the remote plugin stagesunder; a hello whose bundle or protocol differs from the bridge's is onlylogged for now (#55 acts on it). `jobs` (live `job_start` frames), `external`and `slots` (job slots still waited for) let a fresh port owner take over jobownership and admission.
## session layer
after the handshake every frame is `{"seq": n, "ack": m, "frame": {...}}`, ora bare `{"ack": m}`. `ack` is cumulative: everything up to it arrived.
- **kernel side** (`priv/python/albedo_link.py`): an in-memory outbox of unacknowledged frames, resent after each hello. `mirror` (per handle), `trace` (per cell) and `jobs` coalesce latest-wins; everything else must be delivered. bounded at 4096 frames / 64 MiB: coalescing frames go first, then the oldest, counted in the next hello's `dropped`. a daemon frame at or below the last seq applied is a replay and is dropped; each applied frame is acknowledged at once.- **daemon side** (`python/link.gleam`, tables from the python extension's `kernel_links` schema migration): `kernel_links` (one row per session: kernel id, token, run dir, workspace, modules, pid/pgid/leader, epoch, last seq sent), `kernel_outbox` keyed (session, kernel, seq), bounded at 2048 frames / 64 MiB per kernel, and `kernel_calls`. the port owner persists each frame before writing it, deletes what the kernel acknowledges, and acknowledges kernel frames within 100 ms. its own last-seen seq lives in memory: after a daemon restart it accepts everything the kernel resends, and the dedupe rules below make that safe.
dedupe:
- host call id: a call already running in this daemon is ignored (its reply goes out through the outbox); one answered already is ignored the same way; one an earlier daemon started without answering is answered `{ok: false, code: "unknown"}`, never run again. a call's row goes once the kernel acknowledges its reply.- execute, by cell id: a finished cell (the newest 16) resends its `done`; a queued or running one is ignored. callers must not reuse cell ids.- interrupt, reply, release are idempotent already.
## drops
- the bridge dies: the port owner reattaches with backoff (50 ms doubling to 2 s, 40 tries); exit 3 or a refusal means the kernel is gone and the old abandon path reaps it. the cell keeps running meanwhile.- the cell deadline passes, or an interrupt goes unanswered for 2 s, while detached: the interrupt waits in the outbox and the caller gets `kernel.Detached`, which leaves the journaled cell `started` (the model is told to check `cells.info`). a `done` that arrives for a cell nobody waits for any more is journaled through the `cells.finish` host route.- the daemon crashes: the bridge sees EOF and exits; the kernel waits.- graceful shutdown: `runtime.detach_kernels` first, so closing sessions and stopping the runtime let kernels go instead of ending them.- daemon start: `runtime.resume_kernels` reattaches every recorded kernel in the background, one at a time through the boot slots, so late results and job wakes arrive without waiting for the session. a session that asks for its kernel meanwhile waits for that attach, and gets a fresh boot if there was nothing to attach to. an attached kernel is `runtime.resumed`, and the session keeps its namespace without the disk restore. a recorded kernel whose workspace or module set no longer fits is stopped and replaced (#55 will swap it at idle instead).
explicit drops still end the kernel at once: reset, release by the idlesweep, `/cd`, session close, extension reloads (the replacement boots besidethe live kernel, never attached to it).
## lifetime
a kernel with nothing attached for its grace period reaps its job groups(plugin cleanup) and exits, unless a job or cell is still live. the grace is`ALBEDO_KERNEL_GRACE_SECONDS` read by the daemon (default 3600) and sent inevery attach; before its first attach a kernel waits 60 s. a kernel alsoexits within a second of its run directory disappearing, which is how testhomes clean up after themselves. the e2e harness sets a 30 s grace, the gleamtest home 20 s.
## not yet
- after a daemon restart, job groups of a kernel that died hard (SIGKILL) while nobody was attached are not reaped: the recorded pid could be reused.- job slots a reattached kernel's running jobs held are not re-counted in the new daemon's pool; only slots still waited for are asked for again.19 unmodified lines20212223242425262728133 unmodified linesExpand all19 unmodified lines`runtime.start` installs the extension registry before `server.prepare_storage`.for each installed extension, `extension.install` first calls its tableinitialiser, then applies its contributed `SchemaMigration` callbacks on thestore-owned SQLite connection, in plugin order. work contributes `cwd.apply`;paperclips contributes `title.apply`, `scope.apply`, and `reply.apply`. their ledgers create tables but do not runpaperclips contributes `title.apply`, `scope.apply`, and `reply.apply`; python contributes `kernel_links.apply`, the detached kernel's link tables (see [kernel](kernel.md)), which `link.create` also ensures for embedded runtimes that boot kernels without the extension. their ledgers create tables but do not runupgrades themselves. all installed owners upgrade, even when disabled forsessions; session selection and live reload never rerun migrations.
`server.prepare_storage` then runs before sessions start:133 unmodified linesExpand all4 unmodified lines56789910111213142 unmodified lines171819202122232425261 unmodified line4 unmodified lines%% The directory is the run's scratch space: <root>/gleam-<pid> under the root%% the Python suites share (test/scratch.py). It is also TMPDIR, so every file a%% test writes is removed with it when the run ends, and a run that died first%% is swept by the next Python suite, which removes the <name>-<pid> directories%% of processes that no longer exist.%% of processes that no longer exist. Kernels keep their run directories under%% it too, and leave when it disappears.-module(albedo_test_home).-export([isolate/0, cleanup/0]).
isolate() ->2 unmodified lines ok = filelib:ensure_path(Dir), persistent_term:put(?MODULE, Dir), true = os:putenv("ALBEDO_HOME", Dir), true = os:putenv("TMPDIR", Dir), %% A detached kernel a test never stopped ends soon after, not in an hour. true = os:putenv("ALBEDO_KERNEL_GRACE_SECONDS", "20"), nil.
cleanup() -> _ = file:del_dir_r(persistent_term:get(?MODULE)),1 unmodified line123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124"""Copy a detached kernel's frames between this process's stdio and its socket.
albedo_bridge.py start <run_dir> <modules-json> albedo_bridge.py attach <run_dir>
The daemon talks to this process as it once talked to the kernel itself, over4-byte-length-framed stdio, and writes the attach frame first. `start` readsthat frame to learn the kernel's token, starts the albedo_kernel.py beside thisfile in its own session (the token travels over the kernel's stdin, neverargv), and waits for its socket; `attach` connects to a kernel that is already running. Either way thisprocess announces the bundle it runs from, then copies bytes both ways untilone side closes. Killing it never touches the kernel.
Exit status 3 means no kernel is there to attach to, so the daemon can tell akernel that is gone from a connection that merely dropped."""
from __future__ import annotations
import errnoimport osimport socketimport subprocessimport sysimport threadingimport timefrom collections.abc import Callable
import albedo_bundleimport albedo_link
GONE = 3START_TIMEOUT = 10.0 # seconds a starting kernel has to bind its socket
def connect(run_dir: str) -> socket.socket | None: """The kernel's socket, or None when nothing listens there.""" connection = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) try: os.chdir(run_dir) connection.connect(albedo_link.SOCKET) return connection except OSError as error: connection.close() if error.errno in (errno.ENOENT, errno.ECONNREFUSED, errno.ENOTDIR): return None raise
KERNEL = os.path.join(os.path.dirname(os.path.abspath(__file__)), "albedo_kernel.py")
def start(run_dir: str, modules: str, token: str) -> socket.socket: os.makedirs(run_dir, mode=0o700, exist_ok=True) with open(os.path.join(run_dir, "kernel.log"), "ab") as log: kernel = subprocess.Popen( [sys.executable, "-u", KERNEL, "--run", run_dir, modules], stdin=subprocess.PIPE, stdout=subprocess.DEVNULL, stderr=log, start_new_session=True, close_fds=True, ) assert kernel.stdin is not None kernel.stdin.write(token.encode() + b"\n") kernel.stdin.close() deadline = time.monotonic() + START_TIMEOUT while time.monotonic() < deadline: connection = connect(run_dir) if connection is not None: return connection if kernel.poll() is not None: break time.sleep(0.02) sys.stderr.write( f"albedo_bridge: kernel did not start (see {run_dir}/kernel.log)\n" ) sys.exit(GONE)
def stdin(size: int) -> bytes: return os.read(0, size)
def stdout(data: bytes) -> int: return os.write(1, data)
def pump(source: Callable[[int], bytes], sink: Callable[[bytes], int | None]) -> None: try: while data := source(65536): albedo_link.write_all(sink, data) except OSError: pass os._exit(0)
def main(argv: list[str]) -> None: mode, run_dir = argv[1], argv[2] if mode == "start": attach = albedo_link.read_frame(stdin) if not isinstance(attach, dict) or not isinstance(attach.get("attach"), dict): sys.exit(2) token = attach["attach"].get("token") if not isinstance(token, str): sys.exit(2) connection = start(run_dir, argv[3], token) first = albedo_link.encode(attach) else: found = connect(run_dir) if found is None: sys.exit(GONE) connection, first = found, None albedo_link.write_frame( stdout, albedo_link.encode({"bridge": {"bundle": albedo_bundle.digest()}}) ) if first is not None: albedo_link.write_frame(connection.send, first) threading.Thread(target=pump, args=(stdin, connection.send), daemon=True).start() pump(connection.recv, stdout)
if __name__ == "__main__": main(sys.argv)12345678910111213141516171819202122"""Content hash of the packaged python tree.
The remote plugin stages a bundle under this hash, and a detached kernel andits bridge announce it, so the daemon can tell when a kernel runs older codethan the daemon's own tree."""
from __future__ import annotations
import hashlibfrom pathlib import Path
ROOT = Path(__file__).resolve().parent
def digest(root: Path = ROOT) -> str: """sha256 over every .py file's relative path and bytes, in path order.""" hasher = hashlib.sha256() for path in sorted(root.rglob("*.py")): hasher.update(str(path.relative_to(root)).encode()) hasher.update(path.read_bytes()) return hasher.hexdigest()1234567891011 unmodified lines222324252627282930313233343536373835394041425 unmodified lines48495051495051525354555657585915 unmodified lines757677787980818283848586878815 unmodified lines104105106107989910010110210810911010510610710810911011111211311411111211311436 unmodified lines15115215315415515615715815916016116316216316416516616716816917017117217317417517617717817918018118218318418516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222328 unmodified lines2522532542552562572582592602612622632642652661245 unmodified linesExpand all15121513151415151479148014811482148314841485148614871488148914901516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115724 unmodified lines1577157815791580158115821583158415851513158615871588158916 unmodified lines1606160716081609161016111612161316148 unmodified lines1623162416251626162716281629163016318 unmodified linesfrom __future__ import annotations
import astimport albedo_apiimport albedo_bundleimport albedo_linkfrom albedo_protocol import parse_incomingimport albedo_procimport albedo_shellfrom collections.abc import Awaitable, Callable, Iterable, Sequence11 unmodified linesimport jsonimport osimport reprlibimport selectimport shutilimport signalimport pickleimport typesimport structimport sysimport tempfileimport threadingimport timeimport tracebackimport uuidimport weakref
MAX_FRAME = 8 * 1024 * 1024CLEANUP_DEADLINE = ( 1.5 # seconds: plugins must finish cleanup inside the supervisor's patience)PREVIEW = 64 * 10245 unmodified linesMAX_IMAGE_BYTES = 5 * 1024 * 1024CELL: contextvars.ContextVar[Capture | None] = contextvars.ContextVar( "cell", default=None)CONTROL_IN = os.fdopen(os.dup(0), "rb", buffering=0)CONTROL_OUT = os.fdopen(os.dup(1), "wb", buffering=0)SEND_LOCK = threading.Lock()LINK: albedo_link.StdioLink | albedo_link.SocketLinkRUN_DIR: str | None = None # a detached kernel's run directory, None over stdioWATCH_INTERVAL = 1.0 # seconds between a detached kernel's lifetime checksFIRST_ATTACH_GRACE = 60.0 # seconds a detached kernel waits for its first attachLOOP = asyncio.new_event_loop()asyncio.set_event_loop(LOOP)
albedo_proc.reap_stopped_safely(LOOP)15 unmodified linesactive: asyncio.Task[object] | None = Noneactive_capture: Capture | None = Noneinterrupt_capture: Capture | None = NoneARCHIVES: collections.OrderedDict[str, Capture] = collections.OrderedDict()# Finished executions by cell id: a replayed execute answers with its result# instead of running the cell twice. Queued or running ones are in EXECUTING.FINISHED: collections.OrderedDict[str, dict[str, object]] = collections.OrderedDict()FINISHED_LIMIT = 16EXECUTING: set[str] = set()READY = threading.Event()NAMESPACE: dict[str, object] = {"__name__": "__main__"}DEFINITIONS: collections.OrderedDict[str, str] = collections.OrderedDict()CLEANUP: list[albedo_api.Cleanup] = []HANDLES: list[type[object]] = [asyncio.Task]15 unmodified lines )
def send(value: dict[str, object]) -> None: data = json.dumps(value, ensure_ascii=True).encode() with SEND_LOCK: remaining = memoryview(struct.pack(">I", len(data)) + data) while remaining: remaining = remaining[CONTROL_OUT.write(remaining) :] LINK.send(value)
def read_exact(size: int) -> bytes: data = bytearray() while len(data) < size: chunk = CONTROL_IN.read(size - len(data)) if not chunk: raise EOFError() data.extend(chunk) return bytes(data)
async def cleanup() -> None: """Run plugin cleanup; a callback may return an awaitable to clean up asynchronously.""" for close in reversed(CLEANUP): try:36 unmodified lines } ) except (OSError, ValueError): pass # the owner disconnected; it independently tracks our groups if RUN_DIR is not None: # No socket left behind: an attach from now on learns we are gone. shutil.rmtree(RUN_DIR, ignore_errors=True) # The interpreter and its subprocesses own a separate process group. os.killpg(os.getpid(), signal.SIGKILL)
def reader():def receive(frame: object) -> None: """Act on one frame from the owner, on the link's reader thread.""" global interrupt_capture try: while True: size = struct.unpack(">I", read_exact(4))[0] if size > MAX_FRAME: raise ValueError("control frame too large") message = parse_incoming(json.loads(read_exact(size))) if message["type"] == "shutdown": die() elif message["type"] == "interrupt": capture = active_capture if capture is not None and capture.id == message["id"]: interrupt_capture = capture capture.interruption = message.get("reason", "cancelled") os.kill(os.getpid(), signal.SIGINT) else: task = OWNER_TASKS.get(message["id"]) if task is not None: LOOP.call_soon_threadsafe(task.cancel) else: _ = LOOP.call_soon_threadsafe(deliver, message) except Exception: message = parse_incoming(frame) except ValueError: die() return if message["type"] == "shutdown": die() elif message["type"] == "interrupt": capture = active_capture if capture is not None and capture.id == message["id"]: interrupt_capture = capture capture.interruption = message.get("reason", "cancelled") os.kill(os.getpid(), signal.SIGINT) else: task = OWNER_TASKS.get(message["id"]) if task is not None: LOOP.call_soon_threadsafe(task.cancel) else: _ = LOOP.call_soon_threadsafe(deliver, message)
def reader() -> None: try: LINK.serve(receive) except Exception: pass die()
def watch() -> None: """A detached kernel ends itself: when its run directory is gone (its home was removed), or when nothing attached for the grace period and no job or cell is still running.""" assert isinstance(LINK, albedo_link.SocketLink) and RUN_DIR is not None while True: time.sleep(WATCH_INTERVAL) if not os.path.isdir(RUN_DIR): die() busy = active is not None or not QUEUE.empty() or LINK.jobs.live() > 0 if not busy and LINK.idle_for() > LINK.grace: die()
def hello() -> dict[str, object]: """What a detached kernel tells each attach about itself.""" pgid = os.getpgid(0) return { "pid": os.getpid(), "pgid": pgid if pgid == os.getpid() else None, "leader": albedo_proc.leader_token(os.getpid()), "ready": READY.is_set(), "slots": [id for id, (slot, _) in JOB_SLOTS.items() if not slot.done()], }
def deliver(message: albedo_api.Incoming) -> None: if message["type"] == "job_slot": entry = JOB_SLOTS.get(message["id"]) if entry is not None and not entry[0].done():28 unmodified lines } ) elif message["type"] == "release": LIVE.pop(message["handle"], None) elif message["type"] == "execute" and message["id"] in FINISHED: send(FINISHED[message["id"]]) elif message["type"] == "execute" and message["id"] in EXECUTING: pass # already queued or running; its done frame answers both else: if message["type"] == "execute": EXECUTING.add(message["id"]) QUEUE.put_nowait(cast(albedo_api.Execute | albedo_api.State, message))
class WorkError(Exception):1245 unmodified linesExpand all active_capture = None CELL.reset(token) duration = round(LOOP.time() - began, 3) deliver_trace(capture) send( { "type": "done", "id": capture.id, "status": status, "duration": duration, "output": capture.preview(status), "value": value, "truncated": capture.seen > PREVIEW, "images": capture.encoded_images(), } ) done: dict[str, object] = { "type": "done", "id": capture.id, "status": status, "duration": duration, "output": capture.preview(status), "value": value, "truncated": capture.seen > PREVIEW, "images": capture.encoded_images(), } FINISHED[capture.id] = done while len(FINISHED) > FINISHED_LIMIT: FINISHED.popitem(last=False) EXECUTING.discard(capture.id) send(done) capture.images.clear()
def flush(timeout: float) -> None: """Give the owner a moment to attach and take what we sent, before a detached kernel that failed to start leaves.""" if isinstance(LINK, albedo_link.SocketLink): deadline = time.monotonic() + timeout while time.monotonic() < deadline and ( not LINK.attached() or LINK.outbox.entries ): time.sleep(0.02)
def open_link(argv: list[str]) -> list[str]: """Stdio for `kernel.py <modules>`; a socket for `kernel.py --run <dir> <modules>`, whose bridge writes the attach token on our stdin. Returns the module list.""" global LINK, RUN_DIR if argv[1:2] != ["--run"]: LINK = albedo_link.StdioLink() return cast(list[str], json.loads(argv[1])) RUN_DIR = argv[2] token = sys.stdin.buffer.readline().decode().strip() LINK = albedo_link.SocketLink( RUN_DIR, token, hello, bundle=albedo_bundle.digest(), grace=FIRST_ATTACH_GRACE, ) return cast(list[str], json.loads(argv[3]))
def main(): global NATIVE_FD if os.getpgrp() != os.getpid(): os.setsid() modules = open_link(sys.argv) _ = os.dup2(os.open(os.devnull, os.O_RDONLY), 0) read_fd, write_fd = os.pipe() _ = os.dup2(write_fd, 1) _ = os.dup2(write_fd, 2)4 unmodified lines sys.stdin = open(os.devnull) sys.stdout = sys.stderr = Stream() threading.Thread(target=native_output, args=(read_fd,), daemon=True).start() threading.Thread(target=reader, daemon=True).start() if RUN_DIR is not None: threading.Thread(target=watch, daemon=True).start() _ = signal.signal(signal.SIGINT, interrupt) albedo_shell.install() # refusals precede observational audit hooks albedo_trace.install(CELL.get) modules = cast(list[str], json.loads(sys.argv[1])) api = albedo_api.PythonApi( version=2, loop=LOOP, host=host,16 unmodified lines try: LOOP.run_until_complete(albedo_api.load_plugins(modules, api, NAMESPACE)) except Exception as error: send({"type": "startup_error", "message": str(error)}) flush(5.0) die() return # Match an interactive Python session only after loading trusted plugins. sys.path[0] = ""8 unmodified lines "pgid": pgid if pgid == os.getpid() else None, "leader": albedo_proc.leader_token(os.getpid()), } ) READY.set() task = LOOP.create_task(serve()) while not task.done(): try: LOOP.run_until_complete(task)8 unmodified lines504 unmodified linesExpand all505506507508509510511512513514515494 unmodified linesExpand all504 unmodified linesExpand all ALBEDO_NO_BROWSER="1", ALBEDO_PARENT_PID=str(os.getpid()), ERL_FLAGS=TEST_VM_FLAGS, ALBEDO_IDLE_SECONDS="10", # Kernels outlive a stopped daemon only this long; one whose home # is removed leaves at once. ALBEDO_KERNEL_GRACE_SECONDS="30", ALBEDO_MCP_SECRET="configured-secret", ALBEDO_MCP_CLOSED=str(self.root / "closed"), ALBEDO_MCP_AMBIENT="must-not-reach-the-server", )494 unmodified linesExpand all934 unmodified linesExpand all93593693793893993994094194294394460 unmodified lines934 unmodified linesExpand all note.startswith( '<system-note origin="daemon restart">albedo restarted' ) ) # The detached kernel outlives the crash: no reset notice. self.assertIn("<system-note>The python kernel", note) self.assertNotIn("<system-note>The python kernel", note) markers = [ e for e in restored if e.get("type") == "user" and e.get("source") == "daemon restart"60 unmodified lines112345633 unmodified lines404142434344454647484 unmodified lines5354555656575859606113 unmodified lines757677787879808182835 unmodified lines"""Daemon and Python crashes inject exactly one reset notice without corrupting history.""""""A Python crash injects exactly one reset notice without corrupting history; adaemon crash leaves the detached kernel alive, so it injects none."""
import jsonimport urllib.errorimport unittest33 unmodified lines self.app.prompt(session, prompt).close() self.app.idle(session) return latest_user(self.provider.requests[-1]["request"])
def test_daemon_restart_and_kernel_crash_emit_one_shot_reset_notices(self): def test_kernel_crash_emits_one_shot_reset_notice_and_daemon_crash_none(self): app = self.app session = app.session() self.assertEqual(self.turn(session, "first prompt"), "first prompt") unused = app.session()4 unmodified lines ) with self.assertRaises(urllib.error.HTTPError) as rejected: app.prompt(session, " ").close() self.assertEqual(rejected.exception.code, 409) self.assertEqual(self.turn(session, "after reset"), "after reset" + NOTICE) self.assertEqual(self.turn(session, "after restart"), "after restart") self.assertEqual(self.turn(session, "ordinary turn"), "ordinary turn") self.assertEqual(self.turn(session, "lose kernel"), "lose kernel") self.assertEqual( self.turn(session, "recover kernel"), "recover kernel" + NOTICE13 unmodified lines self.assertEqual( users, [ "first prompt", "after reset", "after restart", "ordinary turn", "lose kernel", "recover kernel", "still alive",5 unmodified lines