oskar/homelab-codex-ws
1
0
Fork 0
Code Issues Pull requests Actions Packages Projects Releases Wiki Activity

feat(kb-mail): adapter IMAP + model stanu synca + migracja 005

Browse source 
Krok 7 fazy mailowej, warstwa wspoldzielona. Realizuje decyzje (a), (b), (f)
reconu kb/audits/mail-sync-2026-08-06.md (zatwierdzone przez operatora
2026-08-06): jeden adapter IMAP na oba konta, stan synca jako tabela w bazie.

Nowe moduly w packages/kb-mail:
- imap.py    — ImapAccount/ImapClient nad stdlib imaplib (zero nowych zaleznosci).
               Foldery otwierane READ-ONLY (EXAMINE) i pobierane przez BODY.PEEK[],
               zeby job nie ustawial \Seen na skrzynce operatora. Wybor folderu po
               atrybucie SPECIAL-USE, nigdy po nazwie — Gmail lokalizuje
               "[Gmail]/All Mail". search_from_uid filtruje zakres po stronie
               klienta, bo n:* zwraca ostatnia wiadomosc takze gdy przedzial pusty.
- sync_state.py — tabela mail_sync_state + czyste funkcje: plan_folder_sync
               (pierwszy tick / przyrost / uniewaznienie UIDVALIDITY) i
               contiguous_last_uid (kursor przesuwa sie tylko po nieprzerwanym
               ciagu sukcesow — bledna wiadomosc jest ponawiana, nie przeskakiwana).
- headers.py / message.py — parse_headers(+fallback) z gmail-header-backfill oraz
               message_id/parse_date/parse_attachments/eml_ref z gmail-bulk-import,
               przeniesione zamiast skopiowane. Klucz dedup musi pochodzic z jednej
               implementacji: kazdy insert przyrostowki trafia na 225 030 istniejacych
               id. Stare joby re-eksportuja te nazwy — ich CLI i testy bez zmian.

kb_mail.db.insert_envelope zwraca teraz command tag (+ rows_affected,
envelope_source) — bez tego nie da sie odroznic zwyklego duplikatu od kolizji
Message-ID miedzy kontami (recon §2.4).

Migracja 005_mail_sync_state.sql: addytywna, klucz (account, folder).

Testy: 285 passed (111 kb-mail w tym 37 adaptera IMAP na fake serwerze i 26
planera kursora; 174 istniejace suity jobow bez zmian po ekstrakcji).

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
oskar 2026-08-06 14:26:51 +02:00
parent 8031396f04
commit c65f0f2075
12 changed files with 1580 additions and 219 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

98
jobs/gmail-bulk-import/src/gmail_bulk_import/importer.py
View file

@ -29,12 +29,10 @@ from __future__ import annotations
import argparse
import asyncio
import email.policy
import hashlib
import json
import mailbox
import sys
from datetime import datetime, timezone
from email.utils import parsedate_to_datetime
from datetime import datetime
from pathlib import Path
from typing import Optional
@ -43,94 +41,26 @@ import structlog
from kb_mail.archive import save_eml
from kb_mail.envelope import Envelope
from kb_mail.text import sanitize_surrogates as _sanitize
# id / ts / attachment-manifest derivation moved to packages/kb-mail when jobs/mail-imap-sync
# needed the identical four functions (recon kb/audits/mail-sync-2026-08-06.md §2.1). The
# dedup key MUST come from ONE implementation: every insert the poller makes is matched
# against the 225 030 ids this job wrote. Aliased to the old private names so the rest of this
# module and its tests are untouched.
from kb_mail.message import EPOCH as _EPOCH
from kb_mail.message import eml_ref as _archive_ref
from kb_mail.message import message_id as _message_id # noqa: F401
from kb_mail.message import parse_attachments as _parse_attachments # noqa: F401
from kb_mail.message import parse_date as _parse_date # noqa: F401
_log = structlog.get_logger(__name__)
BATCH_SIZE = 500
_EPOCH = datetime(1970, 1, 1, tzinfo=timezone.utc)
# Mirrors archive._UNSAFE — used to reconstruct raw_ref for already-archived messages.
_ARCHIVE_UNSAFE = str.maketrans({"/": "_", "\\": "_", ":": "_", "<": "", ">": ""})
def _eml_ref(envelope_id: str, ts: datetime) -> str:
"""Compute the archive-relative .eml path without writing (mirrors save_eml logic)."""
safe_id = envelope_id.translate(_ARCHIVE_UNSAFE)
return f"gmail/{ts.year:04d}/{ts.month:02d}/{safe_id}.eml"
def _message_id(msg: mailbox.mboxMessage) -> str:
"""Return a stable envelope id from Message-ID header or SHA-256 content hash.
compat32 .get() returns an email.header.Header (not str) when the raw value
holds 8-bit bytes — str() + _sanitize keeps the id jsonb-safe and gives
.strip() a real string to work on. An unguarded .strip() on a Header raised
AttributeError and, because _message_id runs before the per-message try,
killed the whole run (the archive has proven 8-bit header bytes).
"""
mid = _sanitize(str(msg.get("Message-ID", ""))).strip().strip("<>")
if mid:
return mid
return "sha256-" + hashlib.sha256(msg.as_bytes()).hexdigest()[:32]
def _parse_date(msg: mailbox.mboxMessage) -> datetime:
"""Parse Date header to UTC-aware datetime. Returns epoch as last resort.
str() first: a compat32 8-bit Date surfaces as an email.header.Header, and
parsedate_to_datetime() raises on a Header (→ needless epoch_fallback) even
when str(header) is perfectly parseable.
"""
date_str = str(msg.get("Date", ""))
if date_str:
try:
dt = parsedate_to_datetime(date_str)
if dt.tzinfo is None:
dt = dt.replace(tzinfo=timezone.utc)
return dt.astimezone(timezone.utc)
except Exception:
pass
return _EPOCH
def _parse_attachments(msg: mailbox.mboxMessage) -> list[dict]:
"""Return attachment descriptors from MIME structure.
A part is an attachment if it has a filename, Content-Disposition: attachment,
or is not inline text/plain or text/html.
Errors on individual parts are logged and skipped — never propagated.
"""
attachments = []
for part in msg.walk():
if part.get_content_maintype() == "multipart":
continue
content_type = part.get_content_type()
disposition = part.get_content_disposition() or ""
filename = part.get_filename()
if (
content_type in ("text/plain", "text/html")
and disposition != "attachment"
and not filename
):
continue
try:
payload = part.get_payload(decode=True)
if payload is None:
continue
attachments.append({
"type": "attachment",
"filename": filename,
"content_type": content_type,
"size": len(payload),
"sha256": hashlib.sha256(payload).hexdigest(),
})
except Exception:
_log.warning("attachment_parse_failed",
filename=filename, content_type=content_type)
return attachments
"""Archive-relative .eml path for an already-archived gmail message (source is fixed here;
`kb_mail.message.eml_ref` takes it as a parameter for the multi-account poller)."""
return _archive_ref(envelope_id, "gmail", ts)
async def _insert_batch(conn: asyncpg.Connection, envs: list[Envelope]) -> None:

136
jobs/gmail-header-backfill/src/gmail_header_backfill/backfill.py
View file

