feat(hooks): own the notmuch hooks, and keep sent mail out of the inbox

The post-new hook and its rule store move here from the companion mailctl
project, which is being retired. Nothing else was shared between the two, so
this is a plain move: mailrules.py is stdlib-only and post-new imports only
it.

With that in hand, the hook learns the one thing it could not know before.
notmuch's new.tags applies `inbox` to every file it indexes, and it cannot
tell an arrival from the copy this application files into a sent folder after
a send, so sent mail turned up in the inbox view and in any hand-typed
tag:inbox search. Drafts arrived the same way, through the composer's
autosave. 786 messages were affected on the developer's own index.

qtmaildirconf.py reads the sent and drafts folders out of qtmaildir.conf, so
adding an account fixes itself. Reading the application's own config is not
the cross-repo coupling it would have been last week: this repo owns the hook
now.

Three properties are load-bearing:

  - it is NOT a relaxation of PROTECTED_REMOVALS, which is about a rule
    removing `inbox` from mail whose provenance the hook cannot judge. Here
    the provenance is the file's own path, and `inbox` was never true of it.
  - only `inbox`. maildir.synchronize_flags is true, so removing `unread`
    would rewrite Maildir filenames and reach the server on the next mbsync.
  - an empty folder list means NOTHING, never an empty query, which notmuch
    reads as "match everything". A system with no qtmaildir config must be
    left alone rather than have every new message stripped.

Trash is deliberately not in the list: Delete leaves `inbox` on a trashed
message so Restore can put it back where it came from.

The three Python suites run under ctest rather than beside it as scripts
someone remembers to run, since this code tags real mail unattended on every
sync.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Q2koFevoSxTLhfexJTZWQd

1 files changed, 278 insertions, 0 deletions

