Initial commit: agent-safe notmuch wrapperHEADmaster

mailctl is a deliberately narrow CLI wrapper around notmuch, built so an
AI coding agent can search and organize local mail with no ability to
send. There is no SMTP, reply, or compose code path in the tool.

Safety model:
- reads run freely, mutations are gated
- tag changes are dry-run until --apply
- cross-account mutations need an explicit --all-accounts
- destructive tags need --apply and --confirm-destructive
- bulk mutations are capped by --max-messages
- every applied mutation is audited to a local log

Accounts are not in the source. They load from
~/.config/mailctl/accounts.json and are validated against both the
schema and the actual maildirs on disk at import time, so a typo cannot
produce a query matching nothing or a draft under the wrong identity.

Ships with mailsync.sh (a separate mbsync + notmuch new driver, meant
for cron) and mail-organize, a Claude Code skill that makes mailctl the
only sanctioned interface to the user's mail.

Licensed GPLv2-only.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

1 files changed, 252 insertions, 0 deletions

diff --git a/test_mailctl.py b/test_mailctl.py
new file mode 100755
index 0000000..c6a50b9
--- /dev/null
+++ b/test_mailctl.py
@@ -0,0 +1,252 @@
+#!/usr/bin/env python3
+#
+# Copyright (C) 2026 Danilo M. <danix@danix.xyz>
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License version 2 as
+# published by the Free Software Foundation.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, write to the Free Software
+# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
+"""Self-checks for the bits of mailctl with real logic in them:
+
+  - the senders address merge: notmuch dedupes on name-addr, so one address
+    appears once per display name it ever used, and we merge on the address.
+  - subject_terms: tokenizing/stopword-filtering subject lines, counting each
+    term once per subject.
+  - validate_accounts: the config gate. Every rejection case here is a typo
+    that would otherwise silently produce a query matching nothing, or a
+    draft written under the wrong identity.
+
+Run: ./test_mailctl.py
+"""
+
+import io
+import json
+import os
+import tempfile
+from contextlib import redirect_stdout
+from pathlib import Path
+from unittest.mock import patch
+
+# mailctl validates its account config at import time, so a valid one has to
+# exist before the import below. Point it at a throwaway maildir tree.
+_tmp = tempfile.TemporaryDirectory()
+FIXTURE_ROOT = Path(_tmp.name)
+(FIXTURE_ROOT / "mail" / "acct-a" / "Drafts").mkdir(parents=True)
+(FIXTURE_ROOT / "mail" / "acct-b").mkdir(parents=True)
+FIXTURE_CONFIG = FIXTURE_ROOT / "accounts.json"
+FIXTURE_CONFIG.write_text(json.dumps({"accounts": {
+    "acct-a": {"maildir": "acct-a", "address": "a@example.org",
+               "drafts": "Drafts"},
+    "acct-b": {"maildir": "acct-b", "address": "b@example.org",
+               "drafts": None},
+}}))
+os.environ["MAILCTL_CONFIG"] = str(FIXTURE_CONFIG)
+os.environ["MAILCTL_MAIL_ROOT"] = str(FIXTURE_ROOT / "mail")
+
+import mailctl  # noqa: E402  (must follow the env vars above)
+
+MAIL = FIXTURE_ROOT / "mail"
+
+
+def senders_json(raw, **overrides):
+    """Run cmd_senders against a canned notmuch reply, return parsed JSON."""
+    opts = {"query": "*", "account": None, "top": None, "json": True}
+    opts.update(overrides)
+    args = type("Args", (), opts)()
+    buf = io.StringIO()
+    with patch.object(mailctl, "run_notmuch", return_value=json.dumps(raw)), \
+            redirect_stdout(buf):
+        mailctl.cmd_senders(args)
+    return json.loads(buf.getvalue())
+
+
+def test_merges_on_address_keeping_longest_name():
+    got = senders_json([
+        {"name": "", "address": "a@x", "count": 112},
+        {"name": "DPReview", "address": "a@x", "count": 353},
+        {"name": "B", "address": "b@x", "count": 5},
+    ])
+    assert got == [
+        {"address": "a@x", "name": "DPReview", "count": 465},
+        {"address": "b@x", "name": "B", "count": 5},
+    ], got
+
+
+def test_top_truncates_after_sorting():
+    got = senders_json([
+        {"name": "small", "address": "s@x", "count": 1},
+        {"name": "big", "address": "b@x", "count": 99},
+    ], top=1)
+    assert [a["address"] for a in got] == ["b@x"], got
+
+
+def test_empty():
+    assert senders_json([]) == []
+
+
+def test_subject_terms_counts_each_term_once_per_subject():
+    # "promo" three times in one subject must not outrank "sconto" in two
+    got = mailctl.subject_terms(["Promo promo PROMO", "sconto", "Sconto!"])
+    assert got["promo"] == 1, got
+    assert got["sconto"] == 2, got
+
+
+def test_subject_terms_drops_stopwords_digits_and_emoji():
+    got = mailctl.subject_terms(["🔥 Le offerte di oggi for you 2024 ⏳"])
+    assert set(got) == {"offerte"}, got
+
+
+def test_subject_terms_keeps_accented_words():
+    got = mailctl.subject_terms(["Località e novità"])
+    assert set(got) == {"località", "novità"}, got
+
+
+def test_subject_terms_empty():
+    assert mailctl.subject_terms([]) == {}
+    assert mailctl.subject_terms(["", "🔥"]) == {}
+
+
+def rejects(accounts_value, expect_in_message):
+    """Assert a config is refused, and that the message names the problem.
+
+    Checking the message matters as much as the exit: these fire on the user's
+    own typo, and a rejection that doesn't say which account and which field
+    is barely better than a silent wrong answer.
+    """
+    buf = io.StringIO()
+    try:
+        with patch("sys.stderr", buf):
+            mailctl.validate_accounts(accounts_value, MAIL)
+    except SystemExit as e:
+        assert e.code == 2, f"expected exit 2, got {e.code}"
+        msg = buf.getvalue()
+        assert expect_in_message in msg, f"want {expect_in_message!r} in:\n{msg}"
+        return
+    raise AssertionError(f"config was accepted but should not be: {accounts_value}")
+
+
+def good(**overrides):
+    spec = {"maildir": "acct-a", "address": "a@example.org", "drafts": "Drafts"}
+    spec.update(overrides)
+    return {"accounts": {"acct-a": spec}}
+
+
+def test_valid_config_returns_both_maps():
+    accounts, drafts = mailctl.validate_accounts(json.loads(
+        FIXTURE_CONFIG.read_text()), MAIL)
+    assert accounts == {"acct-a": ("acct-a", "a@example.org"),
+                        "acct-b": ("acct-b", "b@example.org")}, accounts
+    assert drafts == {"acct-a": "Drafts", "acct-b": None}, drafts
+
+
+def test_rejects_misspelled_maildir():
+    # the typo this whole gate exists for: scoping would match zero mail
+    rejects(good(maildir="acct-A"), "does not exist")
+
+
+def test_rejects_misspelled_drafts_dir():
+    # would write a draft into a folder mbsync never syncs back
+    rejects(good(drafts="Bozze"), "does not exist")
+
+
+def test_rejects_absolute_and_traversing_maildir():
+    rejects(good(maildir="/etc"), "plain subdirectory")
+    rejects(good(maildir="../../etc"), "plain subdirectory")
+
+
+def test_rejects_bad_address():
+    rejects(good(address="a@example"), "not a valid email address")
+    rejects(good(address="not-an-address"), "not a valid email address")
+
+
+def test_rejects_unknown_field():
+    # catches "addresss"/"maildirs" style typos instead of ignoring them
+    rejects(good(adress="a@example.org"), "unknown field")
+
+
+def test_rejects_missing_required_field():
+    spec = good()
+    del spec["accounts"]["acct-a"]["address"]
+    rejects(spec, 'missing required field "address"')
+
+
+def test_rejects_duplicate_maildir():
+    rejects({"accounts": {
+        "one": {"maildir": "acct-a", "address": "a@example.org", "drafts": None},
+        "two": {"maildir": "acct-a", "address": "b@example.org", "drafts": None},
+    }}, "is used by 2 accounts")
+
+
+def test_rejects_structural_problems():
+    rejects([], "must be a JSON object")
+    rejects({}, 'missing top-level "accounts" key')
+    rejects({"accounts": {}}, "non-empty")
+    rejects({"accounts": {"acct-a": "acct-a"}}, "must be an object")
+    rejects({"accounts": {"Acct A": good()["accounts"]["acct-a"]}},
+            "invalid name")
+
+
+def test_message_names_the_offending_account():
+    buf = io.StringIO()
+    try:
+        with patch("sys.stderr", buf):
+            mailctl.validate_accounts({"accounts": {
+                "fine": {"maildir": "acct-b", "address": "b@example.org"},
+                "broken": {"maildir": "nope", "address": "c@example.org"},
+            }}, MAIL)
+    except SystemExit:
+        pass
+    assert '"broken"' in buf.getvalue(), buf.getvalue()
+
+
+def test_drafts_null_is_allowed():
+    accounts, drafts = mailctl.validate_accounts(good(drafts=None), MAIL)
+    assert drafts == {"acct-a": None}, drafts
+
+
+def test_drafts_field_is_optional():
+    spec = good()
+    del spec["accounts"]["acct-a"]["drafts"]
+    _, drafts = mailctl.validate_accounts(spec, MAIL)
+    assert drafts == {"acct-a": None}, drafts
+
+
+def test_load_accounts_rejects_bad_json():
+    bad = FIXTURE_ROOT / "bad.json"
+    bad.write_text("{not json")
+    buf = io.StringIO()
+    try:
+        with patch("sys.stderr", buf):
+            mailctl.load_accounts(bad, MAIL)
+    except SystemExit as e:
+        assert e.code == 2
+        assert "not valid JSON" in buf.getvalue(), buf.getvalue()
+        return
+    raise AssertionError("bad JSON was accepted")
+
+
+def test_load_accounts_missing_file_explains_how_to_create_it():
+    buf = io.StringIO()
+    try:
+        with patch("sys.stderr", buf):
+            mailctl.load_accounts(FIXTURE_ROOT / "absent.json", MAIL)
+    except SystemExit as e:
+        assert e.code == 2
+        assert "no config at" in buf.getvalue(), buf.getvalue()
+        return
+    raise AssertionError("missing config was accepted")
+
+
+if __name__ == "__main__":
+    for name, fn in sorted(globals().items()):
+        if name.startswith("test_") and callable(fn):
+            fn()
+    print("ok")