diff --git a/.agents/skills/lib_py/herdr_config.py b/.agents/skills/lib_py/herdr_config.py new file mode 100644 index 0000000..69f1500 --- /dev/null +++ b/.agents/skills/lib_py/herdr_config.py @@ -0,0 +1,196 @@ +"""Issue #45: the config file MAM's isolated herdr servers start with. + +herdr restores a named session's saved layout when its server starts and, by +default, relaunches each saved agent with herdr's own resume command - without +the permission flags MAM adds, and even for agents MAM deleted while the server +was down. `[session] resume_agents_on_restore = false` turns that off. The +shim's headless bootstrap (lib.sh) hands the server `HERDR_CONFIG_PATH` pointing +at the file built here: the user's own herdr config with only that key forced +to false. The user's file is never modified. + + * Source: the file herdr itself would read - HERDR_CONFIG_PATH, else + $XDG_CONFIG_HOME/herdr/config.toml, else ~/.config/herdr/config.toml. A + HERDR_CONFIG_PATH naming a file generated here (every MAM pane inherits it + from the server) is followed back to the source recorded in that file. + * Output: `mam-server-config.toml` next to the source, so relative paths in + it (herdr resolves them against the config file's directory) keep their + meaning; `<.mam>/herdr-server-config.toml` only when that directory is not + writable. + * The merge edits text (stdlib has no TOML writer) and is then re-parsed: the + key must read false and everything else must equal the source. A merge that + fails the check falls back to a file holding only the key - herdr would + drop a broken file whole and restore again, silently. + +CLI: `python3 -m lib_py.herdr_config server-config [--fallback-dir DIR]` writes +the file and prints its path; warnings go to stderr; exit 1 when no file could +be written. +""" +from __future__ import annotations + +import argparse +import os +import re +import sys +import tempfile +import tomllib +from typing import List, Mapping, Optional, Tuple + +MARKER = "# generated by multi-agent-mux (Issue #45) - do not edit" +SOURCE_PREFIX = "# source: " +OUTPUT_NAME = "mam-server-config.toml" +FALLBACK_NAME = "herdr-server-config.toml" +KEY = "resume_agents_on_restore" +MINIMAL = f"[session]\n{KEY} = false\n" + +_SESSION_HEADER = re.compile(r"^\s*\[\s*session\s*\]\s*(#.*)?$") +_ANY_HEADER = re.compile(r"^\s*\[") +_KEY_LINE = re.compile(rf"^\s*{KEY}\s*=") + + +def _generated_source(path: str) -> Optional[str]: + """The source recorded in a file generated here; '' when it is one without + a source line; None when it is not a generated file.""" + try: + with open(path, encoding="utf-8") as f: + if f.readline().rstrip("\n") != MARKER: + return None + line = f.readline().rstrip("\n") + except OSError: + return None + return line[len(SOURCE_PREFIX):] if line.startswith(SOURCE_PREFIX) else "" + + +def user_config_path(env: Mapping[str, str]) -> str: + """The config file herdr reads under `env`, seen through generated files.""" + explicit = env.get("HERDR_CONFIG_PATH", "") + if explicit: + source = _generated_source(explicit) + if source is None: + return explicit + if source: + return source + xdg = env.get("XDG_CONFIG_HOME", "") + if xdg: + # herdr takes a relative XDG_CONFIG_HOME against its cwd, too. + return os.path.join(os.path.abspath(xdg), "herdr", "config.toml") + return os.path.join(env.get("HOME", os.path.expanduser("~")), ".config", "herdr", "config.toml") + + +def _without_key(d: dict) -> dict: + d = dict(d) + session = dict(d.get("session") or {}) + session.pop(KEY, None) + if session: + d["session"] = session + else: + d.pop("session", None) + return d + + +def _merge_text(user_text: str) -> str: + lines = user_text.splitlines() + header = next((i for i, l in enumerate(lines) if _SESSION_HEADER.match(l)), None) + if header is None: + sep = "" if user_text.endswith("\n") else "\n" + return f"{user_text}{sep}\n{MINIMAL}" + end = next((i for i in range(header + 1, len(lines)) if _ANY_HEADER.match(lines[i])), len(lines)) + key_line = next((i for i in range(header + 1, end) if _KEY_LINE.match(lines[i])), None) + if key_line is None: + lines.insert(header + 1, f"{KEY} = false") + else: + lines[key_line] = f"{KEY} = false" + return "\n".join(lines) + "\n" + + +def server_config_text(user_text: Optional[str]) -> Tuple[str, List[str]]: + """(the TOML body for the server, warnings). The body always parses and + always has session.resume_agents_on_restore = false.""" + if not user_text or not user_text.strip(): + return MINIMAL, [] + try: + user = tomllib.loads(user_text) + except tomllib.TOMLDecodeError as e: + return MINIMAL, [f"herdr config does not parse ({e}); the MAM server uses only {KEY} = false"] + if (user.get("session") or {}).get(KEY) is False: + return user_text, [] + merged = _merge_text(user_text) + try: + result = tomllib.loads(merged) + ok = result["session"][KEY] is False and _without_key(result) == _without_key(user) + except (tomllib.TOMLDecodeError, KeyError, TypeError): + ok = False + if not ok: + return MINIMAL, [f"could not set [session] {KEY} in the herdr config (dotted key or inline table?); " + f"the MAM server uses only {KEY} = false, without the rest of your settings"] + return merged, [] + + +def _write_atomic(path: str, text: str, mode: int) -> None: + d = os.path.dirname(path) + os.makedirs(d, exist_ok=True) + fd, tmp = tempfile.mkstemp(dir=d, prefix=".mam-server-config.", suffix=".tmp") + try: + with os.fdopen(fd, "w", encoding="utf-8") as f: + f.write(text) + os.chmod(tmp, mode) + os.replace(tmp, path) + except BaseException: + try: + os.unlink(tmp) + except OSError: + pass + raise + + +def write_server_config(env: Mapping[str, str], fallback_dir: Optional[str] = None) -> Tuple[Optional[str], List[str]]: + """Write the server config; (its path or None, warnings).""" + source = user_config_path(env) + warnings: List[str] = [] + user_text: Optional[str] = None + mode = 0o644 + try: + with open(source, encoding="utf-8") as f: + user_text = f.read() + mode = os.stat(source).st_mode & 0o777 + except FileNotFoundError: + pass + except (OSError, UnicodeDecodeError) as e: + # herdr cannot use it either, so it runs on defaults: same as no file. + warnings.append(f"cannot read herdr config {source}: {e}") + body, merge_warnings = server_config_text(user_text) + warnings += merge_warnings + text = f"{MARKER}\n{SOURCE_PREFIX}{source}\n{body}" + target = os.path.join(os.path.dirname(source), OUTPUT_NAME) + try: + _write_atomic(target, text, mode) + return target, warnings + except OSError as e: + warnings.append(f"cannot write {target}: {e}") + if fallback_dir: + target = os.path.join(fallback_dir, FALLBACK_NAME) + try: + _write_atomic(target, text, mode) + warnings.append(f"wrote {target} instead; relative paths in the herdr config may not resolve") + return target, warnings + except OSError as e: + warnings.append(f"cannot write {target}: {e}") + return None, warnings + + +def main(argv=None) -> int: + p = argparse.ArgumentParser(description="herdr config for MAM's isolated servers (Issue #45)") + sub = p.add_subparsers(dest="cmd", required=True) + sc = sub.add_parser("server-config", help="write the server config and print its path") + sc.add_argument("--fallback-dir", default=None, help="where to write when the config dir is not writable") + args = p.parse_args(argv) + path, warnings = write_server_config(os.environ, args.fallback_dir) + for w in warnings: + print(f"WARNING: {w}", file=sys.stderr) + if not path: + return 1 + print(path) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/test_issue45_herdr_config.py b/tests/test_issue45_herdr_config.py new file mode 100644 index 0000000..add63e4 --- /dev/null +++ b/tests/test_issue45_herdr_config.py @@ -0,0 +1,192 @@ +"""Issue #45 (A): the herdr config MAM's isolated servers start with. + +lib_py/herdr_config.py builds the user's herdr config with only +[session] resume_agents_on_restore forced to false. herdr drops a config that +does not parse and runs on defaults - restore on - so every result here is +re-parsed, and fed to the real `herdr config check` when herdr is installed. +That restore really is off on a live server is test_issue45_restore.py. +""" +from __future__ import annotations + +import os +import shutil +import stat +import subprocess +import sys +import tomllib +from pathlib import Path + +import pytest + +REPO_ROOT = Path(__file__).resolve().parent.parent +sys.path.insert(0, str(REPO_ROOT / ".agents" / "skills")) + +from lib_py import herdr_config as hc # noqa: E402 + +HERDR = shutil.which("herdr") + +# (name, user config text or None, expect the user's other settings kept, expect a warning) +CASES = [ + ("missing", None, True, False), + ("empty", "", True, False), + ("broken", "[session]\n[session]\n", False, True), + ("no_session", "[ui]\nconfirm_close = false\n", True, False), + ("no_session_no_newline", "[ui]\nconfirm_close = false", True, False), + ("session_without_key", "[session]\n# a comment\n\n[ui]\nconfirm_close = false\n", True, False), + ("key_true", "[ui]\nconfirm_close = false\n[session]\nresume_agents_on_restore = true # mine\n", True, False), + ("key_false", "[session]\nresume_agents_on_restore=false\n", True, False), + ("sub_table_only", "[session.extra]\na = 1\n", True, False), + ("dotted_key", "session.resume_agents_on_restore = true\n", False, True), + ("inline_table", "session = { resume_agents_on_restore = true }\n", False, True), + ("header_inside_string", 'banner = """\n[session]\n"""\n', False, True), +] + + +def _without_key(d): + return hc._without_key(d) + + +@pytest.mark.parametrize("name,user,kept,warned", CASES, ids=[c[0] for c in CASES]) +def test_merge_always_parses_with_restore_off(name, user, kept, warned): + text, warnings = hc.server_config_text(user) + parsed = tomllib.loads(text) + assert parsed["session"]["resume_agents_on_restore"] is False + assert bool(warnings) == warned, warnings + if kept and user: + assert _without_key(parsed) == _without_key(tomllib.loads(user)) + if not kept: + assert text == hc.MINIMAL + + +def test_existing_false_is_left_byte_for_byte(): + user = "[session]\nresume_agents_on_restore=false # keep\n" + assert hc.server_config_text(user) == (user, []) + + +def test_a_bad_value_type_falls_back_to_minimal(): + # herdr rejects a non-bool here as a parse error of the whole file. + text, warnings = hc.server_config_text('[session]\nresume_agents_on_restore = "no"\n') + assert tomllib.loads(text)["session"]["resume_agents_on_restore"] is False + # tomllib reads the string fine, so the merge replaces the line instead. + assert warnings == [] + + +def _env(tmp_path, **kw): + env = {"HOME": str(tmp_path / "home")} + env.update(kw) + return env + + +def test_source_is_home_config_by_default(tmp_path): + assert hc.user_config_path(_env(tmp_path)) == str(tmp_path / "home" / ".config" / "herdr" / "config.toml") + + +def test_source_follows_xdg_config_home(tmp_path, monkeypatch): + assert hc.user_config_path(_env(tmp_path, XDG_CONFIG_HOME=str(tmp_path / "x"))) == \ + str(tmp_path / "x" / "herdr" / "config.toml") + monkeypatch.chdir(tmp_path) # herdr takes a relative one against its cwd + assert hc.user_config_path(_env(tmp_path, XDG_CONFIG_HOME="rel")) == \ + str(tmp_path / "rel" / "herdr" / "config.toml") + + +def test_source_is_herdr_config_path_unless_generated(tmp_path): + user = tmp_path / "custom.toml" + user.write_text("[ui]\nconfirm_close = false\n") + env = _env(tmp_path, HERDR_CONFIG_PATH=str(user), XDG_CONFIG_HOME=str(tmp_path / "x")) + assert hc.user_config_path(env) == str(user) + + # A MAM pane inherits HERDR_CONFIG_PATH=; a server bootstrapped + # from there must still build from the user's file, not from the copy. + generated, _ = hc.write_server_config(env) + assert generated == str(tmp_path / hc.OUTPUT_NAME) + assert hc.user_config_path(dict(env, HERDR_CONFIG_PATH=generated)) == str(user) + + # A generated file without a source line falls through to XDG / HOME. + bare = tmp_path / "bare.toml" + bare.write_text(hc.MARKER + "\n" + hc.MINIMAL) + assert hc.user_config_path(dict(env, HERDR_CONFIG_PATH=str(bare))) == \ + str(tmp_path / "x" / "herdr" / "config.toml") + + +def test_written_next_to_source_with_its_mode(tmp_path): + cfg_dir = tmp_path / "home" / ".config" / "herdr" + cfg_dir.mkdir(parents=True) + user = cfg_dir / "config.toml" + user.write_text('[ui.sound]\npath = "sounds/done.wav"\n') + user.chmod(0o600) + path, warnings = hc.write_server_config(_env(tmp_path)) + assert path == str(cfg_dir / hc.OUTPUT_NAME) and warnings == [] + text = Path(path).read_text() + assert text.splitlines()[:2] == [hc.MARKER, hc.SOURCE_PREFIX + str(user)] + parsed = tomllib.loads(text) + # same directory, so the relative sound path resolves to the same file + assert parsed["ui"]["sound"]["path"] == "sounds/done.wav" + assert parsed["session"]["resume_agents_on_restore"] is False + assert stat.S_IMODE(os.stat(path).st_mode) == 0o600 + assert user.read_text() == '[ui.sound]\npath = "sounds/done.wav"\n' # untouched + assert [p.name for p in cfg_dir.iterdir() if p.name.endswith(".tmp")] == [] + + +def test_written_without_any_user_config(tmp_path): + path, warnings = hc.write_server_config(_env(tmp_path)) + assert path == str(tmp_path / "home" / ".config" / "herdr" / hc.OUTPUT_NAME) and warnings == [] + assert stat.S_IMODE(os.stat(path).st_mode) == 0o644 + assert tomllib.loads(Path(path).read_text()) == {"session": {"resume_agents_on_restore": False}} + + +def test_rewritten_on_every_call(tmp_path): + cfg_dir = tmp_path / "home" / ".config" / "herdr" + cfg_dir.mkdir(parents=True) + (cfg_dir / "config.toml").write_text("[ui]\nconfirm_close = true\n") + path, _ = hc.write_server_config(_env(tmp_path)) + (cfg_dir / "config.toml").write_text("[ui]\nconfirm_close = false\n") + hc.write_server_config(_env(tmp_path)) + assert tomllib.loads(Path(path).read_text())["ui"]["confirm_close"] is False + + +@pytest.mark.skipif(os.geteuid() == 0, reason="root ignores directory permissions") +def test_unwritable_config_dir_falls_back_to_mam_dir(tmp_path): + cfg_dir = tmp_path / "home" / ".config" / "herdr" + cfg_dir.mkdir(parents=True) + (cfg_dir / "config.toml").write_text("[ui]\nconfirm_close = false\n") + cfg_dir.chmod(0o555) + try: + mam = tmp_path / ".mam" + mam.mkdir() + path, warnings = hc.write_server_config(_env(tmp_path), str(mam)) + assert path == str(mam / hc.FALLBACK_NAME) + assert any("relative paths" in w for w in warnings), warnings + assert tomllib.loads(Path(path).read_text())["ui"]["confirm_close"] is False + assert hc.write_server_config(_env(tmp_path))[0] is None # no fallback given + finally: + cfg_dir.chmod(0o755) + + +def test_cli_prints_the_path(tmp_path): + env = dict(os.environ, HOME=str(tmp_path), PYTHONPATH=str(REPO_ROOT / ".agents" / "skills")) + env.pop("HERDR_CONFIG_PATH", None) + env.pop("XDG_CONFIG_HOME", None) + cfg_dir = tmp_path / ".config" / "herdr" + cfg_dir.mkdir(parents=True) + (cfg_dir / "config.toml").write_text("[session]\n[session]\n") + res = subprocess.run([sys.executable, "-m", "lib_py.herdr_config", "server-config"], + capture_output=True, text=True, env=env) + assert res.returncode == 0, res.stderr + assert res.stdout == str(cfg_dir / hc.OUTPUT_NAME) + "\n" + assert "WARNING: herdr config does not parse" in res.stderr + + +@pytest.mark.skipif(not HERDR, reason="herdr is not installed") +@pytest.mark.parametrize("name,user,kept,warned", CASES, ids=[c[0] for c in CASES]) +def test_real_herdr_accepts_every_result(tmp_path, name, user, kept, warned): + text, _ = hc.server_config_text(user) + cfg = tmp_path / "server.toml" + cfg.write_text(hc.MARKER + "\n" + text) + env = {k: v for k, v in os.environ.items() if not k.startswith("HERDR_")} + env.update(HOME=str(tmp_path), HERDR_CONFIG_PATH=str(cfg)) + res = subprocess.run([HERDR, "config", "check"], capture_output=True, text=True, env=env) + out = res.stdout + res.stderr + assert "parse error" not in out, out + # the fixtures use only keys herdr knows, apart from the made-up ones below + issues = [l for l in out.splitlines()[1:] if l.strip()] + assert all(l.startswith("unknown config key") and ("extra" in l or "banner" in l) for l in issues), out