@ -34,19 +34,18 @@ from __future__ import annotations
import argparse
import asyncio
import email
import email.policy
import email.utils
import json
import os
import sys
from pathlib import Path
from typing import Optional
import asyncpg
import structlog
from kb_mail.text import sanitize_surrogates as _sanitize
# Both parsers moved to packages/kb-mail when jobs/mail-imap-sync needed them too (recon
# kb/audits/mail-sync-2026-08-06.md §6 poz. 3). Re-exported here so this job's CLI, its tests,
# and every existing `from gmail_header_backfill.backfill import parse_headers` keep working.
from kb_mail.headers import parse_headers, parse_headers_fallback # noqa: F401
_log = structlog.get_logger(__name__)
@ -64,133 +63,6 @@ WHERE id = $1
"""
def parse_headers(raw: bytes) -> dict:
"""Parse Gmail .eml headers into the `{"type": "headers", ...}` shape (plan §4.1).
`from`/`to`/`cc` are decoded via `email.policy.default` (resolves RFC 2047
encoded-words to real Unicode) and then split into {name, address} pairs with
`email.utils.getaddresses`. `delivered_to` is kept as a list of raw (decoded but
unparsed) strings — `Delivered-To` may repeat per hop, and every occurrence is
kept (see plan §1.8/§4.1). `date_raw` comes from a separate compat32 parse: the
default policy's DateHeader reformats the header (e.g. corrects the weekday name,
zero-pads the day) instead of preserving the literal original text.
"""
msg = email.message_from_bytes(raw, policy=email.policy.default)
msg_compat = email.message_from_bytes(raw, policy=email.policy.compat32)
from_headers = msg.get_all("From") or []
if len(from_headers) > 1:
_log.warning("headers.multiple_from", count=len(from_headers))
from_obj: Optional[dict] = None
if from_headers:
from_addrs = email.utils.getaddresses([str(from_headers[0])])
if from_addrs and from_addrs[0][1]:
name, address = from_addrs[0]
from_obj = {"name": name or None, "address": address}
to_list = [
{"name": name or None, "address": address}
for name, address in email.utils.getaddresses(
[str(h) for h in (msg.get_all("To") or [])]
)
if address
]
cc_list = [
{"name": name or None, "address": address}
for name, address in email.utils.getaddresses(
[str(h) for h in (msg.get_all("Cc") or [])]
)
if address
]
delivered_to = [str(h) for h in (msg.get_all("Delivered-To") or [])]
subject_header = msg.get("Subject")
subject = str(subject_header) if subject_header is not None else None
# compat32 .get() returns an email.header.Header (not str) when the raw
# value contains 8-bit bytes — str() + _sanitize keeps it JSON/jsonb-safe.
# An unguarded Header here crashed the original full run mid-slice
# (TypeError at json.dumps), silently losing the rest of the slice.
date_header = msg_compat.get("Date")
date_raw = _sanitize(str(date_header)) if date_header is not None else None
return {
"type": "headers",
"from": from_obj,
"to": to_list,
"cc": cc_list,
"delivered_to": delivered_to,
"subject": subject,
"date_raw": date_raw,
}
def parse_headers_fallback(raw: bytes) -> dict:
"""Degraded parse for messages the typed (policy.default) path rejects.
Real-world triggers found in the archive (diagnosis 2026-07-14, 9 of
225 030): RFC 2047 encoded-words that decode to text with CR/LF in a
display name (ValueError in headerregistry), group syntax in To:
("unlisted-recipients:;"), and a _header_value_parser bug on malformed
display names (fixed in newer CPython, present on PIHA's 3.11).
Everything comes from a compat32 parse: address fields are split into
{name, address} by email.utils.getaddresses over the RAW header text — no
RFC 2047 decoding, so names may keep literal =?...?= encoded-words.
Subject and delivered_to stay raw strings too. The returned shape is
exactly parse_headers()'s §4.1 shape — only parse QUALITY degrades, never
the schema. Callers count these separately (stats["parsed_fallback"]).
"""
msg = email.message_from_bytes(raw, policy=email.policy.compat32)
from_headers = [str(h) for h in (msg.get_all("From") or [])]
if len(from_headers) > 1:
_log.warning("headers.multiple_from", count=len(from_headers))
from_obj: Optional[dict] = None
if from_headers:
from_addrs = email.utils.getaddresses([from_headers[0]])
if from_addrs and from_addrs[0][1]:
name, address = from_addrs[0]
from_obj = {"name": _sanitize(name) or None, "address": _sanitize(address)}
to_list = [
{"name": _sanitize(name) or None, "address": _sanitize(address)}
for name, address in email.utils.getaddresses(
[str(h) for h in (msg.get_all("To") or [])]
)
if address
]
cc_list = [
{"name": _sanitize(name) or None, "address": _sanitize(address)}
for name, address in email.utils.getaddresses(
[str(h) for h in (msg.get_all("Cc") or [])]
)
if address
]
delivered_to = [_sanitize(str(h)) for h in (msg.get_all("Delivered-To") or [])]
subject_header = msg.get("Subject")
subject = _sanitize(str(subject_header)) if subject_header is not None else None
date_raw = _sanitize(msg.get("Date"))
return {
"type": "headers",
"from": from_obj,
"to": to_list,
"cc": cc_list,
"delivered_to": delivered_to,
"subject": subject,
"date_raw": date_raw,
}
def _decode_jsonb(value: object) -> object:
"""asyncpg may return jsonb as a str or an already-decoded object."""
if value is None:

31
packages/kb-mail/src/kb_mail/db.py
View file

@ -12,10 +12,15 @@ from .envelope import Envelope
_log = structlog.get_logger(__name__)
async def insert_envelope(conn: asyncpg.Connection, env: Envelope) -> None:
"""Insert envelope into kb-postgres. Silently ignores duplicate ids (ON CONFLICT DO NOTHING)."""
async def insert_envelope(conn: asyncpg.Connection, env: Envelope) -> str:
"""Insert envelope into kb-postgres. Silently ignores duplicate ids (ON CONFLICT DO NOTHING).
Returns the postgres command tag (`INSERT 0 1` on a real insert, `INSERT 0 0` when the id
was already present) so a caller that needs to tell "new" from "already had it" can, via
`rows_affected()`. Callers that don't care simply ignore it — this used to return None.
"""
geo_param = json.dumps(env.geo) if env.geo is not None else None
await conn.execute(
tag = await conn.execute(
"""
INSERT INTO envelope (id, source, ts, geo, raw_ref, entities)
VALUES ($1, $2, $3, $4::jsonb, $5, $6::jsonb)
@ -29,6 +34,26 @@ async def insert_envelope(conn: asyncpg.Connection, env: Envelope) -> None:
json.dumps(env.entities),
)
_log.info("envelope.inserted", id=env.id, source=env.source)
return tag
def rows_affected(command_tag: str) -> int:
"""Row count out of a postgres command tag (`INSERT 0 1` -> 1)."""
return int(command_tag.rsplit(" ", 1)[-1])
async def envelope_source(conn: asyncpg.Connection, envelope_id: str) -> Optional[str]:
"""The `source` of an existing envelope, or None if there is no such row.
Used to classify a suppressed insert. `envelope.id` is a bare Message-ID and therefore
GLOBALLY unique across accounts, so a mail present in both mailboxes (list traffic, a CC
to both addresses, a forward) is stored once under whichever source inserted it first —
correct for the index, but it means "new fastmail mails" is systematically undercounted by
the shared part. The recon's answer (§2.4) is to keep the behaviour and make the
phenomenon visible: count it as `envelopes_conflict_other_source` instead of letting it
hide inside an ordinary duplicate count.
"""
return await conn.fetchval("SELECT source FROM envelope WHERE id = $1", envelope_id)
async def get_envelope(conn: asyncpg.Connection, envelope_id: str) -> Optional[Envelope]:

168
packages/kb-mail/src/kb_mail/headers.py Normal file
View file

@ -0,0 +1,168 @@
"""Header parsing shared by every job that turns `.eml` bytes into an
`entities[type=headers]` object (kb/phases/kb-m5-faza2.md §4.1).
Extracted from `gmail_header_backfill.backfill` when `jobs/mail-imap-sync` needed the same
two functions (recon `kb/audits/mail-sync-2026-08-06.md` §6 poz. 3): a poller that inserts
envelopes without headers would give every new mail the chunk prefix
`Temat: (brak tematu) | Od: ?` — silently, with no error, just worse retrieval (§2.5 i).
Same reason `kb_mail.chunking` was extracted in Krok 0: one tested implementation, not a copy.
`gmail_header_backfill.backfill` re-exports both names, so its own tests and CLI are unchanged.
"""
from __future__ import annotations
import email
import email.policy
import email.utils
from typing import Optional
import structlog
from .text import sanitize_surrogates as _sanitize
_log = structlog.get_logger(__name__)
def parse_headers(raw: bytes) -> dict:
"""Parse `.eml` headers into the `{"type": "headers", ...}` shape (plan §4.1).
`from`/`to`/`cc` are decoded via `email.policy.default` (resolves RFC 2047
encoded-words to real Unicode) and then split into {name, address} pairs with
`email.utils.getaddresses`. `delivered_to` is kept as a list of raw (decoded but
unparsed) strings — `Delivered-To` may repeat per hop, and every occurrence is
kept (see plan §1.8/§4.1). `date_raw` comes from a separate compat32 parse: the
default policy's DateHeader reformats the header (e.g. corrects the weekday name,
zero-pads the day) instead of preserving the literal original text.
"""
msg = email.message_from_bytes(raw, policy=email.policy.default)
msg_compat = email.message_from_bytes(raw, policy=email.policy.compat32)
from_headers = msg.get_all("From") or []
if len(from_headers) > 1:
_log.warning("headers.multiple_from", count=len(from_headers))
from_obj: Optional[dict] = None
if from_headers:
from_addrs = email.utils.getaddresses([str(from_headers[0])])
if from_addrs and from_addrs[0][1]:
name, address = from_addrs[0]
from_obj = {"name": name or None, "address": address}
to_list = [
{"name": name or None, "address": address}
for name, address in email.utils.getaddresses(
[str(h) for h in (msg.get_all("To") or [])]
)
if address
]
cc_list = [
{"name": name or None, "address": address}
for name, address in email.utils.getaddresses(
[str(h) for h in (msg.get_all("Cc") or [])]
)
if address
]
delivered_to = [str(h) for h in (msg.get_all("Delivered-To") or [])]
subject_header = msg.get("Subject")
subject = str(subject_header) if subject_header is not None else None
# compat32 .get() returns an email.header.Header (not str) when the raw
# value contains 8-bit bytes — str() + _sanitize keeps it JSON/jsonb-safe.
# An unguarded Header here crashed the original full backfill run mid-slice
# (TypeError at json.dumps), silently losing the rest of the slice.
date_header = msg_compat.get("Date")
date_raw = _sanitize(str(date_header)) if date_header is not None else None
return {
"type": "headers",
"from": from_obj,
"to": to_list,
"cc": cc_list,
"delivered_to": delivered_to,
"subject": subject,
"date_raw": date_raw,
}
def parse_headers_fallback(raw: bytes) -> dict:
"""Degraded parse for messages the typed (policy.default) path rejects.
Real-world triggers found in the archive (diagnosis 2026-07-14, 9 of
225 030): RFC 2047 encoded-words that decode to text with CR/LF in a
display name (ValueError in headerregistry), group syntax in To:
("unlisted-recipients:;"), and a _header_value_parser bug on malformed
display names (fixed in newer CPython, present on PIHA's 3.11).
Everything comes from a compat32 parse: address fields are split into
{name, address} by email.utils.getaddresses over the RAW header text — no
RFC 2047 decoding, so names may keep literal =?...?= encoded-words.
Subject and delivered_to stay raw strings too. The returned shape is
exactly parse_headers()'s §4.1 shape — only parse QUALITY degrades, never
the schema. Callers count these separately (stats["parsed_fallback"] /
stats["headers_fallback"]).
"""
msg = email.message_from_bytes(raw, policy=email.policy.compat32)
from_headers = [str(h) for h in (msg.get_all("From") or [])]
if len(from_headers) > 1:
_log.warning("headers.multiple_from", count=len(from_headers))
from_obj: Optional[dict] = None
if from_headers:
from_addrs = email.utils.getaddresses([from_headers[0]])
if from_addrs and from_addrs[0][1]:
name, address = from_addrs[0]
from_obj = {"name": _sanitize(name) or None, "address": _sanitize(address)}
to_list = [
{"name": _sanitize(name) or None, "address": _sanitize(address)}
for name, address in email.utils.getaddresses(
[str(h) for h in (msg.get_all("To") or [])]
)
if address
]
cc_list = [
{"name": _sanitize(name) or None, "address": _sanitize(address)}
for name, address in email.utils.getaddresses(
[str(h) for h in (msg.get_all("Cc") or [])]
)
if address
]
delivered_to = [_sanitize(str(h)) for h in (msg.get_all("Delivered-To") or [])]
subject_header = msg.get("Subject")
subject = _sanitize(str(subject_header)) if subject_header is not None else None
date_raw = _sanitize(msg.get("Date"))
return {
"type": "headers",
"from": from_obj,
"to": to_list,
"cc": cc_list,
"delivered_to": delivered_to,
"subject": subject,
"date_raw": date_raw,
}
def parse_headers_resilient(raw: bytes) -> tuple[dict, bool]:
"""`parse_headers` with the compat32 fallback already wired in.
Returns `(headers_entity, used_fallback)`. The poller inserts envelopes one at a time and
must never lose a mail to a header quirk, so it always takes this path; the backfill job
keeps its own inline try/except because it counts the two outcomes into different slice
stats. A message neither parser can handle raises — the caller decides whether that is a
skipped message or a failed run.
"""
try:
return parse_headers(raw), False
except Exception as typed_exc:
entity = parse_headers_fallback(raw)
_log.info("headers.parsed_fallback", typed_error=repr(typed_exc))
return entity, True

