diff --git a/.agents/skills/lib.sh b/.agents/skills/lib.sh index 35959f8..36efd5b 100644 --- a/.agents/skills/lib.sh +++ b/.agents/skills/lib.sh @@ -235,9 +235,10 @@ except Exception: # relaunch every saved agent with herdr's own resume command (no permission # flags, deleted agents included). It gets HERDR_CONFIG_PATH = the user's # herdr config with [session] resume_agents_on_restore = false, rebuilt on - # every start (lib_py/herdr_config.py). Without that file the server still - # starts - with restore on, as before. - MAM_SERVER_CFG=$(python3 -m lib_py.herdr_config server-config --fallback-dir "$_SHIM_DIR/.." || true) + # every start and checked by this herdr's `config check` + # (lib_py/herdr_config.py). Without that file the server still starts - + # with restore on, as before. + MAM_SERVER_CFG=$(python3 -m lib_py.herdr_config server-config --fallback-dir "$_SHIM_DIR/.." --herdr "$REAL_HERDR" || true) export MAM_SERVER_CFG if [ -z "$MAM_SERVER_CFG" ]; then echo "WARNING: restore-on-restart stays enabled for herdr session $_MAM_SESSION (no server config)" >&2 diff --git a/.agents/skills/lib_py/herdr_config.py b/.agents/skills/lib_py/herdr_config.py index 1d20f55..dc5469c 100644 --- a/.agents/skills/lib_py/herdr_config.py +++ b/.agents/skills/lib_py/herdr_config.py @@ -14,22 +14,30 @@ to false. The user's file is never modified. 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. + meaning. Never the source itself and never a file this module did not + write for the same source: then a name with a hash of the source path. + `<.mam>/herdr-server-config.toml` only when the directory is not writable. + * The source is parsed, the key set, and the result serialized; the result + must parse back to exactly the source plus the key (any TOML spelling of + the key - header, dotted, inline - works). Then herdr's own `config check` + must accept the written file: herdr drops a file it cannot use whole + ("using defaults") and would restore again, silently. Anything that fails + falls back to a file holding only the key, which is checked the same way. -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. +CLI: `python3 -m lib_py.herdr_config server-config [--fallback-dir DIR] +[--herdr PATH]` writes the file and prints its path; warnings go to stderr; +exit 1 when no usable file could be written. """ from __future__ import annotations import argparse +import datetime +import hashlib +import json +import math import os import re +import subprocess import sys import tempfile from typing import List, Mapping, Optional, Tuple @@ -40,27 +48,30 @@ OUTPUT_NAME = "mam-server-config.toml" FALLBACK_NAME = "herdr-server-config.toml" KEY = "resume_agents_on_restore" MINIMAL = f"[session]\n{KEY} = false\n" +LOSES_SETTINGS = f"the MAM server uses only {KEY} = false, without the rest of your settings" try: import tomllib except ImportError: # Python < 3.11: no parser to check a merge with tomllib = None -_SESSION_HEADER = re.compile(r"^\s*\[\s*session\s*\]\s*(#.*)?$") -_ANY_HEADER = re.compile(r"^\s*\[") -_KEY_LINE = re.compile(rf"^\s*{KEY}\s*=") +_BARE_KEY = re.compile(r"^[A-Za-z0-9_-]+$") +# Where a pane's server, socket and workspace come from; `config check` must not +# be steered by the caller's pane. +_SCOPE_ENV = ("HERDR_WORKSPACE_ID", "HERDR_PANE_ID", "HERDR_TAB_ID", "HERDR_SOCKET_PATH") 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.""" + a source line; None when it is not a generated file (or unreadable).""" try: - with open(path, encoding="utf-8") as f: - if f.readline().rstrip("\n") != MARKER: - return None - line = f.readline().rstrip("\n") + with open(path, "rb") as f: + first, second = f.readline(), f.readline() except OSError: return None + if first.rstrip(b"\n").decode("utf-8", "replace") != MARKER: + return None + line = second.rstrip(b"\n").decode("utf-8", "replace") return line[len(SOURCE_PREFIX):] if line.startswith(SOURCE_PREFIX) else "" @@ -80,56 +91,116 @@ def user_config_path(env: Mapping[str, str]) -> str: 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 +# --- TOML writer for what tomllib returns ----------------------------------- + +def _str(s: str) -> str: + # json's escapes (\" \\ \b \f \n \r \t \uXXXX) are TOML's; DEL is not + # escaped by json but is not allowed raw in a TOML string. + return json.dumps(s, ensure_ascii=False).replace("\x7f", "\\u007f") -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" +def _key(k: str) -> str: + return k if _BARE_KEY.match(k) else _str(k) + + +def _value(v) -> str: + if isinstance(v, bool): + return "true" if v else "false" + if isinstance(v, int): + return str(v) + if isinstance(v, float): + if math.isnan(v): + return "nan" + if math.isinf(v): + return "inf" if v > 0 else "-inf" + return repr(v) + if isinstance(v, str): + return _str(v) + if isinstance(v, (datetime.datetime, datetime.date, datetime.time)): + return v.isoformat() + if isinstance(v, list): + return "[" + ", ".join(_value(x) for x in v) + "]" + if isinstance(v, dict): + return "{" + ", ".join(f"{_key(k)} = {_value(x)}" for k, x in v.items()) + "}" + raise TypeError(f"unexpected TOML value {type(v).__name__}") + + +def dumps(doc: dict) -> str: + """Top-level values, then one [table] per top-level table with its members + written inline - valid TOML for any document tomllib can return.""" + lines = [f"{_key(k)} = {_value(v)}" for k, v in doc.items() if not isinstance(v, dict)] + for k, v in doc.items(): + if isinstance(v, dict): + lines.append(f"[{_key(k)}]") + lines += [f"{_key(kk)} = {_value(vv)}" for kk, vv in v.items()] 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.""" + """(the TOML body for the server, warnings). The body always parses, has + session.resume_agents_on_restore = false and otherwise equals the source - + or is MINIMAL, with a warning, when the source cannot be used.""" if not user_text or not user_text.strip(): return MINIMAL, [] if tomllib is None: - return MINIMAL, [f"this Python has no tomllib (3.11+); the MAM server uses only {KEY} = false, " - "without the rest of your settings"] + return MINIMAL, [f"this Python has no tomllib (3.11+); {LOSES_SETTINGS}"] 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) + return MINIMAL, [f"herdr config does not parse ({e}); {LOSES_SETTINGS}"] + session = user.get("session", {}) + if not isinstance(session, dict): + # herdr rejects this file as a whole too. + return MINIMAL, [f"herdr config has a non-table `session`; {LOSES_SETTINGS}"] + expected = dict(user, session=dict(session, **{KEY: False})) try: - result = tomllib.loads(merged) - ok = result["session"][KEY] is False and _without_key(result) == _without_key(user) - except (tomllib.TOMLDecodeError, KeyError, TypeError): + text = dumps(expected) + ok = tomllib.loads(text) == expected + except (TypeError, tomllib.TOMLDecodeError): 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, [] + return MINIMAL, [f"could not rewrite the herdr config faithfully; {LOSES_SETTINGS}"] + return text, [] + + +# --- herdr's own check ------------------------------------------------------ + +def herdr_rejects(path: str, herdr: str) -> Optional[str]: + """None when herdr would use the config at `path`, else herdr's diagnostics. + herdr 0.9.1 drops a file whole only on a parse or schema error, which it + reports as "config parse error: ...; using defaults"; other issues (an + unknown key, an unsupported sound file) cost only that one setting.""" + env = {k: v for k, v in os.environ.items() if k not in _SCOPE_ENV} + env["HERDR_CONFIG_PATH"] = path + res = subprocess.run([herdr, "config", "check"], env=env, capture_output=True, text=True, timeout=30) + out = (res.stdout + res.stderr).strip() + if res.returncode == 0 or not ("parse error" in out or "using defaults" in out): + return None + return out + + +# --- output ----------------------------------------------------------------- + +def _same_file(a: str, b: str) -> bool: + try: + return os.path.samefile(a, b) + except OSError: + return os.path.realpath(a) == os.path.realpath(b) + + +def output_path(source: str) -> str: + """Beside the source: OUTPUT_NAME, unless that is the source itself or a + file this module did not write for this source; then a per-source name.""" + d = os.path.dirname(source) + tag = hashlib.sha256(os.path.realpath(source).encode()).hexdigest()[:12] + for name in (OUTPUT_NAME, f"mam-server-config-{tag}.toml"): + target = os.path.join(d, name) + if _same_file(target, source): + continue + if os.path.lexists(target) and _generated_source(target) not in ("", source): + continue + return target + raise OSError(f"no free output name beside {source}") def _write_atomic(path: str, text: str, mode: int) -> None: @@ -149,7 +220,28 @@ def _write_atomic(path: str, text: str, mode: int) -> None: raise -def write_server_config(env: Mapping[str, str], fallback_dir: Optional[str] = None) -> Tuple[Optional[str], List[str]]: +def _write_checked(target: str, source: str, body: str, mode: int, herdr: Optional[str], + warnings: List[str]) -> bool: + """Write body (then MINIMAL if herdr rejects it); True when herdr accepts + what is on disk (or there is no herdr to ask).""" + for candidate in ((body, MINIMAL) if body != MINIMAL else (MINIMAL,)): + _write_atomic(target, f"{MARKER}\n{SOURCE_PREFIX}{source}\n{candidate}", mode) + if not herdr: + return True + try: + why = herdr_rejects(target, herdr) + except (OSError, subprocess.SubprocessError) as e: + warnings.append(f"cannot run `{herdr} config check` ({e}); the config is unchecked") + return True + if why is None: + return True + tail = "" if candidate == MINIMAL else f"; {LOSES_SETTINGS}" + warnings.append(f"herdr rejects the config built from {source}: {why}{tail}") + return False + + +def write_server_config(env: Mapping[str, str], fallback_dir: Optional[str] = None, + herdr: 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] = [] @@ -163,22 +255,22 @@ def write_server_config(env: Mapping[str, str], fallback_dir: Optional[str] = No 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}") + warnings.append(f"cannot read herdr config {source}: {e}; {LOSES_SETTINGS}") 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 + target = output_path(source) + if _write_checked(target, source, body, mode, herdr, warnings): + return target, warnings + return None, warnings except OSError as e: - warnings.append(f"cannot write {target}: {e}") + warnings.append(f"cannot write beside {source}: {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 + if _write_checked(target, source, body, mode, herdr, warnings): + 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 @@ -189,8 +281,9 @@ def main(argv=None) -> int: 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") + sc.add_argument("--herdr", default=None, help="the real herdr binary, to `config check` the result") args = p.parse_args(argv) - path, warnings = write_server_config(os.environ, args.fallback_dir) + path, warnings = write_server_config(os.environ, args.fallback_dir, args.herdr) for w in warnings: print(f"WARNING: {w}", file=sys.stderr) if not path: diff --git a/tests/test_issue45_herdr_config.py b/tests/test_issue45_herdr_config.py index 3ee1429..bb8a066 100644 --- a/tests/test_issue45_herdr_config.py +++ b/tests/test_issue45_herdr_config.py @@ -30,20 +30,32 @@ CASES = [ ("missing", None, True, False), ("empty", "", True, False), ("broken", "[session]\n[session]\n", False, True), + ("session_not_a_table", "session = 1\n[ui]\nconfirm_close = false\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), + # valid spellings a line edit cannot reach: the rest must survive (review R2) + ("dotted_key", "session.resume_agents_on_restore = true\n[ui]\nconfirm_close = false\n", True, False), + ("inline_table", "session = { resume_agents_on_restore = true }\n[ui]\nconfirm_close = false\n", True, False), + ("header_inside_string", 'banner = """\n[session]\n"""\n[ui]\nconfirm_close = false\n', True, False), + ("value_kinds", 'top = "a\\"b\\u007f\\u00e9"\n"odd key" = [1, 2.5, inf, true]\n' + 'when = 1979-05-27T07:32:00Z\n[ui.sound]\npath = "s/done.wav"\n' + '[[keys.bind]]\nkey = "a"\n[[keys.bind]]\nkey = "b"\n', True, False), ] def _without_key(d): - return hc._without_key(d) + d = dict(d) + session = dict(d.get("session") or {}) + session.pop("resume_agents_on_restore", None) + if session: + d["session"] = session + else: + d.pop("session", None) + return d @pytest.mark.parametrize("name,user,kept,warned", CASES, ids=[c[0] for c in CASES]) @@ -58,17 +70,10 @@ def test_merge_always_parses_with_restore_off(name, user, kept, warned): 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. +def test_a_non_bool_restore_value_is_replaced(): + # herdr would reject the whole file for this; the key is ours to set anyway. 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 == [] + assert tomllib.loads(text) == {"session": {"resume_agents_on_restore": False}} and warnings == [] def test_without_tomllib_restore_is_still_off(monkeypatch): @@ -187,14 +192,101 @@ def test_cli_prints_the_path(tmp_path): @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): + if kept and user: # a source herdr itself takes must stay one it takes + src = tmp_path / "source.toml" + src.write_text(user) + assert hc.herdr_rejects(str(src), HERDR) is None 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 + assert hc.herdr_rejects(str(cfg), HERDR) is None + + +@pytest.mark.skipif(not HERDR, reason="herdr is not installed") +def test_herdr_rejects_tells_schema_errors_from_harmless_issues(tmp_path): + bad = tmp_path / "bad.toml" + bad.write_text('[ui]\nconfirm_close = "no"\n') + assert "using defaults" in hc.herdr_rejects(str(bad), HERDR) + soft = tmp_path / "soft.toml" + soft.write_text('made_up = 1\n[ui.sound]\npath = "x.wav"\n') + assert hc.herdr_rejects(str(soft), HERDR) is None + + +def _cli(tmp_path, source: Path, *extra) -> subprocess.CompletedProcess: + env = {k: v for k, v in os.environ.items() if not k.startswith("HERDR_") and k != "XDG_CONFIG_HOME"} + env.update(HOME=str(tmp_path), HERDR_CONFIG_PATH=str(source), + PYTHONPATH=str(REPO_ROOT / ".agents" / "skills")) + return subprocess.run([sys.executable, "-m", "lib_py.herdr_config", "server-config", *extra], + capture_output=True, text=True, env=env) + + +@pytest.mark.skipif(not HERDR, reason="herdr is not installed") +@pytest.mark.parametrize("body,kept", [ + (b"[ui]\nconfirm_close = false\n", True), # control: valid, kept + (b'[ui]\nconfirm_close = "no"\n', False), # tomllib yes, herdr's schema no (review R1) + (b"session = 1\n[ui]\nconfirm_close = false\n", False), # non-table session (R1) + (b"[ui]\nconfirm_close = false\n# \xff\xfe\n", False), # not UTF-8 (R1) +], ids=["valid", "schema-error", "session-scalar", "invalid-utf8"]) +def test_cli_always_hands_herdr_a_restore_off_config_it_accepts(tmp_path, body, kept): + source = tmp_path / "custom.toml" + source.write_bytes(body) + res = _cli(tmp_path, source, "--herdr", HERDR) + assert res.returncode == 0, res.stderr + out = Path(res.stdout.strip()) + assert out.parent == tmp_path and out != source + assert hc.herdr_rejects(str(out), HERDR) is None + parsed = tomllib.loads(out.read_text()) + assert parsed["session"]["resume_agents_on_restore"] is False + if kept: + assert parsed["ui"]["confirm_close"] is False and res.stderr == "" + else: + assert parsed == {"session": {"resume_agents_on_restore": False}} and "WARNING" in res.stderr + assert source.read_bytes() == body + + +@pytest.mark.skipif(not HERDR, reason="herdr is not installed") +def test_a_config_herdr_rejects_is_never_left_in_place(tmp_path): + # the merge itself is fine for tomllib; only herdr's schema says no + source = tmp_path / "config.toml" + source.write_text('[ui]\nconfirm_close = "no"\n[session]\nresume_agents_on_restore = true\n') + path, warnings = hc.write_server_config({"HERDR_CONFIG_PATH": str(source)}, herdr=HERDR) + assert tomllib.loads(Path(path).read_text()) == {"session": {"resume_agents_on_restore": False}} + assert any("herdr rejects" in w for w in warnings), warnings + + +# --- the user's file is never the output (review R3) ----------------------- + +def test_source_named_like_the_output_is_not_overwritten(tmp_path): + source = tmp_path / hc.OUTPUT_NAME + source.write_bytes(b"[ui]\nconfirm_close = false\n") + path, _ = hc.write_server_config({"HERDR_CONFIG_PATH": str(source)}) + assert source.read_bytes() == b"[ui]\nconfirm_close = false\n" + assert Path(path).parent == tmp_path and Path(path) != source + assert tomllib.loads(Path(path).read_text())["ui"]["confirm_close"] is False + # and the next start finds the same file again + assert hc.write_server_config({"HERDR_CONFIG_PATH": str(source)})[0] == path + + +def test_an_alias_of_the_source_is_not_overwritten(tmp_path): + source = tmp_path / "config.toml" + source.write_bytes(b"[ui]\nconfirm_close = false\n") + (tmp_path / hc.OUTPUT_NAME).symlink_to(source) + path, _ = hc.write_server_config({"HERDR_CONFIG_PATH": str(source)}) + assert source.read_bytes() == b"[ui]\nconfirm_close = false\n" + assert os.path.islink(tmp_path / hc.OUTPUT_NAME) + assert not os.path.samefile(path, source) + + +def test_a_foreign_file_with_the_output_name_is_kept(tmp_path): + source = tmp_path / "config.toml" + source.write_text("[ui]\nconfirm_close = false\n") + foreign = tmp_path / hc.OUTPUT_NAME + foreign.write_bytes(b"# someone else's\n") + path, _ = hc.write_server_config({"HERDR_CONFIG_PATH": str(source)}) + assert foreign.read_bytes() == b"# someone else's\n" and Path(path) != foreign + # a file generated for another source is someone else's too + other = tmp_path / "other.toml" + other.write_text("") + other_path, _ = hc.write_server_config({"HERDR_CONFIG_PATH": str(other)}) + assert other_path not in (path, str(foreign)) + assert hc.user_config_path({"HERDR_CONFIG_PATH": path}) == str(source)