1 files changed, 563 insertions, 0 deletions
diff --git a/mailctl.py b/mailctl.py
new file mode 100755
index 0000000..be4cbcf
--- /dev/null
+++ b/mailctl.py
@@ -0,0 +1,563 @@
+#!/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.
+"""
+mailctl - a deliberately narrow CLI wrapper around notmuch, meant for
+agent use (Claude Code, opencode, etc.) to search and organize mail.
+
+Design goals:
+  - No SMTP, no send, no reply, no compose. Not "discouraged", not present
+    in the code at all. An agent can't do what the tool has no code path for.
+  - Read operations (search/show/tags/count/senders/subjects) run freely, --account is
+    optional there, defaulting to a global search across all mailboxes.
+  - Mutating operations (tag) are scoped to a single account by default.
+    Touching more than one account requires the explicit --all-accounts
+    flag, there's no accidental global mutation.
+  - Tag mutations default to dry-run: they print what WOULD change and
+    require --apply to actually touch the index.
+  - Anything that looks destructive (removing 'inbox', adding 'deleted' or
+    'trash'/'spam') requires --apply AND --confirm-destructive, and gets
+    logged regardless of account scope.
+  - Every applied mutation is appended to ~/.local/state/mailctl/audit.log
+    with a timestamp, so there's a plain-text trail of what an agent changed.
+
+  - The account map (maildir names and real From addresses) is NOT in this
+    file. It loads from ~/.config/mailctl/accounts.json, override with
+    MAILCTL_CONFIG. The config is validated against its schema and against
+    the actual maildirs on disk at import time; any problem is a hard exit
+    before argparse runs, so a typo can't reach a query or a draft.
+
+This is a starting skeleton, not a finished tool. In particular:
+  - The audit log doesn't distinguish "run by agent" vs "run by you",
+    worth adding an --actor tag if that distinction matters to you.
+"""
+
+import argparse
+import json
+import os
+import re
+import socket
+import subprocess
+import sys
+import time
+from collections import Counter
+from datetime import datetime, timezone
+from email.message import EmailMessage
+from email.utils import make_msgid, formatdate
+from pathlib import Path
+
+AUDIT_LOG = Path.home() / ".local" / "state" / "mailctl" / "audit.log"
+MAIL_ROOT = Path(os.environ.get("MAILCTL_MAIL_ROOT", Path.home() / "Mail"))
+
+CONFIG_PATH = Path(
+    os.environ.get("MAILCTL_CONFIG",
+                   Path.home() / ".config" / "mailctl" / "accounts.json")
+)
+
+# The account map is real addresses and maildir names, so it lives outside the
+# repo. Same closed-set guarantee as when it was hardcoded: it is validated
+# against the schema AND against the disk at import time, and mailctl refuses
+# to run at all if anything is off. A typo can't silently produce a path
+# filter that matches nothing, or a draft sent from the wrong identity.
+ACCOUNT_KEY_RE = re.compile(r"^[a-z0-9][a-z0-9._-]*$")
+ADDRESS_RE = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$")
+ACCOUNT_FIELDS = {"maildir", "address", "drafts"}
+
+
+def config_error(problem, *hints):
+    """Refuse to run. Config problems are typos in the user's own file, so the
+    message names the file, the problem, and how to fix it."""
+    print(f"mailctl: bad config at {CONFIG_PATH}\n  {problem}", file=sys.stderr)
+    for h in hints:
+        print(f"  {h}", file=sys.stderr)
+    sys.exit(2)
+
+
+def validate_accounts(raw, mail_root):
+    """Check a parsed config against the schema and the filesystem.
+
+    Returns (accounts, drafts_subdir) in the same shape the rest of the tool
+    used when these were module constants. Calls config_error (exits) on the
+    first problem found.
+    """
+    if not isinstance(raw, dict):
+        config_error(f"top level must be a JSON object, got {type(raw).__name__}",
+                     'example: {"accounts": {"work": {...}}}')
+    accounts_raw = raw.get("accounts")
+    if accounts_raw is None:
+        config_error('missing top-level "accounts" key',
+                     f"found instead: {', '.join(sorted(raw)) or '(empty file)'}")
+    if not isinstance(accounts_raw, dict) or not accounts_raw:
+        config_error('"accounts" must be a non-empty JSON object')
+
+    accounts, drafts = {}, {}
+    for key, spec in accounts_raw.items():
+        where = f'account "{key}"'
+        if not ACCOUNT_KEY_RE.match(key):
+            config_error(f"{where}: invalid name",
+                         "names are lowercase letters, digits, . _ - and must "
+                         "not start with a separator")
+        if not isinstance(spec, dict):
+            config_error(f"{where}: must be an object, got {type(spec).__name__}")
+
+        unknown = set(spec) - ACCOUNT_FIELDS
+        if unknown:
+            config_error(f"{where}: unknown field(s) {', '.join(sorted(unknown))}",
+                         f"valid fields: {', '.join(sorted(ACCOUNT_FIELDS))}")
+        for required in ("maildir", "address"):
+            if required not in spec:
+                config_error(f"{where}: missing required field \"{required}\"")
+
+        maildir = spec["maildir"]
+        if not isinstance(maildir, str) or not maildir:
+            config_error(f"{where}: \"maildir\" must be a non-empty string")
+        if maildir.startswith("/") or ".." in Path(maildir).parts:
+            config_error(f"{where}: \"maildir\" must be a plain subdirectory "
+                         f"of {mail_root}, got {maildir!r}")
+        if not (mail_root / maildir).is_dir():
+            config_error(f"{where}: maildir {mail_root / maildir} does not exist",
+                         "check the spelling against what mbsync actually synced")
+
+        address = spec["address"]
+        if not isinstance(address, str) or not ADDRESS_RE.match(address):
+            config_error(f"{where}: \"address\" is not a valid email address: "
+                         f"{address!r}")
+
+        drafts_dir = spec.get("drafts")
+        if drafts_dir is not None:
+            if not isinstance(drafts_dir, str) or not drafts_dir:
+                config_error(f"{where}: \"drafts\" must be a non-empty string "
+                             "or null (null = account has no synced Drafts)")
+            if not (mail_root / maildir / drafts_dir).is_dir():
+                config_error(
+                    f"{where}: Drafts maildir "
+                    f"{mail_root / maildir / drafts_dir} does not exist",
+                    "it must match the mbsync Patterns for this account "
+                    '(e.g. "[Gmail]/Bozze" vs "Drafts"), or set it to null')
+
+        accounts[key] = (maildir, address)
+        drafts[key] = drafts_dir
+
+    dupes = Counter(a for a, _ in accounts.values())
+    for maildir, n in dupes.items():
+        if n > 1:
+            config_error(f"maildir {maildir!r} is used by {n} accounts",
+                         "each account needs its own maildir, or scoping "
+                         "silently matches the wrong mail")
+    return accounts, drafts
+
+
+def load_accounts(path=None, mail_root=None):
+    path = Path(path) if path else CONFIG_PATH
+    mail_root = mail_root or MAIL_ROOT
+    try:
+        text = path.read_text()
+    except FileNotFoundError:
+        print(f"mailctl: no config at {path}\n"
+              "  Create it with one entry per account:\n"
+              '    {"accounts": {"work": {"maildir": "work-mbsync-dir",\n'
+              '                           "address": "you@example.org",\n'
+              '                           "drafts": "Drafts"}}}\n'
+              '  "drafts" may be null if that account has no synced Drafts '
+              "folder.\n"
+              "  Override the location with MAILCTL_CONFIG=/path/to/file.",
+              file=sys.stderr)
+        sys.exit(2)
+    except OSError as e:
+        config_error(f"cannot read: {e}")
+    try:
+        raw = json.loads(text)
+    except json.JSONDecodeError as e:
+        config_error(f"not valid JSON: {e}")
+    return validate_accounts(raw, mail_root)
+
+ACCOUNTS, DRAFTS_SUBDIR = load_accounts()
+
+DESTRUCTIVE_TAGS = {"deleted", "trash", "spam"}
+PROTECTED_REMOVALS = {"inbox"}
+DEFAULT_MAX_MESSAGES = 5000  # abort a tag --apply touching more than this unless --max-messages raises it
+
+# Function words filtered out of `subjects` term counts. Italian and English
+# only, because that's what this mail is; not a general-purpose stopword list.
+# ponytail: a hand-rolled set, no NLP dependency for what is word counting.
+# Add words here when a useless term keeps topping a real query.
+STOPWORDS = {
+    # Italian
+    "di", "il", "la", "le", "lo", "gli", "un", "una", "uno", "del", "della",
+    "dei", "delle", "dello", "al", "alla", "ai", "alle", "allo", "da", "dal",
+    "dalla", "in", "nel", "nella", "con", "su", "sul", "sulla", "per", "tra",
+    "fra", "che", "chi", "cui", "non", "come", "piu", "più", "anche", "sono",
+    "sei", "hai", "ha", "ho", "essere", "questo", "questa", "questi",
+    "queste", "quello", "quella", "tuo", "tua", "tuoi", "tue", "mio", "mia",
+    "ti", "si", "ci", "ne", "se", "ma", "così", "cosa", "tutto", "tutti",
+    "già", "ora", "oggi", "solo", "ed", "od", "sta", "fa", "qui", "te",
+    "lì", "là", "dove", "quando", "molto", "ancora", "sempre", "poi",
+    # English (words under 2 chars are dropped by length, not listed here)
+    "the", "an", "of", "to", "on", "at", "for", "and", "or", "but",
+    "is", "are", "was", "were", "be", "been", "your", "you", "my", "it", "its",
+    "this", "that", "these", "those", "with", "from", "by", "as", "we", "our",
+    "has", "have", "had", "will", "can", "not", "new", "now", "all", "more",
+    "re", "fwd",
+}
+
+
+def run_notmuch(args, capture=True):
+    cmd = ["notmuch"] + args
+    result = subprocess.run(cmd, capture_output=capture, text=True)
+    if result.returncode != 0:
+        print(f"notmuch error: {result.stderr.strip()}", file=sys.stderr)
+        sys.exit(result.returncode)
+    return result.stdout if capture else None
+
+
+def scoped_query(query, account):
+    """Wrap a user query with a path: filter for the given account key.
+    Returns the query unchanged if account is None (global scope)."""
+    if account is None:
+        return query
+    if account not in ACCOUNTS:
+        print(f"Unknown account '{account}'. Valid accounts: "
+              f"{', '.join(ACCOUNTS)}", file=sys.stderr)
+        sys.exit(1)
+    subdir, _address = ACCOUNTS[account]
+    return f'path:"{subdir}/**" and ({query})'
+
+
+def account_choices_help():
+    return "one of: " + ", ".join(ACCOUNTS)
+
+
+def cmd_search(args):
+    q = scoped_query(args.query, args.account)
+    out = run_notmuch(["search", "--format=json", "--output=summary", q])
+    results = json.loads(out)
+    if args.json:
+        print(json.dumps(results, indent=2))
+        return
+    for r in results:
+        date = r.get("date_relative", "")
+        frm = r.get("authors", "")
+        subj = r.get("subject", "(no subject)")
+        tags = ",".join(r.get("tags", []))
+        print(f"{date:>12}  {frm:<30.30}  {subj:<60.60}  [{tags}]")
+    scope = args.account or "all accounts"
+    print(f"\n{len(results)} thread(s)  [scope: {scope}]", file=sys.stderr)
+
+
+def cmd_show(args):
+    out = run_notmuch(["show", "--format=json", args.query])
+    print(out)
+
+
+def cmd_tags(args):
+    out = run_notmuch(["search", "--output=tags", "*"])
+    print(out.strip())
+
+
+def cmd_count(args):
+    q = scoped_query(args.query, args.account)
+    out = run_notmuch(["count", q])
+    scope = args.account or "all accounts"
+    print(f"{out.strip()}  [scope: {scope}]")
+
+
+def cmd_senders(args):
+    q = scoped_query(args.query, args.account)
+    out = run_notmuch(["address", "--output=sender", "--output=count",
+                       "--format=json", q])
+    # notmuch dedupes on name-addr, so one address shows up once per display
+    # name it ever used. Merge on the address, keeping the longest name seen.
+    merged = {}
+    for a in json.loads(out):
+        e = merged.setdefault(a["address"], {"address": a["address"],
+                                             "name": "", "count": 0})
+        e["count"] += a["count"]
+        if len(a["name"]) > len(e["name"]):
+            e["name"] = a["name"]
+    results = sorted(merged.values(), key=lambda a: a["count"], reverse=True)
+    if args.top:
+        results = results[:args.top]
+    if args.json:
+        print(json.dumps(results, indent=2))
+        return
+    for a in results:
+        print(f"{a['count']:>7}  {a['address']:<45.45}  {a['name']:.35}")
+    scope = args.account or "all accounts"
+    print(f"\n{len(results)} sender(s)  [scope: {scope}]", file=sys.stderr)
+
+
+def subject_terms(subjects):
+    """Count word frequency across subject lines, ignoring stopwords.
+
+    Returns a Counter. Mail here is mixed Italian/English and heavy on emoji
+    and marketing punctuation, so tokens are lowercased word characters only
+    (emoji and '...' fall out), and single characters plus pure digits are
+    dropped as noise.
+    """
+    counts = Counter()
+    for subj in subjects:
+        seen = set()
+        for word in re.findall(r"\w+", subj.lower()):
+            if len(word) < 2 or word.isdigit() or word in STOPWORDS:
+                continue
+            seen.add(word)
+        # count each term once per subject, so one shouty repeated word in a
+        # single subject can't outrank a term used across many messages
+        counts.update(seen)
+    return counts
+
+
+def cmd_subjects(args):
+    q = scoped_query(args.query, args.account)
+    out = run_notmuch(["search", "--format=json", "--output=summary", q])
+    results = json.loads(out)
+    subjects = [r.get("subject") or "" for r in results]
+    terms = subject_terms(subjects).most_common(args.top)
+
+    if args.json:
+        print(json.dumps([{"term": t, "threads": n} for t, n in terms],
+                         indent=2))
+        return
+    total = len(subjects)
+    for term, n in terms:
+        pct = 100 * n / total if total else 0
+        print(f"{n:>7}  {pct:>5.1f}%  {term}")
+    scope = args.account or "all accounts"
+    print(f"\n{len(terms)} term(s) across {total} thread(s)  "
+          f"[scope: {scope}]", file=sys.stderr)
+
+
+def cmd_tag(args):
+    adds = args.add or []
+    removes = args.remove or []
+
+    if not adds and not removes:
+        print("Nothing to do: specify --add and/or --remove", file=sys.stderr)
+        sys.exit(1)
+
+    # --- account scoping gate, applies before anything else ---
+    if not args.account and not args.all_accounts:
+        print("Refusing: 'tag' needs either --account NAME (recommended) "
+              "or --all-accounts (explicit, for a deliberate cross-account "
+              f"change). {account_choices_help()}", file=sys.stderr)
+        sys.exit(1)
+
+    if args.account and args.account not in ACCOUNTS:
+        print(f"Unknown account '{args.account}'. {account_choices_help()}",
+              file=sys.stderr)
+        sys.exit(1)
+
+    q = scoped_query(args.query, args.account)  # None if --all-accounts
+
+    is_destructive = bool(
+        set(adds) & DESTRUCTIVE_TAGS or set(removes) & PROTECTED_REMOVALS
+    )
+
+    count = run_notmuch(["count", q]).strip()
+    tag_expr = [f"+{t}" for t in adds] + [f"-{t}" for t in removes]
+    scope = args.account or "ALL ACCOUNTS"
+
+    print(f"Scope:   {scope}")
+    print(f"Query:   {args.query}")
+    print(f"Effective query: {q}")
+    print(f"Matches: {count} message(s)")
+    print(f"Change:  {' '.join(tag_expr)}")
+
+    if is_destructive:
+        print("\n[!] This includes a destructive tag change "
+              f"({DESTRUCTIVE_TAGS | PROTECTED_REMOVALS} related).")
+
+    n = int(count)
+    if n > args.max_messages:
+        print(f"\n[cap] {n} matches exceeds --max-messages "
+              f"({args.max_messages}).")
+
+    if not args.apply:
+        print("\nDry run only. Re-run with --apply to actually change tags.")
+        return
+
+    if n > args.max_messages:
+        print(f"\nRefusing to apply: {n} messages exceeds the "
+              f"--max-messages cap ({args.max_messages}). Re-run with a "
+              "higher --max-messages if this is intended.", file=sys.stderr)
+        sys.exit(1)
+
+    if is_destructive and not args.confirm_destructive:
+        print("\nRefusing to apply: destructive change needs "
+              "--apply AND --confirm-destructive.", file=sys.stderr)
+        sys.exit(1)
+
+    run_notmuch(["tag"] + tag_expr + ["--", q], capture=False)
+    log_mutation(scope, args.query, tag_expr, int(count))
+    print(f"\nApplied. {count} message(s) affected.")
+    print("Note: if synchronize_flags is on, this will also update Maildir "
+          "flags and may propagate to the IMAP server on next mbsync run.")
+
+
+def cmd_draft(args):
+    if args.account not in ACCOUNTS:
+        print(f"Unknown account '{args.account}'. {account_choices_help()}",
+              file=sys.stderr)
+        sys.exit(1)
+
+    subdir, from_addr = ACCOUNTS[args.account]
+    drafts_subdir = DRAFTS_SUBDIR.get(args.account)
+    if drafts_subdir is None:
+        print(f"Account '{args.account}' has no synced Drafts folder "
+              "(not in its mbsync Patterns). Pick a different account "
+              "or add Drafts to that account's sync config first.",
+              file=sys.stderr)
+        sys.exit(1)
+
+    drafts_path = MAIL_ROOT / subdir / drafts_subdir
+    if not drafts_path.is_dir():
+        print(f"Expected Drafts maildir not found at {drafts_path}. "
+              "Has this account been synced yet?", file=sys.stderr)
+        sys.exit(1)
+
+    body = args.body
+    if args.body_file:
+        body = Path(args.body_file).read_text()
+    if body is None:
+        body = sys.stdin.read()
+
+    msg = EmailMessage()
+    msg["From"] = from_addr
+    msg["To"] = args.to
+    if args.cc:
+        msg["Cc"] = args.cc
+    msg["Subject"] = args.subject
+    msg["Date"] = formatdate(localtime=True)
+    msg["Message-ID"] = make_msgid()
+    msg.set_content(body)
+
+    # Maildir atomic write: build the full file in tmp/, then rename
+    # (not copy) into new/. Any reader (neomutt, notmuch, mbsync) only
+    # ever sees either "not there yet" or "fully written", never a
+    # partial file.
+    unique = f"{int(time.time())}.M{os.getpid()}P{id(msg) % 100000}.{socket.gethostname()}"
+    tmp_path = drafts_path / "tmp" / unique
+    new_path = drafts_path / "new" / unique
+
+    tmp_path.write_bytes(msg.as_bytes())
+    os.rename(tmp_path, new_path)
+
+    print(f"Draft written: {new_path}")
+    print(f"From:    {from_addr}")
+    print(f"To:      {args.to}")
+    print(f"Subject: {args.subject}")
+    print("\nThis is a LOCAL draft only. It will appear on the server "
+          "(and other devices) after the next mbsync run. Nothing has "
+          "been sent, review and send it yourself in neomutt.")
+
+
+def log_mutation(scope, query, tag_expr, count):
+    AUDIT_LOG.parent.mkdir(parents=True, exist_ok=True)
+    ts = datetime.now(timezone.utc).isoformat(timespec="seconds")
+    with open(AUDIT_LOG, "a") as f:
+        f.write(f"{ts}\tscope={scope}\tquery={query!r}\t"
+                f"change={' '.join(tag_expr)}\tcount={count}\n")
+
+
+def build_parser():
+    p = argparse.ArgumentParser(
+        prog="mailctl",
+        description="Read-heavy, agent-safe wrapper around notmuch. "
+                     "No send/reply/compose capability exists in this tool.",
+    )
+    sub = p.add_subparsers(dest="command", required=True)
+
+    sp = sub.add_parser("search", help="search mail, notmuch query syntax")
+    sp.add_argument("query")
+    sp.add_argument("--account", choices=list(ACCOUNTS),
+                     help="scope to one account, default is global")
+    sp.add_argument("--json", action="store_true")
+    sp.set_defaults(func=cmd_search)
+
+    sp = sub.add_parser("show", help="show a full thread/message")
+    sp.add_argument("query", help="e.g. thread:0000... or id:...")
+    sp.set_defaults(func=cmd_show)
+
+    sp = sub.add_parser("tags", help="list all tags currently in use")
+    sp.set_defaults(func=cmd_tags)
+
+    sp = sub.add_parser("count", help="count messages matching a query")
+    sp.add_argument("query")
+    sp.add_argument("--account", choices=list(ACCOUNTS),
+                     help="scope to one account, default is global")
+    sp.set_defaults(func=cmd_count)
+
+    sp = sub.add_parser("senders", help="rank senders by message count")
+    sp.add_argument("query")
+    sp.add_argument("--account", choices=list(ACCOUNTS),
+                     help="scope to one account, default is global")
+    sp.add_argument("--top", type=int, metavar="N",
+                     help="show only the top N senders")
+    sp.add_argument("--json", action="store_true")
+    sp.set_defaults(func=cmd_senders)
+
+    sp = sub.add_parser("subjects", help="rank subject terms by how many "
+                                          "threads use them")
+    sp.add_argument("query")
+    sp.add_argument("--account", choices=list(ACCOUNTS),
+                     help="scope to one account, default is global")
+    sp.add_argument("--top", type=int, default=25, metavar="N",
+                     help="show only the top N terms (default 25)")
+    sp.add_argument("--json", action="store_true")
+    sp.set_defaults(func=cmd_subjects)
+
+    sp = sub.add_parser("tag", help="add/remove tags (dry-run unless --apply)")
+    sp.add_argument("query")
+    sp.add_argument("--account", choices=list(ACCOUNTS),
+                     help="required unless --all-accounts is given")
+    sp.add_argument("--all-accounts", action="store_true",
+                     help="explicit opt-in to a cross-account mutation")
+    sp.add_argument("--add", action="append", metavar="TAG")
+    sp.add_argument("--remove", action="append", metavar="TAG")
+    sp.add_argument("--apply", action="store_true",
+                     help="actually apply the change, default is dry-run")
+    sp.add_argument("--confirm-destructive", action="store_true",
+                     help="required in addition to --apply for "
+                          "deleted/trash/spam or removing 'inbox'")
+    sp.add_argument("--max-messages", type=int, default=DEFAULT_MAX_MESSAGES,
+                     metavar="N",
+                     help=f"abort --apply if the match count exceeds N "
+                          f"(default {DEFAULT_MAX_MESSAGES})")
+    sp.set_defaults(func=cmd_tag)
+
+    sp = sub.add_parser("draft", help="write a draft to the account's Drafts "
+                                       "folder for later review/sending in "
+                                       "neomutt. Never sends anything.")
+    sp.add_argument("--account", required=True, choices=list(ACCOUNTS),
+                     help="which identity/mailbox to draft into")
+    sp.add_argument("--to", required=True)
+    sp.add_argument("--cc")
+    sp.add_argument("--subject", required=True)
+    sp.add_argument("--body", help="draft body text")
+    sp.add_argument("--body-file", help="read body from a file instead")
+    sp.set_defaults(func=cmd_draft)
+
+    return p
+
+
+def main():
+    parser = build_parser()
+    args = parser.parse_args()
+    args.func(args)
+
+
+if __name__ == "__main__":
+    main()