fix(python): release obsolete job output watchers
Use weak callbacks and clear obsolete watchers while preserving early-read notifications and retention limits. Benchmark: retained memory falls 73%, from 5.02 to 1.35 MiB after 256 jobs.
117 unmodified linesExpand all11811912012112212312412512612771 unmodified lines117 unmodified linesExpand allKernel 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.
`run(program, *args, cwd=, env=, stdin=)` starts one program, without a shell, as a supervised job that owns its process group. `run` is synchronous and returns the handle; awaiting the handle waits for the program, so the instructions teach `job = run(...)` then `await job` — `job = await run(...)` would leave nothing bound when a cell deadline interrupts the wait, and the model would recover the running job through `jobs` or start it twice; 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 `run` 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.
there is no shell, so shell syntax has python spellings: `cd dir &&` is `cwd=`, `NAME=value` is `env=`, `| tail -n N` and `| head -n N` are `job.tail(lines=N)` and `job.head(lines=N)`, stderr always joins stdout, and chains, loops, and globs are python over the job's output. `a | b` is `run(a…).pipe(b…)`, the same as `run(b…, stdin=job)`: the writer's bytes stream into the reader exactly, a full reader pauses the writer, and once every reader has closed its stdin the kernel stops reading the writer, so its next write fails as it would into a closed shell pipe (`run('yes').pipe('head', '-2')` ends). the writer still captures its own output; piping retires its wake, and the reader's notice names the whole pipeline. a job that already wrote feeds what it retained first; one that wrote past the 1 MiB it retains cannot be piped after the fact. `run("bash", "-c", …)` and its spellings are refused, and so is a cell's own `subprocess`, `os.system`, `os.exec*`, `os.posix_spawn`, or `asyncio.create_subprocess_*` — an audit hook in `albedo_shell` checks whether the nearest non-stdlib frame is cell code. each refusal carries the `run()` call the line means when it is simple enough to read (`cd cli && go test ./... | tail -5` → `job = run('go', 'test', './...', cwd='cli')`, `await job`, then `job.tail(lines=5)`), so the error is a copy-paste answer rather than a wall. this is guidance, not a sandbox: plugins and libraries a cell calls still spawn, and `exec` of a compiled string is not cell code.
71 unmodified lines27 unmodified lines28293031323334353643 unmodified lines808182838485868788369 unmodified linesExpand all458459460461460462463464465464465466467468469470471472473474475476477478479480481482483473484485486487488489490387 unmodified linesExpand all878879880881869882883884885886478 unmodified linesExpand all27 unmodified linesimport tempfileimport threadingimport tracebackimport uuidimport weakref
MAX_FRAME = 8 * 1024 * 1024CLEANUP_DEADLINE = ( 1.5 # seconds: plugins must finish cleanup inside the supervisor's patience43 unmodified lines
def forget_output(id: str) -> None: ARCHIVES.pop(id, None) READ_WATCHERS.pop(id, None)
def show(value: object) -> str: return (369 unmodified linesExpand all ) return capture
READ_WATCHERS: dict[str, Callable[[], None]] = {}READ_WATCHERS: dict[str, Callable[[], None] | weakref.WeakMethod] = {}
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 """Notify reads while the output is retained, without owning a bound method's job."""
def expired(reference: weakref.WeakMethod) -> None: if READ_WATCHERS.get(id) is reference: READ_WATCHERS.pop(id, None)
try: READ_WATCHERS[id] = weakref.WeakMethod(notify, expired) except TypeError: # Functions and other callables need a strong reference to stay usable. 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: up to 1 MiB each, for the 16 most recent cells and 64 most recent jobs.""" capture = retained(id) notify = READ_WATCHERS.pop(id, None) # one read satisfies the watcher notify = READ_WATCHERS.get(id) if isinstance(notify, weakref.WeakMethod): notify = notify() if notify is not None: try: notify() except Exception:387 unmodified linesExpand all ARCHIVES.move_to_end(capture.id) counts = collections.Counter(held.kind for held in ARCHIVES.values()) for key, held in list(ARCHIVES.items()): if counts[held.kind] > LIMITS[held.kind]: del ARCHIVES[key] forget_output(key) counts[held.kind] -= 1
def state_reply(message: albedo_api.State) -> dict[str, object]:478 unmodified linesExpand all123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107"""Watcher ownership and completion/read races need GC and busy-host control that daemon E2E cannot reliably provide."""
from pathlib import Pathimport sysimport textwrapimport unittest
sys.path.insert(0, str(Path(__file__).resolve().parent))from job_wake_test import Owner # noqa: E402
class JobWatchersTest(unittest.TestCase): def setUp(self): self.owner = Owner(["run", "files"]) self.addCleanup(self.owner.close) self.owner.wait_for(lambda frame: frame.get("type") == "ready") self.cells = 0 self.cell( "import __main__ as kernel, asyncio, gc, weakref\n" "from albedo_plugins import run as runner, files as file_plugin" )
def cell(self, code): self.cells += 1 cell_id = f"c{self.cells}" self.owner.send( {"type": "execute", "id": cell_id, "code": textwrap.dedent(code)} ) done = self.owner.wait_for( lambda frame: frame.get("type") == "done" and frame.get("id") == cell_id ) self.assertEqual(done["status"], "ok", done) return done
def test_an_early_read_does_not_consume_the_completion_watcher(self): self.cell(""" job = run('echo', 'result') output.read(job.id) await asyncio.shield(job.task) assert not job._read and not job._awaited assert output.read(job.id).strip() == 'result' assert job._read """)
def test_forgotten_jobs_and_internal_file_commands_release_watchers(self): self.cell(""" job = run('true') await job reference = weakref.ref(job) runner.forget(job) del job await file_plugin._run('printf internal') await asyncio.sleep(0) gc.collect() assert reference() is None assert not jobs and not kernel.READ_WATCHERS assert not [c for c in kernel.ARCHIVES.values() if c.kind == 'job'] """)
def test_eviction_releases_jobs_but_preserves_held_handles(self): self.cell(""" held = run('echo', 'held output') await held disposable = run('true') await disposable reference = weakref.ref(disposable) capture = disposable.capture del disposable for index in range(runner.RETAINED_LIMIT): # Keep this output past job eviction to catch strong watcher ownership. kernel.remember(capture) await run('true') await asyncio.sleep(0) gc.collect() assert reference() is None and capture.id in kernel.ARCHIVES assert capture.id not in kernel.READ_WATCHERS assert held.id not in jobs and held.id not in kernel.ARCHIVES assert held.id not in kernel.READ_WATCHERS assert held.tail().strip() == 'held output' and await held is held """)
def test_other_callables_and_replacement_watchers_remain_usable(self): self.cell(""" capture = kernel.background_capture('callables') reads = [] kernel.watch_output(capture.id, lambda: reads.append('function')) gc.collect() output.read(capture.id) class Callback: def __call__(self): reads.append('callable') callback = Callback() kernel.watch_output(capture.id, callback) output.read(capture.id) kernel.watch_output(capture.id, callback.__call__) kernel.watch_output(capture.id, lambda: reads.append('replacement')) del callback gc.collect() output.read(capture.id) assert reads == ['function', 'callable', 'replacement'] kernel.forget_output(capture.id) assert capture.id not in kernel.READ_WATCHERS """)
if __name__ == "__main__": unittest.main()