413
packages/kb-mail/src/kb_mail/imap.py Normal file
View file

@ -0,0 +1,413 @@
"""IMAP transport for the mail pillar — one adapter, two accounts (gmail + fastmail).
Realizes decisions (a) and (b) of `kb/audits/mail-sync-2026-08-06.md`, approved by the
operator 2026-08-06: both live mailboxes are read over plain IMAP with an app password, not
Gmail API and not JMAP. The reason is written down in `kb/subsystems/kb-overview.md` §61 —
*"protokół, nie provider"* — and it has a concrete payoff here: one class, one test suite, one
8-bit-header hardening path, instead of two transports producing identical rows.
Deliberately synchronous. `imaplib` is stdlib (PIHA runs 3.11.2), so the poller adds **zero
dependencies**; at ~37 messages a day an async client would buy nothing but a third-party
package to keep patched. The job runs these calls under `asyncio.to_thread` so its asyncpg
side stays async.
Two properties this module is responsible for, both easy to get silently wrong:
* **Never mutate the mailbox.** Folders are opened with `EXAMINE` (`readonly=True`) and
bodies fetched with `BODY.PEEK[]`. A plain `SELECT` + `FETCH RFC822` would set `\\Seen` on
every message the poller touches — turning a read-only KB job into something that visibly
edits the operator's inbox.
* **Never hardcode a folder name.** Gmail localizes `[Gmail]/All Mail` (Polish UI:
`[Gmail]/Wszystkie`), so the folder is resolved by its SPECIAL-USE attribute (`\\All`) from
the `LIST` response. Literal names stay available for Fastmail's classic `INBOX`/`Archive`/
`Sent` layout (recon Decyzja (e)).
Folder names are carried as the exact string the server returned. Non-ASCII names are IMAP
modified-UTF-7 on the wire and are neither decoded nor re-encoded here — they only ever need
to round-trip back to the server and into `mail_sync_state.folder`, which they do byte-for-byte.
"""
from __future__ import annotations
import imaplib
import re
import ssl
from dataclasses import dataclass, field
from datetime import date
from typing import Callable, Optional
import structlog
_log = structlog.get_logger(__name__)
DEFAULT_PORT = 993
DEFAULT_TIMEOUT_S = 60.0
INITIAL_MODES = ("new-only", "since", "full")
# IMAP SEARCH dates are `dd-Mon-yyyy` with English month abbreviations regardless of locale.
_IMAP_MONTHS = ("Jan", "Feb", "Mar", "Apr", "May", "Jun",
"Jul", "Aug", "Sep", "Oct", "Nov", "Dec")
_LIST_RE = re.compile(
rb'^\((?P<attrs>[^)]*)\)\s+(?P<delim>"(?:[^"\\]|\\.)*"|NIL)\s+(?P<name>.+)$'
)
_UID_IN_FETCH_RE = re.compile(rb"\bUID\s+(\d+)")
class ImapError(RuntimeError):
"""An IMAP command returned a non-OK status, or a response could not be parsed.
Raised instead of returning a sentinel so a folder whose protocol exchange went wrong
fails loudly and leaves `mail_sync_state` untouched — the next tick then retries from the
same cursor rather than skipping a range nobody ever fetched.
"""
@dataclass(frozen=True)
class ImapAccount:
"""One mailbox's connection details and sync scope.
`name` doubles as `envelope.source` and `mail_sync_state.account` — 'gmail' | 'fastmail'.
`password` is `repr=False` on purpose: this dataclass is passed around the job and dumped
into structlog context in places, and the repo has two documented secret leaks to session
transcripts already (recon Decyzja (c) rule 2).
"""
name: str
host: str
user: str
password: str = field(repr=False)
port: int = DEFAULT_PORT
folders: tuple[str, ...] = () # literal mailbox names (fastmail)
special_use: tuple[str, ...] = () # SPECIAL-USE attributes to resolve (gmail: '\\All')
initial_mode: str = "new-only" # what the FIRST tick on a folder fetches
initial_since: Optional[date] = None # required when initial_mode == 'since'
def __post_init__(self) -> None:
if not self.folders and not self.special_use:
raise ValueError(
f"account {self.name}: no scope — set folders and/or special_use "
f"(recon Decyzja (e): gmail = \\All, fastmail = INBOX,Archive,Sent)"
)
if self.initial_mode not in INITIAL_MODES:
raise ValueError(
f"account {self.name}: initial_mode={self.initial_mode!r} "
f"must be one of {INITIAL_MODES}"
)
if self.initial_mode == "since" and self.initial_since is None:
raise ValueError(
f"account {self.name}: initial_mode='since' requires initial_since (YYYY-MM-DD)"
)
@dataclass(frozen=True)
class FolderStatus:
"""What the server says about a folder right now.
`uidnext` is the UID the *next* delivered message will get, so `uidnext - 1` is the
high-water mark a 'start from now' first tick records without fetching anything.
`messages` is only populated by `status()` (the sizing measurement); `select()` leaves it
None because EXAMINE's EXISTS count is not what the runbook asks for.
"""
name: str
uidvalidity: int
uidnext: int
messages: Optional[int] = None
def imap_date(value: date) -> str:
"""`dd-Mon-yyyy` as IMAP SEARCH wants it (`15-Jun-2026`), month names always English."""
return f"{value.day:02d}-{_IMAP_MONTHS[value.month - 1]}-{value.year:04d}"
def quote_mailbox(name: str) -> str:
"""Quote a mailbox name for the wire. imaplib passes mailbox arguments through verbatim,
so `[Gmail]/All Mail` without quotes is parsed by the server as two arguments."""
escaped = name.replace("\\", "\\\\").replace('"', '\\"')
return f'"{escaped}"'
def _unquote(raw: bytes) -> str:
text = raw.decode("utf-8", errors="replace").strip()
if len(text) >= 2 and text.startswith('"') and text.endswith('"'):
text = text[1:-1].replace('\\"', '"').replace("\\\\", "\\")
return text
def parse_list_line(line) -> Optional[tuple[frozenset[str], str]]:
"""One `LIST` response item -> (lowercased attribute set, mailbox name).
imaplib hands back either a bytes line or, when the server sent the name as a literal,
a `(prefix, literal)` tuple — both shapes occur in the wild, so both are handled.
Returns None for anything that does not parse as a LIST line (servers occasionally
interleave other untagged data).
"""
if isinstance(line, tuple):
prefix, literal = line[0], line[1]
match = _LIST_RE.match(prefix.rstrip(b"{0123456789}").strip())
if match is None:
head = _LIST_RE.match(prefix.strip())
if head is None:
return None
attrs = head.group("attrs")
else:
attrs = match.group("attrs")
name = literal.decode("utf-8", errors="replace")
else:
match = _LIST_RE.match(line.strip())
if match is None:
return None
attrs = match.group("attrs")
name = _unquote(match.group("name"))
attr_set = frozenset(
a.decode("ascii", errors="replace").lower()
for a in attrs.split()
)
return attr_set, name
class ImapClient:
"""Thin, read-only wrapper over `imaplib.IMAP4_SSL` for one account.
Use as a context manager — `__exit__` always attempts LOGOUT, so a folder that raises
mid-sync still closes its connection instead of leaning on the server's idle timeout.
`connection_factory` exists so the whole protocol layer is unit-testable against a fake
IMAP server: the tests exercise this class's real parsing, not a mock of it.
"""
def __init__(
self,
account: ImapAccount,
*,
timeout_s: float = DEFAULT_TIMEOUT_S,
connection_factory: Optional[Callable[[ImapAccount, float], object]] = None,
) -> None:
self.account = account
self.timeout_s = timeout_s
self._factory = connection_factory or _default_connection
self._imap = None
self._selected: Optional[str] = None
# -- lifecycle ---------------------------------------------------------------
def __enter__(self) -> "ImapClient":
self.connect()
return self
def __exit__(self, *exc_info) -> None:
self.logout()
def connect(self) -> None:
self._imap = self._factory(self.account, self.timeout_s)
# LOGIN failures raise imaplib.IMAP4.error; the message carries the server's
# rejection text, never the credential.
self._imap.login(self.account.user, self.account.password)
_log.info("imap.connected", account=self.account.name,
host=self.account.host, port=self.account.port)
def logout(self) -> None:
if self._imap is None:
return
try:
self._imap.logout()
except Exception:
_log.warning("imap.logout_failed", account=self.account.name)
finally:
self._imap = None
self._selected = None
def _conn(self):
if self._imap is None:
raise ImapError(f"account {self.account.name}: not connected")
return self._imap
def _check(self, command: str, result) -> list:
typ, data = result
if typ != "OK":
raise ImapError(
f"account {self.account.name}: {command} returned {typ}: {data!r}"
)
return data
# -- folder discovery --------------------------------------------------------
def list_folders(self) -> list[tuple[frozenset[str], str]]:
data = self._check("LIST", self._conn().list())
parsed = [parse_list_line(line) for line in data if line]
return [p for p in parsed if p is not None]
def resolve_folders(self) -> list[str]:
"""The account's configured scope as concrete mailbox names, in a stable order.
SPECIAL-USE attributes are resolved against `LIST`; a configured attribute the server
does not advertise is an error, not a silent skip — a gmail account whose `\\All`
vanished would otherwise sync nothing while reporting a clean run every hour.
Literal names are trusted as configured (a typo surfaces on SELECT, loudly).
Duplicates are collapsed, so listing a folder that also matches an attribute costs
nothing.
"""
resolved: list[str] = []
if self.account.special_use:
listing = self.list_folders()
for attr in self.account.special_use:
wanted = attr.lower()
matches = [name for attrs, name in listing if wanted in attrs]
if not matches:
advertised = sorted({a for attrs, _ in listing for a in attrs if a.startswith("\\")})
raise ImapError(
f"account {self.account.name}: no folder advertises SPECIAL-USE "
f"{attr!r}; server advertises {advertised}"
)
if len(matches) > 1:
_log.warning("imap.special_use_ambiguous", account=self.account.name,
attribute=attr, matches=matches)
resolved.append(matches[0])
resolved.extend(self.account.folders)
seen: set[str] = set()
unique = []
for name in resolved:
if name not in seen:
seen.add(name)
unique.append(name)
return unique
# -- folder state ------------------------------------------------------------
def status(self, folder: str) -> FolderStatus:
"""`STATUS (MESSAGES UIDNEXT UIDVALIDITY)` — read-only and cheap.
This is the measurement the runbook takes on Fastmail before deciding whether to pull
its history (recon Decyzja (e): decide on the number, do not guess it).
"""
data = self._check(
"STATUS",
self._conn().status(quote_mailbox(folder), "(MESSAGES UIDNEXT UIDVALIDITY)"),
)
blob = b" ".join(part for part in data if isinstance(part, bytes))
return FolderStatus(
name=folder,
uidvalidity=_status_field(blob, b"UIDVALIDITY", folder),
uidnext=_status_field(blob, b"UIDNEXT", folder),
messages=_status_field(blob, b"MESSAGES", folder),
)
def examine(self, folder: str) -> FolderStatus:
"""Open a folder READ-ONLY (EXAMINE) and return its UIDVALIDITY/UIDNEXT.
Both values normally arrive as untagged responses to EXAMINE; a server that omits
UIDNEXT there is covered by falling back to STATUS rather than by guessing, because
`uidnext - 1` is exactly the cursor a first tick persists.
"""
conn = self._conn()
self._check("EXAMINE", conn.select(quote_mailbox(folder), readonly=True))
self._selected = folder
uidvalidity = _untagged_int(conn, "UIDVALIDITY")
uidnext = _untagged_int(conn, "UIDNEXT")
if uidvalidity is None or uidnext is None:
fallback = self.status(folder)
uidvalidity = uidvalidity if uidvalidity is not None else fallback.uidvalidity
uidnext = uidnext if uidnext is not None else fallback.uidnext
# STATUS on some servers implicitly deselects; re-EXAMINE so the UID commands
# below still have a mailbox open.
self._check("EXAMINE", conn.select(quote_mailbox(folder), readonly=True))
return FolderStatus(name=folder, uidvalidity=int(uidvalidity), uidnext=int(uidnext))
# -- searching ---------------------------------------------------------------
def _uid_search(self, *criteria: str) -> list[int]:
data = self._check("UID SEARCH", self._conn().uid("SEARCH", None, *criteria))
uids: list[int] = []
for blob in data:
if not blob:
continue
uids.extend(int(token) for token in blob.split())
return sorted(uids)
def search_all(self) -> list[int]:
return self._uid_search("ALL")
def search_since(self, since: date) -> list[int]:
"""`UID SEARCH SINCE dd-Mon-yyyy` — filters on the server's INTERNALDATE (when the
message arrived), NOT the sender's `Date:` header. That is the right side of the
distinction the recon draws in §2.2 wariant C: a mail delivered today with a
month-old header date is still caught."""
return self._uid_search("SINCE", imap_date(since))
def search_from_uid(self, start_uid: int) -> list[int]:
"""UIDs at or above `start_uid`.
The client-side filter is not redundant: `n:*` is a RANGE, and when `n` exceeds the
highest existing UID the server resolves it as `highest:n` and returns the last
message anyway. Without the filter, the 'new mails this tick' counter would never
read zero on an idle mailbox and every observability claim built on it would be a
small, permanent lie (recon §2.2).
"""
uids = self._uid_search("UID", f"{start_uid}:*")
return [uid for uid in uids if uid >= start_uid]
# -- fetching ----------------------------------------------------------------
def fetch_message(self, uid: int) -> Optional[bytes]:
"""Raw RFC822 bytes for one UID via `BODY.PEEK[]` — never sets `\\Seen`.
Returns None when the UID no longer exists (deleted or expunged between SEARCH and
FETCH — an ordinary race on a live mailbox, not an error).
"""
data = self._check("UID FETCH", self._conn().uid("FETCH", str(uid), "(BODY.PEEK[])"))
for item in data:
if not isinstance(item, tuple) or len(item) < 2:
continue
prefix, payload = item[0], item[1]
match = _UID_IN_FETCH_RE.search(prefix or b"")
# A server that echoes a different UID than requested means the response stream
# is out of step with our request — refuse the payload rather than archive a
# message under the wrong cursor.
if match is not None and int(match.group(1)) != uid:
raise ImapError(
f"account {self.account.name}: FETCH for UID {uid} returned UID "
f"{int(match.group(1))}"
)
if payload:
return bytes(payload)
return None
def _default_connection(account: ImapAccount, timeout_s: float):
return imaplib.IMAP4_SSL(
host=account.host,
port=account.port,
ssl_context=ssl.create_default_context(),
timeout=timeout_s,
)
def _status_field(blob: bytes, key: bytes, folder: str) -> int:
match = re.search(key + rb"\s+(\d+)", blob)
if match is None:
raise ImapError(f"STATUS for {folder!r} has no {key.decode()}: {blob!r}")
return int(match.group(1))
def _untagged_int(conn, key: str) -> Optional[int]:
try:
typ, data = conn.response(key)
except Exception:
return None
if typ != key and typ != "OK":
return None
for blob in data or []:
if not blob:
continue
match = re.search(rb"(\d+)", blob if isinstance(blob, bytes) else bytes(blob))
if match:
return int(match.group(1))
return None

