From 48a5b3f534f697b874293433ae4b108d9029dfd6 Mon Sep 17 00:00:00 2001
From: "nekomimi.pet" <meowskulls@nekomimi.pet>
Date: Wed, 23 Sep 2026 02:17:47 -0400
Subject: [PATCH] fix tools failing to mention capabilities they had

---
 .../{gleam-skill => gleam-design}/SKILL.md    |  0
 .../agents/openai.yaml                        |  0
 .../assets/examples/session_core.gleam        |  0
 .../assets/examples/session_core_test.gleam   |  0
 .../references/repository-studies.md          |  0
 .../references/review-and-tests.md            |  0
 .../references/session-example.md             |  0
 .../references/source-ledger.json             |  0
 .../references/sources.md                     |  0
 .../references/state-and-concurrency.md       |  0
 .../references/types-and-modules.md           |  0
 priv/python/albedo_api.py                     | 41 +++++++++
 priv/python/albedo_kernel.py                  | 31 +++++--
 priv/python/albedo_plugins/bash.py            | 16 +++-
 priv/python/albedo_plugins/files.py           | 59 ++++++-------
 priv/python/albedo_plugins/remote.py          | 12 ++-
 priv/python/albedo_plugins/skills.py          | 18 ++--
 priv/python/albedo_plugins/work.py            | 17 ++--
 .../harness/extensions/bash/extension.gleam   |  2 +-
 .../harness/extensions/files/extension.gleam  |  2 +-
 .../harness/extensions/python/cells.gleam     | 20 +++++
 .../harness/extensions/python/extension.gleam |  4 +-
 .../harness/extensions/python/rpc.gleam       | 84 ++++++++++++++++---
 .../harness/extensions/remote/extension.gleam |  2 +-
 .../harness/extensions/skills/extension.gleam |  2 +-
 .../harness/extensions/work/extension.gleam   |  2 +-
 .../harness/extensions/work/ledger.gleam      |  7 +-
 test/harness/files_plugin_test.py             | 23 ++++-
 test/harness/kernel_api_test.gleam            | 61 ++++++++++++++
 test/harness/work_page_test.gleam             |  4 +-
 30 files changed, 328 insertions(+), 79 deletions(-)
 rename .agents/skills/{gleam-skill => gleam-design}/SKILL.md (100%)
 rename .agents/skills/{gleam-skill => gleam-design}/agents/openai.yaml (100%)
 rename .agents/skills/{gleam-skill => gleam-design}/assets/examples/session_core.gleam (100%)
 rename .agents/skills/{gleam-skill => gleam-design}/assets/examples/session_core_test.gleam (100%)
 rename .agents/skills/{gleam-skill => gleam-design}/references/repository-studies.md (100%)
 rename .agents/skills/{gleam-skill => gleam-design}/references/review-and-tests.md (100%)
 rename .agents/skills/{gleam-skill => gleam-design}/references/session-example.md (100%)
 rename .agents/skills/{gleam-skill => gleam-design}/references/source-ledger.json (100%)
 rename .agents/skills/{gleam-skill => gleam-design}/references/sources.md (100%)
 rename .agents/skills/{gleam-skill => gleam-design}/references/state-and-concurrency.md (100%)
 rename .agents/skills/{gleam-skill => gleam-design}/references/types-and-modules.md (100%)
 create mode 100644 test/harness/kernel_api_test.gleam

diff --git a/.agents/skills/gleam-skill/SKILL.md b/.agents/skills/gleam-design/SKILL.md
similarity index 100%
rename from .agents/skills/gleam-skill/SKILL.md
rename to .agents/skills/gleam-design/SKILL.md
diff --git a/.agents/skills/gleam-skill/agents/openai.yaml b/.agents/skills/gleam-design/agents/openai.yaml
similarity index 100%
rename from .agents/skills/gleam-skill/agents/openai.yaml
rename to .agents/skills/gleam-design/agents/openai.yaml
diff --git a/.agents/skills/gleam-skill/assets/examples/session_core.gleam b/.agents/skills/gleam-design/assets/examples/session_core.gleam
similarity index 100%
rename from .agents/skills/gleam-skill/assets/examples/session_core.gleam
rename to .agents/skills/gleam-design/assets/examples/session_core.gleam
diff --git a/.agents/skills/gleam-skill/assets/examples/session_core_test.gleam b/.agents/skills/gleam-design/assets/examples/session_core_test.gleam
similarity index 100%
rename from .agents/skills/gleam-skill/assets/examples/session_core_test.gleam
rename to .agents/skills/gleam-design/assets/examples/session_core_test.gleam
diff --git a/.agents/skills/gleam-skill/references/repository-studies.md b/.agents/skills/gleam-design/references/repository-studies.md
similarity index 100%
rename from .agents/skills/gleam-skill/references/repository-studies.md
rename to .agents/skills/gleam-design/references/repository-studies.md
diff --git a/.agents/skills/gleam-skill/references/review-and-tests.md b/.agents/skills/gleam-design/references/review-and-tests.md
similarity index 100%
rename from .agents/skills/gleam-skill/references/review-and-tests.md
rename to .agents/skills/gleam-design/references/review-and-tests.md
diff --git a/.agents/skills/gleam-skill/references/session-example.md b/.agents/skills/gleam-design/references/session-example.md
similarity index 100%
rename from .agents/skills/gleam-skill/references/session-example.md
rename to .agents/skills/gleam-design/references/session-example.md
diff --git a/.agents/skills/gleam-skill/references/source-ledger.json b/.agents/skills/gleam-design/references/source-ledger.json
similarity index 100%
rename from .agents/skills/gleam-skill/references/source-ledger.json
rename to .agents/skills/gleam-design/references/source-ledger.json
diff --git a/.agents/skills/gleam-skill/references/sources.md b/.agents/skills/gleam-design/references/sources.md
similarity index 100%
rename from .agents/skills/gleam-skill/references/sources.md
rename to .agents/skills/gleam-design/references/sources.md
diff --git a/.agents/skills/gleam-skill/references/state-and-concurrency.md b/.agents/skills/gleam-design/references/state-and-concurrency.md
similarity index 100%
rename from .agents/skills/gleam-skill/references/state-and-concurrency.md
rename to .agents/skills/gleam-design/references/state-and-concurrency.md
diff --git a/.agents/skills/gleam-skill/references/types-and-modules.md b/.agents/skills/gleam-design/references/types-and-modules.md
similarity index 100%
rename from .agents/skills/gleam-skill/references/types-and-modules.md
rename to .agents/skills/gleam-design/references/types-and-modules.md
diff --git a/priv/python/albedo_api.py b/priv/python/albedo_api.py
index c8ef26e..a88be01 100644
--- a/priv/python/albedo_api.py
+++ b/priv/python/albedo_api.py
@@ -16,6 +16,44 @@ Send = Callable[[dict[str, object]], None]
 Cleanup = Callable[[], object]
 
 
