diff --git a/priv/python/albedo_bridge.py b/priv/python/albedo_bridge.py index ead2474f..8c66f474 100644 --- a/priv/python/albedo_bridge.py +++ b/priv/python/albedo_bridge.py @@ -30,6 +30,7 @@ from __future__ import annotations import contextlib import errno +import json import os import socket import subprocess @@ -141,7 +142,56 @@ def framed() -> list[str]: return [sys.argv[0], *argv] +def spare() -> None: + """Keep a detached fresh interpreter until one adoption or owner EOF.""" + script = os.path.join(os.path.dirname(KERNEL), "albedo_spare.py") + child = subprocess.Popen( + [sys.executable, "-u", script, os.environ.get("ALBEDO_HOME", "")], + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=sys.stderr, + start_new_session=True, + close_fds=True, + ) + adopted = False + try: + assert child.stdout is not None and child.stdin is not None + ready = child.stdout.readline(65536) + if not ready or not isinstance(json.loads(ready).get("ready"), str): + raise RuntimeError("spare exited before imports completed") + sys.stdout.buffer.write(ready) + sys.stdout.buffer.flush() + line = sys.stdin.buffer.readline(1024 * 1024 + 1) + if not line: + return + spec = json.loads(line) + os.makedirs(spec["run_dir"], mode=0o700, exist_ok=True) + spec["environment"] = dict(os.environ) + child.stdin.write((json.dumps(spec) + "\n").encode()) + child.stdin.close() + deadline = time.monotonic() + START_TIMEOUT + while time.monotonic() < deadline: + connection = connect(spec["run_dir"]) + if connection is not None: + connection.close() + adopted = True + print(json.dumps({"adopted": True}), flush=True) + return + if child.poll() is not None: + raise RuntimeError("spare exited during adoption") + time.sleep(0.001) + raise RuntimeError("spare adoption timed out") + finally: + if not adopted: + with contextlib.suppress(ProcessLookupError): + child.kill() + child.wait() + + def main(argv: list[str]) -> None: + if argv[1:2] == ["spare"]: + spare() + return if argv[1:] == ["--frame"]: argv = framed() mode, run_dir = argv[1], argv[2] diff --git a/priv/python/albedo_spare.py b/priv/python/albedo_spare.py new file mode 100644 index 00000000..0486348a --- /dev/null +++ b/priv/python/albedo_spare.py @@ -0,0 +1,62 @@ +"""A fresh interpreter waiting for one kernel identity, never reused.""" + +from __future__ import annotations + +import importlib +import io +import json +import os +import sys + +import albedo_bundle +import albedo_cells +import albedo_proc + +# Import trusted packaged modules only; setup needs the adopting session's host. +PREIMPORT = ( + "run work mail agents paperclips artifacts files memory commands remote browser web_search skills hook" +).split() + + +def main() -> None: + for name in PREIMPORT: + importlib.import_module("albedo_plugins." + name) + bundle = albedo_bundle.digest() + print( + json.dumps( + { + "ready": bundle, + "pid": os.getpid(), + "pgid": os.getpgrp(), + "leader": albedo_proc.leader_token(os.getpid()), + } + ), + flush=True, + ) + line = sys.stdin.buffer.readline(1024 * 1024 + 1) + if len(line) > 1024 * 1024 or not line.endswith(b"\n"): + raise ValueError("invalid spare adoption frame") + spec = json.loads(line) + if bundle != albedo_bundle.digest(): + raise ValueError("spare bundle changed before adoption") + if not isinstance(spec, dict) or not all( + isinstance(spec.get(key), str) for key in ("cwd", "run_dir", "modules", "token") + ): + raise ValueError("invalid spare adoption identity") + environment = spec.get("environment") + if not isinstance(environment, dict) or not all( + isinstance(key, str) and isinstance(value, str) + for key, value in environment.items() + ): + raise ValueError("invalid spare adoption environment") + os.chdir(spec["cwd"]) + os.environ.clear() + os.environ.update(environment) + # The ordinary detached handshake still reads the secret from stdin. + sys.stdin = io.TextIOWrapper(io.BytesIO((spec["token"] + "\n").encode())) + sys.argv = [sys.argv[0], "--run", spec["run_dir"], spec["modules"]] + albedo_cells.main() + + +if __name__ == "__main__": + main() diff --git a/robot-docs/kernel.md b/robot-docs/kernel.md index b4d448d1..dc7ab9b2 100644 --- a/robot-docs/kernel.md +++ b/robot-docs/kernel.md @@ -382,6 +382,25 @@ a kernel run with plain `albedo_kernel.py ` (no `--run`) keeps the old raw stdio protocol and dies with its pipe: the remote plugin's targets use that. +## fresh spares + +`albedo_spare.py` imports the kernel and packaged default plugins without calling +plugin setup or starting kernel threads. A monitored bridge keeps its detached +process waiting on one private stdin adoption frame. Adoption installs the +session's cwd, sanitized launcher environment, module list, run directory, and +token, then starts the ordinary kernel. A spare is consumed once; it never sees +a second agent. Namespace restore still uses the existing state file. + +The native custodian monitors the runtime owner. Stopping an unused spare checks +its process group is gone; an adopted kernel becomes the ordinary link's owned +process and survives daemon detachment as before. A source-bundle change rejects +adoption, cleans the spare, and falls back to a cold launch. `spare_current` only +checks custodian liveness; the adoption checks the source digest again. + +`python.live_work` reads only the last published count of cells and local/remote +jobs, without messaging a blocked owner. Unknown is an error, not zero. A session +with a live job or cell must retain its kernel. + ## handshake the daemon writes `{"attach": {"kernel", "token", "ack", "grace"}}` first. the diff --git a/src/albedo/harness/extensions/python/albedo_python.erl b/src/albedo/harness/extensions/python/albedo_python.erl index 06af99a9..14c0ca17 100644 --- a/src/albedo/harness/extensions/python/albedo_python.erl +++ b/src/albedo/harness/extensions/python/albedo_python.erl @@ -23,11 +23,8 @@ -export([observation/1, stop_recorded/2, stop_jobs/1, job_list/1, job_read/4]). -export([watch/2, unwatch/1]). -export([loop/1, relay_loop/3]). %% hibernate continuations --export([start_spare/1, stop_spare/1, spare_current/1]). +-export([start_spare/1, stop_spare/1, spare_current/1, live_work/1]). -start_spare(_) -> {error, {unavailable, <<"fresh spare startup is not installed">>}}. -stop_spare(_) -> {ok, nil}. -spare_current(_) -> false. -define(PROTOCOL, 1). @@ -56,8 +53,115 @@ spare_current(_) -> false. -define(REMOTE_REATTACH_MAX_MS, 30000). -define(GRACE_MARGIN_S, 60). %% past a remote kernel's grace before it is given up +%% A custodian owns the launcher until a detached spare is adopted or stopped. +start_spare(Owner) -> + Parent = self(), Ref = make_ref(), + {Pid, Mon} = spawn_monitor(fun() -> spare_start(Owner, Parent, Ref) end), + receive + {Ref, Result} -> demonitor(Mon, [flush]), Result; + {'DOWN', Mon, process, Pid, Reason} -> {error, {unavailable, detail(Reason)}} + after ?STARTUP_TIMEOUT -> + Pid ! stop, + demonitor(Mon, [flush]), + {error, {unavailable, <<"spare startup timed out">>}} + end. + +spare_current({spare, Pid}) -> is_process_alive(Pid). + +stop_spare({spare, Pid}) -> + case spare_request(Pid, stop) of + {error, lost} -> {ok, nil}; + Result -> Result + end. + +spare_request(Pid, Request) -> + Alias = alias(), Mon = monitor(process, Pid), + Pid ! {spare_call, Alias, Mon, Request}, + Result = receive + {Mon, Reply} -> Reply; + {'DOWN', Mon, process, Pid, _} -> {error, lost} + after ?STARTUP_TIMEOUT + ?HELPER_WAIT -> {error, <<"spare did not answer">>} + end, + unalias(Alias), demonitor(Mon, [flush]), Result. + +spare_start(Owner, Parent, Ref) -> + process_flag(trap_exit, true), + Watch = monitor(process, Owner), + case paths() of + {error, Reason} -> Parent ! {Ref, {error, Reason}}; + {ok, {Python, Bridge}} -> + Port = open_port({spawn_executable, binary_to_list(Python)}, + [binary, {line, 1048576}, use_stdio, exit_status, hide, + {args, ["-u", binary_to_list(Bridge), "spare", os:getenv("ALBEDO_HOME", "")]}, {env, clean_environment()}]), + receive + {Port, {data, {eol, Data}}} -> + Ready = try json:decode(Data) catch _:_ -> #{} end, + case Ready of + #{<<"ready">> := Bundle, <<"pid">> := Pid} when is_binary(Bundle), is_integer(Pid), Pid > 1 -> + albedo_inspect:label(<<"albedo_spare">>, integer_to_binary(Pid)), + Parent ! {Ref, {ok, {spare, self()}}}, + spare_wait(Port, Watch, target_of(Ready)); + _ -> port_close(Port), Parent ! {Ref, {error, {unavailable, <<"invalid spare readiness">>}}} + end; + {Port, {exit_status, Status}} -> Parent ! {Ref, {error, {unavailable, detail(Status)}}}; + {'DOWN', Watch, process, Owner, _} -> port_close(Port) + after ?STARTUP_TIMEOUT -> + port_close(Port), Parent ! {Ref, {error, {unavailable, <<"spare imports timed out">>}}} + end + end. + +spare_wait(Port, Watch, Target) -> + receive + {spare_call, From, Ref, stop} -> + case spare_cleanup(Port, Target) of + {ok, nil} -> From ! {Ref, {ok, nil}}; + Error -> From ! {Ref, Error}, spare_wait(Port, Watch, Target) + end; + {spare_call, From, Ref, {adopt, Spec}} -> + port_command(Port, [json:encode(Spec), $\n]), + spare_adopting(Port, Watch, Target, From, Ref, false); + stop -> spare_cleanup(Port, Target); + {'DOWN', Watch, process, _, _} -> spare_cleanup(Port, Target); + {Port, {exit_status, _}} -> spare_cleanup(Port, Target); + {'EXIT', Port, _} -> spare_cleanup(Port, Target) + end. + +spare_adopting(Port, Watch, Target, From, Ref, Adopted) -> + receive + {Port, {data, {eol, Data}}} -> + Confirmed = try json:decode(Data) =:= #{<<"adopted">> => true} catch _:_ -> false end, + spare_adopting(Port, Watch, Target, From, Ref, Confirmed); + {Port, {exit_status, 0}} when Adopted -> From ! {Ref, {ok, nil}}; + {Port, {exit_status, _}} -> + case spare_cleanup(Port, Target) of + {ok, nil} -> From ! {Ref, {error, lost}}; + Error -> From ! {Ref, Error} + end; + {'DOWN', Watch, process, _, _} -> + spare_cleanup(Port, Target), From ! {Ref, {error, <<"spare owner stopped">>}} + after ?STARTUP_TIMEOUT -> + spare_cleanup(Port, Target), From ! {Ref, {error, <<"spare adoption timed out">>}} + end. + +spare_cleanup(Port, Target) -> + catch port_close(Port), + verdict(supervise(#{remote => none}, [target_spec(<<"spare">>, Target)], ?TERM_MS, ?KILL_MS)). + +warm_launch(S, Launch) -> + case maps:get(spare, S, none) of + none -> launch(Launch); + {some, {spare, Pid}} -> + Spec = #{cwd => maps:get(cwd, S), run_dir => maps:get(run_dir, S), + modules => maps:get(modules, S), token => maps:get(token, S)}, + case spare_request(Pid, {adopt, Spec}) of + {ok, nil} -> 0; + {error, lost} -> launch(Launch); + {error, _} -> ?GONE + end + end. + %% Boot is the Gleam kernel.Boot record; Link is kernel.Link. -start({boot, Owner, Python, Bridge, Cwd, Host, Modules, Link, RunDir, Kernel, Token, Grace, OutSeq, Owned, Fresh, Remote}) -> +start({boot, Owner, Python, Bridge, Cwd, Host, Modules, Link, RunDir, Kernel, Token, Grace, OutSeq, Owned, Fresh, Remote, Spare}) -> Parent = self(), Ref = make_ref(), {Pid, Mon} = spawn_monitor(fun() -> albedo_inspect:label(<<"albedo_kernel">>, Kernel), @@ -70,7 +174,7 @@ start({boot, Owner, Python, Bridge, Cwd, Host, Modules, Link, RunDir, Kernel, To put(albedo_kernel_observation_table, ObservationTable), S = #{port => none, host => Host, link => Link, owner => Owner, active => none, events => [], groups => Groups, external => 0, awaited_external => 0, external_jobs => [], cells => 0, target => Target, - python => Python, bridge => Bridge, cwd => Cwd, remote => Remote, + python => Python, bridge => Bridge, cwd => Cwd, remote => Remote, spare => Spare, modules => Modules, run_dir => RunDir, kernel => Kernel, token => Token, observation_table => ObservationTable, grace => Grace, in => 0, acked => 0, kack => 0, out => OutSeq, @@ -133,9 +237,10 @@ open_relay(S = #{python := Python, bridge := Bridge, run_dir := RunDir}, Mode) - end, Socket = <>, Kernel = maps:get(kernel, S), + Warm = maps:with([spare, cwd, run_dir, modules, token], S), Relay = spawn(fun() -> albedo_inspect:label(<<"albedo_relay">>, Kernel), - relay(Owner, Launch, Socket) + relay(Owner, Launch, Socket, Warm) end), {ok, handshake(S#{bundle => bundle_digest(filename:dirname(Bridge))}, Relay, [])}. @@ -149,9 +254,9 @@ handshake(S, Port, First) -> %% {self(), {data, Frame}} and {self(), {exit_status, Status}} go out, %% {Owner, {command, Frame}} and {Owner, close} come in. Starting a kernel %% runs the bridge script's launch mode first, which exits once it listens. -relay(Owner, Launch, Socket) -> +relay(Owner, Launch, Socket, Warm) -> Watch = monitor(process, Owner), - Status = case Launch of none -> 0; _ -> launch(Launch) end, + Status = case Launch of none -> 0; _ -> warm_launch(Warm, Launch) end, Connected = Status =:= 0 andalso gen_tcp:connect({local, Socket}, 0, [local, binary, {packet, 4}, {active, true}]), case Connected of @@ -329,6 +434,9 @@ execute(Pid, Id, Data, Timeout) -> call(Pid, {execute, Id, Data, Timeout}). %% Background jobs whose groups are still owned: local job groups plus the %% remote jobs the remote plugin reported through "jobs" frames, plus running %% cells. A released kernel would end them, so the idle sweep keeps it alive. +live_work(Pid) -> + cached_call(Pid, job_count, fun() -> {error, nil} end). + job_count(Pid) -> case cached_call(Pid, job_count, fun() -> call(Pid, job_count) end) of {ok, Count} when is_integer(Count), Count >= 0 -> Count; diff --git a/src/albedo/harness/extensions/python/kernel.gleam b/src/albedo/harness/extensions/python/kernel.gleam index 46a8fd40..3fb3c23b 100644 --- a/src/albedo/harness/extensions/python/kernel.gleam +++ b/src/albedo/harness/extensions/python/kernel.gleam @@ -23,6 +23,10 @@ pub type Kernel pub type Spare +/// Last published live cells and jobs; unknown never means idle. +@external(erlang, "albedo_python", "live_work") +pub fn live_work(kernel: Kernel) -> Result(Int, Nil) + @external(erlang, "albedo_python", "start_spare") pub fn start_spare(owner: process.Pid) -> Result(Spare, Error) @@ -124,6 +128,7 @@ type Boot { owned: String, fresh: Bool, remote: Option(Remote), + spare: Option(Spare), ) } @@ -524,15 +529,15 @@ pub fn open_with_spare( modules: List(String), spare: Option(Spare), ) -> Result(#(Kernel, Opened), Error) { - let opened = open_inner(store, session, cwd, host, modules) - case spare { - Some(spare) -> { - let _ = stop_spare(spare) - Nil - } - None -> Nil + let opened = open_inner(store, session, cwd, host, modules, spare) + let cleanup = case spare { + Some(spare) -> stop_spare(spare) + None -> Ok(Nil) + } + case cleanup { + Ok(_) -> opened + Error(reason) -> Error(Unavailable("spare cleanup failed: " <> reason)) } - opened } fn open_inner( @@ -541,11 +546,18 @@ fn open_inner( cwd: String, host: fn(String, Dynamic) -> String, modules: List(String), + spare: Option(Spare), ) -> Result(#(Kernel, Opened), Error) { use _ <- result.try( reconcile_stages(store, session) |> result.map_error(Unavailable), ) - use #(modules, runs, boot) <- result.try(booter(store, cwd, host, modules)) + use #(modules, runs, boot) <- result.try(booter_with_spare( + store, + cwd, + host, + modules, + spare, + )) case reattach(store, session, cwd, modules, boot) { Ok(kernel) -> Ok(#(kernel, Attached)) Error(Nil) -> { @@ -660,6 +672,23 @@ fn booter( fn(link.Record, Bool) -> Result(Kernel, Error), ), Error, +) { + booter_with_spare(store, cwd, host, modules, None) +} + +fn booter_with_spare( + store: store.Store, + cwd: String, + host: fn(String, Dynamic) -> String, + modules: List(String), + spare: Option(Spare), +) -> Result( + #( + String, + Result(String, Error), + fn(link.Record, Bool) -> Result(Kernel, Error), + ), + Error, ) { use #(python, bridge) <- result.try(paths()) use #(path, runs, reach) <- result.try(place(cwd)) @@ -683,6 +712,7 @@ fn booter( owned: record.owned, fresh: fresh, remote: remote, + spare: spare, )) }), ) diff --git a/test/e2e/albedo_kernel_spare_probe.erl b/test/e2e/albedo_kernel_spare_probe.erl new file mode 100644 index 00000000..8abcab55 --- /dev/null +++ b/test/e2e/albedo_kernel_spare_probe.erl @@ -0,0 +1,38 @@ +%% Exercise fresh-spare adoption through the real kernel/link stack. +-module(albedo_kernel_spare_probe). +-export([run_json/2]). + +run_json(Id, Cwd) -> + [Name] = [N || N <- registered(), lists:prefix("albedo_registry", atom_to_list(N))], + {ok, Runtime} = 'albedo@daemon@registry':host('gleam@erlang@process':named_subject(Name)), + Ledger = 'albedo@harness@runtime':ledger(Runtime), + Owner = 'albedo@daemon@store':owner(Ledger), + Host = fun(_, _) -> <<"{\"ok\":true,\"value\":null}">> end, + {ok, Spare} = albedo_python:start_spare(Owner), + true = albedo_python:spare_current(Spare), + Start = erlang:monotonic_time(microsecond), + {ok, {Kernel, _}} = 'albedo@harness@extensions@python@kernel':open_with_spare( + Ledger, Id, Cwd, Host, [<<"run">>], {some, Spare}), + AdoptUs = erlang:monotonic_time(microsecond) - Start, + false = albedo_python:spare_current(Spare), + {ok, nil} = albedo_python:stop_spare(Spare), + try + Defined = execute(Kernel, <<"define">>, <<"import fractions\nanswer=7\nrows=[1,2,3]\ndef helper(x):\n return fractions.Fraction(x,answer)\nprint('defined')">>), + Path = <>, + {ok, _} = 'albedo@harness@extensions@python@kernel':snapshot(Kernel, Path, 5000), + {ok, nil} = albedo_python:stop(Kernel), + {ok, Spare2} = albedo_python:start_spare(Owner), + {ok, {Restored, _}} = 'albedo@harness@extensions@python@kernel':open_with_spare( + Ledger, Id, Cwd, Host, [<<"run">>], {some, Spare2}), + try + {ok, _} = 'albedo@harness@extensions@python@kernel':restore(Restored, Path, 5000), + Recalled = execute(Restored, <<"recall">>, <<"assert helper(21)==3 and rows==[1,2,3]\nprint('SPARE_OK')">>), + {ok, Idle} = albedo_python:live_work(Restored), + json:encode(#{defined => Defined, recalled => Recalled, live_work => Idle, adopt_us => AdoptUs}) + after albedo_python:stop(Restored) end + after albedo_python:stop(Kernel) end. + +execute(Kernel, Id, Code) -> + Data = iolist_to_binary(json:encode(#{type => <<"execute">>, id => Id, code => Code})), + {ok, Reply} = albedo_python:execute(Kernel, Id, Data, 5000), + json:decode(Reply). diff --git a/test/e2e/kernel_spare_test.py b/test/e2e/kernel_spare_test.py new file mode 100644 index 00000000..1d69cd40 --- /dev/null +++ b/test/e2e/kernel_spare_test.py @@ -0,0 +1,23 @@ +"""Fresh spare adoption restores a namespace through the daemon's real link stack.""" + +from pathlib import Path +import unittest + +from harness import Albedo, exclusive +from inspect_support import Inspection + + +@exclusive +class KernelSpareTests(unittest.TestCase): + def test_spare_is_single_use_and_restores_definitions(self): + with Albedo(prepare=lambda app: app.env.update(ALBEDO_INSPECT="1")) as app: + session = app.session() + probe = Inspection( + app, Path(__file__).with_name("albedo_kernel_spare_probe.erl") + ) + result = probe.call_json( + "run_json", f'[<<"{session}">>, <<"{app.workspace}">>]' + ) + self.assertEqual(result["recalled"]["status"], "ok", result) + self.assertIn("SPARE_OK", result["recalled"]["output"]) + self.assertEqual(result["live_work"], 0) diff --git a/test/manual/agent_density.py b/test/manual/agent_density.py index 0bf4c3f6..1ac12244 100644 --- a/test/manual/agent_density.py +++ b/test/manual/agent_density.py @@ -56,13 +56,18 @@ def processes(app): result = [{"pid": app.connection["pid"], "kind": "daemon"}] for line in listing.splitlines(): pid, command = line.strip().split(None, 1) - if str(app.home) + "/run/" in command and any( - name in command for name in ("albedo_kernel.py", "albedo_bridge.py") + if str(app.home) in command and any( + name in command + for name in ("albedo_kernel.py", "albedo_bridge.py", "albedo_spare.py") ): result.append( { "pid": int(pid), - "kind": "kernel" if "albedo_kernel.py" in command else "bridge", + "kind": "spare" + if "albedo_spare.py" in command + else "kernel" + if "albedo_kernel.py" in command + else "bridge", } ) return result