116
packages/kb-mail/src/kb_mail/message.py Normal file
View file

@ -0,0 +1,116 @@
"""Message-level derivations shared by every job that turns `.eml` bytes into an `Envelope`:
the envelope id, the timestamp, the attachment manifest, and the archive path they imply.
Extracted from `gmail_bulk_import.importer` when `jobs/mail-imap-sync` needed the identical
four derivations (recon `kb/audits/mail-sync-2026-08-06.md` §2.1) — the poller's dedup rests
entirely on producing byte-identical ids to the 225 030 rows the bulk import already wrote, so
a second implementation of `message_id()` is the one thing that must not exist.
`gmail_bulk_import.importer` re-exports these under its old private names, unchanged.
"""
from __future__ import annotations
import email.message
import hashlib
from datetime import datetime, timezone
from email.utils import parsedate_to_datetime
import structlog
from .text import sanitize_surrogates as _sanitize
_log = structlog.get_logger(__name__)
EPOCH = datetime(1970, 1, 1, tzinfo=timezone.utc)
# Mirrors archive._UNSAFE — used to reconstruct raw_ref for already-archived messages
# without writing them again.
_ARCHIVE_UNSAFE = str.maketrans({"/": "_", "\\": "_", ":": "_", "<": "", ">": ""})
def eml_ref(envelope_id: str, source: str, ts: datetime) -> str:
"""The archive-relative `.eml` path `save_eml` would write, computed without writing.
Kept in lockstep with `kb_mail.archive.save_eml` — callers use it on `FileExistsError`
("already archived") to fill `Envelope.raw_ref` for a message they did not just write.
"""
safe_id = envelope_id.translate(_ARCHIVE_UNSAFE)
return f"{source}/{ts.year:04d}/{ts.month:02d}/{safe_id}.eml"
def message_id(msg: email.message.Message) -> str:
"""Return a stable envelope id from the Message-ID header, or a SHA-256 content hash.
This IS the dedup key (`envelope.id`, `ON CONFLICT (id) DO NOTHING`), so its output must
stay byte-identical across jobs and years — 9 of the 225 030 existing rows carry the
`sha256-` form, the rest a bare Message-ID.
compat32 `.get()` returns an `email.header.Header` (not str) when the raw value holds
8-bit bytes — `str()` + `_sanitize` keeps the id jsonb-safe and gives `.strip()` a real
string to work on. An unguarded `.strip()` on a Header raised AttributeError and killed a
whole bulk-import run (the archive has proven 8-bit header bytes).
"""
mid = _sanitize(str(msg.get("Message-ID", ""))).strip().strip("<>")
if mid:
return mid
return "sha256-" + hashlib.sha256(msg.as_bytes()).hexdigest()[:32]
def parse_date(msg: email.message.Message) -> datetime:
"""Parse the Date header to a UTC-aware datetime. Returns `EPOCH` as last resort.
`str()` first: a compat32 8-bit Date surfaces as an `email.header.Header`, and
`parsedate_to_datetime()` raises on a Header (→ needless epoch fallback) even when
`str(header)` is perfectly parseable.
Note this is the SENDER's clock, which is why it must never be used as a sync cursor —
2 559 envelopes in the live corpus sit at epoch 1970 (recon §2.2 wariant C).
"""
date_str = str(msg.get("Date", ""))
if date_str:
try:
dt = parsedate_to_datetime(date_str)
if dt.tzinfo is None:
dt = dt.replace(tzinfo=timezone.utc)
return dt.astimezone(timezone.utc)
except Exception:
pass
return EPOCH
def parse_attachments(msg: email.message.Message) -> list[dict]:
"""Return `entities[type=attachment]` descriptors from the MIME structure.
A part is an attachment if it has a filename, `Content-Disposition: attachment`, or is not
inline `text/plain`/`text/html`. Errors on individual parts are logged and skipped — never
propagated, since one unwalkable part must not cost the whole message.
"""
attachments = []
for part in msg.walk():
if part.get_content_maintype() == "multipart":
continue
content_type = part.get_content_type()
disposition = part.get_content_disposition() or ""
filename = part.get_filename()
if (
content_type in ("text/plain", "text/html")
and disposition != "attachment"
and not filename
):
continue
try:
payload = part.get_payload(decode=True)
if payload is None:
continue
attachments.append({
"type": "attachment",
"filename": filename,
"content_type": content_type,
"size": len(payload),
"sha256": hashlib.sha256(payload).hexdigest(),
})
except Exception:
_log.warning("attachment_parse_failed",
filename=filename, content_type=content_type)
return attachments