+def _ready(value):
+    """`await` on a value that is already here: returns it without suspending."""
+    yield from ()
+    return value
+
+
+class Text(str):
+    """A result that is ready now. Awaiting it is harmless, so a synchronous
+    call reads the same with or without `await`."""
+
+    def __await__(self):
+        return _ready(self)
+
+
+class ReadyList(list):
+    """A list result that is ready now; like Text, it may be awaited."""
+
+    def __await__(self):
+        return _ready(self)
+
+
+class Record(dict):
+    """A result record: `item["id"]` and `item.id` both work, it prints and
+    serializes as the plain dict it is, and it may be awaited like Text."""
+
+    __slots__ = ()
+
+    def __getattr__(self, name: str) -> object:
+        try:
+            return self[name]
+        except KeyError:
+            raise AttributeError(
+                f"{type(self).__name__} has no field {name!r}; its fields are {', '.join(self) or 'none'}") from None
+
+    def __await__(self):
+        return _ready(self)
+
+
 class OutputCapture(Protocol):
     tail_data: bytearray
     seen: int
@@ -43,6 +81,9 @@ class PythonApi:
     # model reads it, so background work can withdraw a completion wake the
     # read already satisfied. Absent when the host pre-dates the hook.
     watch_output: Callable[[str, Callable[[], None]], None] | None = None
+    # Drops one retained output channel, so work a plugin runs for itself (a
+    # search job, say) never crowds the model's own output out of retention.
+    forget_output: Callable[[str], None] | None = None
 
 
 class Execute(TypedDict):
diff --git a/priv/python/albedo_kernel.py b/priv/python/albedo_kernel.py
index 0aedb74..557fe87 100644
--- a/priv/python/albedo_kernel.py
+++ b/priv/python/albedo_kernel.py
@@ -229,8 +229,9 @@ def retained(id: str) -> Capture:
     capture = ARCHIVES.get(id)
     if capture is None:
         raise LookupError(
-            f"no retained output for {id!r}; output.list() names what is kept, and a background "
-            f"job keeps its own output on its handle: jobs[{id!r}].tail()")
+            f"no retained output for {id!r}: only the 16 most recent cells and 64 most recent jobs "
+            f"keep output, so it may have rolled out; output.list() names what is kept, and a "
+            f"background job keeps its own output on its handle: jobs[{id!r}].tail()")
     return capture
 
 
@@ -244,7 +245,8 @@ def watch_output(id: str, notify: Callable[[], None]) -> None:
 
 class Output:
     def read(self, id: str, *, offset: int = 0, limit: int = 4000) -> str:
-        """Read retained output by cell or job id. At most 1 MiB each, 16 recent cells."""
+        """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
         if notify is not None:
@@ -252,7 +254,7 @@ class Output:
                 notify()
             except Exception:
                 pass  # a plugin's bookkeeping must not fail the read
-        return capture.read(offset, limit)
+        return albedo_api.Text(capture.read(offset, limit))
 
     def list(self):
         """Every retained channel: cells, background jobs, and 'native'.
@@ -260,7 +262,8 @@ class Output:
         'native' holds bytes written to fd 1/2 while no cell was running; bytes a
         cell's subprocesses write land in that cell's own output instead.
         """
-        return [{"id": c.id, "kind": c.kind, "bytes": c.seen, "retained": len(c.data)} for c in ARCHIVES.values()]
+        return albedo_api.ReadyList(
+            {"id": c.id, "kind": c.kind, "bytes": c.seen, "retained": len(c.data)} for c in ARCHIVES.values())
 
 
 class Stream(io.TextIOBase):
@@ -374,10 +377,17 @@ class Cells:
             f"\n[{limit} of {len(data)} bytes; the cell has {len(lines)} lines; "
             f"narrow with start_line/end_line or raise limit]")
 
-    async def info(self, id: str) -> dict[str, object]:
-        """Status, parentage and whether the cell started, without its source."""
+    async def info(self, id: str) -> albedo_api.Record:
+        """Status (ok, error, interrupted, started, saved, lost, unavailable),
+        parentage and whether the cell started, without its source."""
         cell = cast(albedo_api.SavedCell, await host("cells.read", {"id": id}))
-        return {key: value for key, value in cell.items() if key != "source"}
+        return albedo_api.Record((key, value) for key, value in cell.items() if key != "source")
+
+    async def list(self, limit: int = 20) -> albedo_api.ReadyList:
+        """This session's cells, newest first: id, status, parent, and first line.
+        Unlike output.list(), it reaches cells whose output has rolled out."""
+        cells = cast(list[dict[str, object]], await host("cells.list", {"limit": limit}))
+        return albedo_api.ReadyList(albedo_api.Record(cell) for cell in cells)
 
     async def trace(self, id: str) -> dict[str, object]:
         """What the cell read, ran and changed: activities plus diffs of files it wrote.
@@ -569,6 +579,10 @@ def _wire_encode(value: object, depth: int = 0) -> object:
         return {"__bytes__": base64.b64encode(bytes(value)).decode("ascii")}
     if isinstance(value, dict) and type(value) is dict:
         return {str(key): _wire_encode(item, depth + 1) for key, item in value.items()}
+    if isinstance(value, albedo_api.Record):
+        cls = type(value)
+        return {"__record__": cls.__module__ + "." + cls.__qualname__,
+                "fields": {str(key): _wire_encode(item, depth + 1) for key, item in value.items()}}
     if isinstance(value, (list, tuple)):
         try:
             if type(value) is not list:
