"""Management discovery must predict runtime selection without loading context."""
import json
import os
import unittest
import urllib.error
from harness import Albedo, Provider, exclusive, text
@exclusive
class CatalogTest(unittest.TestCase):
def setUp(self):
self.provider = Provider(lambda _request: text("ok"))
self.addCleanup(self.provider.close)
self.app = Albedo(self.provider, protocol="responses")
self.app.__enter__()
self.addCleanup(self.app.__exit__, None, None, None)
self.session = self.app.session()
self.route = f"/sessions/{self.session}/catalog"
def request(self, path, body=None):
with self.app.api(path, body) as response:
return json.load(response)
def write_skill(self, base, name, description):
path = base / name / "SKILL.md"
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(f"---\nname: {name}\ndescription: {description}\n---\nbody\n")
return path
def reload(self):
self.request(
f"/sessions/{self.session}/commands",
{"name": "/reload", "args": {"target": "session"}},
)
def test_rejected_and_shadowed_candidates_do_not_hide_valid_runtime_winners(self):
high = self.app.workspace / ".albedo/skills"
low = self.app.workspace / ".agents/skills"
malformed = self.write_skill(high, "fallback", "invalid")
malformed.write_text("---\nname: [\ndescription: bad\n---\n")
fallback = self.write_skill(low, "fallback", "LOWER_WINNER")
mismatch = self.write_skill(high, "mismatch", "invalid")
mismatch.write_text("---\nname: other\ndescription: bad\n---\n")
oversized = self.write_skill(high, "oversized", "invalid")
oversized.write_text("x" * (1048576 + 1))
winner = self.write_skill(high, "duplicate", "HIGHER_WINNER")
shadowed = self.write_skill(low, "duplicate", "SHADOWED_DESCRIPTION")
target = self.write_skill(high / "container", "linked", "LINKED_WINNER")
(high / "linked").symlink_to(target.parent, target_is_directory=True)
catalog = self.request(self.route)
by_source = {row["source"]: row for row in catalog["candidates"]}
for path in (malformed, mismatch, oversized):
row = by_source[str(path)]
self.assertFalse(row["valid"])
self.assertTrue(row["diagnostic"])
self.assertFalse(row["eligible"])
self.assertTrue(by_source[str(fallback)]["eligible"])
self.assertTrue(by_source[str(winner)]["eligible"])
self.assertTrue(by_source[str(shadowed)]["valid"])
self.assertFalse(by_source[str(shadowed)]["eligible"])
self.assertEqual(
by_source[str(shadowed)]["shadowed_by"], by_source[str(winner)]["id"]
)
self.assertNotEqual(
by_source[str(winner)]["id"], by_source[str(shadowed)]["id"]
)
linked = by_source[str(high / "linked/SKILL.md")]
self.assertTrue(linked["valid"])
self.assertEqual(linked["resolved_source"], str(target))
self.reload()
commands = self.request(f"/sessions/{self.session}/commands")
descriptions = {row["name"]: row["description"] for row in commands}
self.assertEqual(descriptions["/fallback"], "LOWER_WINNER")
self.assertEqual(descriptions["/duplicate"], "HIGHER_WINNER")
self.assertEqual(descriptions["/linked"], "LINKED_WINNER")
self.assertTrue({"/mismatch", "/oversized"}.isdisjoint(descriptions))
self.app.prompt(self.session, "inspect selected skills").close()
self.app.idle(self.session)
prompt = self.provider.requests[-1]["request"]["instructions"]
for description in ("LOWER_WINNER", "HIGHER_WINNER", "LINKED_WINNER"):
self.assertIn(description, prompt)
self.assertNotIn("SHADOWED_DESCRIPTION", prompt)
replacement_target = self.write_skill(
high / "replacement", "linked", "RETARGETED_WINNER"
)
(high / "linked").unlink()
(high / "linked").symlink_to(
replacement_target.parent, target_is_directory=True
)
retargeted = self.request(self.route)
self.assertNotEqual(retargeted["revision"], catalog["revision"])
retargeted_row = next(
row for row in retargeted["candidates"] if row["id"] == linked["id"]
)
self.assertEqual(retargeted_row["resolved_source"], str(replacement_target))
self.assertEqual(retargeted_row["description"], "RETARGETED_WINNER")
original_id = by_source[str(malformed)]["id"]
self.write_skill(high, "fallback", "REPAIRED_WINNER")
repaired = self.request(self.route)
self.assertNotEqual(repaired["revision"], catalog["revision"])
row = next(
row for row in repaired["candidates"] if row["source"] == str(malformed)
)
self.assertEqual(row["id"], original_id)
self.assertTrue(row["eligible"])
self.reload()
self.assertEqual(
next(
row["description"]
for row in self.request(f"/sessions/{self.session}/commands")
if row["name"] == "/fallback"
),
"REPAIRED_WINNER",
)
def test_instruction_links_follow_daemon_policy_and_exclude_system_files(self):
directory = self.app.workspace / ".agents"
directory.mkdir()
external = self.app.root / "external.md"
external.write_text("LINKED_INSTRUCTION")
(directory / "linked.md").symlink_to(external)
(directory / "broken.md").symlink_to(self.app.root / "missing")
(directory / "directory.md").symlink_to(
self.app.workspace, target_is_directory=True
)
(directory / "SYSTEM.md").write_text("SYSTEM_REPLACEMENT")
(directory / "APPEND_SYSTEM.md").write_text("SYSTEM_APPEND")
(directory / "oversized.md").write_text("x" * (1048576 + 1))
catalog = self.request(self.route)
instructions = {
row["title"]: row
for row in catalog["candidates"]
if row["kind"] == "instructions"
}
self.assertTrue(instructions[".agents/linked.md"]["eligible"])
oversized = instructions[".agents/oversized.md"]
self.assertFalse(oversized["valid"])
self.assertTrue(oversized["diagnostic"])
self.assertFalse(oversized["eligible"])
for name in ("broken.md", "directory.md", "SYSTEM.md", "APPEND_SYSTEM.md"):
self.assertNotIn(f".agents/{name}", instructions)
self.app.prompt(self.session, "inspect linked instruction").close()
self.app.idle(self.session)
self.assertIn(
"LINKED_INSTRUCTION", self.provider.requests[-1]["request"]["instructions"]
)
def test_workspace_move_changes_catalog_without_using_client_workspace(self):
first = self.write_skill(
self.app.workspace / ".albedo/skills", "first", "FIRST_WORKSPACE"
)
(self.app.workspace / "AGENTS.md").write_text("FIRST_INSTRUCTION")
other = self.app.root / "other-workspace"
other.mkdir()
second = self.write_skill(
other / ".albedo/skills", "second", "SECOND_WORKSPACE"
)
(other / "AGENTS.md").write_text("SECOND_INSTRUCTION")
before = self.request(self.route)
self.assertEqual(before["workspace"], str(self.app.workspace))
self.assertIn(str(first), {row["source"] for row in before["candidates"]})
self.assertNotIn(str(second), {row["source"] for row in before["candidates"]})
self.request(f"/sessions/{self.session}/workspace", {"workspace": str(other)})
after = self.request(self.route)
self.assertEqual(after["workspace"], str(other))
self.assertNotEqual(after["revision"], before["revision"])
self.assertNotIn(str(first), {row["source"] for row in after["candidates"]})
self.assertIn(str(second), {row["source"] for row in after["candidates"]})
commands = {
row["name"] for row in self.request(f"/sessions/{self.session}/commands")
}
self.assertIn("/second", commands)
self.assertNotIn("/first", commands)
preferences = self.request("/settings")["capabilities"]
previous = next(
row for row in before["candidates"] if row["source"] == str(first)
)
with self.assertRaises(urllib.error.HTTPError) as failure:
self.request(
self.route,
{
"revision": before["revision"],
"id": previous["id"],
"scope": "session",
"enabled": False,
},
)
self.assertEqual(failure.exception.code, 409)
self.assertEqual(json.load(failure.exception)["code"], "stale_catalog")
self.assertEqual(self.request("/settings")["capabilities"], preferences)
self.assertEqual(
{row["name"] for row in self.request(f"/sessions/{self.session}/commands")},
commands,
)
def test_stale_updates_do_not_write_preferences_or_reload_disk_changes(self):
skill = self.write_skill(
self.app.workspace / ".albedo/skills", "stale", "PREPARED_DESCRIPTION"
)
self.reload()
initial = self.request(self.route)
row = next(
row for row in initial["candidates"] if row["preference_key"] == "stale"
)
body = {
"revision": initial["revision"],
"id": row["id"],
"scope": "session",
"enabled": False,
}
settings_file = self.app.home / "capabilities.json"
def assert_stale_without_side_effects():
before = settings_file.read_bytes() if settings_file.exists() else None
with self.assertRaises(urllib.error.HTTPError) as failure:
self.request(self.route, body)
self.assertEqual(failure.exception.code, 409)
self.assertEqual(json.load(failure.exception)["code"], "stale_catalog")
self.assertEqual(
settings_file.read_bytes() if settings_file.exists() else None, before
)
commands = self.request(f"/sessions/{self.session}/commands")
self.assertEqual(
next(row["description"] for row in commands if row["name"] == "/stale"),
"PREPARED_DESCRIPTION",
)
original_stat = skill.stat()
original_contents = skill.read_bytes()
replacement = original_contents.replace(b"body", b"edit")
self.assertEqual(len(replacement), len(original_contents))
skill.write_bytes(replacement)
os.utime(skill, ns=(original_stat.st_atime_ns, original_stat.st_mtime_ns))
self.assertEqual(skill.stat().st_mtime_ns, original_stat.st_mtime_ns)
assert_stale_without_side_effects()
current = self.request(self.route)
self.assertNotEqual(current["revision"], initial["revision"])
self.assertEqual(self.request(self.route)["revision"], current["revision"])
body["revision"] = current["revision"]
settings_file.write_text(
json.dumps({"global": {"skills": {"unrelated": False}}})
)
assert_stale_without_side_effects()
body["revision"] = self.request(self.route)["revision"]
self.request(self.route, body)
self.assertNotIn(
"/stale",
{row["name"] for row in self.request(f"/sessions/{self.session}/commands")},
)
acknowledged = next(
row
for row in self.request(self.route)["candidates"]
if row["id"] == body["id"]
)
self.assertFalse(acknowledged["session_override"])
self.assertFalse(acknowledged["eligible"])
before_extension_change = self.request(self.route)
body["revision"] = before_extension_change["revision"]
self.request(
f"/sessions/{self.session}/extensions", {"name": "skills", "enabled": False}
)
disabled = self.request(self.route)
self.assertFalse(disabled["extensions"]["skills"])
self.assertNotEqual(disabled["revision"], before_extension_change["revision"])
before_skills = {
row["id"]
for row in before_extension_change["candidates"]
if row["kind"] == "skills"
}
disabled_skills = [
row for row in disabled["candidates"] if row["kind"] == "skills"
]
self.assertEqual({row["id"] for row in disabled_skills}, before_skills)
self.assertTrue(all(not row["eligible"] for row in disabled_skills))
preferences = settings_file.read_bytes()
commands = self.request(f"/sessions/{self.session}/commands")
with self.assertRaises(urllib.error.HTTPError) as failure:
self.request(self.route, body)
self.assertEqual(failure.exception.code, 409)
self.assertEqual(json.load(failure.exception)["code"], "stale_catalog")
self.assertEqual(settings_file.read_bytes(), preferences)
self.assertEqual(self.request(f"/sessions/{self.session}/commands"), commands)
def test_instruction_content_change_rejects_stale_save_with_restored_mtime(self):
instruction = self.app.workspace / "AGENTS.md"
instruction.write_text("FIRST_INSTRUCTION")
self.reload()
initial = self.request(self.route)
self.assertEqual(self.request(self.route)["revision"], initial["revision"])
row = next(
row for row in initial["candidates"] if row["source"] == str(instruction)
)
row_id = row["id"]
preferences = self.request("/settings")["capabilities"]
original_stat = instruction.stat()
original_contents = instruction.read_bytes()
replacement = original_contents.replace(b"FIRST", b"OTHER")
self.assertEqual(len(replacement), len(original_contents))
instruction.write_bytes(replacement)
os.utime(instruction, ns=(original_stat.st_atime_ns, original_stat.st_mtime_ns))
self.assertEqual(instruction.stat().st_mtime_ns, original_stat.st_mtime_ns)
with self.assertRaises(urllib.error.HTTPError) as failure:
self.request(
self.route,
{
"revision": initial["revision"],
"id": row["id"],
"scope": "session",
"enabled": False,
},
)
self.assertEqual(failure.exception.code, 409)
self.assertEqual(json.load(failure.exception)["code"], "stale_catalog")
self.assertEqual(self.request("/settings")["capabilities"], preferences)
refreshed = self.request(self.route)
self.assertNotEqual(refreshed["revision"], initial["revision"])
self.assertEqual(self.request(self.route)["revision"], refreshed["revision"])
self.assertTrue(
next(
row["eligible"]
for row in refreshed["candidates"]
if row["id"] == row_id
)
)
def test_session_in_daemon_home_has_unique_rows_and_distinct_instruction_choices(
self,
):
home = self.app.root / "user-home"
skill = self.write_skill(home / ".albedo/skills", "home-skill", "HOME_SKILL")
instruction = home / ".agents/shared.md"
instruction.parent.mkdir(exist_ok=True)
instruction.write_text("HOME_INSTRUCTION")
session = self.app.session(home)
route = f"/sessions/{session}/catalog"
catalog = self.request(route)
ids = [row["id"] for row in catalog["candidates"]]
self.assertEqual(len(ids), len(set(ids)))
skills = [row for row in catalog["candidates"] if row["source"] == str(skill)]
self.assertEqual(sum(row["eligible"] for row in skills), 1)
commands = self.request(f"/sessions/{session}/commands")
self.assertEqual(sum(row["name"] == "/home-skill" for row in commands), 1)
instructions = [
row for row in catalog["candidates"] if row["source"] == str(instruction)
]
self.assertEqual(
{row["preference_key"] for row in instructions},
{"project:.agents/shared.md", "global:~/.agents/shared.md"},
)
project = next(
row for row in instructions if row["preference_key"].startswith("project:")
)
self.request(
route,
{
"revision": catalog["revision"],
"id": project["id"],
"scope": "session",
"enabled": False,
},
)
updated = self.request(route)
instructions = [
row for row in updated["candidates"] if row["source"] == str(instruction)
]
self.assertFalse(
next(row["eligible"] for row in instructions if row["id"] == project["id"])
)
self.assertTrue(
next(row["eligible"] for row in instructions if row["id"] != project["id"])
)
self.app.prompt(session, "inspect home instruction scopes").close()
self.app.idle(session)
self.assertEqual(
self.provider.requests[-1]["request"]["instructions"].count(
"HOME_INSTRUCTION"
),
1,
)