167
packages/kb-mail/src/kb_mail/sync_state.py Normal file
View file

@ -0,0 +1,167 @@
"""Per-folder IMAP sync cursor: the `mail_sync_state` table (migration 005) plus the pure
logic that decides what a tick fetches and how far the cursor is allowed to move.
The rules encoded here are the whole correctness story of the poller, so they live as pure
functions with their own tests rather than inline in the job's loop:
* **UIDVALIDITY governs everything.** When the server changes it, every stored UID becomes
meaningless. The only safe response is to sweep the folder and lean on Message-ID dedup —
a poller that trusts `last_uid` across a UIDVALIDITY change loses mail with no error
anywhere (recon `kb/audits/mail-sync-2026-08-06.md` §2.2).
* **The cursor moves only over messages that are fully durable**, archive file and envelope
row both. It moves over a CONTIGUOUS prefix of successes, so a message that failed is
re-fetched next tick instead of being stepped over silently.
* **A first tick is a policy decision, not a default.** `new-only` starts the corpus from
now; `since` closes a known gap (gmail's corpus stops at 2026-06-19); `full` pulls a
mailbox's whole history. Which one is right depends on a measurement the runbook takes.
"""
from __future__ import annotations
from dataclasses import dataclass
from datetime import date, datetime
from typing import Iterable, Optional
import asyncpg
import structlog
from .imap import FolderStatus, ImapAccount
_log = structlog.get_logger(__name__)
# What a plan tells the caller to run against the server.
SEARCH_NONE = "none" # fetch nothing this tick; just record the high-water mark
SEARCH_ALL = "all" # UID SEARCH ALL — full sweep, dedup does the rest
SEARCH_SINCE = "since" # UID SEARCH SINCE <date>
SEARCH_FROM_UID = "from-uid" # UID SEARCH UID <start>:*
@dataclass(frozen=True)
class FolderSyncState:
"""One row of `mail_sync_state`."""
account: str
folder: str
uidvalidity: int
last_uid: int
last_sync_ts: Optional[datetime] = None
@dataclass(frozen=True)
class SyncPlan:
"""What this tick will do to one folder.
`baseline_uid` is the cursor to persist when the plan turns up NO messages — never a
floor to combine with partial results, because on a full sweep it equals `uidnext - 1`
and combining it with a partially-failed batch would step over everything that failed.
"""
mode: str
search: str
baseline_uid: int
start_uid: Optional[int] = None
since: Optional[date] = None
uidvalidity_reset: bool = False
def plan_folder_sync(
account: ImapAccount,
status: FolderStatus,
state: Optional[FolderSyncState],
) -> SyncPlan:
"""Decide this tick's work for one folder from the stored cursor and the server's state."""
high_water = max(status.uidnext - 1, 0)
if state is None:
if account.initial_mode == "new-only":
# Start the corpus here: record the high-water mark, fetch nothing. Everything
# already in the mailbox stays out of KB unless the operator re-runs with a
# different initial_mode after clearing the row.
return SyncPlan(mode="initial-new-only", search=SEARCH_NONE, baseline_uid=high_water)
if account.initial_mode == "since":
return SyncPlan(mode="initial-since", search=SEARCH_SINCE,
baseline_uid=high_water, since=account.initial_since)
return SyncPlan(mode="initial-full", search=SEARCH_ALL, baseline_uid=high_water)
if state.uidvalidity != status.uidvalidity:
_log.warning("imap.uidvalidity_reset", account=account.name, folder=status.name,
stored=state.uidvalidity, server=status.uidvalidity)
return SyncPlan(mode="uidvalidity-reset", search=SEARCH_ALL,
baseline_uid=high_water, uidvalidity_reset=True)
return SyncPlan(mode="incremental", search=SEARCH_FROM_UID,
baseline_uid=state.last_uid, start_uid=state.last_uid + 1)
def contiguous_last_uid(floor: int, attempted: Iterable[int], succeeded: set[int]) -> int:
"""How far the cursor may move: the last UID of the unbroken run of successes.
A failure stops the cursor at the message before it. The consequences are deliberate and
worth naming: nothing is ever lost, the re-fetch costs nothing (dedup), and a message
that fails *permanently* stalls its folder — visibly, as a non-zero error counter every
tick and a cursor that stops moving, with the offending UID in the log. That is strictly
better than the alternative of stepping over it, which drops mail while reporting success.
The runbook documents the manual `UPDATE mail_sync_state` escape hatch for that case.
"""
last = floor
for uid in sorted(attempted):
if uid not in succeeded:
break
last = max(last, uid)
return last
async def get_folder_state(
conn: asyncpg.Connection, account: str, folder: str
) -> Optional[FolderSyncState]:
row = await conn.fetchrow(
"SELECT account, folder, uidvalidity, last_uid, last_sync_ts "
"FROM mail_sync_state WHERE account = $1 AND folder = $2",
account, folder,
)
if row is None:
return None
return FolderSyncState(
account=row["account"],
folder=row["folder"],
uidvalidity=int(row["uidvalidity"]),
last_uid=int(row["last_uid"]),
last_sync_ts=row["last_sync_ts"],
)
async def upsert_folder_state(
conn: asyncpg.Connection, account: str, folder: str, uidvalidity: int, last_uid: int
) -> None:
"""Persist the cursor and stamp `last_sync_ts`.
Called on every tick including empty ones: `last_sync_ts` is how one answers "is this
folder being polled at all", which stays true when nothing has arrived for a week.
"""
await conn.execute(
"""
INSERT INTO mail_sync_state (account, folder, uidvalidity, last_uid, last_sync_ts)
VALUES ($1, $2, $3, $4, now())
ON CONFLICT (account, folder) DO UPDATE
SET uidvalidity = EXCLUDED.uidvalidity,
last_uid = EXCLUDED.last_uid,
last_sync_ts = EXCLUDED.last_sync_ts
""",
account, folder, uidvalidity, last_uid,
)
_log.info("sync_state.saved", account=account, folder=folder,
uidvalidity=uidvalidity, last_uid=last_uid)
async def fetch_all_state(conn: asyncpg.Connection) -> list[FolderSyncState]:
"""Every cursor, for the runbook's inspection queries and the job's summary line."""
rows = await conn.fetch(
"SELECT account, folder, uidvalidity, last_uid, last_sync_ts "
"FROM mail_sync_state ORDER BY account, folder"
)
return [
FolderSyncState(
account=r["account"], folder=r["folder"], uidvalidity=int(r["uidvalidity"]),
last_uid=int(r["last_uid"]), last_sync_ts=r["last_sync_ts"],
)
for r in rows
]

330
packages/kb-mail/tests/test_imap.py Normal file
View file