@@ -861,6 +875,7 @@ def main():
     albedo_trace.install(CELL.get)
     modules = cast(list[str], json.loads(sys.argv[1]))
     api = albedo_api.PythonApi(version=2, loop=LOOP, host=host, HostError=WorkError,
+        forget_output=lambda id: ARCHIVES.pop(id, None) and None,
         capture=background_capture, preview=PREVIEW, send=send, on_shutdown=CLEANUP.append,
         background_handle=HANDLES.append, modules=modules, watch_output=watch_output)
     NAMESPACE.update(cells=Cells(), output=Output())
diff --git a/priv/python/albedo_plugins/bash.py b/priv/python/albedo_plugins/bash.py
index 0b325d2..4e63917 100644
--- a/priv/python/albedo_plugins/bash.py
+++ b/priv/python/albedo_plugins/bash.py
@@ -30,6 +30,7 @@ retained: OrderedDict[str, Job] = OrderedDict()  # finished handles, completion
 send: Send
 host: Callable[[dict[str, object]], Awaitable[object]] | None = None
 watch_output: Callable[[str, Callable[[], None]], None] | None = None
+forget_output: Callable[[str], None] | None = None
 
 ACTIVE_LIMIT = 64       # jobs owning running processes at once
 RETAINED_LIMIT = 64     # finished handles still addressable through `jobs`
@@ -263,6 +264,17 @@ def release(job: Job) -> None:
             del jobs[evicted.id]
 
 
+def forget(job: Job) -> None:
+    """Drop a finished job a plugin ran for its own purposes: it leaves `jobs`
+    and the retained output channels, so internal work never shows up beside,
+    or crowds out, the model's own jobs and cells."""
+    if jobs.get(job.id) is job:
+        del jobs[job.id]
+    retained.pop(job.id, None)
+    if forget_output is not None:
+        forget_output(job.id)
+
+
 def bash(command: object, *, timeout: float = 300) -> Job:
     """Start immediately; await the handle to wait.
 
@@ -301,9 +313,9 @@ async def close() -> None:
 
 def setup(api: PythonApi) -> dict[str, object]:
     global loop, capture_factory, preview_limit, jobs, active, retained, send
-    global host, watch_output
+    global host, watch_output, forget_output
     loop, capture_factory, preview_limit, send = api.loop, api.capture, api.preview, api.send
-    host, watch_output = api.host, api.watch_output
+    host, watch_output, forget_output = api.host, api.watch_output, api.forget_output
     jobs, active, retained = {}, {}, OrderedDict()
     api.background_handle(Job)
     api.on_shutdown(close)
diff --git a/priv/python/albedo_plugins/files.py b/priv/python/albedo_plugins/files.py
index 6e04d11..89ece18 100644
--- a/priv/python/albedo_plugins/files.py
+++ b/priv/python/albedo_plugins/files.py
@@ -18,7 +18,7 @@ from pathlib import Path
 from collections.abc import Awaitable, Callable, Generator
 from typing import Iterable, Sequence
 
-from albedo_api import PythonApi
+from albedo_api import PythonApi, ReadyList, Text
 from albedo_plugins import bash as jobs
 
 READ_LIMIT = 16_000
@@ -50,21 +50,7 @@ class Match:
         return fields
 
 
-def _ready(value):
-    """`await` on a value that is already here: returns it without suspending."""
-    yield from ()
-    return value
-
-
-class Text(str):
-    """A result that is ready now. Awaiting it is harmless, so `files.read(...)`
-    and `await files.read(...)` both work."""
-
-    def __await__(self) -> Generator[object, None, "Text"]:
-        return _ready(self)
-
-
-class Rows(list):
+class Rows(ReadyList):
     """A list that prints one row per line, so a REPL result reads like output.
     Like Text, it may be awaited or used directly."""
 
@@ -72,9 +58,6 @@ class Rows(list):
         super().__init__(items)
         self.truncated = truncated
 
-    def __await__(self) -> Generator[object, None, "Rows"]:
-        return _ready(self)
-
     def __repr__(self) -> str:
         if not self:
             return "[]"
@@ -123,8 +106,13 @@ class Search:
 async def _run(command: str) -> tuple[int | None, str]:
     """One supervised shell job, awaited to completion, with its bounded output."""
     job = jobs.bash(command, timeout=SEARCH_TIMEOUT)
-    await job
-    return job.exit_code, job.tail(jobs.preview_limit)
+    try:
+        await job
+        return job.exit_code, job.tail(jobs.preview_limit)
+    finally:
+        # The search is the plugin's own work, not a job the model started.
+        if job.exit_code is not None or job.timed_out:
+            jobs.forget(job)
 
 
 def _numbered(number: int, lines: Sequence[str]) -> str:
@@ -150,6 +138,10 @@ class Files:
             raise ValueError("start_line >= 1, end_line >= start_line, limit >= 1 line, "
                              "0 < max_chars <= 200000")
         target = Path(path).expanduser()
+        if not target.is_file():
+            if target.is_dir():
+                raise IsADirectoryError(f"{path} is a directory; files.ls({path!r}) lists it")
+            raise FileNotFoundError(_missing(target, path))
         lines = target.read_bytes().decode("utf-8", errors="replace").splitlines()
         if start_line > len(lines):
             return Text(f"[{path} has {len(lines)} lines; nothing at line {start_line}]")
@@ -248,7 +240,8 @@ class Files:
 
     def paths(self, pattern: str | None = None, path: str = ".", *,
               glob: str | None = None, max_results: int = 100, hidden: bool = False) -> Search:
-        """File names, not contents; the same ripgrep-or-Python split.
+        """File names, not contents; the same ripgrep-or-Python split. A pattern
+        with *, ? or [ is a glob over names; other text matches anywhere in the path.
         Await it: `await files.paths(pattern)`."""
         return Search("paths", lambda: self._paths(pattern, path, glob, max_results, hidden))
 
@@ -264,9 +257,8 @@ class Files:
             flags += ["-g", glob]
         command = " ".join(shlex.quote(part) for part in ["rg", *flags, str(target)])
         _, output = await _run(command)
-        matcher = re.compile(re.escape(pattern), re.IGNORECASE) if pattern else None
-        found = [line for line in output.splitlines()
-                 if line and (matcher is None or matcher.search(line))]
+        matches = _name_matcher(pattern)
+        found = [line for line in output.splitlines() if line and matches(line)]
         return Rows(found[:max_results], truncated=len(found) > max_results)
 
     def edit(self, path: str, old_str: str, new_str: str, line_hint: int | None = None) -> Text:
