Add VPS IMAP sorting automation

Prompts used since the previous commit:
- Hieronder staat je oorspronkelijke plan. Volgens mij moet stap 'Add VPS automation' nog worden uitgevoerd, inclusief de testen.

Special observations:
- Added a testable IMAP sorting daemon, shared IMAP operation helpers, systemd service template, and SSH deployment script.
- Local tests passed; no live VPS deployment or mailbox-affecting daemon test was run.
- verplaats_log.json remains modified from the user's run and was intentionally not staged.

.gitignore

@@ -4,3 +4,5 @@ __pycache__/
 .DS_Store
 config.json
 **/*.eml
+sort_mail_daemon_state.json
+verplaats_log_*.json

deploy_vps.sh

@@ -0,0 +1,68 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+VPS_HOST="${VPS_HOST:-vps.austalius.nl}"
+VPS_USER="${VPS_USER:?Set VPS_USER to the non-root sudo SSH user}"
+APP_USER="${APP_USER:-mailcat}"
+APP_DIR="${APP_DIR:-/opt/mailcat}"
+SERVICE_NAME="${SERVICE_NAME:-mailcat-sort-backup}"
+ACCOUNT="${ACCOUNT:-backup}"
+
+ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+ARCHIVE="$(mktemp -t mailcat-deploy.XXXXXX.tar.gz)"
+
+cleanup() {
+  rm -f "$ARCHIVE"
+}
+trap cleanup EXIT
+
+tar \
+  --exclude ".git" \
+  --exclude "__pycache__" \
+  --exclude "mailbox" \
+  --exclude "config.json" \
+  --exclude "*.log" \
+  --exclude "backup_log.json" \
+  --exclude "verplaats_log*.json" \
+  --exclude "sort_mail_daemon_state.json" \
+  --exclude "mailing_lists_cache*.json" \
+  --exclude "*.pyc" \
+  -C "$ROOT_DIR" \
+  -czf "$ARCHIVE" .
+
+scp "$ARCHIVE" "$VPS_USER@$VPS_HOST:/tmp/mailcat-deploy.tar.gz"
+
+ssh "$VPS_USER@$VPS_HOST" \
+  "APP_USER='$APP_USER' APP_DIR='$APP_DIR' SERVICE_NAME='$SERVICE_NAME' ACCOUNT='$ACCOUNT' bash -s" <<'REMOTE'
+set -euo pipefail
+
+if ! id "$APP_USER" >/dev/null 2>&1; then
+  sudo useradd --system --home "$APP_DIR" --shell /usr/sbin/nologin "$APP_USER"
+fi
+
+sudo mkdir -p "$APP_DIR" /var/lib/mailcat /var/log/mailcat
+sudo tar -xzf /tmp/mailcat-deploy.tar.gz -C "$APP_DIR"
+sudo chown -R "$APP_USER:$APP_USER" "$APP_DIR" /var/lib/mailcat /var/log/mailcat
+sudo chmod 750 "$APP_DIR"
+
+if [ ! -f "$APP_DIR/config.json" ]; then
+  echo "Missing $APP_DIR/config.json on VPS. Create it with mode 600 before starting the service." >&2
+fi
+sudo chmod 600 "$APP_DIR/config.json" 2>/dev/null || true
+sudo chown "$APP_USER:$APP_USER" "$APP_DIR/config.json" 2>/dev/null || true
+
+sudo install -m 0644 "$APP_DIR/systemd/mailcat-sort.service" "/etc/systemd/system/$SERVICE_NAME.service"
+sudo sed -i \
+  -e "s#__APP_USER__#$APP_USER#g" \
+  -e "s#__APP_DIR__#$APP_DIR#g" \
+  -e "s#__ACCOUNT__#$ACCOUNT#g" \
+  "/etc/systemd/system/$SERVICE_NAME.service"
+
+sudo systemctl daemon-reload
+sudo systemctl enable "$SERVICE_NAME.service"
+
+echo "Deployment complete."
+echo "Create/check $APP_DIR/config.json, then run:"
+echo "  sudo systemctl start $SERVICE_NAME.service"
+echo "  sudo journalctl -u $SERVICE_NAME.service -f"
+REMOTE

mail_imap_ops.py

@@ -0,0 +1,88 @@
+"""Shared IMAP operations used by mailcat sorters."""
+
+from __future__ import annotations
+
+import email
+import email.header
+from email.message import Message
+
+from imap_utils import quote_mailbox
+from mail_routes import PREFIX, quarter_folder, route_from_subject
+
+
+HEADER_FETCH = "(BODY.PEEK[HEADER.FIELDS (FROM SUBJECT DATE)])"
+
+
+def decode_header_value(value) -> str:
+    """Decode an RFC 2047 header value into display text."""
+    if not value:
+        return ""
+    try:
+        parts = email.header.decode_header(value)
+        out = []
+        for raw, charset in parts:
+            if isinstance(raw, bytes):
+                out.append(raw.decode(charset or "utf-8", errors="replace"))
+            else:
+                out.append(str(raw))
+        return " ".join(out).strip()
+    except Exception:
+        return str(value or "")
+
+
+def fetch_body(data) -> bytes | None:
+    """Return the first bytes payload from an IMAP FETCH response."""
+    if not data:
+        return None
+    for item in data:
+        if isinstance(item, tuple) and len(item) >= 2 and isinstance(item[1], bytes):
+            return item[1]
+    return None
+
+
+def parse_header(raw_header: bytes) -> Message:
+    """Parse an IMAP header payload into an email Message."""
+    return email.message_from_bytes(raw_header)
+
+
+def destination_from_header(raw_header: bytes) -> str | None:
+    """Return the full IMAP destination folder for a fetched header."""
+    msg = parse_header(raw_header)
+    from_raw = decode_header_value(msg.get("From", ""))
+    subject = decode_header_value(msg.get("Subject", ""))
+    date_str = msg.get("Date", "")
+
+    destination = route_from_subject(from_raw, subject)
+    if destination == "__FACTUUR_DATE__":
+        destination = quarter_folder(date_str)
+    if destination:
+        return PREFIX + destination
+    return None
+
+
+def ensure_mailbox(mail, folder: str) -> bool:
+    """Create a mailbox and its parents if they do not already exist."""
+    parts = folder.split(".")
+    for index in range(1, len(parts) + 1):
+        parent = ".".join(parts[:index])
+        status, _ = mail.create(quote_mailbox(parent))
+        if status not in {"OK", "NO"}:
+            return False
+    return True
+
+
+def move_message(mail, uid: bytes, src: str, dst: str):
+    """Copy a message to dst and mark it deleted in src."""
+    select_status, select_data = mail.select(quote_mailbox(src))
+    if select_status != "OK":
+        return False, ("select_source", select_status, select_data)
+
+    copy_status, copy_data = mail.uid("COPY", uid, quote_mailbox(dst))
+    if copy_status != "OK":
+        return False, ("copy", copy_status, copy_data)
+
+    store_status, store_data = mail.uid("STORE", uid, "+FLAGS", "\\Deleted")
+    if store_status != "OK":
+        return False, ("store_deleted", store_status, store_data)
+
+    return True, None

restart_prompt.md

@@ -89,6 +89,12 @@ Scripts using shared IMAP folder handling:
 - `kopieer_naar_backup.py`
 - `download_mailbox.py`
 - `dagelijks_overzicht.py`
+- `sort_mail_daemon.py`
+
+Shared operation helpers:
+
+- `mail_imap_ops.py` centralizes header decoding, FETCH payload extraction, routing fetched headers to full IMAP destinations, destination mailbox creation, and copy/delete move semantics for live sorting.
+- `verplaats_bestaand.py` still contains its own historical-sorter retry/logging flow; do not refactor it casually because it has already been exercised against the backup mailbox.
 
 `verplaats_bestaand.py` current behavior:
 
@@ -127,6 +133,24 @@ Scripts using shared IMAP folder handling:
 - Verifies the destination folder is selectable before dedupe and append work.
 - Does not mark a folder complete if any failure occurs while copying that folder.
 
+`sort_mail_daemon.py` current behavior:
+
+- Intended VPS automation entry point for IMAP sorting.
+- Defaults to `--account backup` and `--folder INBOX`.
+- Uses the same shared routing policy through `mail_imap_ops.destination_from_header()`.
+- Maintains a JSON state file of processed UIDs per folder and resets that state when UIDVALIDITY changes.
+- Supports `--once` for local/non-daemon smoke tests and `--dry-run` for non-mutating checks.
+- Watches for new mail with IMAP IDLE. On Python versions with public `imaplib.IMAP4.idle()` it uses that API; otherwise it uses a conservative private-IDLE fallback and reconnects on failures.
+- Logs to `sort_mail_daemon.log` by default, or to the path passed with `--log-file`.
+- Ensures the destination mailbox exists before moving a message, then moves by UID COPY plus UID STORE `\Deleted`, followed by expunge after a scan.
+
+VPS deployment files:
+
+- `deploy_vps.sh` packages the repo excluding `.git`, `config.json`, logs, runtime JSON state/log files, bytecode, and local mailbox data; uploads to `vps.austalius.nl`; installs under `/opt/mailcat` by default; creates/uses a non-root system user named `mailcat` by default; installs a systemd service; enables but does not start the service.
+- The deploy script requires `VPS_USER` to be set to the non-root sudo SSH user. Optional overrides: `VPS_HOST`, `APP_USER`, `APP_DIR`, `SERVICE_NAME`, and `ACCOUNT`.
+- `systemd/mailcat-sort.service` is a template consumed by `deploy_vps.sh`; after placeholder replacement it runs `sort_mail_daemon.py --account backup --folder INBOX` and stores state/logs under `/var/lib/mailcat` and `/var/log/mailcat`.
+- `config.json` remains ignored and must be created manually on the VPS with mode `600` before starting the service.
+
 ## Current Backup Test Status
 
 The user reported that `backup@australius.nl` was empty except for folder structure, so the stale `backup_log.json` state was reset and the real mailbox state was inspected.
@@ -193,6 +217,9 @@ Notable policy risks:
 Latest local code verification:
 
 - `python3 -m py_compile *.py`
+- `python3 -m py_compile *.py tests/*.py`
+- `python3 -m unittest discover -s tests`
+- `bash -n deploy_vps.sh`
 - Explicit `route_from_subject()` checks for the latest INBOX candidate domains, including negative checks for `zuiderzee.net` and `gmail.com`.
 - `git diff --check`
 
@@ -209,24 +236,22 @@ Latest live mailbox verification:
 - The user ran the previous dry-run sorter version against `backup@australius.nl`. It reached `INBOX.Archief.2022.verzonden` after completing `INBOX.Archief.2022.inkomend`, then the IMAP server closed the connection with `Server shutting down.`. The current code has reconnect/retry handling for that failure mode.
 - The user started an actual sorter run on `backup@australius.nl`; it began moving messages from `INBOX` and then crashed on an unexpected IMAP `FETCH` response shape: `AttributeError: 'int' object has no attribute 'decode'`. The local `verplaats_log.json` contains processed `INBOX:<uid>` entries from that run and should be preserved if the user resumes sorting. The current code fixes this parser crash.
 - A later read-only INBOX inspection of the `backup` account found 1,908 unmatched messages and identified additional candidate sender domains. No mailbox-affecting script was run by Codex.
+- After the latest routing refinements, the user reported the backup sorter ended with `Totaal gepland: 0 | Totaal geen match: 2218 | Totaal overgeslagen: 8219 | Totaal fouten: 0`. Treat this as a successful backup historical-sorter run with no immediate recovery work needed.
+- VPS deployment and daemon live tests have not yet been run by Codex.
 
 ## What To Do Next
 
 Recommended next work:
 
-1. Ask the user to run a non-destructive dry-run sorter audit on `backup@australius.nl`: `python3 verplaats_bestaand.py --account backup --audit`, then review planned moves by source and destination.
+1. Ask the user for the VPS non-root sudo SSH username.
-2. Review the audit output for suspicious high-volume destinations, missing matches, and unexpected `Afmelden.*` or invoice routes before approving any actual sorting.
+2. Before starting the service, ensure `/opt/mailcat/config.json` on the VPS contains the `backup` account credentials and has mode `600`.
-3. Add shared operation helpers where useful:
+3. Deploy with a command like `VPS_USER=<ssh-user> ./deploy_vps.sh`.
-   - UID fetch wrappers.
+4. On the VPS, start and inspect the service:
-   - safe copy/delete/expunge helper.
+   - `sudo systemctl start mailcat-sort-backup.service`
-   - Message-ID dedupe helper reused by live automation.
+   - `sudo systemctl status mailcat-sort-backup.service`
-4. Add `List-ID` header support to mailing-list routing if the audit shows sender/domain matching is too coarse.
+   - `sudo journalctl -u mailcat-sort-backup.service -f`
-5. Add VPS automation only after dry-run and actual backup sorting behavior are approved:
+5. Send controlled test messages to `backup@australius.nl` for invoices, clear AI providers, a newsletter/service route, and unmatched mail; verify expected folder moves and daemon logs.
-   - create an IMAP IDLE daemon for `backup@australius.nl`
+6. Add `List-ID` header support to mailing-list routing if live daemon tests show sender/domain matching is too coarse.
-   - deploy via SSH to `vps.austalius.nl`
-   - use a non-root sudo user
-   - install as a systemd service
-   - keep secrets in an ignored `config.json` with restrictive permissions
 
 ## Persistent File Rule
 

sort_mail_daemon.py

@@ -0,0 +1,246 @@
+#!/usr/bin/env python3
+"""IMAP IDLE daemon for routing new mail through mailcat rules."""
+
+from __future__ import annotations
+
+import argparse
+import imaplib
+import json
+import logging
+import select
+import signal
+import sys
+import time
+from pathlib import Path
+
+import config
+from imap_utils import quote_mailbox
+from mail_imap_ops import HEADER_FETCH, destination_from_header, ensure_mailbox, fetch_body, move_message
+
+DEFAULT_STATE_FILE = Path(__file__).parent / "sort_mail_daemon_state.json"
+DEFAULT_LOG_FILE = Path(__file__).parent / "sort_mail_daemon.log"
+
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(description="Sort new IMAP mail using mailcat routes.")
+    parser.add_argument("--account", default="backup", help="config.json account name")
+    parser.add_argument("--folder", default="INBOX", help="folder to watch")
+    parser.add_argument("--state-file", type=Path, default=DEFAULT_STATE_FILE)
+    parser.add_argument("--log-file", type=Path, default=DEFAULT_LOG_FILE)
+    parser.add_argument("--idle-timeout", type=int, default=1740)
+    parser.add_argument("--reconnect-delay", type=int, default=30)
+    parser.add_argument("--once", action="store_true", help="process once and exit")
+    parser.add_argument("--dry-run", action="store_true", help="log intended moves without changing IMAP")
+    return parser
+
+
+def load_state(path: Path) -> dict:
+    if not path.exists():
+        return {"folder_state": {}}
+    try:
+        state = json.loads(path.read_text(encoding="utf-8"))
+    except json.JSONDecodeError:
+        return {"folder_state": {}}
+    state.setdefault("folder_state", {})
+    return state
+
+
+def save_state(path: Path, state: dict):
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps(state, indent=2, ensure_ascii=False), encoding="utf-8")
+
+
+def folder_state(state: dict, folder: str) -> dict:
+    folders = state.setdefault("folder_state", {})
+    entry = folders.setdefault(folder, {})
+    entry.setdefault("uidvalidity", None)
+    entry.setdefault("processed_uids", [])
+    return entry
+
+
+def reset_on_uidvalidity_change(state: dict, folder: str, uidvalidity: str | None):
+    entry = folder_state(state, folder)
+    if entry.get("uidvalidity") != uidvalidity:
+        entry["uidvalidity"] = uidvalidity
+        entry["processed_uids"] = []
+
+
+def processed_set(state: dict, folder: str) -> set[str]:
+    return set(folder_state(state, folder).get("processed_uids", []))
+
+
+def mark_processed(state: dict, folder: str, uid: str):
+    entry = folder_state(state, folder)
+    processed = set(entry.get("processed_uids", []))
+    processed.add(uid)
+    entry["processed_uids"] = sorted(processed, key=lambda value: int(value) if value.isdigit() else value)
+
+
+def setup_logging(log_file: Path):
+    log_file.parent.mkdir(parents=True, exist_ok=True)
+    logging.basicConfig(
+        level=logging.INFO,
+        format="%(asctime)s %(levelname)s %(message)s",
+        handlers=[logging.FileHandler(log_file, encoding="utf-8"), logging.StreamHandler(sys.stdout)],
+    )
+
+
+def connect(account_name: str):
+    cfg = config.load(account_name)
+    mail = imaplib.IMAP4_SSL(cfg.imap_host, cfg.imap_port)
+    mail.login(cfg.username, cfg.password)
+    logging.info("connected account=%s username=%s host=%s", cfg.name, cfg.username, cfg.imap_host)
+    return mail
+
+
+def select_folder(mail, folder: str) -> str | None:
+    status, data = mail.select(quote_mailbox(folder))
+    if status != "OK":
+        raise RuntimeError(f"Cannot select {folder}: {data!r}")
+    _status, data = mail.response("UIDVALIDITY")
+    if data and data[0]:
+        return data[0].decode(errors="replace") if isinstance(data[0], bytes) else str(data[0])
+    return None
+
+
+def uid_search_all(mail) -> list[bytes]:
+    status, data = mail.uid("SEARCH", None, "ALL")
+    if status != "OK" or not data or not data[0]:
+        return []
+    return [uid for uid in data[0].split() if uid]
+
+
+def process_folder(mail, state: dict, folder: str, dry_run: bool) -> dict[str, int]:
+    uidvalidity = select_folder(mail, folder)
+    reset_on_uidvalidity_change(state, folder, uidvalidity)
+    processed = processed_set(state, folder)
+    counts = {"seen": 0, "matched": 0, "moved": 0, "unmatched": 0, "failed": 0, "skipped": 0}
+
+    for uid in uid_search_all(mail):
+        uid_text = uid.decode(errors="replace")
+        counts["seen"] += 1
+        if uid_text in processed:
+            counts["skipped"] += 1
+            continue
+
+        status, data = mail.uid("FETCH", uid, HEADER_FETCH)
+        raw_header = fetch_body(data)
+        if status != "OK" or raw_header is None:
+            counts["failed"] += 1
+            logging.warning("fetch_failed folder=%s uid=%s status=%s data=%r", folder, uid_text, status, data)
+            continue
+
+        destination = destination_from_header(raw_header)
+        if not destination or destination == folder:
+            counts["unmatched"] += 1
+            mark_processed(state, folder, uid_text)
+            logging.info("no_match folder=%s uid=%s", folder, uid_text)
+            continue
+
+        counts["matched"] += 1
+        if dry_run:
+            logging.info("dry_move folder=%s uid=%s destination=%s", folder, uid_text, destination)
+            mark_processed(state, folder, uid_text)
+            continue
+
+        if not ensure_mailbox(mail, destination):
+            counts["failed"] += 1
+            logging.error("ensure_destination_failed folder=%s uid=%s destination=%s", folder, uid_text, destination)
+            continue
+
+        ok, failure = move_message(mail, uid, folder, destination)
+        if ok:
+            counts["moved"] += 1
+            mark_processed(state, folder, uid_text)
+            logging.info("moved folder=%s uid=%s destination=%s", folder, uid_text, destination)
+        else:
+            counts["failed"] += 1
+            logging.error("move_failed folder=%s uid=%s destination=%s failure=%r", folder, uid_text, destination, failure)
+
+    if not dry_run:
+        mail.expunge()
+    return counts
+
+
+def wait_for_mail(mail, timeout: int) -> bool:
+    if hasattr(mail, "idle"):
+        with mail.idle(duration=timeout) as idler:
+            for typ, _data in idler:
+                if typ == "EXISTS":
+                    return True
+        return False
+    return wait_for_mail_private_idle(mail, timeout)
+
+
+def wait_for_mail_private_idle(mail, timeout: int) -> bool:
+    tag = mail._new_tag()
+    mail.send(tag + b" IDLE\r\n")
+    continuation = mail.readline()
+    if not continuation.startswith(b"+"):
+        raise RuntimeError(f"IDLE rejected: {continuation!r}")
+
+    changed = False
+    try:
+        ready, _, _ = select.select([mail.sock], [], [], timeout)
+        if ready:
+            line = mail.readline()
+            changed = b"EXISTS" in line or b"RECENT" in line
+    finally:
+        mail.send(b"DONE\r\n")
+        while True:
+            line = mail.readline()
+            if line.startswith(tag):
+                break
+    return changed
+
+
+def run(args) -> int:
+    setup_logging(args.log_file)
+    stop = {"requested": False}
+
+    def request_stop(_signum, _frame):
+        stop["requested"] = True
+
+    signal.signal(signal.SIGTERM, request_stop)
+    signal.signal(signal.SIGINT, request_stop)
+
+    state = load_state(args.state_file)
+
+    while not stop["requested"]:
+        mail = None
+        try:
+            mail = connect(args.account)
+            counts = process_folder(mail, state, args.folder, args.dry_run)
+            save_state(args.state_file, state)
+            logging.info("scan_complete folder=%s counts=%s", args.folder, counts)
+            if args.once:
+                return 0
+
+            while not stop["requested"]:
+                select_folder(mail, args.folder)
+                changed = wait_for_mail(mail, args.idle_timeout)
+                if changed:
+                    counts = process_folder(mail, state, args.folder, args.dry_run)
+                    save_state(args.state_file, state)
+                    logging.info("idle_scan_complete folder=%s counts=%s", args.folder, counts)
+                else:
+                    logging.info("idle_timeout folder=%s", args.folder)
+        except Exception:
+            logging.exception("daemon_error reconnecting_after=%ss", args.reconnect_delay)
+            if args.once:
+                return 1
+            time.sleep(args.reconnect_delay)
+        finally:
+            if mail is not None:
+                try:
+                    mail.logout()
+                except Exception:
+                    pass
+
+    save_state(args.state_file, state)
+    logging.info("stopped")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(run(build_parser().parse_args()))

systemd/mailcat-sort.service

@@ -0,0 +1,21 @@
+[Unit]
+Description=Mailcat IMAP sorting daemon
+After=network-online.target
+Wants=network-online.target
+
+[Service]
+Type=simple
+User=__APP_USER__
+Group=__APP_USER__
+WorkingDirectory=__APP_DIR__
+ExecStart=/usr/bin/python3 __APP_DIR__/sort_mail_daemon.py --account __ACCOUNT__ --folder INBOX --state-file /var/lib/mailcat/sort_mail_daemon_state.json --log-file /var/log/mailcat/sort_mail_daemon.log
+Restart=always
+RestartSec=30
+NoNewPrivileges=true
+PrivateTmp=true
+ProtectSystem=full
+ProtectHome=true
+ReadWritePaths=__APP_DIR__ /var/lib/mailcat /var/log/mailcat
+
+[Install]
+WantedBy=multi-user.target

tests/test_sort_mail_daemon.py

@@ -0,0 +1,50 @@
+import unittest
+
+from mail_imap_ops import destination_from_header
+from sort_mail_daemon import mark_processed, processed_set, reset_on_uidvalidity_change
+
+
+class SortMailDaemonTests(unittest.TestCase):
+    def test_destination_from_header_routes_ai_provider(self):
+        header = (
+            b"From: Claude Team <hello@claude.com>\r\n"
+            b"Subject: Product update\r\n"
+            b"Date: Tue, 01 Jul 2025 10:00:00 +0000\r\n"
+            b"\r\n"
+        )
+        self.assertEqual(destination_from_header(header), "INBOX.Diensten.AI.Claude")
+
+    def test_destination_from_header_routes_invoice_by_quarter(self):
+        header = (
+            b"From: Billing <billing@example.com>\r\n"
+            b"Subject: Your invoice\r\n"
+            b"Date: Tue, 15 Apr 2025 10:00:00 +0000\r\n"
+            b"\r\n"
+        )
+        self.assertEqual(destination_from_header(header), "INBOX.Financieel.Facturen.2025.Q2")
+
+    def test_destination_from_header_returns_none_for_unmatched(self):
+        header = (
+            b"From: Person <person@example.invalid>\r\n"
+            b"Subject: Hello\r\n"
+            b"Date: Tue, 01 Jul 2025 10:00:00 +0000\r\n"
+            b"\r\n"
+        )
+        self.assertIsNone(destination_from_header(header))
+
+    def test_uidvalidity_change_resets_processed_uids(self):
+        state = {"folder_state": {"INBOX": {"uidvalidity": "1", "processed_uids": ["10"]}}}
+        reset_on_uidvalidity_change(state, "INBOX", "2")
+        self.assertEqual(processed_set(state, "INBOX"), set())
+        self.assertEqual(state["folder_state"]["INBOX"]["uidvalidity"], "2")
+
+    def test_mark_processed_is_idempotent(self):
+        state = {"folder_state": {}}
+        mark_processed(state, "INBOX", "2")
+        mark_processed(state, "INBOX", "2")
+        mark_processed(state, "INBOX", "1")
+        self.assertEqual(state["folder_state"]["INBOX"]["processed_uids"], ["1", "2"])
+
+
+if __name__ == "__main__":
+    unittest.main()