adeelahmad · Oct 5, 2026 · Oct 5, 2026
diff --git a/docs/processing.md b/docs/processing.md
@@ -254,6 +254,16 @@ importing chosen ones and watching work as above.
   Important), and a message several mailboxes show (one Message-ID) becomes one resource. A watch remembers the last
   message it saw in each mailbox and asks only for newer ones. Lens only reads: mailboxes are opened read-only and
   messages fetched without marking them read.
+- **Routing rules** (email watches only): a shared inbox can feed several namespaces. Each rule has conditions on who
+  sent a message (`from`), who it was for (`to`: To, Cc, and the address the inbox received it at, so mail to an alias
+  like support@ is told apart), its `subject`, its mailing list (`list`, the List-Id) or its `mailbox`, and sends
+  matching messages to a namespace, or skips them. Rules are tried in order and the first whose conditions all match
+  decides; a message no rule matches goes to the watch's own namespace, as it does with no rules. Patterns ignore case;
+  one without `*` or `?` matches anywhere in the text (`acme.com`), one with them must match all of it
+  (`*@acme.com`), and a condition can list several patterns, any of which may match. A rule names a namespace that
+  exists (rules never create one); mail for a namespace deleted since goes to the watch's. Rules apply to messages that
+  arrive after they're saved. `POST /api/v1/watches/routes/preview` shows where the latest messages would go before
+  saving, and a scan's stats count the messages `routed` elsewhere.
 - **Calendar feed (iCal)**: the calendar's iCal address (`https://` or `webcal://`; a user and password if it asks for
   one). The address is kept encrypted like a password and never shown again, since a private calendar's address is
   all it takes to read it. It is fetched the way web pages are captured: public addresses only (and the networks in

diff --git a/fastapi_backend/app/api/v1/routes/sources.py b/fastapi_backend/app/api/v1/routes/sources.py
@@ -15,6 +15,8 @@
 from app.schemas.sources import (
     Backend,
     BrowseEntry,
+    RoutedMessage,
+    RoutePreviewRequest,
     Source,
     SourceCreate,
     SourceCreated,
@@ -147,6 +149,18 @@ def preview_watch(body: WatchPreviewRequest, user: AdminWriter, db: Db, cfg: Cfg
     )
 
 
+@router.post("/watches/routes/preview")
+def preview_routes(body: RoutePreviewRequest, user: AdminWriter, db: Db, cfg: Cfg) -> list[RoutedMessage]:
+    """Where the latest messages of an email account would go under these routing rules, before saving them."""
+    try:
+        rows = sources.preview_routes(db, cfg, body.source, body.path, [r.model_dump() for r in body.routes], body.namespace)
+    except KeyError:
+        raise HTTPException(400, "no such source") from None
+    except (ValueError, RuntimeError) as e:
+        raise HTTPException(400, str(e)) from None
+    return [RoutedMessage(**r) for r in rows]
+
+
 @router.patch("/watches/{wid}")
 def update_watch(wid: int, body: WatchUpdate, user: AdminWriter, db: Db) -> Ok:
     _watch(db, wid)

diff --git a/fastapi_backend/app/domain/feeds.py b/fastapi_backend/app/domain/feeds.py
@@ -24,6 +24,7 @@
 import contextlib
 import datetime as dt
 import email
+import email.utils
 import imaplib
 import re
 import ssl
@@ -51,6 +52,8 @@
 # not taken when a whole account is watched: what can't be opened, the bin, junk and drafts, and the mailboxes that
 # only show messages kept elsewhere (Gmail's All Mail, Starred and Important)
 SKIP_MAILBOXES = {"\\noselect", "\\nonexistent", "\\trash", "\\junk", "\\drafts", "\\all", "\\flagged", "\\important"}
+# read with each message's listing (headers only): enough to name it, place it, and route it
+HEADERS = ("SUBJECT", "DATE", "MESSAGE-ID", "FROM", "TO", "CC", "DELIVERED-TO", "X-ORIGINAL-TO", "LIST-ID")
 MAX_CALENDAR = 50 * 1024 * 1024
 FEED_SECONDS = 60  # a calendar fetched for a listing is reused this long (a scan reads each of its events)
 _FEEDS: dict = {}
@@ -191,7 +194,7 @@ def _messages(conn, mailbox, base, after=None):
     if not count:
         return []
     last = int(after["uid"]) if after and int(after.get("validity") or -1) == validity else 0
-    typ, data = conn.uid("FETCH", f"{last + 1}:*", "(UID RFC822.SIZE INTERNALDATE BODY.PEEK[HEADER.FIELDS (SUBJECT DATE MESSAGE-ID)])")
+    typ, data = conn.uid("FETCH", f"{last + 1}:*", f"(UID RFC822.SIZE INTERNALDATE BODY.PEEK[HEADER.FIELDS ({' '.join(HEADERS)})])")
     data = _ok(typ, data, f"reading {mailbox}")
     out, i = [], 0
     while i < len(data):
@@ -231,11 +234,40 @@ def _messages(conn, mailbox, base, after=None):
                 sent or arrived,
                 key=f"message-id:{msgid}" if msgid else None,
                 cursor={"mailbox": mailbox, "validity": validity, "uid": int(uid.group(1))},
+                mail=_addressing(h, subject, mailbox),
             )
         )
     return sorted(out, key=lambda e: e["modified"] or "", reverse=True)
 
 