@@ -335,11 +327,22 @@ def _fallback_find(pattern, targets: list[Path], glob, context, max_results,
     return Rows(results)
 
 
+def _name_matcher(pattern: str | None) -> Callable[[str], bool]:
+    """A pattern with *, ? or [ is a glob over the file name (or the whole path
+    when it has a /); anything else matches as case-insensitive text anywhere
+    in the path, so "*.md" and "readme" both mean what they look like."""
+    if not pattern:
+        return lambda _path: True
+    folded = pattern.lower()
+    if any(char in pattern for char in "*?["):
+        return lambda path: fnmatch.fnmatch((path if "/" in pattern else Path(path).name).lower(), folded)
+    return lambda path: folded in path.lower()
+
+
 def _fallback_paths(pattern, path: Path, glob, max_results, hidden) -> Rows:
-    matcher = re.compile(re.escape(pattern), re.IGNORECASE) if pattern else None
+    matches = _name_matcher(pattern)
     found = [str(file) for file in _walk(path, hidden)
-             if (glob is None or fnmatch.fnmatch(file.name, glob))
-             and (matcher is None or matcher.search(str(file)))]
+             if (glob is None or fnmatch.fnmatch(file.name, glob)) and matches(str(file))]
     return Rows(found[:max_results], truncated=len(found) > max_results)
 
 
diff --git a/priv/python/albedo_plugins/remote.py b/priv/python/albedo_plugins/remote.py
index b9dfd98..a0691f7 100644
--- a/priv/python/albedo_plugins/remote.py
+++ b/priv/python/albedo_plugins/remote.py
@@ -29,7 +29,7 @@ import uuid
 from pathlib import Path
 from typing import Any
 
-from albedo_api import PythonApi
+from albedo_api import PythonApi, ReadyList
 
 CONNECT_TIMEOUT = 15  # seconds before an unreachable target gives up
 HANDSHAKE_TIMEOUT = 30  # seconds for the remote kernel to say ready
@@ -230,6 +230,12 @@ def wire_decode(value: Any) -> Any:
                     **{key: wire_decode(item) for key, item in value["fields"].items()})
             except Exception:
                 return {key: wire_decode(item) for key, item in value["fields"].items()}
+        if set(value) == {"__record__", "fields"} and isinstance(value.get("__record__"), str):
+            fields = {key: wire_decode(item) for key, item in value["fields"].items()}
+            try:
+                return _load_class(value["__record__"])(fields)
+            except Exception:
+                return fields
         if set(value) == {"__list__", "items"} and isinstance(value.get("__list__"), str):
             items = [wire_decode(item) for item in value["items"]]
             try:
@@ -1029,9 +1035,9 @@ class Remote:
         return connection
 
     @staticmethod
-    def connections() -> tuple[RemoteConnection, ...]:
+    def connections() -> ReadyList:
         """Every open connection, kernel or degraded."""
-        return tuple(connections)
+        return ReadyList(connections)
 
     @staticmethod
     async def close_all() -> None:
diff --git a/priv/python/albedo_plugins/skills.py b/priv/python/albedo_plugins/skills.py
index 8ffa2fb..6ccfb5d 100644
--- a/priv/python/albedo_plugins/skills.py
+++ b/priv/python/albedo_plugins/skills.py
@@ -1,15 +1,17 @@
-from typing import TypedDict, cast
+from typing import cast
 
-from albedo_api import Host, PythonApi
+from albedo_api import Host, PythonApi, Record
 
 
-class SkillResources(TypedDict):
+class SkillResources(Record):
+    """resources.resources and resources["resources"] both work."""
     resources: list[str]
     truncated: bool
     diagnostics: list[str]
 
 
-class SkillPage(TypedDict):
+class SkillPage(Record):
+    """page.content and page["content"] both work."""
     path: str
     encoding: str
     content: str
@@ -24,7 +26,7 @@ host: Host
 class Skills:
     async def resources(self, name: str) -> SkillResources:
         """List resource names without reading or executing their contents."""
-        return cast(SkillResources, await host("skills.resources", {"name": name}))
+        return SkillResources(cast(dict, await host("skills.resources", {"name": name})))
 
     async def read(
         self,
@@ -35,8 +37,8 @@ class Skills:
         limit: int = 16_384,
     ) -> SkillPage:
         """Read one bounded page from a cataloged skill resource."""
-        return cast(
-            SkillPage,
+        return SkillPage(cast(
+            dict,
             await host(
                 "skills.read",
                 {
@@ -46,7 +48,7 @@ class Skills:
                     "limit": limit,
                 },
             ),
-        )
+        ))
 
 
 def setup(api: PythonApi) -> dict[str, object]:
diff --git a/priv/python/albedo_plugins/work.py b/priv/python/albedo_plugins/work.py
index bbc3b46..eabd2fb 100644
--- a/priv/python/albedo_plugins/work.py
+++ b/priv/python/albedo_plugins/work.py
@@ -1,10 +1,11 @@
 from __future__ import annotations
 
-from albedo_api import Host, PythonApi
-from typing import TypedDict, cast
+from albedo_api import Host, PythonApi, Record
+from typing import cast
 
 
-class WorkItem(TypedDict):
+class WorkItem(Record):
+    """One ledger item: item.id and item["id"] both work."""
     id: int
     title: str
     notes: str
@@ -19,23 +20,23 @@ host: Host
 
 class Work:
     async def list(self, after: int = 0, limit: int = 50) -> list[WorkItem]:
-        return cast(list[WorkItem], await host("work.list", {"after": after, "limit": limit}))
+        return [WorkItem(item) for item in cast(list[dict], await host("work.list", {"after": after, "limit": limit}))]
 
     async def get(self, id: int) -> WorkItem:
-        return cast(WorkItem, await host("work.get", {"id": id}))
+        return WorkItem(cast(dict, await host("work.get", {"id": id})))
 
     async def create(self, title: str, notes: str = "", parent: int | None = None) -> WorkItem:
-        return cast(WorkItem, await host("work.create", {"title": title, "notes": notes, "parent": parent}))
+        return WorkItem(cast(dict, await host("work.create", {"title": title, "notes": notes, "parent": parent})))
 
     async def update(self, id: int, *, revision: int, **changes: object) -> WorkItem:
         allowed = {"title", "notes", "status", "session", "run"}
         if changes.keys() - allowed:
             raise ValueError("unknown work fields: " + repr(changes.keys() - allowed))