@ -0,0 +1,330 @@
"""Unit tests for the IMAP adapter — no network, no real imaplib connection.
`ImapClient` is exercised against `_FakeIMAP`, a stand-in for `imaplib.IMAP4_SSL` that speaks
real IMAP response shapes (bytes lines, `(prefix, literal)` FETCH tuples, untagged responses).
The point is to test this module's OWN parsing rather than a mock of it — every response shape
here is one imaplib actually hands back.
"""
from __future__ import annotations
from datetime import date
import pytest
from kb_mail.imap import (
DEFAULT_PORT,
ImapAccount,
ImapClient,
ImapError,
imap_date,
parse_list_line,
quote_mailbox,
)
class _FakeIMAP:
"""Minimal IMAP4_SSL stand-in. Records commands so tests can assert on the wire traffic —
notably that folders are opened READ-ONLY and bodies fetched with BODY.PEEK[]."""
def __init__(
self,
*,
list_data=None,
messages=None,
uidvalidity=42,
uidnext=101,
status_line=None,
search_ok=True,
select_ok=True,
untagged=True,
):
self.list_data = list_data if list_data is not None else [
b'(\\HasNoChildren) "/" "INBOX"',
b'(\\HasNoChildren \\All) "/" "[Gmail]/Wszystkie"',
b'(\\HasNoChildren \\Sent) "/" "[Gmail]/Wys&AWI-ane"',
]
self.messages = messages or {}
self.uidvalidity = uidvalidity
self.uidnext = uidnext
self.status_line = status_line
self.search_ok = search_ok
self.select_ok = select_ok
self.untagged = untagged
self.commands: list[tuple] = []
self.logged_in = False
self.logged_out = False
def login(self, user, password):
self.commands.append(("LOGIN", user))
self.logged_in = True
return ("OK", [b"LOGIN completed"])
def logout(self):
self.logged_out = True
return ("BYE", [b"logging out"])
def list(self, directory='""', pattern="*"):
self.commands.append(("LIST",))
return ("OK", self.list_data)
def select(self, mailbox, readonly=False):
self.commands.append(("SELECT", mailbox, readonly))
if not self.select_ok:
return ("NO", [b"no such mailbox"])
return ("OK", [str(len(self.messages)).encode()])
def status(self, mailbox, names):
self.commands.append(("STATUS", mailbox, names))
line = self.status_line or (
f'"{mailbox}" (MESSAGES {len(self.messages)} '
f"UIDNEXT {self.uidnext} UIDVALIDITY {self.uidvalidity})"
).encode()
return ("OK", [line])
def response(self, key):
if not self.untagged:
return (key, [None])
if key == "UIDVALIDITY":
return (key, [str(self.uidvalidity).encode()])
if key == "UIDNEXT":
return (key, [str(self.uidnext).encode()])
return (key, [None])
def uid(self, command, *args):
self.commands.append(("UID", command) + args)
if command == "SEARCH":
if not self.search_ok:
return ("NO", [b"search failed"])
criteria = args[1:]
return ("OK", [self._search(criteria)])
if command == "FETCH":
uid = int(args[0])
raw = self.messages.get(uid)
if raw is None:
return ("OK", [None])
prefix = f"1 (UID {uid} BODY[] {{{len(raw)}}}".encode()
return ("OK", [(prefix, raw), b")"])
raise AssertionError(f"unexpected UID command {command}")
def _search(self, criteria):
uids = sorted(self.messages)
if criteria and criteria[0] == "UID":
start = int(criteria[1].split(":")[0])
hits = [u for u in uids if u >= start]
# Real servers resolve `n:*` as a range and return the highest message even when
# n is beyond it. Reproduced deliberately — the client-side filter exists for this.
if not hits and uids:
hits = [uids[-1]]
return " ".join(str(u) for u in hits).encode()
return " ".join(str(u) for u in uids).encode()
def _account(**kwargs) -> ImapAccount:
base = dict(name="gmail", host="imap.example.com", user="u@example.com",
password="secret", special_use=("\\All",))
base.update(kwargs)
return ImapAccount(**base)
def _client(fake: _FakeIMAP, account=None) -> ImapClient:
account = account or _account()
client = ImapClient(account, connection_factory=lambda acc, timeout: fake)
client.connect()
return client
class TestImapAccount:
def test_defaults_to_implicit_tls_port(self):
assert _account().port == DEFAULT_PORT
def test_password_is_not_in_repr(self):
# The repo has two documented secret leaks to session transcripts (recon Decyzja (c)).
assert "secret" not in repr(_account())
def test_rejects_account_with_no_scope(self):
with pytest.raises(ValueError, match="no scope"):
ImapAccount(name="x", host="h", user="u", password="p")
def test_rejects_unknown_initial_mode(self):
with pytest.raises(ValueError, match="initial_mode"):
_account(initial_mode="yesterday")
def test_since_mode_requires_a_date(self):
with pytest.raises(ValueError, match="initial_since"):
_account(initial_mode="since")
def test_since_mode_with_date_is_valid(self):
assert _account(initial_mode="since", initial_since=date(2026, 6, 15)).initial_since
class TestHelpers:
def test_imap_date_uses_english_month_abbreviations(self):
assert imap_date(date(2026, 6, 15)) == "15-Jun-2026"
def test_imap_date_zero_pads_the_day(self):
assert imap_date(date(2026, 12, 5)) == "05-Dec-2026"
def test_quote_mailbox_wraps_names_with_spaces(self):
assert quote_mailbox("[Gmail]/All Mail") == '"[Gmail]/All Mail"'
def test_quote_mailbox_escapes_quotes_and_backslashes(self):
assert quote_mailbox('we"ird\\name') == '"we\\"ird\\\\name"'
class TestParseListLine:
def test_quoted_name_with_attributes(self):
attrs, name = parse_list_line(b'(\\HasNoChildren \\All) "/" "[Gmail]/All Mail"')
assert name == "[Gmail]/All Mail"
assert "\\all" in attrs
def test_unquoted_atom_name(self):
attrs, name = parse_list_line(b'(\\HasNoChildren) "/" INBOX')
assert name == "INBOX"
def test_nil_delimiter(self):
attrs, name = parse_list_line(b'(\\Noselect) NIL "Archive"')
assert name == "Archive"
def test_literal_name_arrives_as_a_tuple(self):
attrs, name = parse_list_line((b'(\\HasNoChildren \\Archive) "/" {7}', b"Archiwa"))
assert name == "Archiwa"
assert "\\archive" in attrs
def test_unparseable_line_returns_none(self):
assert parse_list_line(b"* SOMETHING ELSE") is None
class TestResolveFolders:
def test_special_use_is_matched_by_attribute_not_name(self):
# The Polish-UI name is what makes hardcoding '[Gmail]/All Mail' a silent zero-mail sync.
fake = _FakeIMAP()
assert _client(fake).resolve_folders() == ["[Gmail]/Wszystkie"]
def test_literal_folders_are_used_as_configured(self):
fake = _FakeIMAP()
account = _account(name="fastmail", special_use=(),
folders=("INBOX", "Archive", "Sent"))
assert _client(fake, account).resolve_folders() == ["INBOX", "Archive", "Sent"]
def test_missing_special_use_raises_instead_of_syncing_nothing(self):
fake = _FakeIMAP(list_data=[b'(\\HasNoChildren) "/" "INBOX"'])
with pytest.raises(ImapError, match="SPECIAL-USE"):
_client(fake).resolve_folders()
def test_duplicates_between_attribute_and_literal_are_collapsed(self):
fake = _FakeIMAP()
account = _account(special_use=("\\All",), folders=("[Gmail]/Wszystkie", "INBOX"))
assert _client(fake, account).resolve_folders() == ["[Gmail]/Wszystkie", "INBOX"]
def test_no_list_call_when_only_literal_folders_are_configured(self):
fake = _FakeIMAP()
account = _account(special_use=(), folders=("INBOX",))
_client(fake, account).resolve_folders()
assert not any(c[0] == "LIST" for c in fake.commands)
class TestExamine:
def test_opens_folder_read_only(self):
# A plain SELECT + FETCH RFC822 would set \Seen on the operator's mail.
fake = _FakeIMAP(messages={1: b"raw"})
_client(fake).examine("INBOX")
select_cmds = [c for c in fake.commands if c[0] == "SELECT"]
assert select_cmds and select_cmds[0][2] is True
def test_returns_uidvalidity_and_uidnext_from_untagged_responses(self):
fake = _FakeIMAP(uidvalidity=7, uidnext=55)
status = _client(fake).examine("INBOX")
assert (status.uidvalidity, status.uidnext) == (7, 55)
assert not any(c[0] == "STATUS" for c in fake.commands)
def test_falls_back_to_status_when_server_omits_untagged_values(self):
fake = _FakeIMAP(uidvalidity=9, uidnext=30, untagged=False)
status = _client(fake).examine("INBOX")
assert (status.uidvalidity, status.uidnext) == (9, 30)
assert any(c[0] == "STATUS" for c in fake.commands)
# ... and re-opens the mailbox, since STATUS can deselect it.
assert len([c for c in fake.commands if c[0] == "SELECT"]) == 2
def test_non_ok_select_raises(self):
fake = _FakeIMAP(select_ok=False)
with pytest.raises(ImapError, match="EXAMINE"):
_client(fake).examine("Nope")
class TestStatus:
def test_parses_message_count_for_the_sizing_measurement(self):
fake = _FakeIMAP(messages={1: b"a", 2: b"b"}, uidnext=3, uidvalidity=11)
status = _client(fake).status("INBOX")
assert (status.messages, status.uidnext, status.uidvalidity) == (2, 3, 11)
def test_missing_field_raises_rather_than_defaulting(self):
fake = _FakeIMAP(status_line=b'"INBOX" (MESSAGES 2)')
with pytest.raises(ImapError, match="has no UIDVALIDITY"):
_client(fake).status("INBOX")
class TestSearch:
def test_search_all_returns_sorted_uids(self):
fake = _FakeIMAP(messages={3: b"c", 1: b"a", 2: b"b"})
assert _client(fake).search_all() == [1, 2, 3]
def test_search_since_sends_an_imap_date(self):
fake = _FakeIMAP(messages={1: b"a"})
_client(fake).search_since(date(2026, 6, 15))
search = [c for c in fake.commands if c[:2] == ("UID", "SEARCH")][0]
assert search[-2:] == ("SINCE", "15-Jun-2026")
def test_search_from_uid_filters_the_range_trailer(self):
# `n:*` past the end returns the LAST message on real servers; without the
# client-side filter every idle tick would report one "new" mail forever.
fake = _FakeIMAP(messages={1: b"a", 2: b"b"})
assert _client(fake).search_from_uid(3) == []
def test_search_from_uid_returns_only_new_uids(self):
fake = _FakeIMAP(messages={1: b"a", 2: b"b", 3: b"c"})
assert _client(fake).search_from_uid(2) == [2, 3]
def test_non_ok_search_raises(self):
fake = _FakeIMAP(messages={1: b"a"}, search_ok=False)
with pytest.raises(ImapError, match="UID SEARCH"):
_client(fake).search_all()
class TestFetch:
def test_returns_raw_bytes_and_uses_body_peek(self):
fake = _FakeIMAP(messages={5: b"From: a@b\r\n\r\nhello"})
assert _client(fake).fetch_message(5) == b"From: a@b\r\n\r\nhello"
fetch = [c for c in fake.commands if c[:2] == ("UID", "FETCH")][0]
assert fetch[-1] == "(BODY.PEEK[])"
def test_missing_uid_returns_none_not_an_error(self):
# Expunged between SEARCH and FETCH — an ordinary race on a live mailbox.
fake = _FakeIMAP(messages={5: b"x"})
assert _client(fake).fetch_message(6) is None
def test_mismatched_uid_in_response_raises(self):
fake = _FakeIMAP(messages={5: b"x"})
client = _client(fake)
fake.uid = lambda command, *args: ("OK", [(b"1 (UID 9 BODY[] {1}", b"x"), b")"])
with pytest.raises(ImapError, match="returned UID 9"):
client.fetch_message(5)
class TestLifecycle:
def test_context_manager_logs_in_and_out(self):
fake = _FakeIMAP()
with ImapClient(_account(), connection_factory=lambda acc, t: fake):
assert fake.logged_in
assert fake.logged_out
def test_logout_runs_even_when_the_body_raises(self):
fake = _FakeIMAP()
with pytest.raises(RuntimeError):
with ImapClient(_account(), connection_factory=lambda acc, t: fake):
raise RuntimeError("boom")
assert fake.logged_out
def test_commands_before_connect_raise(self):
client = ImapClient(_account(), connection_factory=lambda acc, t: _FakeIMAP())
with pytest.raises(ImapError, match="not connected"):
client.search_all()