+def _addresses(h, *names):
+    """The addresses in these headers, lowercased, each once (Delivered-To and the like are read as plain text)."""
+    out = []
+    for n in names:
+        try:
+            values = [str(v) for v in h.get_all(n) or []]
+        except (TypeError, ValueError):  # a header too broken to read gives no addresses
+            continue
+        out += [a.strip().lower() for _, a in email.utils.getaddresses(values) if "@" in a]
+    return list(dict.fromkeys(out))
+
+
+def _addressing(h, subject, mailbox):
+    """What routing rules (mail_routes.py) look at: who sent it, who it was for (To, Cc, and the address a shared
+    inbox received it at), its subject, mailing list and mailbox."""
+    try:
+        listed = str(h.get("list-id") or "").strip()
+    except (ValueError, TypeError):
+        listed = ""
+    return {
+        "from": _addresses(h, "from"),
+        "to": _addresses(h, "to", "cc", "delivered-to", "x-original-to"),
+        "subject": subject,
+        "list": listed,
+        "mailbox": mailbox,
+    }
+
+
 MESSAGE_RX = re.compile(r"^(?P<mailbox>.+)/(?P<validity>\d+)-(?P<uid>\d+)\.eml$")
 
 

diff --git a/fastapi_backend/app/domain/mail_routes.py b/fastapi_backend/app/domain/mail_routes.py
@@ -0,0 +1,110 @@
+"""Routing rules for a watched email account: which namespace each new message goes to.
+
+A shared inbox (support@, info@, a family address) often carries mail for several namespaces. A watch on an IMAP
+source can carry rules, tried in order; the first whose conditions all match a message decides where it goes: into
+another namespace, or nowhere (skipped). A message no rule matches goes to the watch's own namespace, as it does with
+no rules at all, and so does one whose rule names a namespace that has since been deleted.
+
+A condition looks at one part of the message (FIELDS) and holds one or more patterns, any of which may match. Patterns
+ignore case; one without * or ? matches anywhere in the text (`acme.com` matches `ana@acme.com`), one with them must
+match all of it (`*@acme.com`). `to` covers To, Cc and the address the inbox received the message at (Delivered-To,
+X-Original-To), which is how mail sent to an alias of a shared inbox is told apart.
+
+Rules are kept on the watch with namespace ids (a renamed namespace keeps its mail), and shown with names.
+"""
+
+from __future__ import annotations
+
+import fnmatch
+
+from . import store
+
+FIELDS = {
+    "from": "who sent it",
+    "to": "who it was for: To, Cc, or the address the inbox received it at",
+    "subject": "its subject",
+    "list": "its mailing list (List-Id)",
+    "mailbox": "the mailbox it is in",
+}
+MAX_RULES = 100
+MAX_PATTERNS = 50
+
+
+def _patterns(field, value):
+    vals = [value] if isinstance(value, str) else value
+    if not isinstance(vals, list) or not all(isinstance(v, str) for v in vals):
+        raise ValueError(f"{field} is a pattern or a list of patterns")
+    vals = [v.strip() for v in vals if v.strip()]
+    if not vals or len(vals) > MAX_PATTERNS:
+        raise ValueError(f"{field} needs between 1 and {MAX_PATTERNS} patterns")
+    return vals
+
+
+def check(db, rules):
+    """Rules as the API takes them ({match: {field: patterns}, namespace} or {match, skip: true}), checked, with
+    namespace ids in place of names. A namespace must exist already: a rule never creates one."""
+    if rules is None:
+        return []
+    if not isinstance(rules, list) or len(rules) > MAX_RULES:
+        raise ValueError(f"routes is a list of at most {MAX_RULES} rules")
+    out = []
+    for i, r in enumerate(rules, 1):
+        if not isinstance(r, dict):
+            raise ValueError(f"rule {i} isn't a rule")
+        match = r.get("match") or {}
+        if not isinstance(match, dict) or not match:
+            raise ValueError(f"rule {i} needs at least one condition: {', '.join(FIELDS)}")
+        if set(match) - set(FIELDS):
+            raise ValueError(f"rule {i}: conditions are on {', '.join(FIELDS)}")
+        rule = {"match": {f: _patterns(f, v) for f, v in match.items()}}
+        if r.get("skip"):
+            if r.get("namespace"):
+                raise ValueError(f"rule {i} either skips messages or sends them to a namespace, not both")
+            rule["skip"] = True
+        elif r.get("namespace"):
+            try:
+                rule["space"] = store.ns_id(db, r["namespace"], create=False)
+            except KeyError:
+                raise ValueError(f"rule {i}: there's no namespace {r['namespace']}") from None
+        else:
+            raise ValueError(f"rule {i} needs a namespace, or skip")
+        if r.get("name"):
+            rule["name"] = str(r["name"]).strip()[:100]
+        out.append(rule)
+    return out
+
+
+def view(rules, names):
+    """Stored rules as the API shows them: namespaces by name (None for one deleted since)."""
+    out = []
+    for r in rules or []:
+        shown = {"match": r["match"], "skip": bool(r.get("skip")), "namespace": None if r.get("skip") else names.get(r.get("space"))}
+        out.append({**shown, "name": r["name"]} if r.get("name") else shown)
+    return out
+
+
+def _hit(pattern, text):
+    p, t = pattern.lower(), (text or "").lower()
+    return fnmatch.fnmatchcase(t, p) if any(c in p for c in "*?[") else p in t
+
+
+def matches(rule, mail):
+    for field, patterns in rule["match"].items():
+        got = mail.get(field)
+        texts = got if isinstance(got, list) else [got or ""]
+        if not any(_hit(p, t) for p in patterns for t in texts):
+            return False
+    return True
+
+
+def route(rules, mail, home, spaces):
+    """Where a message goes: (namespace id, or None to skip it; the number of the rule that decided, or None).
+    `home` is the watch's namespace; `spaces` the namespace ids that exist."""
+    if not mail:
+        return home, None
+    for i, r in enumerate(rules or [], 1):
+        if matches(r, mail):
+            if r.get("skip"):
+                return None, i
+            return (r["space"], i) if r.get("space") in spaces else (home, i)
+    return home, None
diff --git a/fastapi_backend/app/domain/sources.py b/fastapi_backend/app/domain/sources.py
@@ -4,7 +4,8 @@
 this machine (only inside sources.local_roots); or, without rclone, an email account (IMAP) or a calendar feed (iCal),
 whose messages and events are shown as files (feeds.py). A watch maps a folder on a source to a namespace: new audio, video,
 documents (PDFs) and images become resources queued for the whole pipeline, their files staying on the source; new
