"""Agent files: the frontmatter, the body kept byte for byte, plugins by entry-point name (a third-party one the same as a built-in), sharing.""" from __future__ import annotations from importlib.metadata import EntryPoint from pathlib import Path from typing import Any import pytest from noeta.agents.agent_files import ( AgentFileError, build_agents, parse_agent_file, read_agent_dir, ) from noeta.core import registry from noeta.core.extension import Extension FULL = """--- name: reviewer description: Reviews a diff. model: some-model one_shot: true effort: high tools: [Read, Grep] plugins: - fs: {tools: [Read, Grep]} - instructions color: blue --- You review. Twice. """ def test_parse_keeps_the_body_and_reads_every_field() -> None: f = parse_agent_file(FULL, source="x") assert f.name == "reviewer" and f.description == "Reviews a diff." assert f.prompt == "You review.\n\nTwice.\n" assert f.model == "some-model" and f.one_shot and f.effort == "high" assert f.tools == ("Read", "Grep") assert f.plugins == (("fs", {"tools": ["Read", "Grep"]}), ("instructions", {})) assert f.extra == {"color": "blue"} def test_claude_code_agent_file() -> None: f = parse_agent_file("---\nname: helper\ndescription: Helps.\ntools: Read, Bash\n" "model: sonnet\n---\nHelp.\n") assert f.tools == ("Read", "Bash") and f.model == "sonnet" and f.plugins is None def test_identity_line_defaults_and_goes_first_as_its_own_cached_block() -> None: from noeta.agents.agent_files import DEFAULT_IDENTITY from noeta.core.loop.request import build_request from noeta.core.loop.state import TaskState own = parse_agent_file("---\nname: a\nidentity: You are A.\nplugins: []\n---\nBody.\n") plain = parse_agent_file("---\nname: b\nplugins: []\n---\nBody.\n") a, b = build_agents([own, plain], model="m") assert own.identity == "You are A." and a.identity == "You are A." assert b.identity == DEFAULT_IDENTITY request = build_request(TaskState(), model="m", system_prompt=b.system_prompt, identity=b.identity, parts=[], tools=[]) assert [(p.text, p.cache) for p in request.system] == [ (DEFAULT_IDENTITY, True), ("Body.\n", True)] @pytest.mark.parametrize("text", [ "no frontmatter\n", "---\nname: x\n", "---\n: [\n---\nbody", "---\n- a\n---\nbody", "---\ndescription: d\n---\nbody", "---\nname: x\nplugins: fs\n---\n", "---\nname: x\nplugins: [{fs: [1]}]\n---\n", "---\nname: x\none_shot: maybe\n---\n", ]) def test_bad_files(text: str) -> None: with pytest.raises(AgentFileError): parse_agent_file(text) def test_read_agent_dir_skips_bad_files(tmp_path: Path) -> None: (tmp_path / "good.md").write_text("---\ndescription: d\n---\nP\n") (tmp_path / "bad.md").write_text("nothing\n") [f] = read_agent_dir(tmp_path) assert f.name == "good" # the file name when the frontmatter has none assert read_agent_dir(tmp_path / "missing") == [] class Greeter(Extension): name = "greeter" def __init__(self, greeting: str = "hi") -> None: self.greeting = greeting def test_a_third_party_plugin_by_name(monkeypatch: pytest.MonkeyPatch) -> None: eps = {"greeter": EntryPoint("greeter", f"{__name__}:Greeter", registry.PLUGINS)} real = registry._group def group(name: str) -> dict[str, Any]: return eps if name == registry.PLUGINS else real(name) monkeypatch.setattr(registry, "_group", group) files = [parse_agent_file("---\nname: a\nplugins: [{greeter: {greeting: hello}}]\n---\nP\n"), parse_agent_file("---\nname: b\nplugins: [{greeter: {greeting: hello}}]\n---\nP\n"), parse_agent_file("---\nname: c\nplugins: [greeter]\n---\nP\n")] a, b, c = build_agents(files, model="m", plugins={"greeter": {"greeting": "hey"}, "tool_search": False, "structured_output": False}) assert a.extensions[0] is b.extensions[0] # same arguments, one instance assert a.extensions[0].greeting == "hello" # type: ignore[attr-defined] assert c.extensions[0].greeting == "hey" # type: ignore[attr-defined] with pytest.raises(registry.UnknownName, match="available: greeter"): build_agents([parse_agent_file("---\nname: d\nplugins: [nope]\n---\n")], model="m") def test_settings_defaults_additions_and_models() -> None: shared = Greeter() files = [parse_agent_file("---\nname: a\nplugins: [greeter, fs]\n---\nP\n"), parse_agent_file("---\nname: b\nmodel: other\n---\nQ\n")] a, b = build_agents(files, model="m", plugins={"greeter": shared, "fs": False, "tool_search": False, "structured_output": False}, default_plugins=[("greeter", {})]) assert list(a.extensions) == [shared] and a.model == "m" assert list(b.extensions) == [shared] and b.model == "other" a, b = build_agents(files, model="m", plugins={"fs": {"enabled": False}, "greeter": shared, "tool_search": False, "structured_output": False}, add={"b": [("instructions", {})], "*": [("greeter", {})]}) assert [e.name for e in a.extensions] == ["greeter"] # not added twice assert [e.name for e in b.extensions] == ["instructions", "greeter"] with pytest.raises(ValueError, match="enabled"): build_agents(files, model="m", plugins={"fs": {"enabled": "no"}}) with pytest.raises(TypeError): build_agents(files, model="m", plugins={"fs": 3}) # type: ignore[dict-item] def test_claude_code_fields_map_onto_the_agent(caplog: pytest.LogCaptureFixture) -> None: f = parse_agent_file( "---\nname: fixer\ndescription: Fixes.\ntools: Read, Bash(npm:*), Edit\n" "disallowedTools: [Write]\nmodel: haiku\npermissionMode: acceptEdits\nmaxTurns: 7\n" "omitClaudeMd: true\nisolation: worktree\nskills: [a]\nflavour: x\n---\nFix.\n", source="f.md") assert f.tools == ("Read", "Bash", "Edit") and f.denied_tools == ("Write",) assert f.permission_mode == "acceptEdits" and f.max_turns == 7 and f.omit_instructions assert f.extra == {"isolation": "worktree", "skills": ["a"], "flavour": "x"} logged = caplog.text assert "'flavour' is ignored" in logged and "'skills'" not in logged assert "isolation" not in logged [agent] = build_agents([f], model="host", plugins={"skills": False, "tool_search": False, "structured_output": False}, default_plugins=[("instructions", {})]) assert agent.model == "claude-haiku-5-5" and agent.denied_tools == ("Write",) # omitClaudeMd keeps only the managed instructions (Claude Code 2.1.312). [instructions] = agent.extensions assert instructions.name == "instructions" assert instructions.managed_only and not instructions.discovery # type: ignore[attr-defined] assert agent.step_budget == 7 # What only the sub-agent plugin acts on stays in the file's fields. assert agent.extra["permissionMode"] == "acceptEdits" assert agent.extra["model"] == "haiku" assert agent.extra["isolation"] == "worktree" [plain] = build_agents([parse_agent_file("---\nname: p\ndescription: d\n---\nx\n")], model="host", default_plugins=[]) assert plain.model == "host" and plain.extra == {"model": "inherit"} def test_tools_star_and_inherit() -> None: f = parse_agent_file("---\nname: a\ntools: '*'\nmodel: inherit\n---\n") assert f.tools is None and f.model is None [agent] = build_agents([f], model="host") assert agent.model == "host" and agent.step_budget == 1_000 @pytest.mark.parametrize("text", [ "---\nname: a:b\n---\n", "---\nname: a\nmaxTurns: 0\n---\n", "---\nname: a\nomitClaudeMd: sometimes\n---\n", "---\nname: a\npermissionMode: [x]\n---\n", ]) def test_bad_claude_code_fields(text: str) -> None: with pytest.raises(AgentFileError): parse_agent_file(text) def test_claude_code_fields_become_plugin_arguments(caplog: pytest.LogCaptureFixture) -> None: from noeta.agents.agent_files import plugin_overlays f = parse_agent_file( "---\nname: rev\nskills: a, b\nmemory: project\nautoCompactWindow: 120k\n" "mcpServers:\n - github\n - local: {command: srv}\n" "hooks: {Stop: [{hooks: [{type: command, command: 'true'}]}]}\n" "initialPrompt: Start here.\nbackground: true\ncacheTtl: 1h\n---\nP\n", source="overlay-rev.md") stop = {"hooks": [{"type": "command", "command": "true"}]} assert plugin_overlays(f) == [ ("skills", {"preload": ["a", "b"]}), ("mcp", {"agent_servers": {"local": {"command": "srv"}}}), ("hooks", {"hooks": {"Stop": [stop], "SubagentStop": [stop]}}), ("memory", {"agent_scope": "project", "agent_name": "rev"}), ("compaction", {"auto_compact_window": 120_000}), ] assert "'cacheTtl' is not applied" in caplog.text assert "initialPrompt" not in caplog.text and "background" not in caplog.text caplog.clear() parse_agent_file("---\nname: rev\ncacheTtl: 1h\n---\nP\n", source="overlay-rev.md") assert "cacheTtl" not in caplog.text # once per file def test_a_critical_reminder_goes_with_every_model_call() -> None: """``criticalSystemReminder_EXPERIMENTAL`` (Claude Code's ``--agents`` field): the environment plugin sends it with each call.""" import asyncio from noeta.agents.agent_files import plugin_overlays from noeta.backends.stores.sqlite import SqliteStore from noeta.core.agent import Agent from noeta.core.client import Client from noeta.core.runner import Runtime from noeta.plugins.environment import Environment from noeta.testing import ScriptedProvider, reply f = parse_agent_file("---\nname: rev\ncriticalSystemReminder_EXPERIMENTAL: Stay read-only." "\n---\nP\n", source="crit.md") assert plugin_overlays(f) == [("environment", {"critical_reminder": "Stay read-only."})] async def go() -> list[list[str]]: store = SqliteStore(":memory:") provider = ScriptedProvider([reply("a"), reply("b")]) async def no_sleep(_: float) -> None: return None agent = Agent(name="a", model="m", extensions=[ Environment(today=lambda: "2026-11-01", critical_reminder="Stay read-only.")]) rt = Runtime(store=store, agents={"a": agent}, provider_for=lambda _m: provider, env_for=lambda _s: None, sleep=no_sleep) # type: ignore[arg-type,return-value] client = Client(store) tid = await client.create("a", "one") await rt.run_until_idle() await client.send(tid, "two") await rt.run_until_idle() await store.close() return [[m.text for m in r.messages] for r in provider.requests] first, second = asyncio.run(go()) assert first.count("Stay read-only.") == 1 and second.count("Stay read-only.") == 2 def test_overlays_add_to_the_hosts_arguments_and_the_plugin() -> None: host_hooks = {"Stop": [{"hooks": [{"type": "command", "command": "host"}]}]} f = parse_agent_file( "---\nname: rev\nskills: [a]\n" "hooks: {Stop: [{hooks: [{type: command, command: own}]}]}\n---\nP\n") [agent] = build_agents([f], model="m", plugins={"hooks": {"hooks": host_hooks}}, default_plugins=[("hooks", {}), ("compaction", {})]) assert [e.name for e in agent.extensions] == ["hooks", "tool_search", "structured_output", "skills", "compaction"] hooks = agent.extensions[0] assert [h.command for m in hooks.matchers["Stop"] # type: ignore[attr-defined] for h in m.hooks] == ["host", "own"] assert agent.extensions[3].preload == ["a"] # type: ignore[attr-defined] [off] = build_agents([f], model="m", plugins={"skills": False, "tool_search": False, "structured_output": False}, default_plugins=[("hooks", {})]) assert [e.name for e in off.extensions] == ["hooks"] def test_agent_mcp_servers_come_first() -> None: from noeta.plugins.mcp import Mcp f = parse_agent_file("---\nname: rev\nmcpServers:\n - local: {command: own}\n---\nP\n") [agent] = build_agents([f], model="m", plugins={"mcp": {"servers": { "local": {"command": "host"}, "other": {"command": "x"}}}, "tool_search": False, "structured_output": False}, default_plugins=[]) [mcp] = agent.extensions assert isinstance(mcp, Mcp) assert [(s.alias, s.command) for s in mcp._agent_servers] == [("local", ["own"])] @pytest.mark.parametrize("text", [ "---\nname: a\nmemory: team\n---\n", "---\nname: a\nmcpServers: x\n---\n", "---\nname: a\nhooks: [x]\n---\n", "---\nname: a\ninitialPrompt: [x]\n---\n", "---\nname: a\nautoCompactWindow: lots\n---\n", ]) def test_bad_applied_fields(text: str) -> None: with pytest.raises(AgentFileError): parse_agent_file(text) def test_every_agent_gets_tool_search_unless_the_host_turns_it_off() -> None: f = parse_agent_file("---\nname: a\nplugins: [fs, compaction]\n---\nP\n") [agent] = build_agents([f], model="m") assert [e.name for e in agent.extensions] == ["fs", "tool_search", "structured_output", "compaction"] [off] = build_agents([f], model="m", plugins={"tool_search": False, "structured_output": False}) assert [e.name for e in off.extensions] == ["fs", "compaction"]