94
packages/kb-mail/tests/test_message.py Normal file
View file

@ -0,0 +1,94 @@
"""Tests for the message-level derivations extracted out of gmail-bulk-import.
The derivations themselves have been exercised by that job's suite (and by 225 030 real
messages) since June; what is new here is that they are now shared, and that `eml_ref` takes
`source` as a parameter instead of hardcoding `gmail/`. These tests pin the properties the
poller depends on: ids identical to the ones already in the table, and archive paths that
agree with `save_eml` for BOTH sources.
"""
from __future__ import annotations
import email
import email.policy
from datetime import datetime, timezone
from pathlib import Path
from kb_mail.archive import save_eml
from kb_mail.message import EPOCH, eml_ref, message_id, parse_attachments, parse_date
def _msg(raw: bytes):
return email.message_from_bytes(raw, policy=email.policy.compat32)
class TestMessageId:
def test_strips_angle_brackets(self):
assert message_id(_msg(b"Message-ID: <abc@example.com>\r\n\r\nbody")) == "abc@example.com"
def test_missing_header_falls_back_to_a_content_hash(self):
mid = message_id(_msg(b"Subject: no id\r\n\r\nbody"))
assert mid.startswith("sha256-")
assert len(mid) == len("sha256-") + 32
def test_content_hash_is_stable_for_identical_bytes(self):
raw = b"Subject: no id\r\n\r\nbody"
assert message_id(_msg(raw)) == message_id(_msg(raw))
def test_eight_bit_header_bytes_do_not_raise(self):
# compat32 hands back an email.header.Header here, not a str — an unguarded
# .strip() on it killed a full bulk-import run once.
mid = message_id(_msg(b"Message-ID: <\xc4\x85bc@example.com>\r\n\r\nbody"))
assert mid
class TestParseDate:
def test_parses_to_utc(self):
ts = parse_date(_msg(b"Date: Mon, 15 Jun 2026 14:00:00 +0200\r\n\r\nbody"))
assert ts == datetime(2026, 6, 15, 12, 0, tzinfo=timezone.utc)
def test_missing_date_is_epoch(self):
assert parse_date(_msg(b"Subject: x\r\n\r\nbody")) == EPOCH
def test_unparseable_date_is_epoch(self):
assert parse_date(_msg(b"Date: not a date at all\r\n\r\nbody")) == EPOCH
def test_naive_date_is_assumed_utc(self):
ts = parse_date(_msg(b"Date: Mon, 15 Jun 2026 14:00:00 -0000\r\n\r\nbody"))
assert ts.tzinfo is timezone.utc
class TestParseAttachments:
def test_inline_text_is_not_an_attachment(self):
assert parse_attachments(_msg(b"Content-Type: text/plain\r\n\r\nhello")) == []
def test_attachment_manifest_carries_sha256_and_size(self):
raw = (
b'Content-Type: multipart/mixed; boundary="b"\r\n\r\n--b\r\n'
b"Content-Type: text/plain\r\n\r\nhello\r\n--b\r\n"
b"Content-Type: application/pdf\r\n"
b'Content-Disposition: attachment; filename="doc.pdf"\r\n'
b"Content-Transfer-Encoding: base64\r\n\r\nJVBERg==\r\n--b--\r\n"
)
[att] = parse_attachments(_msg(raw))
assert att["type"] == "attachment"
assert att["filename"] == "doc.pdf"
assert att["size"] == 4 # decoded payload, not the base64 text
assert len(att["sha256"]) == 64
class TestEmlRef:
def test_layout_is_source_year_month(self):
ts = datetime(2026, 6, 15, tzinfo=timezone.utc)
assert eml_ref("abc@example.com", "fastmail", ts) == "fastmail/2026/06/abc@example.com.eml"
def test_unsafe_characters_are_translated(self):
ts = datetime(2026, 6, 15, tzinfo=timezone.utc)
assert eml_ref("<a/b:c>", "gmail", ts) == "gmail/2026/06/a_b_c.eml"
async def test_agrees_with_save_eml_for_a_new_source(self, tmp_path: Path):
# The two must not drift: on FileExistsError the caller fills raw_ref from eml_ref
# for a file save_eml wrote on an earlier run.
ts = datetime(2026, 6, 15, tzinfo=timezone.utc)
written = await save_eml(tmp_path, "id/with:chars", "fastmail", ts, b"raw")
assert written == eml_ref("id/with:chars", "fastmail", ts)
assert (tmp_path / written).read_bytes() == b"raw"

37
packages/kb-mail/tests/test_migration.py
View file

@ -87,3 +87,40 @@ def test_migration_002_uses_idempotent_ddl():
if line.strip().upper().startswith("CREATE INDEX"):
assert "IF NOT EXISTS" in line.upper(), \
f"CREATE INDEX must be idempotent: {line.strip()}"
def test_migration_005_exists():
assert (INIT_DIR / "005_mail_sync_state.sql").exists(), \
f"Expected {INIT_DIR / '005_mail_sync_state.sql'} to exist"
def test_migration_005_creates_mail_sync_state_idempotently():
sql = (INIT_DIR / "005_mail_sync_state.sql").read_text()
assert "CREATE TABLE IF NOT EXISTS mail_sync_state" in sql
def test_migration_005_has_all_cursor_columns():
sql = (INIT_DIR / "005_mail_sync_state.sql").read_text()
for col in ("account", "folder", "uidvalidity", "last_uid", "last_sync_ts"):
assert col in sql, f"Column '{col}' missing from 005_mail_sync_state.sql"
def test_migration_005_is_keyed_per_account_and_folder():
# Folder scope is per account (recon Decyzja (e)): gmail syncs one \All folder,
# fastmail syncs three. A single-column key could not express that.
sql = (INIT_DIR / "005_mail_sync_state.sql").read_text()
assert "PRIMARY KEY (account, folder)" in sql
def test_migration_005_uidvalidity_and_last_uid_are_not_nullable():
sql = (INIT_DIR / "005_mail_sync_state.sql").read_text()
for col in ("uidvalidity", "last_uid"):
line = next(l for l in sql.splitlines() if l.strip().startswith(col))
assert "NOT NULL" in line.upper(), f"{col} must be NOT NULL but found: {line.strip()}"
def test_migration_005_is_additive():
sql = (INIT_DIR / "005_mail_sync_state.sql").read_text().upper()
for table in ("ENVELOPE", "DOCUMENT_CHUNK", "DOCUMENT_SUMMARY"):
assert f"ALTER TABLE {table}" not in sql
assert f"DROP TABLE {table}" not in sql

182
packages/kb-mail/tests/test_sync_state.py Normal file
View file