-        return cast(WorkItem, await host("work.update", {"id": id, "revision": revision, **changes}))
+        return WorkItem(cast(dict, await host("work.update", {"id": id, "revision": revision, **changes})))
 
     async def delete(self, id: int, *, revision: int) -> WorkItem:
         """Remove an item at the revision you last read; one with sub-items stays."""
-        return cast(WorkItem, await host("work.delete", {"id": id, "revision": revision}))
+        return WorkItem(cast(dict, await host("work.delete", {"id": id, "revision": revision})))
 
 
 def setup(api: PythonApi) -> dict[str, object]:
diff --git a/src/albedo/harness/extensions/bash/extension.gleam b/src/albedo/harness/extensions/bash/extension.gleam
index 07079cb..2ad127c 100644
--- a/src/albedo/harness/extensions/bash/extension.gleam
+++ b/src/albedo/harness/extensions/bash/extension.gleam
@@ -5,7 +5,7 @@ import albedo/harness/extension as harness_extension
 import gleam/dynamic/decode
 import gleam/json
 
-const instructions = "bash(command, timeout=300) starts a background job (timeout in seconds) and returns its handle right away. Await the handle for completion, or keep it in a variable. jobs is a dict of job id to handle for this session. A job's output lives on its handle (job.tail(n=4000) for the last n characters; job.command is what ran) and stays readable in full with output.read(job.id, offset=0, limit=4000); await job.stop() ends it. job.exit_code, job.duration, and job.timed_out report the exit status, wall seconds, and deadline state once the command ends (job.poll() and job.returncode are subprocess-style spellings of job.exit_code). A job that finishes with its result unread wakes the session by itself, so polling a handle is optional: leave it in a variable, end the cell, and a notice naming the job arrives when it lands. Awaiting the job, reading its result, or stopping it first means no wake."
+const instructions = "bash(command, timeout=300) starts a background job (timeout in seconds) and returns its handle right away. Await the handle for completion, or keep it in a variable. jobs is a dict of job id to handle for this session. A job's output lives on its handle (job.tail(n=4000) for the last n characters; job.command is what ran) and stays readable in full with output.read(job.id, offset=0, limit=4000); await job.stop() ends it. job.exit_code, job.duration, and job.timed_out report the exit status, wall seconds, and deadline state once the command ends (job.poll() and job.returncode are subprocess-style spellings of job.exit_code). A job that finishes with its result unread wakes the session by itself, so polling a handle is optional: leave it in a variable, end the cell, and a notice naming the job arrives once the session is idle. A job that finishes while you are still working is reported after your turn ends, unless you read it first. Awaiting the job, reading its result, or stopping it first means no wake."
 
 pub fn extension() -> harness_extension.Extension {
   harness_extension.Extension(
diff --git a/src/albedo/harness/extensions/files/extension.gleam b/src/albedo/harness/extensions/files/extension.gleam
index 946389a..023bb06 100644
--- a/src/albedo/harness/extensions/files/extension.gleam
+++ b/src/albedo/harness/extensions/files/extension.gleam
@@ -5,7 +5,7 @@ pub fn extension() -> extension.Extension {
     "files",
     "Read, search, and exactly edit workspace files from Python.",
     "files",
-    "files.read(path, start_line=, end_line=, limit=) returns complete numbered lines; limit counts lines, and a separate max_chars= budget (default 16000 characters) stops very large windows and says where to resume, files.ls(path, pattern=) lists one directory, files.write(path, content) creates or replaces a whole file, and files.edit(path, old, new) replaces one exact unique string, reporting every occurrence with line ranges when it is not unique; retry it with line_hint=<line>. These return immediately, and awaiting them also works. await files.find(pattern, path_or_paths, glob=, context=N, literal=, case_sensitive=, max_results=50) searches contents (a regex unless literal=True) with N surrounding lines and await files.paths(pattern) searches names; both run a background search and must be awaited. Prefer these over shell equivalents, and start any other command with bash(command): never use subprocess, os.system, or os.popen, which spawn processes this session cannot supervise.",
+    "files.read(path, start_line=, end_line=, limit=) returns complete numbered lines; limit counts lines, and a separate max_chars= budget (default 16000 characters) stops very large windows and says where to resume, files.ls(path, pattern=) lists one directory, files.write(path, content) creates or replaces a whole file, and files.edit(path, old, new) replaces one exact unique string, reporting every occurrence with line ranges when it is not unique; retry it with line_hint=<line>. These return immediately, and awaiting them also works. await files.find(pattern, path_or_paths, glob=, context=N, literal=, case_sensitive=, max_results=50) searches contents (a regex unless literal=True) with N surrounding lines and await files.paths(pattern) searches names; both run a background search and must be awaited. Prefer these over shell equivalents, and start any other command with bash(command). subprocess, os.system, and os.popen are not blocked, but what they start is unsupervised: it is not stopped with the session or a timeout, keeps no retained output, and never wakes you, so use bash instead.",
     ["python", "bash"],
   )
 }
diff --git a/src/albedo/harness/extensions/python/cells.gleam b/src/albedo/harness/extensions/python/cells.gleam
index f636aa3..748cfe1 100644
--- a/src/albedo/harness/extensions/python/cells.gleam
+++ b/src/albedo/harness/extensions/python/cells.gleam
@@ -136,6 +136,26 @@ pub fn get(storage: store.Store, id: String) -> Result(Cell, String) {
   }
 }
 
+/// This session's cells, newest first, at most `limit` of them.
+pub fn recent(
+  storage: store.Store,
+  session: String,
+  limit: Int,
+) -> Result(List(Cell), String) {
+  use ids <- result.try(
+    store.query(storage, fn(db) {
+      sqlight.query(
+        "SELECT id FROM cells WHERE session=? ORDER BY rowid DESC LIMIT ?",
+        db,
+        [sqlight.text(session), sqlight.int(limit)],
+        decode.field(0, decode.string, decode.success),
+      )
+      |> result.map_error(fn(e) { e.message })
+    }),
+  )
+  list.try_map(ids, get(storage, _))
+}
+
 @external(erlang, "albedo_native", "new_id")
 fn new_id() -> String
 
