links: link workspaces into one project, sharing memory and work items
/link adds or removes peers in a workspace_links group. each member keeps writing its own memory and work items; reads cover the group (memory snapshot, grep and search; every work ledger read and change), so a gone member loses nothing and unlinking undoes nothing. a member's folder is checked where it is: a remote one with one exists gather over the shared ControlMaster, every member at once. a host that doesn't answer reads as connecting, sign in or unreachable, never gone, and opening a session never waits on a host. a link or unlink queues a note in every open session of the workspaces it touches; a link note carries the new members' memory snapshot, so an open session sees it without rebuilding its prompt. memory moves daemon-side: the kernel's memory object calls the memory host route through a new blocking host_now, so a remote session's notes land in the daemon's ALBEDO_HOME instead of the host's. Signed-off-by: dawn <90008@klbr.net>
40 unmodified lines4142434445454647484943 unmodified lines40 unmodified lines
- **`src/albedo/daemon/`** — session actors and durable state: submission/run/turn plumbing (`session_*.gleam`), the append-only transcript and conversation store (SQLite via sqlight), the event bus, projections, the family (agents, mail, bus), folders, quota, usage, the reaper, persisted settings, the image store, and `server.gleam`, the HTTP entry point.- **`src/albedo/daemon/migrations/`** — existing SQLite upgrades, with ordered startup data steps in `daemon/migrations.gleam`; schema additions remain called from owning initialisers (see `robot-docs/migrations.md`).- **`src/albedo/harness/`** — the model-facing half: `loop.gleam` (the tool loop), `tool.gleam`, `extension.gleam` and `extensions.gleam` (plugin interfaces and composition), `compaction.gleam`, `runtime.gleam` (boots the built-ins), plus credentials, oauth, rotation, session settings mutations, shared settings persistence, extension settings, and the usage feed.- **`src/albedo/harness/extensions/`** — one directory per extension. Enabled by default: `python`, `run`, `work`, `mail`, `agents`, `schedule`, `paperclips`, `files`, `memory`, `instructions`, `commands`, `skills`, `models`, `openai`, `codex`, `antigravity`, `alibaba`, `claude`, `rolling`, `snapcompact-memory`, `lcm-memory`, `notes`, `remote`, `browser`. Installed but off until enabled: `mcp`, `view`, `proxy`, `webhooks`, `warm`, `snapcompact`, `lcm`. The provider extensions are ordered before `models` on purpose (their own model lists win), and the memory/compaction pairs are ordered so archives cover past folds.- **`src/albedo/harness/extensions/`** — one directory per extension. Enabled by default: `python`, `run`, `work`, `mail`, `agents`, `schedule`, `paperclips`, `files`, `memory`, `links`, `instructions`, `commands`, `skills`, `models`, `openai`, `codex`, `antigravity`, `alibaba`, `claude`, `rolling`, `snapcompact-memory`, `lcm-memory`, `notes`, `remote`, `browser`. Installed but off until enabled: `mcp`, `view`, `proxy`, `webhooks`, `warm`, `snapcompact`, `lcm`. The provider extensions are ordered before `models` on purpose (their own model lists win), and the memory/compaction pairs are ordered so archives cover past folds.- **`src/albedo/openai_api/`** — the provider transport every extension shares: request/turn types, SSE framing, stream reducers, replay after restarts, chat/responses encodings. A provider with a foreign wire format encodes an `openai_api.Exchange` and supplies only its own reducer (antigravity does this); HTTP, retries, limits, and callbacks stay shared.
The extension record is the core abstraction everything composes through:
43 unmodified lines43 unmodified lines444546474849505152533 unmodified lines43 unmodified lines- `/raise-cap [on|off] [model]` — raise a model's context window to the provider's maximum, or restore its default (user only). Without a state it toggles; without a model it uses the session's. The choice is saved per model in `raisedCaps` and applies to every session on that model from its next request. See [raised caps](models.md#raised-caps).
the python extension contributes `/kernel [upgrade]` — whether the session's kernel runs older code than the daemon (bundle, module set, or protocol) and what keeps it, or, with `upgrade` (user only, between turns), swap it now past its live jobs (`KernelReport` / `KernelUpgrade` state ops). See [kernel](kernel.md#version-skew).
the links extension contributes `/link [add|remove] [workspace]` — the workspaces linked with this one, as a page (`here`, `gone` for a member whose folder is missing, checked on its host for a remote one, or why the host can't say), or, user only, link one or unlink one; every open session in the workspaces it touches gets a note either way. See [workspaces](workspaces.md#linked-workspaces).
skills contributes one command per cataloged skill; `/tree`, `/fork`, `/login`, and the other picker-style entries stay presentation-only in the CLI.
## CLI page lifecycle
3 unmodified lines1 unmodified line23456678910212 unmodified linesExpand all1 unmodified line
Persisted preferences are owned by the daemon and changed through the [settings API](settings.md). The CLI refreshes them on use and never writes settings files.
an extension is a named bundle of plugins. it can contribute any number of context, tool, command, managed, models, provider, login, service, migration, compaction, notes, clean, or fold plugins. `python`, `run`, `work`, `mail`, `agents`, `schedule`, `paperclips`, `files`, `memory`, `instructions`, `commands`, `skills`, `models`, `openai`, `codex`, `antigravity`, `alibaba`, `claude`, `rolling`, `snapcompact-memory`, `lcm-memory`, `notes`, `remote`, and `browser` are enabled by default; `mcp`, `view`, [`proxy`](proxy.md), `webhooks`, [`warm`](cache-warming.md), and the `snapcompact` and `lcm` compaction strategies are installed and disabled until enabled. plugin contributions compose inside them.an extension is a named bundle of plugins. it can contribute any number of context, tool, command, managed, models, provider, login, service, migration, compaction, notes, clean, or fold plugins. `python`, `run`, `work`, `mail`, `agents`, `schedule`, `paperclips`, `files`, `memory`, `links`, `instructions`, `commands`, `skills`, `models`, `openai`, `codex`, `antigravity`, `alibaba`, `claude`, `rolling`, `snapcompact-memory`, `lcm-memory`, `notes`, `remote`, and `browser` are enabled by default; `mcp`, `view`, [`proxy`](proxy.md), `webhooks`, [`warm`](cache-warming.md), and the `snapcompact` and `lcm` compaction strategies are installed and disabled until enabled. plugin contributions compose inside them.
## select extensions
open `/extensions` in a session to inspect installed extensions and enable or disable them. selection is saved per session. enabling a compaction strategy disables the previously selected strategy in the same reload. to replace the active strategy, enable the new one; disabling the only active strategy is rejected. required extensions must stay enabled, and changes require an idle session.212 unmodified linesExpand all19 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`; 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 runpaperclips contributes `title.apply`, `scope.apply`, and `reply.apply`; the links extension's initialiser creates `workspace_links` (see [workspaces](workspaces.md#linked-workspaces)), which `links.link` also ensures; 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 all37 unmodified lines383940414242434445468 unmodified lines55565758595960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121190 unmodified linesExpand all37 unmodified linescached per host for the daemon's life. without ssh the user stays.
a remote session's python kernel and `run` jobs run on its host(kernel.md, "remote kernels"). what is only a key works as it is: the workledger and paperclips scope, recent folders, session search by cwd, familyledger and paperclips scope, workspace links, recent folders, session search by cwd, familymoves. what the daemon itself reads from a workspace goes through thelocation, and never looks for a remote path on its own disk:
- `/fs/list`, `/fs/repo` and `/fs/preview` take `host:/abs` and `host:~`8 unmodified lines skill's `location` is its mirror path. while the host is out of reach only the home directories count, and the instructions warn and the skills catalog diagnoses `project … skipped: can't reach chernobog: …`.- memory is a daemon-side file keyed by the workspace string, so a remote workspace gets its own. workspace gets its own. the kernel's `memory` object reaches it through the `memory` host route wherever the kernel runs (`memory.read`, `.save`, `.append`, `.journal`, `.documents`; synchronous over the kernel's `host_now`), so a remote session's notes land in the daemon's `$ALBEDO_HOME/memories`, never the host's.
## linked workspaces
a workspace can be linked with others that hold the same project, such as onerepository checked out here and on chernobog (`src/albedo/harness/links.gleam`,the `links` extension). links are peers: every member keeps writing its ownmemory and work items, and reads cover the whole group.
- storage: `workspace_links(workspace, grp)` in the daemon's sqlite. linking two workspaces merges their groups; unlinking one takes only that row out, and a group left with one member is dropped. nothing else changes, so unlinking (or linking again) loses and repeats nothing.- memory: the snapshot a session opens with shows its own memory, then each linked member's under `## linked workspace <location>`, within the same 8000-character budget. `memory.read()` is this workspace's file only; `grep` and `search` cover the group, a linked file named `[<location>] memory.md`. `save`, `append` and `journal` write here.- work ledger: `list`, `get`, `update` and `delete` reach every member's items, an update or delete stays in the item's own workspace, and `create` files under this one. each item carries `workspace`; the `/work` page says where an item from another member was filed.- vents are one ledger for every workspace already, so links change nothing there.- the model is told it is linked (the `links` context, naming each member, and marking one whose folder is gone).- linking or unlinking queues a note in every open session of every workspace it touches, worded for that side ("linked this workspace with X" / "unlinked X" / "unlinked this workspace from Y, Z"). a link note carries the newly linked members' memory as a new session's snapshot holds it (`albedo_memory:linked`), so a session that opened before the link reads the same text without rebuilding its prompt and losing its prompt cache; its snapshot catches up at the next rebuild. open means the session has a live actor (and so a cached composition); any other session composes its snapshot afresh when it opens. live reads (`grep`, `search`, the work ledger) follow the group on every call anyway.- a member's folder is checked where it is: a local one on the daemon's disk, a remote one with one `exists` gather on its host over the shared ControlMaster (`folders.exists`). a host that does not answer is never read as a gone folder. every member is checked at once, so slow hosts wait side by side. the `/link` page waits up to 5 s for a host and then says why (`connecting`, `sign in`, `unreachable`, with ssh's words); the session-start context takes only a recent probe, starting one in the background, so opening a session never waits on a host.- `/link` (user only for changes) lists the members: this one `here`, a member whose folder no longer exists `gone`, a remote one whose host can't say as above. `add <workspace>` takes what a session can be created in (an existing local directory, or `host:/abs` / `host:~/x`) and refuses this workspace and one already linked; `remove <workspace>` takes a listed one, this one included (leaving the group). a gone member stays, read as before, until it is removed.
a worktree or jj workspace of the same repository is not linked by itselfyet; that and suggesting links from shared history are later work.
the tui shows a remote workspace as `label:path`, the path folded underthat host's own home once a listing or probe reported it: the chat headerreads `✦ albedo on chernobog:~/proj/albedo`, folds only the path, keeps thehost in every layout that shows a place, and colors the host with a stable190 unmodified linesExpand all134 unmodified linesExpand all13513613713813914014114214314414550 unmodified lines134 unmodified linesExpand all attach_image: Callable[[bytes], str] | None = None # Local shell admission, shared across the daemon's kernels. Completion # releases it through the existing verified `job` cleanup frame. job_slot: Callable[[str, Callable[[], None]], Awaitable[None]] | None = None # A host call that blocks until its answer, for synchronous APIs over # routes that answer at once. Absent when the host pre-dates it. host_now: Callable[[str, dict[str, object]], object] | None = None
class PythonPlugin(Protocol): def setup(50 unmodified lines8 unmodified lines910111213131415161718129 unmodified linesExpand all14814915015115115215315415515655 unmodified lines8 unmodified linescommands the daemon planned, file sizes, and project files. Everything comesback in one ssh round trip per question.
Request: `{route, dir, plans, separators, deadline, counted}`, `dir` absoluteand normalised. `route` is list, repo, preview or project.and normalised. `route` is list, repo, preview or project, or exists, whichanswers `directory` alone.
Snapshot: `{directory, entries: {dir: [[name, dir, modified]]}, exists:[path], outputs: {key: stdout}, sizes: {path: bytes}, files: {relative:base64}}`. An output's key is its program and arguments joined by NUL; only129 unmodified linesExpand all
def gather(request: dict) -> dict[str, object]: route, directory = request["route"], request["dir"] snapshot: dict[str, object] = {"directory": os.path.isdir(directory)} if not snapshot["directory"]: if not snapshot["directory"] or route == "exists": return snapshot if route == "project": snapshot["files"] = project(directory) return snapshot55 unmodified lines57 unmodified lines585960616263646566676869112 unmodified linesExpand all182183184185186187188189190191192193120 unmodified linesExpand all3143153163173183193203213223233243253263273283293303313323333343353363373383393403413423431279 unmodified linesExpand all16231624162516261627162816291630163146 unmodified lines57 unmodified lines
albedo_proc.reap_stopped_safely(LOOP)QUEUE: asyncio.Queue[albedo_api.Execute | albedo_api.State] = asyncio.Queue()PENDING: dict[str, asyncio.Future[albedo_api.HostReply]] = {}# Host calls a synchronous API waits for; the reader thread hands their reply# over directly, since the loop it would go through is the one waiting.WAITING: dict[str, tuple[threading.Event, list[albedo_api.HostReply]]] = {}HOST_NOW_TIMEOUT = 30.0JOB_SLOTS: dict[str, tuple[asyncio.Future[None], Callable[[], None]]] = {}HOST_SLOTS = asyncio.Semaphore(32)OWNER_CALLS = asyncio.Semaphore(32) # concurrent tool calls from the connection ownerOWNER_TASKS: dict[str, asyncio.Task[object]] = {} # interruptable by invoke id112 unmodified linesExpand all else: task = OWNER_TASKS.get(message["id"]) if task is not None: LOOP.call_soon_threadsafe(task.cancel) elif message["type"] == "reply" and message["id"] in WAITING: arrived, answer = WAITING[message["id"]] answer.append(message["value"]) arrived.set() else: _ = LOOP.call_soon_threadsafe(deliver, message)
120 unmodified linesExpand all return answer["value"] raise WorkError(answer["code"], answer["message"])
def host_now(method: str, args: dict[str, object]) -> object: """A host call for a synchronous API: blocks the calling thread, the loop's included, until the reply. Only for calls the host answers at once; an interrupt still ends the wait.""" key = uuid.uuid4().hex arrived = threading.Event() answers: list[albedo_api.HostReply] = [] WAITING[key] = (arrived, answers) send({"type": "call", "id": key, "method": method, "args": args}) try: if not arrived.wait(HOST_NOW_TIMEOUT): raise WorkError( "timeout", f"{method} had no answer within {HOST_NOW_TIMEOUT:.0f}s" ) finally: _ = WAITING.pop(key, None) answer = answers[0] if answer["ok"]: return answer["value"] raise WorkError(answer["code"], answer["message"])
class Capture: def __init__( self, id: str, kind: str = "cell", max_edge: int | None = None ) -> None:1279 unmodified linesExpand all api = albedo_api.PythonApi( version=2, loop=LOOP, host=host, host_now=host_now, HostError=WorkError, forget_output=forget_output, capture=background_capture, preview=PREVIEW,46 unmodified lines123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174"""Linked workspaces: one project in several places.
A workspace linked with another reads that one's memory and work items aswell as its own, while everything it writes is filed under itself. A memberwhose folder is gone keeps what it had until the user removes it, andunlinking deletes nothing."""
import jsonimport reimport shutilimport unittestimport urllib.error
from harness import Albedo, Provider, python, text
class LinkTests(unittest.TestCase): def setUp(self): self.cells = []
def script(request): if request["input"][-1].get("role") == "user" and self.cells: return python(self.cells.pop(0)) return text("ok")
self.provider = Provider(script) self.addCleanup(self.provider.close) self.app = Albedo(self.provider, protocol="responses") self.app.__enter__() self.addCleanup(self.app.__exit__, None, None, None)
def cell(self, session, code): self.cells.append(code) self.app.prompt(session, "run it").close() self.app.idle(session) results = [ json.loads(event["result"]) for event in self.app.events(session) if event.get("type") == "tool" and event.get("name") == "python" ] self.assertEqual(results[-1]["status"], "ok", results[-1]) return results[-1]
def link(self, session, **args): with self.app.api( f"/sessions/{session}/commands", {"name": "/link", "args": args} ) as response: return json.load(response)["result"]
def memory_file(self, workspace): slug = re.sub(r"[^A-Za-z0-9]", "-", str(workspace)) return self.app.home / "memories" / slug / "memory.md"
def test_linked_workspaces_read_each_other_and_write_their_own(self): laptop, server = self.app.root / "laptop", self.app.root / "server" laptop.mkdir() server.mkdir() first = self.app.session(laptop) self.cell( first, "memory.append('the cache key is the prefix hash')\n" "await work.create('ship the picker')", )
here = self.app.session(server) added = self.link(here, action="add", details=str(laptop)) self.assertIn("linked with " + str(laptop), added["message"]) rows = self.link(here)["page"]["rows"] self.assertEqual( [(row["id"], row["badge"]) for row in rows], [(str(server), "here"), (str(laptop), "")], )
# A session opened after the link starts with the linked memory and # is told it is linked. later = self.app.session(server) read = self.cell( later, "memory.append('chernobog runs the gate')\n" "await work.create('profile the gate')\n" "print(memory.grep('cache key'))\n" "print(sorted((i.title, i.workspace) for i in await work.list()))", ) self.assertIn( "[" + str(laptop) + "] memory.md:1: the cache key", read["output"] ) self.assertIn(f"('ship the picker', '{laptop}')", read["output"]) self.assertIn(f"('profile the gate', '{server}')", read["output"]) instructions = self.provider.requests[-1]["request"]["instructions"] self.assertIn("the cache key is the prefix hash", instructions) self.assertIn("This workspace is linked with " + str(laptop), instructions) # Writes stayed where they were made. self.assertEqual( self.memory_file(server).read_text(), "chernobog runs the gate\n" ) self.assertEqual( self.memory_file(laptop).read_text(), "the cache key is the prefix hash\n" )
# The laptop checkout goes away: it is shown as gone and still read. shutil.rmtree(laptop) rows = self.link(here)["page"]["rows"] self.assertEqual(rows[1]["badge"], "gone") kept = self.cell(later, "print(memory.grep('cache key'))") self.assertIn("the cache key", kept["output"])
# Removing it stops the reads and deletes nothing. self.link(here, action="remove", details=str(laptop)) self.assertEqual(self.link(here)["page"]["rows"], []) alone = self.cell( later, "print(memory.grep('cache key'))\n" "print([i.title for i in await work.list()])", ) self.assertNotIn("the cache key", alone["output"]) self.assertNotIn("ship the picker", alone["output"]) self.assertTrue(self.memory_file(laptop).exists())
def test_open_sessions_in_the_group_are_told_with_the_memory_they_now_read(self): laptop, server = self.app.root / "desk", self.app.root / "rack" laptop.mkdir() server.mkdir() on_laptop = self.app.session(laptop) self.cell(on_laptop, "memory.append('the cache key is the prefix hash')") here, beside = self.app.session(server), self.app.session(server) self.cell(beside, "1") unopened = self.app.session(server)
added = self.link(here, action="add", details=str(laptop)) self.assertIn("told 3 open sessions", added["message"])
def next_request(session): self.app.prompt(session, "hi").close() self.app.idle(session) request = self.provider.requests[-1]["request"] return json.dumps(request["input"]), request["instructions"]
# Sessions in this workspace opened before the link: the laptop's # memory arrives in the note, not in a rebuilt prompt. for session in (here, beside): told, instructions = next_request(session) self.assertIn("linked this workspace with " + str(laptop), told) self.assertIn("the cache key is the prefix hash", told) self.assertNotIn("the cache key", instructions) told, _ = next_request(on_laptop) self.assertIn("linked this workspace with " + str(server), told) # One that never opened composes its snapshot when it does. told, instructions = next_request(unopened) self.assertNotIn("linked this workspace", told) self.assertIn("the cache key is the prefix hash", instructions)
self.link(here, action="remove", details=str(laptop)) told, _ = next_request(on_laptop) self.assertIn("unlinked this workspace from " + str(server), told) told, _ = next_request(beside) self.assertIn("The user unlinked " + str(laptop), told)
def test_a_link_is_refused_for_this_workspace_and_missing_folders(self): workspace = self.app.root / "alone" workspace.mkdir() session = self.app.session(workspace) for details, reason in ( (str(workspace), "that is this workspace"), (str(self.app.root / "nowhere"), "must be an existing absolute directory"), ): with self.assertRaises(urllib.error.HTTPError) as refused: self.link(session, action="add", details=details) self.assertIn(reason, refused.exception.read().decode()) self.assertEqual(self.link(session)["page"]["rows"], [])
if __name__ == "__main__": unittest.main()7 unmodified lines891011121314151623 unmodified lines404142434445464748183 unmodified linesExpand all232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318170 unmodified linesExpand all7 unmodified lines"""
import jsonimport osimport reimport shutilimport signalimport subprocessimport sys23 unmodified linesficase "$target" in gone) echo "ssh: connect to host gone port 22: Connection refused" >&2; exit 255 ;; locked) echo "locked: Permission denied (publickey)." >&2; exit 255 ;; slow-*) sleep 3; echo "ssh: connect to host $target port 22: Connection refused" >&2; exit 255 ;;esacpath={bin}:/usr/bin:/bin[ "$target" = oldhost ] && path={old}:$pathexec env -i HOME={home} PATH="$path" SHELL={bin}/loginsh /bin/sh -c "$*"183 unmodified linesExpand all self.app.restart(crash=True) after = self.cell(session, "survivor += 1\n(survivor, os.getpid())") self.assertEqual(after["value"], f"(42, {first['value']})")
def test_a_remote_sessions_memory_is_the_daemons_and_links_with_a_local_one(self): remote = f"fakehost:{self.project}" session = self.create(remote)["id"] self.cell(session, "memory.append('the gate runs on fakehost')") slug = re.sub(r"[^A-Za-z0-9]", "-", remote) written = self.app.home / "memories" / slug / "memory.md" self.assertEqual(written.read_text(), "the gate runs on fakehost\n") self.assertFalse((self.remote_home / ".albedo").exists())
# The same project checked out here, linked: it reads the remote one. checkout = self.app.root / "checkout" checkout.mkdir() local = self.create(str(checkout))["id"] with self.app.api( f"/sessions/{local}/commands", {"name": "/link", "args": {"action": "add", "details": remote}}, ) as response: self.assertIn("linked with", json.load(response)["result"]["message"]) found = self.cell(local, "print(memory.grep('gate'))") self.assertIn(f"[{remote}] memory.md:1: the gate runs", found["output"])
def test_a_linked_remote_folder_is_checked_on_its_host(self): checkout = self.app.root / "checkout" checkout.mkdir() local = self.create(str(checkout))["id"]
def link(**args): with self.app.api( f"/sessions/{local}/commands", {"name": "/link", "args": args} ) as response: return json.load(response)["result"]
def badges(): rows = link()["page"]["rows"][1:] return {row["id"]: (row["badge"], row["detail"]) for row in rows}
present = f"fakehost:{self.project}" missing = f"fakehost:{self.remote_home}/missing" for member in (present, missing, "gone:/srv/app"): link(action="add", details=member) found = badges() self.assertEqual(found[present], ("", "")) self.assertEqual(found[missing][0], "gone") badge, detail = found["gone:/srv/app"] self.assertEqual(badge, "unreachable") self.assertIn("Connection refused", detail)
# A host that stops answering is not a gone folder. self.down.touch() found = badges() self.assertEqual(found[present][0], "unreachable") self.assertEqual(found[missing][0], "unreachable") self.down.unlink()
# A session opening with the host's probe fresh is told what is gone. self.cell(self.create(str(checkout))["id"], "1") instructions = self.provider.requests[-1]["request"]["instructions"] self.assertIn(f"{missing} (its folder is gone)", instructions) self.assertNotIn(f"{present} (its folder is gone)", instructions)
def test_linked_hosts_are_checked_side_by_side(self): checkout = self.app.root / "slow-checkout" checkout.mkdir() local = self.create(str(checkout))["id"]
def link(**args): with self.app.api( f"/sessions/{local}/commands", {"name": "/link", "args": args} ) as response: return json.load(response)["result"]
for host in ("slow-a", "slow-b"): link(action="add", details=f"{host}:/srv") # Each host takes 3 s to refuse; one after the other would be 6. began = time.monotonic() rows = link()["page"]["rows"][1:] self.assertLess(time.monotonic() - began, 5.5) self.assertEqual([row["badge"] for row in rows], ["unreachable"] * 2)
def test_a_remote_home_is_resolved_and_stored_absolute(self): created = self.create("fakehost:~/proj") self.assertEqual(created["workspace"], f"fakehost:{self.project}") with self.assertRaises(urllib.error.HTTPError) as refused:170 unmodified linesExpand all112345678951011789121312131415161718192020212122232324252624252630272829303132333438394041424344354647484950515253545556575836373839616263646566674041427071727374434445464776777879808148495051525354555657585960616283848586636465666791926869709495967172739974757677787980818283841091108586878889909111711811912012192939495962 unmodified lines99100101102103132133104105106107108109110137138139140141142143111112113114"""Project-scoped Markdown memory with literal and full-text recall.""""""Project-scoped Markdown memory with literal and full-text recall.
The files live on the daemon's machine, wherever this kernel runs, and everycall goes through the memory host route. Reads of a linked workspace's memorycome back with it; writes always land in this workspace's own."""
from __future__ import annotations
from contextlib import contextmanagerfrom collections.abc import Callablefrom datetime import dateimport fcntlimport osfrom pathlib import Pathimport reimport sqlite3import tempfilefrom typing import Iteratorfrom typing import cast
from albedo_api import PythonApi, Textfrom albedo_plugins.files import Rows
class Memory: def __init__(self, root: Path) -> None: self.root = root def __init__(self, host: Callable[[str, dict[str, object]], object]) -> None: self._host = host
@property def path(self) -> Path: return self.root / "memory.md"
def read(self, max_chars: int = 16000) -> Text: if not 0 < max_chars <= 200_000: raise ValueError("0 < max_chars <= 200000") content = self.path.read_text() if self.path.exists() else "" content = cast(str, self._host("memory.read", {})) suffix = ( "\n[truncated; raise max_chars to read more]" if len(content) > max_chars else "" ) return Text(content[:max_chars] + suffix)
@contextmanager def _locked(self) -> Iterator[None]: self.root.mkdir(parents=True, exist_ok=True) with (self.root / ".lock").open("a+b") as lock: fcntl.flock(lock, fcntl.LOCK_EX) yield
def save(self, text: str) -> Text: """Atomically replace curated memory, serializing with append.""" if len(text.encode("utf-8")) > 1_048_576: raise ValueError("memory.md cannot exceed 1 MiB") with self._locked(): fd, name = tempfile.mkstemp(dir=self.root, prefix=".memory-") try: with os.fdopen(fd, "w", encoding="utf-8") as file: file.write(text) os.replace(name, self.path) finally: if os.path.exists(name): os.unlink(name) return Text(str(self.path)) """Atomically replace curated memory.""" return Text(cast(str, self._host("memory.save", {"text": text})))
def append(self, text: str) -> Text: with self._locked(): size = self.path.stat().st_size if self.path.exists() else 0 extra = len(text.rstrip("\n").encode("utf-8")) + 1 + (size > 0) if size + extra > 1_048_576: raise ValueError("memory.md cannot exceed 1 MiB") self._append(self.path, text) return Text(str(self.path)) return Text(cast(str, self._host("memory.append", {"text": text})))
def journal(self, text: str) -> Text: target = self.root / "journal" / f"{date.today().isoformat()}.md" with self._locked(): target.parent.mkdir(exist_ok=True) self._append(target, text) return Text(str(target)) day = date.today().isoformat() return Text( cast(str, self._host("memory.journal", {"date": day, "text": text})) )
@staticmethod def _append(path: Path, text: str) -> None: with path.open("a", encoding="utf-8") as file: if path.stat().st_size: file.write("\n") file.write(text.rstrip("\n") + "\n") def _documents(self) -> tuple[list[tuple[str, str]], bool]: """(name, text) for every memory and journal file this workspace reads; a linked workspace's names lead with it.""" answer = cast(dict, self._host("memory.documents", {})) named = [ ( document["path"] if document["workspace"] == answer["own"] else f"[{document['workspace']}] {document['path']}", document["text"], ) for document in answer["documents"] ] return named, bool(answer["truncated"])
def _documents(self) -> list[Path]: paths = [self.path, *sorted((self.root / "journal").glob("*.md"))] return [path for path in paths if path.is_file()]
def grep(self, term: str, limit: int = 20) -> Rows: if not term or not 1 <= limit <= 100: raise ValueError("term must be nonempty and 1 <= limit <= 100") documents, cut = self._documents() matches = [] for path in self._documents(): for number, line in enumerate(path.read_text().splitlines(), 1): for name, content in documents: for number, line in enumerate(content.splitlines(), 1): if term.casefold() in line.casefold(): matches.append( f"{path.relative_to(self.root)}:{number}: {line[:500]}" ) matches.append(f"{name}:{number}: {line[:500]}") if len(matches) == limit: return Rows(matches, truncated=True) return Rows(matches) return Rows(matches, truncated=cut)
def search(self, query: str, limit: int = 20) -> Rows: """FTS5 ranks paragraphs, including multiline entries, without a stale index.""" if not query.strip() or not 1 <= limit <= 100: raise ValueError("query must be nonempty and 1 <= limit <= 100") documents, cut = self._documents() with sqlite3.connect(":memory:") as db: db.execute( "CREATE VIRTUAL TABLE notes USING fts5(path UNINDEXED, line UNINDEXED, body)" ) for path in self._documents(): content = path.read_text() for name, content in documents: for match in re.finditer( r"(?s)\S.*?(?=\n[ \t]*\n|\s*\Z)", content, re.S ): line = content.count("\n", 0, match.start()) + 1 db.execute( "INSERT INTO notes VALUES (?, ?, ?)", ( str(path.relative_to(self.root)), line, match.group()[:100_000], ), (name, line, match.group()[:100_000]), ) try: rows = db.execute( "SELECT path, line, snippet(notes, 2, '[', ']', '…', 16) "2 unmodified lines ).fetchall() except sqlite3.OperationalError as exc: raise ValueError(f"invalid full-text query: {exc}") from exc return Rows( ( f"{path}:{line}: {snippet.replace(chr(10), ' ')[:500]}" for path, line, snippet in rows f"{path}:{line}: {snippet.replace(chr(10), ' ')[:500]}" for path, line, snippet in rows ), truncated=cut, )
async def setup(api: PythonApi) -> dict[str, object]: workspace = await api.host("session.cwd", {}) if not isinstance(workspace, str): raise ValueError("session.cwd did not return a workspace path") home = Path(os.environ.get("ALBEDO_HOME") or Path.home() / ".albedo") slug = re.sub(r"[^A-Za-z0-9]", "-", workspace) return {"memory": Memory(home / "memories" / slug)}def setup(api: PythonApi) -> dict[str, object]: if api.host_now is None: raise RuntimeError("this host cannot answer synchronous memory calls") return {"memory": Memory(api.host_now)}