@ -0,0 +1,182 @@
"""Unit tests for the sync cursor: the plan a tick makes, and how far the cursor may move.
These two pure functions carry the whole correctness argument of the poller — a wrong branch
here loses mail with no error anywhere — so they are tested directly, without a DB. The
asyncpg helpers are tested against an in-memory fake connection, the same style as
`jobs/gmail-header-backfill/tests`.
"""
from __future__ import annotations
from datetime import date, datetime, timezone
import pytest
from kb_mail.imap import FolderStatus, ImapAccount
from kb_mail.sync_state import (
SEARCH_ALL,
SEARCH_FROM_UID,
SEARCH_NONE,
SEARCH_SINCE,
FolderSyncState,
contiguous_last_uid,
fetch_all_state,
get_folder_state,
plan_folder_sync,
upsert_folder_state,
)
def _account(**kwargs) -> ImapAccount:
base = dict(name="gmail", host="h", user="u", password="p", special_use=("\\All",))
base.update(kwargs)
return ImapAccount(**base)
def _status(uidvalidity=42, uidnext=101) -> FolderStatus:
return FolderStatus(name="[Gmail]/All Mail", uidvalidity=uidvalidity, uidnext=uidnext)
def _state(uidvalidity=42, last_uid=90) -> FolderSyncState:
return FolderSyncState(account="gmail", folder="[Gmail]/All Mail",
uidvalidity=uidvalidity, last_uid=last_uid)
class TestPlanFirstTick:
def test_new_only_fetches_nothing_and_records_the_high_water_mark(self):
plan = plan_folder_sync(_account(initial_mode="new-only"), _status(uidnext=101), None)
assert plan.search == SEARCH_NONE
assert plan.baseline_uid == 100
assert plan.mode == "initial-new-only"
def test_new_only_on_an_empty_mailbox_records_zero_not_minus_one(self):
plan = plan_folder_sync(_account(initial_mode="new-only"), _status(uidnext=1), None)
assert plan.baseline_uid == 0
def test_since_mode_carries_the_configured_date(self):
account = _account(initial_mode="since", initial_since=date(2026, 6, 15))
plan = plan_folder_sync(account, _status(), None)
assert plan.search == SEARCH_SINCE
assert plan.since == date(2026, 6, 15)
def test_full_mode_sweeps_everything(self):
plan = plan_folder_sync(_account(initial_mode="full"), _status(), None)
assert plan.search == SEARCH_ALL
assert plan.mode == "initial-full"
def test_first_tick_is_never_flagged_as_a_uidvalidity_reset(self):
plan = plan_folder_sync(_account(initial_mode="full"), _status(), None)
assert plan.uidvalidity_reset is False
class TestPlanIncremental:
def test_resumes_from_the_uid_after_the_cursor(self):
plan = plan_folder_sync(_account(), _status(uidvalidity=42), _state(last_uid=90))
assert plan.search == SEARCH_FROM_UID
assert plan.start_uid == 91
assert plan.mode == "incremental"
def test_baseline_is_the_stored_cursor_so_an_empty_tick_changes_nothing(self):
plan = plan_folder_sync(_account(), _status(), _state(last_uid=90))
assert plan.baseline_uid == 90
def test_initial_mode_is_ignored_once_a_cursor_exists(self):
# Otherwise flipping initial_mode to 'full' in .env would re-sweep every folder
# on the next tick instead of only affecting folders with no state.
account = _account(initial_mode="since", initial_since=date(2026, 1, 1))
plan = plan_folder_sync(account, _status(), _state())
assert plan.search == SEARCH_FROM_UID
class TestPlanUidvalidityReset:
def test_changed_uidvalidity_forces_a_full_sweep(self):
plan = plan_folder_sync(_account(), _status(uidvalidity=99), _state(uidvalidity=42))
assert plan.search == SEARCH_ALL
assert plan.uidvalidity_reset is True
assert plan.mode == "uidvalidity-reset"
def test_stored_cursor_is_discarded_not_reused_as_a_floor(self):
# The stored last_uid means nothing under a new UIDVALIDITY; keeping it as a floor
# would skip every message whose new UID happens to fall below it.
plan = plan_folder_sync(_account(), _status(uidvalidity=99, uidnext=51),
_state(uidvalidity=42, last_uid=9000))
assert plan.start_uid is None
assert plan.baseline_uid == 50
class TestContiguousLastUid:
def test_all_succeeded_advances_to_the_last_uid(self):
assert contiguous_last_uid(90, [91, 92, 93], {91, 92, 93}) == 93
def test_stops_before_the_first_failure(self):
assert contiguous_last_uid(90, [91, 92, 93], {91, 93}) == 91
def test_failure_on_the_first_uid_leaves_the_cursor_where_it_was(self):
assert contiguous_last_uid(90, [91, 92], {92}) == 90
def test_nothing_attempted_returns_the_floor(self):
assert contiguous_last_uid(90, [], set()) == 90
def test_sparse_uid_sets_advance_over_gaps(self):
# A SINCE search returns non-contiguous UIDs; skipped-by-design UIDs are not failures.
assert contiguous_last_uid(0, [95, 97, 100], {95, 97, 100}) == 100
def test_unsorted_input_is_ordered_before_walking(self):
assert contiguous_last_uid(0, [93, 91, 92], {91, 92}) == 92
def test_full_sweep_floor_of_zero_with_an_early_failure_keeps_the_cursor_at_zero(self):
# This is why baseline_uid (uidnext-1) must NOT be used as a floor for partial
# results: it would step straight over everything that failed.
assert contiguous_last_uid(0, [1, 2, 3], {2, 3}) == 0
class _FakeConn:
def __init__(self, row=None, rows=None):
self.row = row
self.rows = rows or []
self.executed: list[tuple] = []
self.queries: list[tuple] = []
async def fetchrow(self, query, *params):
self.queries.append((query, params))
return self.row
async def fetch(self, query, *params):
self.queries.append((query, params))
return self.rows
async def execute(self, query, *params):
self.executed.append((query, params))
return "INSERT 0 1"
class TestStateHelpers:
async def test_get_folder_state_returns_none_when_absent(self):
assert await get_folder_state(_FakeConn(), "gmail", "INBOX") is None
async def test_get_folder_state_decodes_a_row(self):
ts = datetime(2026, 8, 6, tzinfo=timezone.utc)
conn = _FakeConn(row={"account": "gmail", "folder": "INBOX", "uidvalidity": 42,
"last_uid": 90, "last_sync_ts": ts})
state = await get_folder_state(conn, "gmail", "INBOX")
assert (state.uidvalidity, state.last_uid, state.last_sync_ts) == (42, 90, ts)
async def test_upsert_is_an_on_conflict_update(self):
conn = _FakeConn()
await upsert_folder_state(conn, "gmail", "INBOX", 42, 91)
query, params = conn.executed[-1]
assert "ON CONFLICT (account, folder) DO UPDATE" in query
assert params == ("gmail", "INBOX", 42, 91)
async def test_upsert_stamps_last_sync_ts_server_side(self):
# last_sync_ts answers "is this folder being polled at all" — it must advance on
# empty ticks too, so it is set by the statement rather than passed in.
conn = _FakeConn()
await upsert_folder_state(conn, "gmail", "INBOX", 42, 91)
assert "now()" in conn.executed[-1][0]
async def test_fetch_all_state_is_ordered(self):
conn = _FakeConn(rows=[{"account": "fastmail", "folder": "INBOX", "uidvalidity": 1,
"last_uid": 2, "last_sync_ts": None}])
states = await fetch_all_state(conn)
assert "ORDER BY account, folder" in conn.queries[-1][0]
assert states[0].account == "fastmail"

27
services/kb-postgres/init/005_mail_sync_state.sql Normal file
View file

@ -0,0 +1,27 @@
-- KB spine: mail_sync_state — per-folder IMAP sync cursor for jobs/mail-imap-sync
-- Additive: does not modify 001_envelope.sql .. 004_summaries.sql.
-- Version: 005 — mail_sync_state (faza mailowa Krok 7, recon kb/audits/mail-sync-2026-08-06.md
-- §2.2 wariant A + Decyzja (f), zatwierdzona przez operatora 2026-08-06)
--
-- Why a table and not a file in /opt/homelab/state/: the sync cursor and the envelopes it
-- describes must restore together or not at all. A state file surviving a DB restore makes the
-- poller silently skip everything between the restored envelope set and the file's last_uid —
-- a failure with no symptom, noticed months later as missing mail. Backing it into the same
-- database buys that invariant for the price of this one migration.
--
-- Keyed (account, folder) because the folder scope is per account (Decyzja (e)): gmail syncs a
-- single SPECIAL-USE \All folder, fastmail syncs INBOX + Archive + Sent. Adding a folder is
-- then a config change, not a migration — Message-ID dedup absorbs any overlap.
--
-- account matches envelope.source ('gmail' | 'fastmail'); folder is the IMAP mailbox name
-- exactly as the server returned it in LIST (never a hardcoded literal — Gmail localizes
-- '[Gmail]/All Mail', so the name is resolved by SPECIAL-USE attribute at runtime).
CREATE TABLE IF NOT EXISTS mail_sync_state (
account TEXT NOT NULL, -- == envelope.source: gmail | fastmail
folder TEXT NOT NULL, -- IMAP mailbox name as returned by LIST
uidvalidity BIGINT NOT NULL, -- server's UIDVALIDITY; a change invalidates last_uid
last_uid BIGINT NOT NULL, -- highest UID processed to completion (archive + envelope durable)
last_sync_ts TIMESTAMPTZ NOT NULL DEFAULT now(),
PRIMARY KEY (account, folder)
);