-transcripts are imported and queued for analysis. Credentials are stored encrypted and handed to rclone in a private
+transcripts are imported and queued for analysis. A watch on an email account can carry routing rules that send some
+messages to other namespaces, or skip them (mail_routes.py). Credentials are stored encrypted and handed to rclone in a private
 temporary config file per call; OAuth tokens rclone refreshes are saved.
 """
 
@@ -21,7 +22,7 @@
 import subprocess
 import tempfile
 
-from . import convert, deletion, feeds, ingest, jobs, settings, store
+from . import convert, deletion, feeds, ingest, jobs, mail_routes, settings, store
 
 R = store.R
 BACKENDS = {
@@ -70,6 +71,7 @@
     "steps": None,
     "pipeline": None,
     "enabled": True,
+    "routes": [],  # an email watch's routing rules (mail_routes.py); none: everything goes to its namespace
 }
 
 
@@ -367,7 +369,7 @@ def list_sources(db):
 
 # ---------- watched folders ----------
 def _check_watch(opts):
-    out = {k: opts[k] for k in WATCH if k in opts}
+    out = {k: opts[k] for k in WATCH if k in opts and k != "routes"}
     if out.get("kinds", WATCH["kinds"]) not in TAKES:
         raise ValueError(f"kinds is one of: {', '.join(TAKES)}")
     if not isinstance(out.get("poll_minutes", 5), int) or out.get("poll_minutes", 5) < 1:
@@ -384,9 +386,20 @@ def _check_watch(opts):
     return out
 
 
+def _routes(db, src, opts):
+    """The routing rules in `opts`, checked, as the watch keeps them ({} when they aren't being set)."""
+    if "routes" not in opts:
+        return {}
+    routes = mail_routes.check(db, opts["routes"])
+    if routes and src["type"] != "imap":
+        raise ValueError("routing rules are for watches on an email account")
+    return {"routes": routes}
+
+
 def create_watch(db, cfg, sid, path, space, user=None, **opts):
     src = get(db, sid)
     p = check_path(cfg, src, path)
+    routes = _routes(db, src, opts)
     wid = db.next_id("watch_path")
     db.q(
         "CREATE $r CONTENT $d",
@@ -395,6 +408,7 @@ def create_watch(db, cfg, sid, path, space, user=None, **opts):
             {
                 **WATCH,
                 **_check_watch(opts),
+                **routes,
                 "source": sid,
                 "path": p,
                 "space": space,
@@ -408,7 +422,9 @@ def create_watch(db, cfg, sid, path, space, user=None, **opts):
 
 
 def update_watch(db, wid, **opts):
-    db.q("UPDATE $r MERGE $d", r=R("watch_path", wid), d=_check_watch(opts))
+    w = db.one("SELECT source FROM $r", r=R("watch_path", wid))
+    routes = _routes(db, get(db, w["source"]), opts) if w else {}
+    db.q("UPDATE $r MERGE $d", r=R("watch_path", wid), d={**_check_watch(opts), **routes})
     if {"kinds", "include", "exclude"} & set(opts):  # what it takes changed: look through the mailboxes again
         db.q("UPDATE $r SET cursor = NONE", r=R("watch_path", wid))
 
@@ -419,12 +435,44 @@ def remove_watch(db, wid):
 
 def list_watches(db, spaces=None):
     q = (
-        "SELECT record::id(id) AS id, source, path, space, kinds, poll_minutes, stable_seconds, backfill, include, exclude, steps, enabled, "
+        "SELECT record::id(id) AS id, source, path, space, kinds, poll_minutes, stable_seconds, backfill, include, exclude, steps, enabled, routes, "
         "last_scan_at, next_scan_at, last_stats, last_error FROM watch_path"
     )
     rows = db.rows(q + (" WHERE space IN $sp" if spaces is not None else ""), sp=sorted(spaces or []))
     names, srcs = store.space_names(db), {s["id"]: s["name"] for s in db.rows("SELECT record::id(id) AS id, name FROM storage_source")}
-    return [{**r, "namespace": names.get(r["space"]), "source_name": srcs.get(r["source"])} for r in rows]
+    return [
+        {**r, "namespace": names.get(r["space"]), "source_name": srcs.get(r["source"]), "routes": mail_routes.view(r.get("routes"), names)}
+        for r in rows
+    ]
+
+
+def preview_routes(db, cfg, sid, path, routes, home, limit=50):
+    """Where the latest messages of an email watch would go under these rules (as the API takes them), before
+    they're saved: one row per message, newest first, with the namespace (None: skipped) and the rule that decided.
+    `home` is the name of the watch's namespace, which may not exist yet."""
+    src = get(db, sid)
+    if src["type"] != "imap":
+        raise ValueError("routing rules are for watches on an email account")
+    rules = _routes(db, src, {"routes": routes})["routes"]
+    names = store.space_names(db)
+    files = sorted(list_files(db, cfg, src, path), key=lambda f: f.get("modified") or "", reverse=True)[:limit]
+    out = []
+    for f in files:
+        to, rule = mail_routes.route(rules, f.get("mail"), "", set(names))  # "": the watch's own namespace, `home`
+        mail = f.get("mail") or {}
+        out.append(
+            {
+                "path": f["path"],
+                "title": f.get("title"),
+                "when": f.get("when"),
+                "from": mail.get("from", []),
+                "to": mail.get("to", []),
+                "namespace": home if to == "" else names.get(to),
+                "skipped": to is None,
+                "rule": rule,
+            }
+        )
+    return out
 
 
 def _when(s):
@@ -533,14 +581,18 @@ def kind_of(cfg, w, f):
 def poll_watch(db, cfg, wid, log=print):
     w = db.one(
         "SELECT record::id(id) AS id, source, path, space, kinds, poll_minutes, stable_seconds, backfill, include, exclude, steps, pipeline, last_scan_at, "
-        "cursor FROM $r",
+        "cursor, routes FROM $r",
         r=R("watch_path", wid),
     )
     src = get(db, w["source"])
     first, now = not w.get("last_scan_at"), dt.datetime.now(dt.timezone.utc)
     known = {r["path"]: r for r in db.rows("SELECT path, size, modified, status FROM remote_file WHERE watch = $w", w=wid)}
-    gone = deletion.gone_remote(db, w["space"])  # recordings someone deleted stay deleted
+    gone = {w["space"]: deletion.gone_remote(db, w["space"])}  # recordings someone deleted stay deleted
     stats = {"seen": 0, "new": 0, "waiting": 0, "skipped": 0, "errors": 0}
+    routes = w.get("routes") or []
+    spaces = set(store.space_names(db)) if routes else set()
+    if routes:
+        stats["routed"] = 0
     files, held = list_files(db, cfg, src, w["path"], w.get("cursor")), set()
     for f in files:
         kind = kind_of(cfg, w, f)
@@ -552,7 +604,12 @@ def poll_watch(db, cfg, wid, log=print):
             continue
         key = R("remote_file", f"{wid}-{hashlib.sha1(f['path'].encode()).hexdigest()[:20]}")
         row = {"watch": wid, "path": f["path"], "size": f["size"], "modified": f["modified"], "seen_at": store.now()}
-        if (first and not w.get("backfill")) or (src["id"], f["path"]) in gone:
+        space, rule = mail_routes.route(routes, f.get("mail"), w["space"], spaces) if routes else (w["space"], None)
+        if rule:
+            row.update(rule=rule, space=space)
+        if space is not None and space not in gone:
+            gone[space] = deletion.gone_remote(db, space)
+        if space is None or (first and not w.get("backfill")) or (src["id"], f["path"]) in gone[space]:
             db.q("UPSERT $k CONTENT $d", k=key, d={**row, "status": "skipped"})
             stats["skipped"] += 1
         elif (now - _when(f["modified"])).total_seconds() < (w.get("stable_seconds") or 0):
@@ -561,9 +618,11 @@ def poll_watch(db, cfg, wid, log=print):
             held.add(f["path"])
         else:
             try:
-                rid, _job = _ingest(db, cfg, src, f, kind, w["space"], f"watch:{wid}", w.get("steps"), w.get("pipeline"))
+                rid, _job = _ingest(db, cfg, src, f, kind, space, f"watch:{wid}", w.get("steps"), w.get("pipeline"))
                 db.q("UPSERT $k CONTENT $d", k=key, d={**row, "status": "queued", "recording": rid})
                 stats["new"] += 1
+                if space != w["space"]:
+                    stats["routed"] += 1
             except Exception as e:  # noqa: BLE001 - one bad file must not stop the scan
                 db.q("UPSERT $k CONTENT $d", k=key, d={**row, "status": "error", "error": f"{type(e).__name__}: {e}"[:300]})
                 stats["errors"] += 1