diff --git a/assets/hooks/test_mailrules.py b/assets/hooks/test_mailrules.py
new file mode 100755
index 0000000..b1f31c9
--- /dev/null
+++ b/assets/hooks/test_mailrules.py
@@ -0,0 +1,278 @@
+#!/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 mailrules.py, the shared tagging-rule store.
+
+The risk in this file is the format, not the notmuch calls: a rule that
+silently loses a field on save, or one that sorts into the wrong stage,
+mis-tags real mail on the next sync and does it quietly.
+
+Run: ./test_mailrules.py
+"""
+
+import json
+import tempfile
+from pathlib import Path
+
+import mailrules
+
+
+def write_rules(tmp, payload):
+    path = Path(tmp) / "rules.json"
+    path.write_text(json.dumps(payload))
+    return path
+
+
+def test_loads_a_rule():
+    with tempfile.TemporaryDirectory() as tmp:
+        path = write_rules(tmp, {
+            "version": 1,
+            "rules": [
+                {
+                    "id": "notify-forge",
+                    "stage": 50,
+                    "enabled": True,
+                    "add": ["notify/forge"],
+                    "remove": [],
+                    "query": "from:notifications@example.com",
+                    "note": "All repositories, not one project.",
+                }
+            ],
+        })
+        store = mailrules.load(path)
+        assert store.warnings == [], store.warnings
+        assert len(store.rules) == 1
+        rule = store.rules[0]
+        assert rule.id == "notify-forge"
+        assert rule.stage == 50
+        assert rule.enabled is True
+        assert rule.add == ["notify/forge"]
+        assert rule.remove == []
+        assert rule.query == "from:notifications@example.com"
+        assert rule.note == "All repositories, not one project."
+
+
+def test_defaults_are_applied():
+    """stage, enabled, remove and note are all optional in the file."""
+    with tempfile.TemporaryDirectory() as tmp:
+        path = write_rules(tmp, {
+            "version": 1,
+            "rules": [{"id": "minimal", "add": ["x"],
+                       "query": "from:someone@example.com"}],
+        })
+        store = mailrules.load(path)
+        assert store.warnings == [], store.warnings
+        rule = store.rules[0]
+        assert rule.stage == 50
+        assert rule.enabled is True
+        assert rule.remove == []
+        assert rule.note == ""
+
+
+def test_a_bad_rule_is_dropped_and_the_rest_survive():
+    """One malformed rule must not stop the others. The hook runs every ten
+    minutes on real mail; losing all tagging because of one typo is worse
+    than losing one rule."""
+    with tempfile.TemporaryDirectory() as tmp:
+        path = write_rules(tmp, {
+            "version": 1,
+            "rules": [
+                {"id": "good", "add": ["x"], "query": "from:a@example.com"},
+                {"id": "no-query", "add": ["y"]},
+                {"id": "no-tags", "query": "from:b@example.com"},
+                {"id": "bad id!", "add": ["z"], "query": "from:c@example.com"},
+                {"add": ["w"], "query": "from:d@example.com"},
+            ],
+        })
+        store = mailrules.load(path)
+        assert [r.id for r in store.rules] == ["good"]
+        assert len(store.warnings) == 4, store.warnings
+        joined = " ".join(store.warnings)
+        assert "no-query" in joined
+        assert "no-tags" in joined
+        assert "bad id!" in joined
+
+
+def test_duplicate_ids_keep_the_first():
+    with tempfile.TemporaryDirectory() as tmp:
+        path = write_rules(tmp, {
+            "version": 1,
+            "rules": [
+                {"id": "dup", "add": ["first"], "query": "from:a@example.com"},
+                {"id": "dup", "add": ["second"], "query": "from:b@example.com"},
+            ],
+        })
+        store = mailrules.load(path)
+        assert len(store.rules) == 1
+        assert store.rules[0].add == ["first"]
+        assert any("dup" in w for w in store.warnings)
+
+
+def test_a_missing_file_is_empty_not_an_error():
+    """qtmaildir must open on a machine that has never written this file."""
+    with tempfile.TemporaryDirectory() as tmp:
+        store = mailrules.load(Path(tmp) / "absent.json")
+        assert store.rules == []
+        assert store.warnings == []
+        assert store.missing is True
+
+
+def test_unparseable_json_warns_and_yields_no_rules():
+    with tempfile.TemporaryDirectory() as tmp:
+        path = Path(tmp) / "rules.json"
+        path.write_text("{not json")
+        store = mailrules.load(path)
+        assert store.rules == []
+        assert len(store.warnings) == 1
+        assert store.failed is True
+
+
+def test_a_newer_format_version_is_refused():
+    """Guessing at semantics a later version defined is how a rule silently
+    changes meaning. Refuse instead."""
+    with tempfile.TemporaryDirectory() as tmp:
+        path = write_rules(tmp, {
+            "version": 2,
+            "rules": [{"id": "x", "add": ["a"], "query": "from:a@example.com"}],
+        })
+        store = mailrules.load(path)
+        assert store.rules == []
+        assert store.failed is True
+        assert any("version" in w for w in store.warnings)
+
+
+def test_ordered_sorts_by_stage_then_file_position():
+    """Account tags must run before topic rules. Ties keep file order, so
+    the file still reads as a sequence."""
+    with tempfile.TemporaryDirectory() as tmp:
+        path = write_rules(tmp, {
+            "version": 1,
+            "rules": [
+                {"id": "topic-b", "stage": 50, "add": ["b"],
+                 "query": "from:b@example.com"},
+                {"id": "account", "stage": 10, "add": ["acct"],
+                 "query": "path:\"work/**\""},
+                {"id": "topic-a", "stage": 50, "add": ["a"],
+                 "query": "from:a@example.com"},
+            ],
+        })
+        store = mailrules.load(path)
+        assert [r.id for r in mailrules.ordered(store.rules)] == [
+            "account", "topic-b", "topic-a"]
+
+
+def test_ordered_skips_disabled_rules():
+    with tempfile.TemporaryDirectory() as tmp:
+        path = write_rules(tmp, {
+            "version": 1,
+            "rules": [
+                {"id": "on", "add": ["a"], "query": "from:a@example.com"},
+                {"id": "off", "add": ["b"], "query": "from:b@example.com",
+                 "enabled": False},
+            ],
+        })
+        store = mailrules.load(path)
+        assert [r.id for r in mailrules.ordered(store.rules)] == ["on"]
+        # The disabled rule is still LOADED, so a UI can show and re-enable it.
+        assert [r.id for r in store.rules] == ["on", "off"]
+
+
+def test_save_round_trips_unknown_fields():
+    """The neutrality guarantee. If this tool strips a field qtmaildir
+    added, the file is this tool's file that qtmaildir may read."""
+    with tempfile.TemporaryDirectory() as tmp:
+        path = write_rules(tmp, {
+            "version": 1,
+            "future_top_level": {"set_by": "another tool"},
+            "rules": [{
+                "id": "keeper",
+                "add": ["x"],
+                "query": "from:a@example.com",
+                "future_field": [1, 2, 3],
+            }],
+        })
+        store = mailrules.load(path)
+        assert store.rules[0].unknown == {"future_field": [1, 2, 3]}
+
+        mailrules.save(store, path)
+
+        raw = json.loads(path.read_text())
+        assert raw["future_top_level"] == {"set_by": "another tool"}
+        assert raw["rules"][0]["future_field"] == [1, 2, 3]
+        assert raw["rules"][0]["id"] == "keeper"
+        assert raw["version"] == 1
+
+
+def test_save_is_atomic():
+    """A reader must never see a half-written file: the hook runs every ten
+    minutes and a truncated read would be a failed sync."""
+    with tempfile.TemporaryDirectory() as tmp:
+        path = Path(tmp) / "rules.json"
+        store = mailrules.Store(rules=[
+            mailrules.Rule(id="a", query="from:a@example.com", add=["x"])])
+        mailrules.save(store, path)
+        # The temp file the write went through must not be left behind.
+        assert [p.name for p in Path(tmp).iterdir()] == ["rules.json"]
+        assert json.loads(path.read_text())["rules"][0]["id"] == "a"
+
+
+def test_save_creates_the_directory():
+    with tempfile.TemporaryDirectory() as tmp:
+        path = Path(tmp) / "nested" / "rules.json"
+        mailrules.save(mailrules.Store(), path)
+        assert path.exists()
+        assert json.loads(path.read_text()) == {"version": 1, "rules": []}
+
+
+def test_scoped_query_parenthesises_the_rule():
+    """Without the parentheses `tag:new and a or b` binds as
+    `(tag:new and a) or b`, and the rule matches every message in the corpus
+    satisfying b rather than only new arrivals. Several real rules are a
+    disjunction of senders, so this is the difference between tagging four
+    messages and tagging four thousand."""
+    rule = mailrules.Rule(
+        id="disjunction",
+        query="from:a@example.com or from:b@example.com",
+        add=["promo"])
+    assert mailrules.scoped_query(rule, "tag:new") == (
+        "tag:new and (from:a@example.com or from:b@example.com)")
+
+
+def test_scoped_query_with_no_scope_is_the_bare_query():
+    """A dry run counts against the whole corpus, which is what makes the
+    same rule answer 'what would this tag on arrival' and 'what does this
+    match in all my mail'."""
+    rule = mailrules.Rule(id="x", query="from:a@example.com", add=["y"])
+    assert mailrules.scoped_query(rule, None) == "from:a@example.com"
+    assert mailrules.scoped_query(rule, "") == "from:a@example.com"
+
+
+def test_tag_arguments():
+    rule = mailrules.Rule(id="x", query="from:a@example.com",
+                          add=["one", "two"], remove=["three"])
+    assert mailrules.tag_arguments(rule) == ["+one", "+two", "-three"]
+
+
+def run_all():
+    for name, fn in sorted(globals().items()):
+        if name.startswith("test_") and callable(fn):
+            fn()
+            print(f"ok  {name}")
+
+
+if __name__ == "__main__":
+    run_all()
+    print("\nall passed")