okami.mom/albedo

git clone https://git.nkp.pet/okami.mom/albedo

nekomimi.petfix kernel output truncation with automatic disk-backed pagination1f92415

main
branchesmainall refs →
3.7 KiB123 linesraw
1"""Every public call in the model's Python namespace is named in the text the
2model is given. Agents otherwise discover the API with dir() and help(), or
3miss parts of it entirely (files.write, cells.last_id, and bash's timeout were
4all real but undocumented)."""
5
6import asyncio
7import importlib
8import inspect
9import re
10from pathlib import Path
11import sys
12import unittest
13
14ROOT = Path(__file__).resolve().parents[2]
15sys.path.insert(0, str(ROOT / "priv" / "python"))
16import albedo_cells as kernel  # noqa: E402
17from albedo_api import PythonApi  # noqa: E402
18
19# Plumbing the kernel or the remote relay calls; not part of the model's API.
20INTERNAL = {"run.Job.claim"}
21
22
23def model_text() -> str:
24    """Every string literal the harness can show the model, joined."""
25    sources = [
26        *ROOT.glob("src/albedo/harness/extensions/*/extension.gleam"),
27        ROOT / "src/albedo/harness/command.gleam",
28    ]
29    literals = []
30    for source in sources:
31        literals += re.findall(r'"((?:[^"\\]|\\.)*)"', source.read_text())
32    return "".join(literals).replace('\\"', '"')
33
34
35def namespace() -> dict[str, object]:
36    loop = asyncio.new_event_loop()
37
38    async def host(method, args):
39        return [] if method in {"commands.list", "mcp.list"} else {}
40
41    api = PythonApi(
42        loop,
43        host,
44        RuntimeError,
45        kernel.background_capture,
46        100,
47        lambda e: None,
48        lambda c: None,
49        lambda _: None,
50    )
51    bindings = {
52        "cells": kernel.Cells(),
53        "output": kernel.Output(),
54        "show_image": kernel.show_image,
55    }
56    for name in [
57        "run",
58        "work",
59        "paperclips",
60        "files",
61        "skills",
62        "commands",
63        "mcp",
64        "remote",
65        "view",
66        "browser",
67    ]:
68        result = importlib.import_module("albedo_plugins." + name).setup(api)
69        bindings.update(
70            loop.run_until_complete(result) if inspect.isawaitable(result) else result
71        )
72    loop.close()
73    return bindings
74
75
76class ApiDocsTest(unittest.TestCase):
77    def test_every_public_call_is_named_for_the_model(self):
78        text = model_text()
79        missing = []
80        for name, value in namespace().items():
81            if isinstance(value, (dict, type)):
82                if name not in text:
83                    missing.append(name)
84                continue
85            if inspect.isfunction(value):
86                if f"{name}(" not in text:
87                    missing.append(f"{name}(")
88                continue
89            for attribute in dir(type(value)) + list(vars(value)):
90                if (
91                    attribute.startswith("_")
92                    or f"{type(value).__module__.split('.')[-1]}.{type(value).__name__}.{attribute}"
93                    in INTERNAL
94                ):
95                    continue
96                if not re.search(
97                    rf"\b{re.escape(name)}\.{re.escape(attribute)}\b", text
98                ):
99                    missing.append(f"{name}.{attribute}")
100        self.assertEqual(
101            sorted(set(missing)), [], "public API the model is never told about"
102        )
103
104    def test_job_handles_are_documented(self):
105        from albedo_plugins import run
106
107        text = model_text()
108        public = [
109            name
110            for name, value in inspect.getmembers(run.Job, callable)
111            if not name.startswith("_") and f"run.Job.{name}" not in INTERNAL
112        ]
113        fields = ["id", "command", "exit_code", "duration", "timed_out"]
114        missing = [
115            name for name in public + fields if not re.search(rf"\bjob\.{name}\b", text)
116        ]
117        self.assertEqual(
118            missing, [], "job handle members the model is never told about"
119        )
120
121
122if __name__ == "__main__":
123    unittest.main()