okami.mom/albedo

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

Aly Raffauffeat(api)!: migrate daemon and clients to protocol 361c43ca

main
branchesmainall refs →
20.5 KiB480 linesraw
1"""Management discovery must predict runtime selection without loading context."""
2
3import json
4import os
5import unittest
6import urllib.error
7
8from harness import Albedo, Provider, exclusive, text
9
10
11class CatalogTest(unittest.TestCase):
12    def setUp(self):
13        self.provider = Provider(lambda _request: text("ok"))
14        self.addCleanup(self.provider.close)
15        self.app = Albedo(self.provider, protocol="responses")
16        self.app.__enter__()
17        self.addCleanup(self.app.__exit__, None, None, None)
18        self.session = self.app.session()
19        self.route = f"/sessions/{self.session}/catalog"
20
21    def request(self, path, body=None, **options):
22        with self.app.api(path, body, **options) as response:
23            return json.load(response)
24
25    def discovery(self, session=None):
26        return self.request(f"/sessions/{session or self.session}/catalog")["discovery"]
27
28    def commands(self, session=None):
29        return self.request(f"/sessions/{session or self.session}/catalog")["loaded"][
30            "commands"
31        ]
32
33    def snapshot(self):
34        return self.request(f"/sessions/{self.session}?tail=0")
35
36    def change(self, patch, session=None):
37        route = f"/sessions/{session or self.session}?view=configuration"
38        with self.app.api(route) as response:
39            json.load(response)
40            revision = response.headers["ETag"]
41        return self.request(
42            route, patch, method="PATCH", headers={"If-Match": revision}
43        )
44
45    def test_reappearing_skill_replaces_its_missing_candidate_choice(self):
46        first = self.write_skill(
47            self.app.workspace / ".albedo/skills", "returning", "FIRST_SOURCE"
48        )
49        catalog = self.discovery()
50        original = next(
51            row for row in catalog["candidates"] if row["source"] == str(first)
52        )
53        self.change(
54            {
55                "catalog_revision": catalog["revision"],
56                "selection": {"skills": {original["id"]: False}},
57            }
58        )
59        self.reload()
60        self.assertNotIn("/returning", {row["slash_name"] for row in self.commands()})
61        first.unlink()
62        replacement = self.write_skill(
63            self.app.workspace / ".agents/skills", "returning", "REPLACEMENT_SOURCE"
64        )
65        catalog = self.discovery()
66        fresh = next(
67            row for row in catalog["candidates"] if row["source"] == str(replacement)
68        )
69        self.assertNotEqual(fresh["id"], original["id"])
70        self.change(
71            {
72                "catalog_revision": catalog["revision"],
73                "selection": {"skills": {fresh["id"]: True}},
74            }
75        )
76        configuration = self.request(f"/sessions/{self.session}?view=configuration")
77        self.assertEqual(configuration["selection"]["skills"], {fresh["id"]: True})
78        self.reload()
79        command = next(
80            row for row in self.commands() if row["slash_name"] == "/returning"
81        )
82        self.assertEqual(command["description"], "REPLACEMENT_SOURCE")
83        catalog = self.discovery()
84        self.change(
85            {
86                "catalog_revision": catalog["revision"],
87                "selection": {"skills": {fresh["id"]: None}},
88            }
89        )
90        self.assertEqual(
91            self.request(f"/sessions/{self.session}?view=configuration")["selection"][
92                "skills"
93            ],
94            {},
95        )
96        self.reload()
97        self.assertIn("/returning", {row["slash_name"] for row in self.commands()})
98
99    def write_skill(self, base, name, description):
100        path = base / name / "SKILL.md"
101        path.parent.mkdir(parents=True, exist_ok=True)
102        path.write_text(f"---\nname: {name}\ndescription: {description}\n---\nbody\n")
103        return path
104
105    def reload(self):
106        self.request(
107            f"/sessions/{self.session}/reload",
108            {"target": "session"},
109        )
110
111    def test_rejected_and_shadowed_candidates_do_not_hide_valid_runtime_winners(self):
112        high = self.app.workspace / ".albedo/skills"
113        low = self.app.workspace / ".agents/skills"
114        malformed = self.write_skill(high, "fallback", "invalid")
115        malformed.write_text("---\nname: [\ndescription: bad\n---\n")
116        fallback = self.write_skill(low, "fallback", "LOWER_WINNER")
117        mismatch = self.write_skill(high, "mismatch", "invalid")
118        mismatch.write_text("---\nname: other\ndescription: bad\n---\n")
119        oversized = self.write_skill(high, "oversized", "invalid")
120        oversized.write_text("x" * (1048576 + 1))
121        winner = self.write_skill(high, "duplicate", "HIGHER_WINNER")
122        shadowed = self.write_skill(low, "duplicate", "SHADOWED_DESCRIPTION")
123        target = self.write_skill(high / "container", "linked", "LINKED_WINNER")
124        (high / "linked").symlink_to(target.parent, target_is_directory=True)
125
126        catalog = self.discovery()
127        by_source = {row["source"]: row for row in catalog["candidates"]}
128        for path in (malformed, mismatch, oversized):
129            row = by_source[str(path)]
130            self.assertFalse(row["valid"])
131            self.assertTrue(row["diagnostic"])
132            self.assertFalse(row["eligible"])
133        self.assertTrue(by_source[str(fallback)]["eligible"])
134        self.assertTrue(by_source[str(winner)]["eligible"])
135        self.assertTrue(by_source[str(shadowed)]["valid"])
136        self.assertFalse(by_source[str(shadowed)]["eligible"])
137        self.assertEqual(
138            by_source[str(shadowed)]["shadowed_by"], by_source[str(winner)]["id"]
139        )
140        self.assertNotEqual(
141            by_source[str(winner)]["id"], by_source[str(shadowed)]["id"]
142        )
143        linked = by_source[str(high / "linked/SKILL.md")]
144        self.assertTrue(linked["valid"])
145        self.assertEqual(linked["resolved_source"], str(target))
146        self.reload()
147        commands = self.commands()
148        descriptions = {row["slash_name"]: row["description"] for row in commands}
149        self.assertEqual(descriptions["/fallback"], "LOWER_WINNER")
150        self.assertEqual(descriptions["/duplicate"], "HIGHER_WINNER")
151        self.assertEqual(descriptions["/linked"], "LINKED_WINNER")
152        self.assertTrue({"/mismatch", "/oversized"}.isdisjoint(descriptions))
153
154        self.app.prompt(self.session, "inspect selected skills").close()
155        self.app.idle(self.session)
156        prompt = self.provider.requests[-1]["request"]["instructions"]
157        for description in ("LOWER_WINNER", "HIGHER_WINNER", "LINKED_WINNER"):
158            self.assertIn(description, prompt)
159        self.assertNotIn("SHADOWED_DESCRIPTION", prompt)
160
161        replacement_target = self.write_skill(
162            high / "replacement", "linked", "RETARGETED_WINNER"
163        )
164        (high / "linked").unlink()
165        (high / "linked").symlink_to(
166            replacement_target.parent, target_is_directory=True
167        )
168        retargeted = self.discovery()
169        self.assertNotEqual(retargeted["revision"], catalog["revision"])
170        retargeted_row = next(
171            row for row in retargeted["candidates"] if row["id"] == linked["id"]
172        )
173        self.assertEqual(retargeted_row["resolved_source"], str(replacement_target))
174        self.assertEqual(retargeted_row["description"], "RETARGETED_WINNER")
175
176        original_id = by_source[str(malformed)]["id"]
177        self.write_skill(high, "fallback", "REPAIRED_WINNER")
178        repaired = self.discovery()
179        self.assertNotEqual(repaired["revision"], catalog["revision"])
180        row = next(
181            row for row in repaired["candidates"] if row["source"] == str(malformed)
182        )
183        self.assertEqual(row["id"], original_id)
184        self.assertTrue(row["eligible"])
185        self.reload()
186        self.assertEqual(
187            next(
188                row["description"]
189                for row in self.commands()
190                if row["slash_name"] == "/fallback"
191            ),
192            "REPAIRED_WINNER",
193        )
194
195    def test_instruction_links_follow_daemon_policy_and_exclude_system_files(self):
196        directory = self.app.workspace / ".agents"
197        directory.mkdir()
198        external = self.app.workspace.with_name(
199            self.app.workspace.name + "-external.md"
200        )
201        external.write_text("LINKED_INSTRUCTION")
202        (directory / "linked.md").symlink_to(external)
203        (directory / "broken.md").symlink_to(self.app.workspace / "missing")
204        (directory / "directory.md").symlink_to(
205            self.app.workspace, target_is_directory=True
206        )
207        (directory / "SYSTEM.md").write_text("SYSTEM_REPLACEMENT")
208        (directory / "APPEND_SYSTEM.md").write_text("SYSTEM_APPEND")
209        (directory / "oversized.md").write_text("x" * (1048576 + 1))
210        catalog = self.discovery()
211        instructions = {
212            row["title"]: row
213            for row in catalog["candidates"]
214            if row["kind"] == "instruction"
215        }
216        self.assertTrue(instructions[".agents/linked.md"]["eligible"])
217        oversized = instructions[".agents/oversized.md"]
218        self.assertFalse(oversized["valid"])
219        self.assertTrue(oversized["diagnostic"])
220        self.assertFalse(oversized["eligible"])
221        for name in ("broken.md", "directory.md", "SYSTEM.md", "APPEND_SYSTEM.md"):
222            self.assertNotIn(f".agents/{name}", instructions)
223        self.app.prompt(self.session, "inspect linked instruction").close()
224        self.app.idle(self.session)
225        self.assertIn(
226            "LINKED_INSTRUCTION", self.provider.requests[-1]["request"]["instructions"]
227        )
228
229    # exclusive: GET /settings validates every profile, so it fails while any concurrent test holds an invalid one
230    @exclusive
231    def test_workspace_move_changes_catalog_without_using_client_workspace(self):
232        first = self.write_skill(
233            self.app.workspace / ".albedo/skills", "first", "FIRST_WORKSPACE"
234        )
235        (self.app.workspace / "AGENTS.md").write_text("FIRST_INSTRUCTION")
236        other = self.app.workspace.with_name(self.app.workspace.name + "-other")
237        other.mkdir()
238        second = self.write_skill(
239            other / ".albedo/skills", "second", "SECOND_WORKSPACE"
240        )
241        (other / "AGENTS.md").write_text("SECOND_INSTRUCTION")
242        before = self.discovery()
243        self.assertEqual(before["workspace"], str(self.app.workspace))
244        self.assertIn(str(first), {row["source"] for row in before["candidates"]})
245        self.assertNotIn(str(second), {row["source"] for row in before["candidates"]})
246        self.change(
247            {
248                "workspace": str(other),
249                "family_revision": self.snapshot()["family_revision"],
250            }
251        )
252        after = self.discovery()
253        self.assertEqual(after["workspace"], str(other))
254        self.assertNotEqual(after["revision"], before["revision"])
255        self.assertNotIn(str(first), {row["source"] for row in after["candidates"]})
256        self.assertIn(str(second), {row["source"] for row in after["candidates"]})
257        self.reload()
258        commands = {row["slash_name"] for row in self.commands()}
259        self.assertIn("/second", commands)
260        self.assertNotIn("/first", commands)
261        preferences = self.snapshot()["configuration_resource"]["value"]["selection"]
262        previous = next(
263            row for row in before["candidates"] if row["source"] == str(first)
264        )
265        with self.assertRaises(urllib.error.HTTPError) as failure:
266            self.change(
267                {
268                    "catalog_revision": before["revision"],
269                    "selection": {"skills": {previous["id"]: False}},
270                }
271            )
272        self.assertEqual(failure.exception.code, 409)
273        self.assertEqual(json.load(failure.exception)["code"], "catalog_changed")
274        self.assertEqual(
275            self.snapshot()["configuration_resource"]["value"]["selection"],
276            preferences,
277        )
278        self.assertEqual(
279            {row["slash_name"] for row in self.commands()},
280            commands,
281        )
282
283    # exclusive: writes global preferences and asserts unchanged global file bytes
284    @exclusive
285    def test_stale_updates_do_not_write_preferences_or_reload_disk_changes(self):
286        skill = self.write_skill(
287            self.app.workspace / ".albedo/skills", "stale", "PREPARED_DESCRIPTION"
288        )
289        self.reload()
290        initial = self.discovery()
291        row = next(
292            row for row in initial["candidates"] if row["preference_key"] == "stale"
293        )
294        body = {
295            "catalog_revision": initial["revision"],
296            "selection": {"skills": {row["id"]: False}},
297        }
298        settings_file = self.app.home / "capabilities.json"
299
300        def assert_stale_without_side_effects():
301            before = settings_file.read_bytes() if settings_file.exists() else None
302            configuration = self.request(f"/sessions/{self.session}?view=configuration")
303            with self.assertRaises(urllib.error.HTTPError) as failure:
304                self.change(body)
305            self.assertEqual(failure.exception.code, 409)
306            self.assertEqual(json.load(failure.exception)["code"], "catalog_changed")
307            self.assertEqual(
308                settings_file.read_bytes() if settings_file.exists() else None, before
309            )
310            self.assertEqual(
311                self.request(f"/sessions/{self.session}?view=configuration"),
312                configuration,
313            )
314            commands = self.commands()
315            self.assertEqual(
316                next(
317                    row["description"]
318                    for row in commands
319                    if row["slash_name"] == "/stale"
320                ),
321                "PREPARED_DESCRIPTION",
322            )
323
324        original_stat = skill.stat()
325        original_contents = skill.read_bytes()
326        replacement = original_contents.replace(b"body", b"edit")
327        self.assertEqual(len(replacement), len(original_contents))
328        skill.write_bytes(replacement)
329        os.utime(skill, ns=(original_stat.st_atime_ns, original_stat.st_mtime_ns))
330        self.assertEqual(skill.stat().st_mtime_ns, original_stat.st_mtime_ns)
331        assert_stale_without_side_effects()
332        current = self.discovery()
333        self.assertNotEqual(current["revision"], initial["revision"])
334        self.assertEqual(self.discovery()["revision"], current["revision"])
335        body["catalog_revision"] = current["revision"]
336        settings_file.write_text(
337            json.dumps({"global": {"skills": {"unrelated": False}}})
338        )
339        assert_stale_without_side_effects()
340        body["catalog_revision"] = self.discovery()["revision"]
341        self.change(body)
342        self.reload()
343        self.assertNotIn(
344            "/stale",
345            {row["slash_name"] for row in self.commands()},
346        )
347        acknowledged = next(
348            candidate
349            for candidate in self.discovery()["candidates"]
350            if candidate["id"] == row["id"]
351        )
352        self.assertFalse(acknowledged["session_override"])
353        self.assertFalse(acknowledged["eligible"])
354
355        before_extension_change = self.discovery()
356        body["catalog_revision"] = before_extension_change["revision"]
357        self.change({"selection": {"extensions": {"skills": False}}})
358        disabled = self.discovery()
359        self.assertFalse(
360            self.snapshot()["selection"]["effective"]["extensions"]["skills"]
361        )
362        self.assertNotEqual(disabled["revision"], before_extension_change["revision"])
363        before_skills = {
364            row["id"]
365            for row in before_extension_change["candidates"]
366            if row["kind"] == "skill"
367        }
368        disabled_skills = [
369            row for row in disabled["candidates"] if row["kind"] == "skill"
370        ]
371        self.assertEqual({row["id"] for row in disabled_skills}, before_skills)
372        self.assertTrue(all(not row["eligible"] for row in disabled_skills))
373        preferences = settings_file.read_bytes()
374        commands = self.commands()
375        with self.assertRaises(urllib.error.HTTPError) as failure:
376            self.change(body)
377        self.assertEqual(failure.exception.code, 409)
378        self.assertEqual(json.load(failure.exception)["code"], "catalog_changed")
379        self.assertEqual(settings_file.read_bytes(), preferences)
380        self.assertEqual(self.commands(), commands)
381
382    # exclusive: GET /settings validates every profile, so it fails while any concurrent test holds an invalid one
383    @exclusive
384    def test_instruction_content_change_rejects_stale_save_with_restored_mtime(self):
385        instruction = self.app.workspace / "AGENTS.md"
386        instruction.write_text("FIRST_INSTRUCTION")
387        self.reload()
388        initial = self.discovery()
389        self.assertEqual(self.discovery()["revision"], initial["revision"])
390        row = next(
391            row for row in initial["candidates"] if row["source"] == str(instruction)
392        )
393        row_id = row["id"]
394        preferences = self.snapshot()["configuration_resource"]["value"]["selection"]
395        original_stat = instruction.stat()
396        original_contents = instruction.read_bytes()
397        replacement = original_contents.replace(b"FIRST", b"OTHER")
398        self.assertEqual(len(replacement), len(original_contents))
399        instruction.write_bytes(replacement)
400        os.utime(instruction, ns=(original_stat.st_atime_ns, original_stat.st_mtime_ns))
401        self.assertEqual(instruction.stat().st_mtime_ns, original_stat.st_mtime_ns)
402        with self.assertRaises(urllib.error.HTTPError) as failure:
403            self.change(
404                {
405                    "catalog_revision": initial["revision"],
406                    "selection": {"instructions": {row["id"]: False}},
407                }
408            )
409        self.assertEqual(failure.exception.code, 409)
410        self.assertEqual(json.load(failure.exception)["code"], "catalog_changed")
411        self.assertEqual(
412            self.snapshot()["configuration_resource"]["value"]["selection"],
413            preferences,
414        )
415        refreshed = self.discovery()
416        self.assertNotEqual(refreshed["revision"], initial["revision"])
417        self.assertEqual(self.discovery()["revision"], refreshed["revision"])
418        self.assertTrue(
419            next(
420                row["eligible"]
421                for row in refreshed["candidates"]
422                if row["id"] == row_id
423            )
424        )
425
426    # exclusive: writes skills and instructions in the daemon user home
427    @exclusive
428    def test_session_in_daemon_home_has_unique_rows_and_distinct_instruction_choices(
429        self,
430    ):
431        home = self.app.root / "user-home"
432        skill = self.write_skill(home / ".albedo/skills", "home-skill", "HOME_SKILL")
433        instruction = home / ".agents/shared.md"
434        instruction.parent.mkdir(exist_ok=True)
435        instruction.write_text("HOME_INSTRUCTION")
436        session = self.app.session(home)
437        catalog = self.discovery(session)
438        ids = [row["id"] for row in catalog["candidates"]]
439        self.assertEqual(len(ids), len(set(ids)))
440        skills = [row for row in catalog["candidates"] if row["source"] == str(skill)]
441        self.assertEqual(sum(row["eligible"] for row in skills), 1)
442        self.request(f"/sessions/{session}/reload", {"target": "session"})
443        commands = self.commands(session)
444        self.assertEqual(sum(row["slash_name"] == "/home-skill" for row in commands), 1)
445        instructions = [
446            row for row in catalog["candidates"] if row["source"] == str(instruction)
447        ]
448        self.assertEqual(
449            {row["preference_key"] for row in instructions},
450            {"project:.agents/shared.md", "global:~/.agents/shared.md"},
451        )
452        project = next(
453            row for row in instructions if row["preference_key"].startswith("project:")
454        )
455        self.change(
456            {
457                "catalog_revision": catalog["revision"],
458                "selection": {"instructions": {project["id"]: False}},
459            },
460            session,
461        )
462        updated = self.discovery(session)
463        instructions = [
464            row for row in updated["candidates"] if row["source"] == str(instruction)
465        ]
466        self.assertFalse(
467            next(row["eligible"] for row in instructions if row["id"] == project["id"])
468        )
469        self.assertTrue(
470            next(row["eligible"] for row in instructions if row["id"] != project["id"])
471        )
472        self.app.api(f"/sessions/{session}/reload", {"target": "session"}).close()
473        self.app.prompt(session, "inspect home instruction scopes").close()
474        self.app.idle(session)
475        self.assertEqual(
476            self.provider.requests[-1]["request"]["instructions"].count(
477                "HOME_INSTRUCTION"
478            ),
479            1,
480        )