okami.mom/albedo

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

nekomimi.petkeep tool results beside their calls: settle out-of-order results on commit, in live history, and on loadc0d0428

main
51.2 KiB1175 linesraw
1"""Extension catalog, commands, live reload, and compaction through a real daemon."""
2
3import json
4import sqlite3
5from pathlib import Path
6import threading
7import time
8import unittest
9import urllib.error
10import urllib.parse
11
12from harness import Albedo, Provider, exclusive, operation_id, python, text
13
14
15SKILL = "---\nname: demo\ndescription: catalog-only fixture description\n---\nBODY_MUST_NOT_AUTOLOAD\n"
16
17
18def scripted(request):
19    messages = request["input"]
20    prompt = next(
21        (
22            str(item.get("content", ""))
23            for item in reversed(messages)
24            if item.get("role") == "user"
25        ),
26        "",
27    )
28    if "<newly-evicted-history>" in prompt:
29        return text("older conversation summary")
30    if messages[-1].get("type") == "function_call_output":
31        return text("done")
32    if "model probe via python" in prompt:
33        return python(
34            "selection = await commands.model()\n"
35            "assert selection['model'] == 'fixture', selection\n"
36            "try:\n"
37            "    await commands.model('unreachable-model')\n"
38            "    assert False, 'switch must refuse from the model'\n"
39            "except CommandsError as error:\n"
40            "    assert 'user action' in str(error), error\n"
41            "print('MODEL_COMMAND_OK')"
42        )
43    if "activate demo via python" in prompt:
44        return python(
45            "activation = await commands.demo('python argument')\n"
46            "assert 'BODY_MUST_NOT_AUTOLOAD' in activation['instructions']\n"
47            "assert activation['arguments'] == 'python argument'\n"
48            "print('SKILL_PYTHON_ACTIVATION_OK')"
49        )
50    if "hot probe before reload" in prompt:
51        return python(
52            "names = [c['name'] for c in await commands.catalog()]\n"
53            "assert '/late' not in names, names\n"
54            "hot_marker = 'kernel-kept-running'\nprint('BEFORE_RELOAD_OK')"
55        )
56    if "hot probe after reload" in prompt:
57        return python(
58            "names = [c['name'] for c in await commands.catalog()]\n"
59            "assert '/late' in names, names\n"
60            "assert hot_marker == 'kernel-kept-running', 'kernel was restarted'\n"
61            "print('AFTER_RELOAD_OK')"
62        )
63    if "hold a cell across a crash" in prompt:
64        return python(
65            "import asyncio\nfrom pathlib import Path\n"
66            "Path('held-cell-started').write_text('')\n"
67            "await asyncio.sleep(30)"
68        )
69    if "lcm tool probe" in prompt:
70        return python("first turn", tool_name="lcm_grep")
71    if "fold list probe" in prompt:
72        return python("", tool_name="lcm_list")
73    return text("older conversation summary")
74
75
76def assert_result_follows_call(test, request):
77    """The call's result comes right after it, and the capability note saved
78    while the call ran comes after that."""
79    items = request["input"]
80    (call,) = [i for i, item in enumerate(items) if item.get("type") == "function_call"]
81    following = items[call + 1]
82    test.assertEqual(
83        (following.get("type"), following.get("call_id")),
84        ("function_call_output", items[call]["call_id"]),
85    )
86    (note,) = [
87        i
88        for i, item in enumerate(items)
89        if "capabilities changed" in str(item.get("content", ""))
90    ]
91    test.assertGreater(note, call + 1)
92
93
94def capability_notes(request):
95    """The capability notes a request carries as inputs of their own; the
96    rolling recap may quote the newest one too."""
97    return [
98        item["content"]
99        for item in request["input"]
100        if item.get("role") == "user"
101        and str(item.get("content", "")).startswith(
102            '<system-note origin="capabilities changed">'
103        )
104    ]
105
106
107class ExtensionTests(unittest.TestCase):
108    def setUp(self):
109        self.provider = Provider(scripted)
110        self.addCleanup(self.provider.close)
111        self.app = Albedo(
112            self.provider,
113            protocol="responses",
114            providers={
115                f"fixture-{self.provider.route}": {
116                    "extension": "openai",
117                    "baseUrl": self.provider.url,
118                    "apiKey": "fixture-key",
119                    "model": "fixture",
120                    "protocol": "responses",
121                }
122            },
123        )
124        self.app.__enter__()
125        self.addCleanup(self.app.__exit__, None, None, None)
126        self.skill = self.app.workspace / ".albedo/skills/demo/SKILL.md"
127        self.skill.parent.mkdir(parents=True)
128        self.skill.write_text(SKILL)
129        self.sid = self.app.session()
130
131    def get(self, path):
132        with self.app.api(path) as response:
133            return json.load(response)
134
135    def post(self, path, body):
136        with self.app.api(path, body) as response:
137            return json.load(response)
138
139    def catalog(self):
140        return self.get(f"/sessions/{self.sid}/catalog")
141
142    def commands(self):
143        return self.catalog()["loaded"]["commands"]
144
145    def reload(self):
146        return self.post(f"/sessions/{self.sid}/reload", {"target": "session"})
147
148    def compact(self, strategy=None):
149        return self.post(
150            f"/sessions/{self.sid}/compaction",
151            {"strategy": strategy} if strategy else {},
152        )
153
154    def change(self, fields, session=None):
155        route = f"/sessions/{session or self.sid}?view=configuration"
156        with self.app.api(route) as response:
157            json.load(response)
158            etag = response.headers["ETag"]
159        with self.app.api(
160            route, fields, method="PATCH", headers={"If-Match": etag}
161        ) as response:
162            return json.load(response)
163
164    def snapshot(self):
165        return self.get(f"/sessions/{self.sid}?tail=0")
166
167    def extensions(self):
168        return self.snapshot()["selection"]["effective"]["extensions"]
169
170    def section_text(self, section):
171        snapshot = self.get(f"/sessions/{self.sid}/context")
172        descriptor = next(row for row in snapshot["sections"] if row["id"] == section)
173        chunks = []
174        for page in range(descriptor["page_count"]):
175            query = urllib.parse.urlencode(
176                {
177                    "view": "section",
178                    "snapshot_id": snapshot["snapshot_id"],
179                    "section_id": section,
180                    "page": page,
181                }
182            )
183            chunks.append(self.get(f"/sessions/{self.sid}/context?{query}")["text"])
184        return "".join(chunks)
185
186    def activate(self, name, *, arguments="", client_id=None):
187        catalog = self.catalog()["discovery"]
188        candidate = next(
189            row
190            for row in catalog["candidates"]
191            if row["kind"] == "skill"
192            and row["preference_key"] == name.removeprefix("/")
193        )
194        body = {
195            "kind": "skill",
196            "candidate_id": candidate["id"],
197            "catalog_revision": catalog["revision"],
198            "arguments": arguments,
199        }
200        if client_id is not None:
201            body["client_id"] = client_id
202        with self.app.api(
203            f"/sessions/{self.sid}/inputs/{operation_id()}", body, method="PUT"
204        ) as response:
205            return json.load(response)
206
207    def turn(self, prompt, session=None, *, expected=1):
208        sid = session or self.sid
209        before = len(self.provider.requests)
210        self.app.prompt(sid, prompt).close()
211        self.app.idle(sid)
212        self.assertEqual(len(self.provider.requests) - before, expected)
213        return [entry["request"] for entry in self.provider.requests[before:]]
214
215    def restart(self):
216        self.app.restart()
217
218    def output(self, request):
219        return next(
220            item["output"]
221            for item in reversed(request["input"])
222            if item.get("type") == "function_call_output"
223        )
224
225    def test_workspace_system_files_replace_base_and_append_before_project_instructions(
226        self,
227    ):
228        (self.app.workspace / "SYSTEM.md").write_text("CUSTOM_BASE_ONLY")
229        (self.app.workspace / "APPEND_SYSTEM.md").write_text("CUSTOM_APPEND_ONLY")
230        (self.app.workspace / "AGENTS.md").write_text("PROJECT_CONVENTION_ONLY")
231        (request,) = self.turn("inspect system files")
232        prompt = request["instructions"]
233        self.assertTrue(prompt.startswith("CUSTOM_BASE_ONLY\n"))
234        self.assertNotIn("You are a coding agent operating inside albedo", prompt)
235        self.assertLess(
236            prompt.index("Local workspace context supplied by an enabled extension"),
237            prompt.index("CUSTOM_APPEND_ONLY"),
238        )
239        self.assertLess(
240            prompt.index('name="commands"'), prompt.index("CUSTOM_APPEND_ONLY")
241        )
242        self.assertLess(
243            prompt.index("CUSTOM_APPEND_ONLY"), prompt.index("PROJECT_CONVENTION_ONLY")
244        )
245        self.assertNotIn("CUSTOM_APPEND_ONLY", json.dumps(request["input"]))
246        (self.app.workspace / "SYSTEM.md").write_text("NEW_BASE_ONLY")
247        (self.app.workspace / "APPEND_SYSTEM.md").write_text("NEW_APPEND_ONLY")
248        self.reload()
249        (pinned,) = self.turn("after system file reload")
250        self.assertEqual(pinned["instructions"], prompt)
251        self.assertIn("NEW_BASE_ONLY", json.dumps(pinned["input"]))
252        self.assertIn("NEW_APPEND_ONLY", json.dumps(pinned["input"]))
253        self.compact()
254        (current,) = self.turn("after prompt compaction")
255        self.assertTrue(current["instructions"].startswith("NEW_BASE_ONLY\n"))
256        self.assertIn("NEW_APPEND_ONLY", current["instructions"])
257        self.assertNotIn("CUSTOM_APPEND_ONLY", current["instructions"])
258
259    # exclusive: writes global home instruction files
260    @exclusive
261    def test_system_files_share_instruction_discovery_paths(self):
262        home = self.app.root / "user-home"
263        locations = [
264            self.app.workspace,
265            self.app.workspace / ".agents",
266            self.app.workspace / ".albedo",
267            home / ".agents",
268            home / ".albedo",
269        ]
270        system = []
271        for i, directory in enumerate(locations):
272            directory.mkdir(exist_ok=True)
273            base = directory / "SYSTEM.md"
274            append = directory / "APPEND_SYSTEM.md"
275            if i:
276                base.write_text(f"BASE_{i}")
277                system.append(base)
278                self.addCleanup(base.unlink, missing_ok=True)
279            append.write_text(f"APPEND_{i}")
280            self.addCleanup(append.unlink, missing_ok=True)
281        (self.app.workspace / "AGENTS.md").write_text("AGENT_CONVENTIONS")
282        (request,) = self.turn("discover all prompt paths")
283        prompt = request["instructions"]
284        self.assertTrue(prompt.startswith("BASE_1\n"))
285        self.assertEqual([prompt.count(f"APPEND_{i}") for i in range(5)], [1] * 5)
286        self.assertEqual([prompt.count(f"BASE_{i}") for i in range(1, 5)], [1, 0, 0, 0])
287        positions = [prompt.index(f"APPEND_{i}") for i in range(5)]
288        self.assertEqual(positions, sorted(positions))
289        self.assertLess(positions[-1], prompt.index("AGENT_CONVENTIONS"))
290
291        root = locations[0] / "SYSTEM.md"
292        root.write_text("BASE_0")
293        self.addCleanup(root.unlink, missing_ok=True)
294        for expected, removed in [
295            (0, root),
296            (1, system[0]),
297            (2, system[1]),
298            (3, system[2]),
299            (4, system[3]),
300        ]:
301            session = self.app.session()
302            (next_request,) = self.turn(f"system priority {expected}", session)
303            self.assertTrue(
304                next_request["instructions"].startswith(f"BASE_{expected}\n")
305            )
306            removed.unlink()
307
308    def test_oversized_agent_instructions_warn_and_allow_turns(self):
309        before = self.app.stream_page(self.sid)
310        (self.app.workspace / "AGENTS.md").write_text("X" * (1024 * 1024 + 1))
311        (self.app.workspace / "CLAUDE.md").write_text("USABLE_CONVENTION")
312        (request,) = self.turn("continue without oversized instructions")
313        self.assertIn("USABLE_CONVENTION", request["instructions"])
314        self.assertNotIn("X" * 128, request["instructions"])
315        events = self.app.stream_page(self.sid, before)["events"]
316        warnings = [
317            event["data"]["text"]
318            for event in events
319            if event["type"] == "note" and "AGENTS.md" in event["data"]["text"]
320        ]
321        self.assertEqual(
322            warnings, ["Warning: AGENTS.md exceeds 1 MiB and was not loaded"]
323        )
324        self.turn("next turn still works")
325
326    # exclusive: writes global home instruction files
327    @exclusive
328    def test_instruction_locations_concatenate_in_scope_and_directory_order(self):
329        home = self.app.root / "user-home"
330        files = [
331            (self.app.workspace / "AGENTS.md", "ROOT_AGENTS"),
332            (self.app.workspace / "cLaUdE.Md", "ROOT_CLAUDE"),
333            (self.app.workspace / ".agents/a.MD", "PROJECT_AGENTS_A"),
334            (self.app.workspace / ".agents/z.md", "PROJECT_AGENTS_Z"),
335            (self.app.workspace / ".albedo/a.md", "PROJECT_ALBEDO"),
336            (home / ".agents/a.md", "GLOBAL_AGENTS"),
337            (home / ".albedo/a.md", "GLOBAL_ALBEDO"),
338        ]
339        for path, marker in reversed(files):
340            path.parent.mkdir(exist_ok=True)
341            path.write_text(marker)
342        (self.app.workspace / "SYSTEM.md").write_text("ROOT_SYSTEM")
343        (self.app.workspace / ".agents/SyStEm.Md").write_text("UNSELECTED_SYSTEM")
344        (self.app.workspace / ".agents/ApPeNd_SyStEm.Md").write_text("APPENDED_SYSTEM")
345        (self.app.workspace / ".agents/ignored.txt").write_text("NON_MARKDOWN")
346
347        (request,) = self.turn("inspect instruction ordering")
348        prompt = request["instructions"]
349        positions = [prompt.index(marker) for _, marker in files]
350        self.assertEqual(positions, sorted(positions))
351        self.assertEqual(
352            [prompt.count(marker) for _, marker in files], [1] * len(files)
353        )
354        self.assertLess(prompt.index("APPENDED_SYSTEM"), positions[0])
355        self.assertEqual(prompt.count("APPENDED_SYSTEM"), 1)
356        self.assertNotIn("UNSELECTED_SYSTEM", prompt)
357        self.assertNotIn("NON_MARKDOWN", prompt)
358        self.assertLess(positions[4], prompt.index("## Global user preferences"))
359        self.assertLess(prompt.index("## Global user preferences"), positions[5])
360        self.assertIn(".agents/a.MD", prompt)
361        self.assertIn("~/.agents/a.md", prompt)
362
363    def test_system_replacement_reads_only_highest_priority_match(self):
364        root = self.app.workspace / "SYSTEM.md"
365        root.write_text("VALID_REPLACEMENT")
366        directory = self.app.workspace / ".agents"
367        directory.mkdir()
368        (directory / "system.md").write_bytes(b"\xff")
369
370        (request,) = self.turn("ignore unreadable lower priority prompt")
371        self.assertTrue(request["instructions"].startswith("VALID_REPLACEMENT\n"))
372        root.unlink()
373        outcome = self.reload()
374        self.assertEqual(outcome["session"]["state"], "failed")
375        error = json.dumps(outcome["session"]["failure"])
376        self.assertIn(".agents/system.md must be UTF-8 text", error)
377
378    # exclusive: writes daemon-wide capabilities.json
379    @exclusive
380    def test_instruction_discovery_limit_applies_before_capability_filtering(self):
381        directory = self.app.workspace / ".agents"
382        directory.mkdir()
383        choices = {}
384        for i in range(128):
385            name = f"instruction-{i:03}.md"
386            (directory / name).write_text("DISABLED_INSTRUCTION")
387            choices[f"project:.agents/{name}"] = False
388        (self.app.home / "capabilities.json").write_text(
389            json.dumps({"global": {"instructions": choices}})
390        )
391        (request,) = self.turn("128 disabled instructions remain usable")
392        self.assertNotIn("DISABLED_INSTRUCTION", request["instructions"])
393
394        (directory / "overflow.md").write_text("OVERFLOW_INSTRUCTION")
395        choices["project:.agents/overflow.md"] = False
396        (self.app.home / "capabilities.json").write_text(
397            json.dumps({"global": {"instructions": choices}})
398        )
399        outcome = self.reload()
400        self.assertEqual(outcome["session"]["state"], "failed")
401        error = json.dumps(outcome["session"]["failure"])
402        self.assertIn("more than 128 instruction files were discovered", error)
403
404    def test_workspace_system_files_are_optional(self):
405        (request,) = self.turn("no custom system files")
406        self.assertTrue(
407            request["instructions"].startswith(
408                "You are a coding agent operating inside albedo"
409            )
410        )
411
412    def test_catalog_is_lazy_and_activation_submits_one_turn(self):
413        initial = self.catalog()
414        self.assertFalse(self.provider.requests)
415        self.assertIsNone(self.get(f"/sessions/{self.sid}/context")["snapshot_id"])
416        (request,) = self.turn("first turn")
417        context = request["instructions"]
418        self.assertIn("catalog-only fixture description", context)
419        self.assertIn(str(self.skill.resolve()), context)
420        self.assertNotIn("BODY_MUST_NOT_AUTOLOAD", json.dumps(request))
421        skills = next(
422            row
423            for row in initial["discovery"]["candidates"]
424            if row["kind"] == "extension" and row["preference_key"] == "skills"
425        )
426        self.assertEqual(skills["dependencies"], ["python", "commands"])
427        self.assertEqual(skills["extension"]["tools"], [])
428        catalog = self.commands()
429        demo = next(row for row in catalog if row["slash_name"] == "/demo")
430        self.assertEqual(demo["delivery"], "input")
431        self.assertTrue({"human", "model"} <= set(demo["caller_permissions"]))
432        before = len(self.provider.requests)
433        receipt = self.activate(
434            "/demo", arguments="one  two", client_id="fixture-client"
435        )
436        self.assertEqual(
437            (receipt["admission"], receipt["client_id"]), ("accepted", "fixture-client")
438        )
439        self.app.idle(self.sid)
440        self.assertEqual(len(self.provider.requests), before + 1)
441        activation = json.dumps(self.provider.requests[-1]["request"]["input"])
442        for marker in ("BODY_MUST_NOT_AUTOLOAD", "one  two", str(self.skill.resolve())):
443            self.assertIn(marker, activation)
444        self.assertEqual(self.commands(), catalog)
445        self.skill.write_text(
446            SKILL.replace("catalog-only fixture description", "changed on disk")
447        )
448        self.assertEqual(self.commands(), catalog)
449        self.skill.write_text(SKILL)
450        before = len(self.provider.requests)
451        self.activate("/demo", arguments="slash argument")
452        self.app.idle(self.sid)
453        self.assertEqual(len(self.provider.requests), before + 1)
454        user = next(
455            item["content"]
456            for item in reversed(self.provider.requests[-1]["request"]["input"])
457            if item.get("role") == "user"
458        )
459        activation = json.loads(user.split("\n", 1)[1])
460        self.assertEqual(
461            (activation["name"], activation["arguments"], activation["source"]),
462            ("demo", "slash argument", str(self.skill.resolve())),
463        )
464        self.assertIn("BODY_MUST_NOT_AUTOLOAD", activation["instructions"])
465
466    # exclusive: writes models.json and /model saves global defaults
467    # exclusive: model catalog seeded on disk
468    @exclusive
469    def test_python_activation_and_user_only_model_switch(self):
470        (self.app.home / "models.json").write_text("{}")
471        requests = self.turn("activate demo via python", expected=2)
472        self.assertIn("SKILL_PYTHON_ACTIVATION_OK", self.output(requests[-1]))
473        session = self.app.session()
474        requests = self.turn("model probe via python", session, expected=2)
475        self.assertIn("MODEL_COMMAND_OK", self.output(requests[-1]))
476        selected = self.change({"model": "switched-model"}, session)["session"]
477        self.assertEqual(
478            (selected["model"], selected["provider_profile"]),
479            ("switched-model", self.app.profile),
480        )
481        selected = self.change({"model": "o3-mini"}, session)["session"]
482        self.assertEqual((selected["model"], selected["effort"]), ("o3-mini", "medium"))
483        with self.assertRaises(urllib.error.HTTPError):
484            self.change({"effort": "bogus"}, session)
485        selected = self.change({"effort": "high"}, session)["session"]
486        self.assertEqual(selected["effort"], "high")
487        self.assertEqual(
488            self.change({"model": "o3-mini", "effort": "low"}, session)["session"][
489                "effort"
490            ],
491            "low",
492        )
493        for fields in (
494            {"model": "o3-mini", "effort": "max"},
495            {"model": "switched-model", "effort": "low"},
496        ):
497            with self.assertRaises(urllib.error.HTTPError):
498                self.change(fields, session)
499        self.assertEqual(self.get(f"/sessions/{session}")["model"], "o3-mini")
500
501    def test_work_page_queues_agent_note_without_starting_turn(self):
502        route = "/extensions/work/items?" + urllib.parse.urlencode(
503            {"workspace": str(self.app.workspace)}
504        )
505        page = self.get(route)["page"]
506        self.assertEqual(page["title"], "work")
507        before = len(self.provider.requests)
508        added = self.post(
509            route, {"title": "write the release notes", "session_id": self.sid}
510        )
511        self.assertEqual(added["notification"]["state"], "queued")
512        self.assertEqual(len(self.provider.requests), before)
513        (request,) = self.turn("anything new on the ledger?")
514        context = json.dumps(request["input"])
515        self.assertIn("The user added work item", context)
516        note = next(
517            item["content"]
518            for item in request["input"]
519            if item.get("role") == "user"
520            and "The user added work item" in item.get("content", "")
521        )
522        self.assertEqual(
523            (note.count("<system-note>"), note.count("</system-note>")), (1, 1)
524        )
525        self.assertLess(
526            context.index("The user added work item"),
527            context.index("anything new on the ledger?"),
528        )
529        self.assertTrue(
530            any(
531                resource["value"]["title"] == "write the release notes"
532                for resource in self.get(route)["items"]
533            )
534        )
535
536    def test_work_resources_are_scoped_to_each_workspace(self):
537        workspaces = [self.app.root / "work-one", self.app.root / "work-two"]
538        for workspace in workspaces:
539            workspace.mkdir()
540            self.app.session(workspace)
541        routes = [
542            "/extensions/work/items?"
543            + urllib.parse.urlencode({"workspace": str(workspace)})
544            for workspace in workspaces
545        ]
546        self.post(routes[0], {"title": "first workspace item"})
547        self.assertIn("first workspace item", json.dumps(self.get(routes[0])))
548        self.assertNotIn("first workspace item", json.dumps(self.get(routes[1])))
549        self.post(routes[1], {"title": "second workspace item"})
550        self.assertNotIn("second workspace item", json.dumps(self.get(routes[0])))
551        self.assertIn("second workspace item", json.dumps(self.get(routes[1])))
552
553    def test_manual_compaction_preserves_durable_history_and_reuses_summary(self):
554        cursor = self.app.stream_page(self.sid)
555        for prompt in ("first turn", "second turn", "third turn", "fourth turn"):
556            self.turn(prompt)
557        original = self.app.history(self.sid)["items"]
558        before = len(self.provider.requests)
559        outcome = self.compact()
560        self.assertEqual(
561            (outcome["effective_strategy"], outcome["state"]), ("rolling", "compacted")
562        )
563        self.assertFalse(outcome["selection_applied"])
564        self.assertIsNone(outcome["failure"])
565        self.assertEqual(outcome["observation"]["strategy"], "rolling")
566        self.assertGreater(outcome["observation"]["evicted_entries"], 0)
567        self.assertEqual(len(self.provider.requests), before + 2)
568        current = self.app.history(self.sid)["items"]
569        self.assertEqual(
570            [
571                entry
572                for entry in current
573                if entry["position"] <= original[-1]["position"]
574            ],
575            original,
576        )
577        (request,) = self.turn("after manual compaction")
578        self.assertIn("older conversation summary", json.dumps(request["input"]))
579        context = self.get(f"/sessions/{self.sid}/context")
580        self.assertEqual(
581            (context["state"], context["compaction"]["status"]), ("ready", "compacted")
582        )
583        events = self.app.stream_page(self.sid, cursor)["events"]
584        compacted = next(
585            event["data"] for event in events if event["type"] == "compacted"
586        )
587        self.assertEqual(compacted, outcome["observation"])
588        self.assertGreater(compacted["evicted_entries"], 0)
589        self.assertIn("older conversation summary", compacted["summary"])
590        self.assertIn("older conversation summary", self.section_text("history"))
591
592    def test_every_shipped_skill_loads(self):
593        self.reload()
594        shipped = sorted(
595            path.name for path in (Path(__file__).parents[2] / "priv/skills").iterdir()
596        )
597        listed = {c["slash_name"] for c in self.commands()}
598        self.assertEqual([name for name in shipped if f"/{name}" not in listed], [])
599
600    def test_builtin_skill_is_listed_and_yields_to_a_workspace_skill_of_the_same_name(
601        self,
602    ):
603        def description():
604            return next(
605                c["description"]
606                for c in self.commands()
607                if c["slash_name"] == "/customize-albedo"
608            )
609
610        self.reload()
611        catalog_route = f"/sessions/{self.sid}/catalog"
612        builtin = next(
613            row
614            for row in self.get(catalog_route)["discovery"]["candidates"]
615            if row["preference_key"] == "customize-albedo"
616        )
617        self.assertTrue(builtin["valid"])
618        self.assertTrue(builtin["eligible"])
619        self.assertEqual(description(), builtin["description"])
620        override = self.app.workspace / ".agents/skills/customize-albedo/SKILL.md"
621        override.parent.mkdir(parents=True)
622        override.write_text(
623            "---\nname: customize-albedo\ndescription: workspace override\n---\nmine\n"
624        )
625        self.reload()
626        self.assertEqual(description(), "workspace override")
627        candidates = [
628            row
629            for row in self.get(catalog_route)["discovery"]["candidates"]
630            if row["preference_key"] == "customize-albedo"
631        ]
632        selected = next(row for row in candidates if row["source"] == str(override))
633        shipped = next(row for row in candidates if row["id"] == builtin["id"])
634        self.assertTrue(selected["eligible"])
635        self.assertTrue(shipped["valid"])
636        self.assertFalse(shipped["eligible"])
637        self.assertEqual(shipped["shadowed_by"], selected["id"])
638        (request,) = self.turn("after override")
639        self.assertNotIn("duplicate skill", json.dumps(request["input"]))
640
641    # exclusive: restarts the daemon
642    @exclusive
643    def test_reload_pins_system_prompt_until_compaction(self):
644        self.turn("first turn")
645        catalog_route = f"/sessions/{self.sid}/catalog"
646        original_catalog = self.get(catalog_route)["discovery"]
647        late = self.app.workspace / ".agents/skills/late/SKILL.md"
648        late.parent.mkdir(parents=True)
649        late.write_text(
650            "---\nname: late\ndescription: added after the session opened\n---\nLATE_BODY\n"
651        )
652        self.skill.write_text(
653            SKILL.replace(
654                "catalog-only fixture description", "updated demo description"
655            )
656        )
657        self.assertNotIn("/late", {c["slash_name"] for c in self.commands()})
658        fresh_catalog = self.get(catalog_route)["discovery"]
659        self.assertNotEqual(fresh_catalog["revision"], original_catalog["revision"])
660        late_row = next(
661            row
662            for row in fresh_catalog["candidates"]
663            if row["preference_key"] == "late"
664        )
665        self.assertTrue(late_row["eligible"])
666        self.assertEqual(
667            self.get(catalog_route)["discovery"]["revision"], fresh_catalog["revision"]
668        )
669        requests = self.turn("hot probe before reload", expected=2)
670        self.assertIn("BEFORE_RELOAD_OK", self.output(requests[-1]))
671        cached = requests[-1]
672        reloaded = self.reload()
673        self.assertEqual(reloaded["session"]["state"], "applied")
674        self.assertIn("/late", {c["slash_name"] for c in self.commands()})
675        current_catalog = self.get(catalog_route)["discovery"]
676        self.assertEqual(current_catalog["revision"], fresh_catalog["revision"])
677        self.assertEqual(
678            next(
679                row["id"]
680                for row in current_catalog["candidates"]
681                if row["preference_key"] == "late"
682            ),
683            late_row["id"],
684        )
685        (request,) = self.turn("after session reload")
686        self.assertEqual(request["instructions"], cached["instructions"])
687        updates = [
688            item["content"]
689            for item in request["input"]
690            if item.get("role") == "user"
691            and str(item.get("content", "")).startswith("<system-note")
692            and "capabilities changed" in item["content"]
693        ]
694        self.assertEqual(len(updates), 1)
695        self.assertIn("commands.catalog()", updates[0])
696        self.assertIn("updated demo description", updates[0])
697        self.assertIn("<name>demo</name>", updates[0])
698        self.assertIn("/late", updates[0])
699        self.assertNotIn("catalog-only fixture description", updates[0])
700        self.assertNotIn("LATE_BODY", updates[0])
701        self.assertNotIn("BODY_MUST_NOT_AUTOLOAD", updates[0])
702        requests = self.turn("hot probe after reload", expected=2)
703        self.assertIn("AFTER_RELOAD_OK", self.output(requests[-1]))
704        newer = self.app.workspace / ".agents/skills/newer/SKILL.md"
705        newer.parent.mkdir(parents=True)
706        newer.write_text(
707            "---\nname: newer\ndescription: another live skill\n---\nNEWER_BODY\n"
708        )
709        late.unlink()
710        self.reload()
711        (request,) = self.turn("after second reload")
712        self.assertEqual(request["instructions"], cached["instructions"])
713        update = next(
714            item["content"]
715            for item in reversed(request["input"])
716            if item.get("role") == "user"
717            and "capabilities changed" in str(item.get("content", ""))
718        )
719        self.assertIn("another live skill", update)
720        self.assertIn("supersedes any earlier version", update)
721        self.assertNotIn("<command>/late</command>", update)
722        self.assertNotIn("  /late ", update)
723        self.assertNotIn("catalog-only fixture description", update)
724        self.assertNotIn("NEWER_BODY", update)
725        self.assertIn("/newer", update)
726        self.assertEqual(
727            sum(
728                "capabilities changed" in str(item.get("content", ""))
729                for item in request["input"]
730                if item.get("role") == "user"
731            ),
732            2,
733        )
734        self.restart()
735        (request,) = self.turn("reload pin survives restart")
736        self.assertEqual(request["instructions"], cached["instructions"])
737        self.turn("one more turn before compaction")
738        self.compact()
739        self.app.idle(self.sid)
740        snapshot = self.section_text("instructions")
741        self.assertNotIn("added after the session opened", snapshot)
742        self.assertIn("updated demo description", snapshot)
743        self.assertIn("another live skill", snapshot)
744        (request,) = self.turn("after compaction following reload")
745        self.assertNotIn("added after the session opened", request["instructions"])
746        self.assertIn("updated demo description", request["instructions"])
747        self.assertIn("another live skill", request["instructions"])
748        (request,) = self.turn("pin stays released")
749        self.assertIn("another live skill", request["instructions"])
750
751    # exclusive: restarts the daemon
752    @exclusive
753    def test_restart_keeps_sent_system_prompt_until_compaction(self):
754        for index in range(3):
755            self.turn(f"older turn {index}")
756        self.compact()
757        self.app.idle(self.sid)
758        (sent,) = self.turn("turn after compaction")
759        self.assertIn("[older conversation summary;", json.dumps(sent["input"]))
760        self.skill.write_text(
761            SKILL.replace(
762                "catalog-only fixture description", "edited while the daemon was down"
763            )
764        )
765        self.restart()
766        (request,) = self.turn("first turn after restart")
767        self.assertEqual(request["instructions"], sent["instructions"])
768        updates = capability_notes(request)
769        self.assertEqual(len(updates), 1)
770        self.assertIn("edited while the daemon was down", updates[0])
771        self.restart()
772        (request,) = self.turn("unchanged restart")
773        self.assertEqual(request["instructions"], sent["instructions"])
774        self.assertEqual(capability_notes(request), updates)
775
776    # exclusive: crashes and restarts the daemon
777    @exclusive
778    def test_a_capability_note_saved_during_a_call_follows_its_result(self):
779        self.app.prompt(self.sid, "hold a cell across a crash").close()
780        started = self.app.workspace / "held-cell-started"
781        deadline = time.monotonic() + 15
782        while not started.exists():
783            self.assertLess(time.monotonic(), deadline, "the held cell never started")
784            time.sleep(0.05)
785        self.skill.write_text(
786            SKILL.replace("catalog-only fixture description", "edited during a call")
787        )
788        before = len(self.provider.requests)
789        # The reattached kernel brings the edited prompt, whose note is saved
790        # before the resumed turn records the interrupted call's result.
791        self.app.restart(crash=True)
792        self.app.idle(self.sid)
793        assert_result_follows_call(self, self.provider.requests[before]["request"])
794        self.restart()
795        (request,) = self.turn("after a clean restart")
796        assert_result_follows_call(self, request)
797
798        def follows_call(db):
799            (call,) = db.execute(
800                "SELECT seq FROM transcript_calls WHERE session=?", (self.sid,)
801            ).fetchone()
802            return [
803                seq
804                for (seq,) in db.execute(
805                    "SELECT seq FROM transcript WHERE session=? AND seq>? ORDER BY seq LIMIT 2",
806                    (self.sid, call),
807                )
808            ]
809
810        # A transcript saved before this fix keeps the note between them.
811        def unsettle(app):
812            columns = "payload,timestamp,provider,thought_ms,row_class,turn_id,calls_indexed,summary_indexed"
813            with sqlite3.connect(app.home / "albedo.sqlite") as db:
814                result, note = follows_call(db)
815                saved = {
816                    seq: db.execute(
817                        f"SELECT {columns} FROM transcript WHERE seq=?", (seq,)
818                    ).fetchone()
819                    for seq in (result, note)
820                }
821                assignments = ",".join(f"{c}=?" for c in columns.split(","))
822                for seq, other in ((result, note), (note, result)):
823                    db.execute(
824                        f"UPDATE transcript SET {assignments} WHERE seq=?",
825                        (*saved[other], seq),
826                    )
827
828        self.app.restart(prepare=unsettle)
829        (request,) = self.turn("after an unsettled restart")
830        assert_result_follows_call(self, request)
831        with sqlite3.connect(self.app.home / "albedo.sqlite") as db:
832            classes = [
833                db.execute(
834                    "SELECT row_class FROM transcript WHERE seq=?", (seq,)
835                ).fetchone()[0]
836                for seq in follows_call(db)
837            ]
838        self.assertEqual(classes, ["other", "user"])
839
840    # exclusive: changes global compaction settings and restarts the daemon
841    @exclusive
842    def test_auto_compaction_releases_pin_on_first_turn_after_restart(self):
843        for index in range(4):
844            self.turn(f"older turn {index}: " + "history " * 8000)
845        self.skill.write_text(
846            SKILL.replace(
847                "catalog-only fixture description", "reloaded skill description"
848            )
849        )
850        self.reload()
851        settings_path = self.app.home / "extensions.json"
852        settings = json.loads(settings_path.read_text())
853        settings["rolling"] = {"contextWindowTokens": 50000}
854        settings_path.write_text(json.dumps(settings))
855        self.restart()
856        before = len(self.provider.requests)
857        self.app.prompt(self.sid, "first turn after restart").close()
858        self.app.idle(self.sid)
859        requests = [entry["request"] for entry in self.provider.requests[before:]]
860        self.assertGreater(len(requests), 1, "expected automatic summarization")
861        self.assertIn("reloaded skill description", requests[-1]["instructions"])
862        self.assertNotIn(
863            "catalog-only fixture description", requests[-1]["instructions"]
864        )
865        self.assertIn("older conversation summary", json.dumps(requests[-1]["input"]))
866
867    def test_failed_snapshot_retains_kernel_and_saved_choice_for_retry(self):
868        self.turn("hot probe before reload", expected=2)
869        before = self.snapshot()["kernel"]["instance_id"]
870        with sqlite3.connect(self.app.home / "albedo.sqlite") as db:
871            (run_dir,) = db.execute(
872                "SELECT run_dir FROM kernel_links WHERE session=?", (self.sid,)
873            ).fetchone()
874        blocked = Path(run_dir) / "namespace.state" / "taken"
875        blocked.mkdir(parents=True)
876        self.change({"selection": {"extensions": {"skills": False}}})
877        refused = self.reload()["session"]
878        self.assertEqual(refused["state"], "failed", refused)
879        self.assertFalse(self.extensions()["skills"], "the desired choice stays saved")
880        self.assertEqual(self.snapshot()["kernel"]["instance_id"], before)
881        self.assertIsNotNone(refused["loaded_revision"])
882        blocked.rmdir()
883        blocked.parent.rmdir()
884        applied = self.reload()["session"]
885        self.assertEqual(applied["state"], "applied", applied)
886        self.assertNotEqual(self.snapshot()["kernel"]["instance_id"], before)
887        self.assertNotIn("/demo", [entry["slash_name"] for entry in self.commands()])
888
889    def test_compaction_applies_saved_strategy_and_keeps_matching_kernel(self):
890        self.turn("first turn")
891        self.turn("second turn")
892        before = self.snapshot()["kernel"]["instance_id"]
893        self.change({"selection": {"extensions": {"lcm": True}}})
894        # Saving is separate from applying. Named compaction must compare with
895        # the loaded strategy, even when the desired choice already matches.
896        report = self.compact("lcm")
897        self.assertTrue(report["selection_applied"], report)
898        self.assertEqual(report["effective_strategy"], "lcm", report)
899        self.assertEqual(report["state"], "compacted", report)
900        after = self.snapshot()["kernel"]["instance_id"]
901        self.assertEqual(after, before)
902
903    # exclusive: restarts the daemon
904    @exclusive
905    def test_live_toggles_guard_dependencies_busy_turns_and_persist(self):
906        self.turn("first turn")
907        installed = self.extensions()
908        for name in ("not-installed", "python"):
909            with self.assertRaises(urllib.error.HTTPError):
910                self.change({"selection": {"extensions": {name: False}}})
911        self.assertEqual(self.extensions(), installed)
912        self.change({"selection": {"extensions": {"skills": False}}})
913        self.reload()
914        disabled = self.extensions()
915        self.assertFalse(disabled["skills"])
916        with self.assertRaises(urllib.error.HTTPError):
917            self.activate("/demo", arguments="must not run")
918        pid = self.app.connection["pid"]
919        (request,) = self.turn("extension disabled")
920        self.assertIn("Context removed: skills", json.dumps(request["input"]))
921        self.assertEqual(
922            json.loads((self.app.home / "daemon.json").read_text())["pid"], pid
923        )
924        entered, release = threading.Event(), threading.Event()
925        self.addCleanup(release.set)
926
927        def held(_request):
928            entered.set()
929            release.wait(30)
930            return text("done")
931
932        self.provider.script = held
933        self.app.prompt(self.sid, "hold this turn").close()
934        self.assertTrue(entered.wait(10))
935        for fields in (
936            {"selection": {"extensions": {"skills": True}}},
937            {"model": "mid-run"},
938        ):
939            with self.assertRaises(urllib.error.HTTPError) as caught:
940                self.change(fields)
941            self.assertEqual(caught.exception.code, 409)
942        with self.assertRaises(urllib.error.HTTPError) as caught:
943            self.compact()
944        self.assertEqual(caught.exception.code, 409)
945        self.assertEqual(self.extensions(), disabled)
946        release.set()
947        self.app.idle(self.sid)
948        self.provider.script = scripted
949        self.restart()
950        self.assertFalse(self.extensions()["skills"])
951        self.skill.write_text(
952            SKILL.replace(
953                "catalog-only fixture description", "refreshed catalog description"
954            )
955        )
956        self.change({"selection": {"extensions": {"skills": True}}})
957        self.reload()
958        refreshed = next(
959            command for command in self.commands() if command["slash_name"] == "/demo"
960        )
961        self.assertEqual(refreshed["description"], "refreshed catalog description")
962        (request,) = self.turn("extension reloaded")
963        self.assertNotIn("BODY_MUST_NOT_AUTOLOAD", request["instructions"])
964
965    # exclusive: restarts the daemon
966    @exclusive
967    def test_compaction_strategies_preserve_lcm_folds_across_switch_and_restart(self):
968        for prompt in ("first turn", "second turn", "third turn", "fourth turn"):
969            self.turn(prompt)
970        refused = self.compact("skills")
971        self.assertEqual(refused["state"], "failed")
972        self.assertFalse(refused["selection_applied"])
973        self.assertIsNotNone(refused["failure"])
974        self.assertFalse(self.extensions()["lcm"])
975        result = self.compact("lcm")
976        self.assertEqual(
977            (result["effective_strategy"], result["state"]), ("lcm", "compacted")
978        )
979        self.assertTrue(result["selection_applied"])
980        self.assertIsNone(result["failure"])
981        self.assertEqual(result["observation"]["strategy"], "lcm")
982        self.assertTrue(self.extensions()["lcm"])
983        self.assertFalse(self.extensions()["rolling"])
984        self.change({"selection": {"extensions": {"lcm": False}}})
985        configured = self.get(f"/sessions/{self.sid}?view=configuration")
986        self.assertFalse(configured["selection"]["extensions"]["lcm"])
987        refused_reload = self.reload()
988        self.assertEqual(refused_reload["session"]["state"], "failed")
989        self.assertIsNotNone(refused_reload["session"]["failure"])
990        (request,) = self.turn("lcm preflight")
991        self.assertTrue(
992            {"lcm_list", "lcm_grep", "lcm_describe", "lcm_expand"}
993            <= {tool["name"] for tool in request["tools"]}
994        )
995        requests = self.turn("lcm tool probe", expected=2)
996        self.assertIn("LCM summary node #", json.dumps(requests[0]["input"]))
997        self.assertIn("source_count", self.output(requests[-1]))
998        self.assertIn("first turn", self.output(requests[-1]))
999        self.change({"selection": {"extensions": {"lcm": True}}})
1000        self.assertEqual(self.reload()["session"]["state"], "applied")
1001        configuration_url = f"/sessions/{self.sid}?view=configuration"
1002        before_conflict = self.get(configuration_url)
1003        with self.assertRaises(urllib.error.HTTPError) as conflict:
1004            self.change({"selection": {"extensions": {"lcm": True, "rolling": True}}})
1005        self.assertEqual(conflict.exception.code, 400)
1006        self.assertEqual(json.load(conflict.exception)["code"], "selection_conflict")
1007        self.assertEqual(self.get(configuration_url), before_conflict)
1008        self.change({"selection": {"extensions": {"rolling": True}}})
1009        self.assertEqual(self.reload()["session"]["state"], "applied")
1010        self.assertTrue(self.extensions()["rolling"])
1011        self.assertFalse(self.extensions()["lcm"])
1012        (request,) = self.turn("folded under rolling")
1013        self.assertIn("LCM summary node #", json.dumps(request["input"]))
1014        requests = self.turn("fold list probe", expected=2)
1015        self.assertGreater(json.loads(self.output(requests[-1]))["total"], 0)
1016        self.assertEqual(
1017            self.get(f"/sessions/{self.sid}/context")["compaction"]["strategy"],
1018            "rolling",
1019        )
1020        self.restart()
1021        self.assertTrue(self.extensions()["rolling"])
1022        self.assertFalse(self.extensions()["lcm"])
1023        (request,) = self.turn("folded after restart")
1024        self.assertIn("LCM summary node #", json.dumps(request["input"]))
1025
1026    def test_rest_extension_edits_compare_the_displayed_resource_at_commit(self):
1027        """Concurrent clients cannot overwrite a newer edit or delete its row."""
1028        cases = [
1029            (
1030                "/extensions/work/items?"
1031                + urllib.parse.urlencode({"workspace": str(self.app.workspace)}),
1032                {"title": "original"},
1033                "title",
1034                "😀" * 1025,
1035            ),
1036            (
1037                "/extensions/paperclips/items",
1038                {"message": "original"},
1039                "reply",
1040                "😀" * 4097,
1041            ),
1042            (
1043                "/extensions/schedule/jobs",
1044                {
1045                    "session_id": self.sid,
1046                    "kind": "once",
1047                    "prompt": "original",
1048                    "delay_seconds": 86400,
1049                },
1050                "prompt",
1051                "😀" * 1025,
1052            ),
1053        ]
1054        before = len(self.provider.requests)
1055        for collection, creation, field, oversize in cases:
1056            with self.subTest(collection=collection):
1057                created = self.post(collection, creation)["resource"]
1058                route = created["url"]
1059                try:
1060                    with self.app.api(route) as response:
1061                        original = json.load(response)
1062                        observed = response.headers["ETag"]
1063                    self.assertEqual(original, created["value"])
1064                    for patch, status in (({field: "unguarded"}, 428),):
1065                        with self.assertRaises(urllib.error.HTTPError) as error:
1066                            self.app.api(route, patch, method="PATCH")
1067                        self.assertEqual(error.exception.code, status)
1068                    for patch in (
1069                        {"revision": "invented"},
1070                        {field: None},
1071                        {field: oversize},
1072                    ):
1073                        with self.assertRaises(urllib.error.HTTPError) as error:
1074                            self.app.api(
1075                                route,
1076                                patch,
1077                                method="PATCH",
1078                                headers={"If-Match": observed},
1079                            )
1080                        self.assertIn(error.exception.code, (400, 413))
1081                    self.assertEqual(self.get(route), original)
1082
1083                    barrier = threading.Barrier(3)
1084                    outcomes = []
1085
1086                    def edit(value):
1087                        barrier.wait()
1088                        try:
1089                            with self.app.api(
1090                                route,
1091                                {field: value},
1092                                method="PATCH",
1093                                headers={"If-Match": observed},
1094                            ) as response:
1095                                outcomes.append((response.status, json.load(response)))
1096                        except urllib.error.HTTPError as error:
1097                            outcomes.append((error.code, json.load(error)))
1098
1099                    threads = [
1100                        threading.Thread(target=edit, args=(value,))
1101                        for value in ("first", "second")
1102                    ]
1103                    for thread in threads:
1104                        thread.start()
1105                    barrier.wait()
1106                    for thread in threads:
1107                        thread.join(timeout=10)
1108                        self.assertFalse(thread.is_alive())
1109                    self.assertEqual(
1110                        sorted(status for status, _ in outcomes), [200, 412]
1111                    )
1112                    winner = next(
1113                        body["resource"] for status, body in outcomes if status == 200
1114                    )
1115                    self.assertEqual(self.get(route), winner["value"])
1116                    with self.assertRaises(urllib.error.HTTPError) as error:
1117                        self.app.api(
1118                            route, method="DELETE", headers={"If-Match": observed}
1119                        )
1120                    self.assertEqual(error.exception.code, 412)
1121                    self.assertEqual(self.get(route), winner["value"])
1122                finally:
1123                    with self.app.api(route) as response:
1124                        json.load(response)
1125                        current = response.headers["ETag"]
1126                    with self.app.api(
1127                        route, method="DELETE", headers={"If-Match": current}
1128                    ) as response:
1129                        self.assertEqual(
1130                            json.load(response)["id"], created["value"]["id"]
1131                        )
1132        self.assertEqual(len(self.provider.requests), before)
1133
1134    def test_links_validator_does_not_reappear_after_link_and_unlink(self):
1135        """An old singleton observation stays stale after membership returns."""
1136        other = self.app.root / "linked-rest-peer"
1137        other.mkdir()
1138        own = str(self.app.workspace)
1139        route = "/extensions/links/groups?" + urllib.parse.urlencode(
1140            {"workspace": own, "view": "configuration"}
1141        )
1142        peer = "/extensions/links/groups?" + urllib.parse.urlencode(
1143            {"workspace": str(other), "view": "configuration"}
1144        )
1145        with self.app.api(route) as response:
1146            before = json.load(response)
1147            old = response.headers["ETag"]
1148        with self.app.api(peer) as response:
1149            json.load(response)
1150            peer_etag = response.headers["ETag"]
1151        with self.app.api(
1152            route,
1153            {"other_workspace": str(other), "other_etag": peer_etag},
1154            headers={"If-Match": old},
1155        ) as response:
1156            merged = json.load(response)["resource"]
1157        self.assertEqual(set(merged["value"]["members"]), {own, str(other)})
1158        unlink = route + "&" + urllib.parse.urlencode({"member": str(other)})
1159        with self.app.api(
1160            unlink, method="DELETE", headers={"If-Match": merged["etag"]}
1161        ) as response:
1162            unlinked = json.load(response)["resource"]
1163        self.assertEqual(unlinked["value"]["members"], before["members"])
1164        self.assertNotEqual(unlinked["etag"], old)
1165        with self.app.api(peer) as response:
1166            json.load(response)
1167            peer_etag = response.headers["ETag"]
1168        with self.assertRaises(urllib.error.HTTPError) as error:
1169            self.app.api(
1170                route,
1171                {"other_workspace": str(other), "other_etag": peer_etag},
1172                headers={"If-Match": old},
1173            )
1174        self.assertEqual(error.exception.code, 412)
1175        self.assertEqual(self.get(route)["members"], before["members"])