diff --git a/src/albedo/harness/extensions/python/extension.gleam b/src/albedo/harness/extensions/python/extension.gleam
index 231ca62..9d2263e 100644
--- a/src/albedo/harness/extensions/python/extension.gleam
+++ b/src/albedo/harness/extensions/python/extension.gleam
@@ -14,7 +14,7 @@ import gleam/result
 pub fn definition() -> types.Tool {
   types.Tool(
     "python",
-    "Execute Python in your persistent session. Always provide both JSON arguments: code (string) and timeout_ms (integer, 1–3600000), even for a one-line call. Top-level await works. Variables stay alive. Only printed output and the final expression return; truncated=true means the output passed the 64 KiB preview, and the rest is readable with output.read(cell_id, offset=65536). Failed cells are retained: await cells.read(id) for bounded source, await cells.info(id) for status without source, await cells.trace(id) for what the cell read, ran and changed. Repair exact unique text without resending the source with await cells.run(id, replacements=[(old,new)]), or pass check=True first to compile the rewrite and get the syntax error back without running anything. A cell that started requires allow_partial=True after checking side effects; never retry blindly.",
+    "Execute Python in your persistent session. Always provide both JSON arguments: code (string) and timeout_ms (integer, 1–3600000), even for a one-line call. Top-level await works. Variables stay alive. Only printed output and the final expression return; truncated=true means the output passed the 64 KiB preview, and the rest (up to 1 MiB) is readable with output.read(cell_id, offset=65536) while the cell is among the 16 most recent; read it before running many more cells, or print less. Failed cells are retained: await cells.read(id) for bounded source, await cells.info(id) for its status (ok, error, interrupted, started, saved) without source, await cells.trace(id) for what the cell read, ran and changed. Repair exact unique text without resending the source with await cells.run(id, replacements=[(old,new)]), or pass check=True first to compile the rewrite and get the syntax error back without running anything. A cell that started requires allow_partial=True after checking side effects; never retry blindly.",
     json.object([
       #("type", json.string("object")),
       #("additionalProperties", json.bool(False)),
@@ -86,7 +86,7 @@ pub fn extension() -> extension.Extension {
     [],
     [
       extension.ToolPlugin(
-        "Python has a persistent namespace, top-level await, cells.read/info/trace and cells.run for saved-source repair (all async), and output.read/output.list for bounded retained output (synchronous). cells.last_id is the id of the latest cell. output.read(id, offset=0, limit=4000) returns up to limit characters. output.list() names every retained channel, which is also how to find earlier cells: cells, background jobs, and 'native' for bytes written to fd 1/2 while no cell was running.",
+        "Python has a persistent namespace, top-level await, cells.read/info/trace and cells.run for saved-source repair (all async), and output.read/output.list for bounded retained output (synchronous; awaiting them also works). cells.last_id is the id of the latest cell, and await cells.list(limit=20) lists this session's cells newest first with their status and first line, even after their output has rolled out. output.read(id, offset=0, limit=4000) returns up to limit characters. output.list() names every retained channel, which is also how to find earlier cells: cells, background jobs, and 'native' for bytes written to fd 1/2 while no cell was running.",
         [extension.Tool(definition(), invoke, recover)],
         [],
         [#("cells", cells.handle)],
diff --git a/src/albedo/harness/extensions/python/rpc.gleam b/src/albedo/harness/extensions/python/rpc.gleam
index c61b7b1..69b24f6 100644
--- a/src/albedo/harness/extensions/python/rpc.gleam
+++ b/src/albedo/harness/extensions/python/rpc.gleam
@@ -3,8 +3,10 @@ import albedo/harness/extensions/python/kernel as python
 import albedo/harness/extensions/work/ledger as work
 import gleam/dynamic/decode
 import gleam/json
-import gleam/option.{None}
+import gleam/list
+import gleam/option.{None, Some}
 import gleam/result
+import gleam/string
 
 pub fn handle(store: work.Store, session: String, request: String) -> String {
   let decoder = {
@@ -13,6 +15,21 @@ pub fn handle(store: work.Store, session: String, request: String) -> String {
     decode.success(#(method, args))
   }
   case json.parse(request, decoder) {
+    Ok(#("cells.list", args)) -> {
+      let limit =
+        decode.run(
+          args,
+          decode.optional_field("limit", 20, decode.int, decode.success),
+        )
+        |> result.unwrap(20)
+      case limit < 1 || limit > 200 {
+        True -> Error("1 <= limit <= 200 required")
+        False ->
+          journal.recent(store, session, limit)
+          |> result.map(json.array(_, summary_json))
+      }
+      |> answer
+    }
     Ok(#(method, args))
       if method == "cells.read"
       || method == "cells.trace"
@@ -21,23 +38,65 @@ pub fn handle(store: work.Store, session: String, request: String) -> String {
       || method == "cells.started"
       || method == "cells.finish"
     -> {
-      let answer = cells(store, session, method, args)
-      case answer {
-        Ok(value) -> json.object([#("ok", json.bool(True)), #("value", value)])
-        Error(message) ->
-          json.object([
-            #("ok", json.bool(False)),
-            #("code", json.string("cell")),
-            #("message", json.string(message)),
-          ])
-      }
-      |> json.to_string
+      cells(store, session, method, args) |> answer
     }
     _ ->
       "{\"ok\":false,\"code\":\"cell\",\"message\":\"unknown cells operation\"}"
   }
 }
 
+fn answer(result: Result(json.Json, String)) -> String {
+  case result {
+    Ok(value) -> json.object([#("ok", json.bool(True)), #("value", value)])
+    Error(message) ->
+      json.object([
+        #("ok", json.bool(False)),
+        #("code", json.string("cell")),
+        #("message", json.string(message)),
+      ])
+  }
+  |> json.to_string
+}
+
+/// A cell without its source: what `cells.info` and `cells.list` show.
+fn summary_json(cell: journal.Cell) -> json.Json {
+  json.object([
+    #("id", json.string(cell.id)),
+    #("status", json.string(status(cell))),
+    #("started", json.bool(cell.started)),
+    #("parent", json.nullable(cell.parent, json.string)),
+    #("finished", json.bool(cell.outcome != None)),
+    #(
+      "first_line",
+      json.string(
+        cell.source
+        |> string.split("\n")
+        |> list.find(fn(line) { string.trim(line) != "" })
+        |> result.unwrap("")
+        |> string.slice(0, 120),
+      ),
+    ),
+  ])
+}
+
+/// ok, error, or interrupted once finished; lost or unavailable when the
+/// kernel could not run it; started when it began without a recorded end
+/// (effects unknown); saved when it never started.
+fn status(cell: journal.Cell) -> String {
+  case cell.outcome, cell.started {
+    Some(Ok(outcome)), _ ->
+      case outcome.status {
+        python.Succeeded -> "ok"
+        python.Failed -> "error"
+        python.Interrupted -> "interrupted"
+      }
+    Some(Error(python.Lost)), _ -> "lost"
+    Some(Error(_)), _ -> "unavailable"
+    None, True -> "started"
+    None, False -> "saved"
+  }
+}
+
 fn parse(args, decoder) {
   decode.run(args, decoder) |> result.replace_error("invalid cell arguments")
 }
@@ -106,5 +165,6 @@ fn to_json(cell: journal.Cell) -> json.Json {
     #("started", json.bool(cell.started)),
     #("parent", json.nullable(cell.parent, json.string)),
     #("finished", json.bool(cell.outcome != None)),
+    #("status", json.string(status(cell))),
   ])
 }
diff --git a/src/albedo/harness/extensions/remote/extension.gleam b/src/albedo/harness/extensions/remote/extension.gleam
index e3f80b8..12aa297 100644
--- a/src/albedo/harness/extensions/remote/extension.gleam
+++ b/src/albedo/harness/extensions/remote/extension.gleam
@@ -31,7 +31,7 @@ pub fn extension() -> harness_extension.Extension {
       <> " still needs await, because the value itself crosses the network; handles and"
       <> " their state never do. await rem.tools() lists what the remote namespace holds;"
       <> " await rem.read(path) and await rem.write(path, content) move one file's text"
-      <> " directly. remote.connections() lists open connections and await"
+      <> " directly. remote.connections() lists open connections (awaiting it also works) and await"
       <> " remote.close_all() ends them all. Failures raise RemoteError."
       <> " The target resolves per call from a host="
       <> " argument, then remote.configure(host, remote_cwd=None), then $ALBEDO_SSH"
diff --git a/src/albedo/harness/extensions/skills/extension.gleam b/src/albedo/harness/extensions/skills/extension.gleam
index 0277009..ae1b231 100644
--- a/src/albedo/harness/extensions/skills/extension.gleam
+++ b/src/albedo/harness/extensions/skills/extension.gleam
@@ -44,7 +44,7 @@ pub fn extension_at(home: String) -> harness_extension.Extension {
   )
 }
 
-const instructions = "Agent Skills are cataloged session commands. Invoke one through the `commands` object (commands.catalog() maps slash names to methods): it returns the skill's instructions as data and never submits a turn or executes bundled scripts. `await skills.resources(name)` returns a dict whose \"resources\" lists bundled resource names, and `await skills.read(name, resource=\"SKILL.md\", offset=0, limit=16384)` reads one bounded page as a dict: page[\"content\"], page[\"next_offset\"], page[\"truncated\"]. Failures raise SkillsError. A user may explicitly run the listed slash command, which submits the activation as one user turn."
+const instructions = "Agent Skills are cataloged session commands. Invoke one through the `commands` object (commands.catalog() maps slash names to methods): it returns the skill's instructions as data and never submits a turn or executes bundled scripts. `await skills.resources(name)` returns a record whose .resources lists bundled resource names, and `await skills.read(name, resource=\"SKILL.md\", offset=0, limit=16384)` reads one bounded page as a record: page.content, page.next_offset, and page.truncated (page[\"content\"] works too). Failures raise SkillsError. A user may explicitly run the listed slash command, which submits the activation as one user turn."
 
 /// One cataloged skill as a session command: both callers resolve the same
 /// activation, and only the delivery differs.
diff --git a/src/albedo/harness/extensions/work/extension.gleam b/src/albedo/harness/extensions/work/extension.gleam
index cb50a65..fdb6ca3 100644
--- a/src/albedo/harness/extensions/work/extension.gleam
+++ b/src/albedo/harness/extensions/work/extension.gleam
@@ -10,7 +10,7 @@ pub fn extension() -> harness_extension.Extension {
     ["python"],
     [
       harness_extension.ToolPlugin(
-        "Humans and agents share this revision-checked work ledger; every call is async. await work.list(after=0, limit=50), work.get(id), work.create(title, notes=\"\", parent=None), work.update(id, revision=, title=, notes=, status=), and work.delete(id, revision=) return plain dicts (item[\"id\"], item[\"revision\"], item[\"status\"]). Status is open, active, blocked, done, or cancelled. Pass the revision you last read; a stale one raises WorkError, so get the item again. Keep execution status separate from work status. People manage the same ledger with /work, and their changes reach you as notes.",
+        "Humans and agents share this revision-checked work ledger; every call is async. await work.list(after=0, limit=50), work.get(id), work.create(title, notes=\"\", parent=None), work.update(id, revision=, title=, notes=, status=), and work.delete(id, revision=) return item records: item.id and item[\"id\"] both work, and so do revision, status, title, notes, and parent. Status is open, active, blocked, done, or cancelled. Pass the revision you last read; a stale one raises WorkError, so get the item again. Keep execution status separate from work status. People manage the same ledger with /work, and their changes reach you as notes.",
         [],
         ["work"],
         [#("work", fn(store, _, request) { rpc.handle(store, request) })],
diff --git a/src/albedo/harness/extensions/work/ledger.gleam b/src/albedo/harness/extensions/work/ledger.gleam
index 568c67f..be1f97f 100644
--- a/src/albedo/harness/extensions/work/ledger.gleam
+++ b/src/albedo/harness/extensions/work/ledger.gleam
@@ -99,7 +99,12 @@ pub fn parse_status(name: String) -> Result(Status, Error) {
     "blocked" -> Ok(Blocked)
     "done" -> Ok(Done)
     "cancelled" -> Ok(Cancelled)
-    _ -> Error(Invalid("unknown work status"))
+    _ ->
+      Error(Invalid(
+        "unknown work status "
+        <> name
+        <> "; use open, active, blocked, done, or cancelled",
+      ))
   }
 }
 
diff --git a/test/harness/files_plugin_test.py b/test/harness/files_plugin_test.py
index adfe940..da349e0 100644
--- a/test/harness/files_plugin_test.py
+++ b/test/harness/files_plugin_test.py
@@ -16,7 +16,7 @@ class FakeJob:
     """What `bash(command)` returns: an awaitable handle owning its output."""
 
     def __init__(self, output, exit_code=0):
-        self.output, self.exit_code = output, exit_code
+        self.output, self.exit_code, self.timed_out = output, exit_code, False
 
     def __await__(self):
         async def settled():
@@ -133,11 +133,15 @@ class FilesPluginTest(unittest.TestCase):
             commands.append(command)
             return FakeJob(matches)
 
+        forgotten = []
         with patch.object(plugin.jobs, "bash", fake_bash), \
+             patch.object(plugin.jobs, "forget", forgotten.append), \
              patch.object(plugin.jobs, "preview_limit", 65_536, create=True), \
              patch.object(plugin, "_which", lambda name: "/usr/bin/rg"):
             found = self.loop.run_until_complete(
                 plugin.Files().find("needle", str(self.root), glob="*.py"))
+        # The search job is the plugin's own; it must not linger in jobs or output.list().
+        self.assertEqual(len(forgotten), 1)
         self.assertEqual([match.to_dict() for match in found],
                          [{"path": "a.py", "line": 7, "text": "needle here"}])
         self.assertIn("rg --json -g '*.py' -e needle", commands[0])
@@ -206,6 +210,23 @@ class FilesPluginTest(unittest.TestCase):
         with self.assertRaises(ValueError):
             files.find("x", context=51)
 
+    # An agent searched paths("*.md"), got [], and concluded nothing matched;
+    # reading a missing file raised a bare FileNotFoundError.
+    def test_path_globs_and_missing_file_diagnostics(self):
+        self.write("notes.md", "x")
+        self.write("sub/README.md", "y")
+        self.write("a.py", "z")
+        files = plugin.Files()
+        with patch.object(plugin, "_which", lambda name: None):
+            globbed = self.loop.run_until_complete(files.paths("*.md", str(self.root)))
+            text = self.loop.run_until_complete(files.paths("readme", str(self.root)))
+        self.assertEqual(sorted(Path(name).name for name in globbed), ["README.md", "notes.md"])
+        self.assertEqual([Path(name).name for name in text], ["README.md"])
+        with self.assertRaisesRegex(FileNotFoundError, "nearby paths: .*notes.md"):
+            files.read(str(self.root/"note.md"))
+        with self.assertRaisesRegex(IsADirectoryError, r"files\.ls"):
+            files.read(str(self.root/"sub"))
+
 
 if __name__ == "__main__":
     unittest.main()
diff --git a/test/harness/kernel_api_test.gleam b/test/harness/kernel_api_test.gleam
new file mode 100644
index 0000000..9e0beb9
--- /dev/null
+++ b/test/harness/kernel_api_test.gleam
@@ -0,0 +1,61 @@
+//// The model-facing Python API in a live kernel, where a fake cannot hide drift.
+
+import albedo/harness/extensions/bash/extension as bash
+import albedo/harness/extensions/files/extension as files
+import albedo/harness/extensions/python/extension as python
+import albedo/harness/extensions/work/extension as work
+import albedo/harness/runtime
+import gleeunit/should
+
+pub fn a_search_leaves_no_job_or_output_channel_test() {
+  let assert Ok(host) =
+    runtime.start_with_extensions(":memory:", [
+      python.extension(),
+      bash.extension(),
+      files.extension(),
+    ])
+  let assert Ok(session) = runtime.open_session(host, "search", "/tmp")
+  let assert Ok(execution) =
+    runtime.execute(
+      host,
+      session,
+      "import os, tempfile\n"
+        <> "root = tempfile.mkdtemp()\n"
+        <> "open(os.path.join(root, 'a.txt'), 'w').write('needle\\n')\n"
+        <> "found = await files.find('needle', root)\n"
+        <> "named = await files.paths('a.txt', root)\n"
+        <> "(len(found), len(named), len(jobs), [c['kind'] for c in output.list() if c['kind'] == 'job'])",
+      30_000,
+    )
+  let assert Ok(outcome) = execution.result
+  outcome.value |> should.equal("(1, 1, 0, [])")
+  runtime.stop(host)
+}
+
+// Cells are listable after their output rolls out, report a status, and the
+// ledger answers records; synchronous output calls may also be awaited.
+pub fn cells_records_and_awaitable_output_test() {
+  let assert Ok(host) =
+    runtime.start_with_extensions(":memory:", [
+      python.extension(),
+      work.extension(),
+    ])
+  let assert Ok(session) = runtime.open_session(host, "api", "/tmp")
+  let assert Ok(_) = runtime.execute(host, session, "1 / 0", 5000)
+  let assert Ok(execution) =
+    runtime.execute(
+      host,
+      session,
+      "cells_now = await cells.list()\n"
+        <> "failed = await cells.info(cells_now[1].id)\n"
+        <> "item = await work.create('probe', 'from a test')\n"
+        <> "listing = await output.list()\n"
+        <> "(len(cells_now), cells_now[1].first_line, failed.status, item.title == item['title'], "
+        <> "type(listing).__name__, await work.delete(item.id, revision=item.revision) == item)",
+      10_000,
+    )
+  let assert Ok(outcome) = execution.result
+  outcome.value
+  |> should.equal("(2, '1 / 0', 'error', True, 'ReadyList', True)")
+  runtime.stop(host)
+}
diff --git a/test/harness/work_page_test.gleam b/test/harness/work_page_test.gleam
index f0ed9b3..2899fa2 100644
--- a/test/harness/work_page_test.gleam
+++ b/test/harness/work_page_test.gleam
@@ -107,7 +107,9 @@ pub fn failed_changes_and_model_calls_queue_nothing_test() {
     command.UserCall,
     dict.from_list([#("action", "status"), #("details", "1 finished")]),
   )
-  |> should.equal(Error("unknown work status"))
+  |> should.equal(Error(
+    "unknown work status finished; use open, active, blocked, done, or cancelled",
+  ))
   process.receive(notes, 0) |> should.equal(Error(Nil))
 }
 
-- 
2.54.0