diff --git a/jobs/mail-body-ingest/src/mail_body_ingest/ingest.py b/jobs/mail-body-ingest/src/mail_body_ingest/ingest.py index 9aff6ea..379e4e9 100644 --- a/jobs/mail-body-ingest/src/mail_body_ingest/ingest.py +++ b/jobs/mail-body-ingest/src/mail_body_ingest/ingest.py @@ -106,6 +106,9 @@ from kb_retrieval.embed import ( _log = structlog.get_logger(__name__) DEFAULT_ARCHIVE_ROOT = Path("/home/oskar/kb/mail/archive") +# Both live mail accounts. `fastmail` joined when jobs/mail-imap-sync started writing that +# source (recon Decyzja (b)/(g)); it is inert until then, since the source has no rows. +DEFAULT_SOURCES = ("gmail", "fastmail") DEFAULT_BATCH_SIZE = 64 DEFAULT_EMBED_RETRIES = 2 DEFAULT_EMBED_BACKOFF_S = 1.0 @@ -366,14 +369,33 @@ def _has_threading(entities: list) -> bool: async def fetch_envelopes( - conn: asyncpg.Connection, since: Optional[datetime], limit: Optional[int], offset: Optional[int] + conn: asyncpg.Connection, + since: Optional[datetime] = None, + limit: Optional[int] = None, + offset: Optional[int] = None, + sources: tuple[str, ...] = DEFAULT_SOURCES, + only_unchunked: bool = False, ) -> list: - """`source='gmail'` envelopes, ordered by id for stable --limit/--offset slicing.""" - query = "SELECT id, ts, raw_ref, entities FROM envelope WHERE source = 'gmail'" - params: list = [] + """Mail envelopes for `sources`, ordered by id for stable --limit/--offset slicing. + + `only_unchunked` restricts the set to envelopes with no `document_chunk` rows at all — + the queue the incremental poller feeds (recon §3.2). It is derived from the data rather + than kept as separate state, which makes it self-healing: an interrupted run, a mail + skipped while Ollama was down, a message inserted with a backdated header — each stays in + the queue until it has chunks, with nothing to reconcile. `--since` deliberately does NOT + play that role: it filters on `envelope.ts`, the SENDER's clock, so a mail delivered today + with a month-old header date would fall outside the window and never be chunked. + """ + query = "SELECT id, ts, raw_ref, entities FROM envelope e WHERE source = ANY($1)" + params: list = [list(sources)] if since is not None: params.append(since) query += f" AND ts >= ${len(params)}" + if only_unchunked: + query += ( + " AND NOT EXISTS (" + "SELECT 1 FROM document_chunk c WHERE c.envelope_id = e.id)" + ) query += " ORDER BY id" if limit is not None: params.append(limit) @@ -384,12 +406,30 @@ async def fetch_envelopes( return await conn.fetch(query, *params) -async def fetch_existing_chunk_keys(conn: asyncpg.Connection, model: str) -> set[tuple[str, int]]: +async def fetch_existing_chunk_keys( + conn: asyncpg.Connection, model: str, envelope_ids: Optional[list[str]] = None +) -> set[tuple[str, int]]: """(envelope_id, chunk_index) pairs already inserted with this model — idempotency + dry-run - preview. The query filters `WHERE model = $1`, so the pair need not redundantly carry model.""" - rows = await conn.fetch( - "SELECT envelope_id, chunk_index FROM document_chunk WHERE model = $1", model - ) + preview. The query filters `WHERE model = $1`, so the pair need not redundantly carry model. + + `envelope_ids` narrows the fetch to the current working set. Unbounded, this reads every + chunk key in the corpus: measured 2026-08-06 at 389 012 rows, ~1.0 s server-side, ~26 MB on + the wire and a 60-90 MB Python set — irrelevant once for a 50k backfill slice, wasteful + every two hours on a node with 2.4 GB available, and growing with the corpus (recon + §2.5 iv). None keeps the old unbounded behaviour for the manual full-corpus runs. + """ + if envelope_ids is not None: + if not envelope_ids: + return set() + rows = await conn.fetch( + "SELECT envelope_id, chunk_index FROM document_chunk " + "WHERE model = $1 AND envelope_id = ANY($2)", + model, envelope_ids, + ) + else: + rows = await conn.fetch( + "SELECT envelope_id, chunk_index FROM document_chunk WHERE model = $1", model + ) return {(r["envelope_id"], r["chunk_index"]) for r in rows} @@ -458,14 +498,16 @@ async def run( limit: Optional[int] = None, offset: Optional[int] = None, apply: bool = False, + sources: tuple[str, ...] = DEFAULT_SOURCES, + only_unchunked: bool = False, batch_size: int = DEFAULT_BATCH_SIZE, max_embed_failures: int = DEFAULT_MAX_EMBED_FAILURES, embed_retries: int = DEFAULT_EMBED_RETRIES, embed_backoff_s: float = DEFAULT_EMBED_BACKOFF_S, embed_timeout_s: float = DEFAULT_EMBED_TIMEOUT_S, ) -> dict: - """Process one --limit/--offset (optionally --since-filtered) slice of `source='gmail'` - envelopes. dry-run (apply=False): parse + quote-strip + classify + chunk + count, zero + """Process one --limit/--offset (optionally --since-filtered) slice of mail envelopes for + `sources`. dry-run (apply=False): parse + quote-strip + classify + chunk + count, zero Ollama calls, zero writes (including the threading UPDATE). Stats must always balance: mails_scanned = missing_file + read_errors + parse_errors + body_empty + mails_chunked @@ -475,8 +517,12 @@ async def run( stats = _new_stats() conn = await asyncpg.connect(dsn) try: - envelopes = await fetch_envelopes(conn, since, limit, offset) - existing = await fetch_existing_chunk_keys(conn, model) + envelopes = await fetch_envelopes( + conn, since, limit, offset, sources=sources, only_unchunked=only_unchunked + ) + existing = await fetch_existing_chunk_keys( + conn, model, envelope_ids=[row["id"] for row in envelopes] + ) session: Optional[aiohttp.ClientSession] = None if apply: @@ -692,6 +738,13 @@ def _parse_since(value: str) -> datetime: return datetime.strptime(value, "%Y-%m-%d").replace(tzinfo=timezone.utc) +def _parse_sources(value: str) -> tuple[str, ...]: + sources = tuple(part.strip() for part in value.split(",") if part.strip()) + if not sources: + raise SystemExit("--sources must name at least one envelope source") + return sources + + def _env_num(name: str, default, cast): """Env override for a numeric CLI default. A malformed value is a loud startup failure, not a silent fallback to the default — a typo'd MAIL_INGEST_BATCH_SIZE must not quietly produce a @@ -707,7 +760,7 @@ def _env_num(name: str, default, cast): def main() -> None: parser = argparse.ArgumentParser( - description="Chunk + embed source='gmail' envelope body content into document_chunk " + description="Chunk + embed mail envelope body content into document_chunk " "(module 5, faza mailowa — plan §5, Krok 2)." ) parser.add_argument("--dsn", default=os.environ.get("KB_DSN"), @@ -718,8 +771,17 @@ def main() -> None: help=f"Ollama base URL (default: {DEFAULT_OLLAMA_URL}, or set OLLAMA_URL)") parser.add_argument("--model", default=os.environ.get("OLLAMA_EMBED_MODEL", DEFAULT_MODEL), help=f"Ollama embedding model (default: {DEFAULT_MODEL})") + parser.add_argument("--sources", type=_parse_sources, default=DEFAULT_SOURCES, + metavar="A,B", + help=f"Comma-separated envelope sources to process " + f"(default: {','.join(DEFAULT_SOURCES)})") + parser.add_argument("--only-unchunked", action="store_true", + help="Only envelopes that have no document_chunk rows at all — the " + "queue mail-imap-sync feeds. This is the cyclic mode; without it " + "the job re-scans the whole slice and relies on the chunk-key set.") parser.add_argument("--since", type=_parse_since, default=None, metavar="YYYY-MM-DD", - help="Only envelopes with ts >= this date (for staged runs, plan Decyzja 9)") + help="Only envelopes with ts >= this date (for staged runs, plan Decyzja 9). " + "Never use it as an incremental cursor — ts is the sender's clock.") parser.add_argument("--limit", type=int, default=None, help="Max envelopes to process (default: all)") parser.add_argument("--offset", type=int, default=0, @@ -772,6 +834,8 @@ def main() -> None: limit=args.limit, offset=args.offset, apply=args.apply, + sources=args.sources, + only_unchunked=args.only_unchunked, batch_size=args.batch_size, max_embed_failures=args.max_embed_failures, embed_retries=args.embed_retries, diff --git a/jobs/mail-body-ingest/tests/test_ingest.py b/jobs/mail-body-ingest/tests/test_ingest.py index 1cf7cac..7bb4c3e 100644 --- a/jobs/mail-body-ingest/tests/test_ingest.py +++ b/jobs/mail-body-ingest/tests/test_ingest.py @@ -304,19 +304,59 @@ class TestFetchHelpers: assert "ORDER BY id" in query assert "LIMIT" not in query and "OFFSET" not in query + async def test_fetch_envelopes_defaults_to_both_live_mail_sources(self): + # Decyzja (g): fastmail must be in the default set from the commit that creates the + # source, or its envelopes chunk correctly and stay invisible. + conn = _FakeConn(envelopes=[]) + await fetch_envelopes(conn) + query, params = conn.queries[-1] + assert "source = ANY($1)" in query + assert params[0] == ["gmail", "fastmail"] + async def test_fetch_envelopes_with_since_limit_offset(self): conn = _FakeConn(envelopes=[]) await fetch_envelopes(conn, since=datetime(2025, 7, 1, tzinfo=timezone.utc), limit=10, offset=5) query, params = conn.queries[-1] - assert "ts >= $1" in query - assert "LIMIT $2" in query - assert "OFFSET $3" in query - assert params == (datetime(2025, 7, 1, tzinfo=timezone.utc), 10, 5) + assert "ts >= $2" in query + assert "LIMIT $3" in query + assert "OFFSET $4" in query + assert params == (["gmail", "fastmail"], datetime(2025, 7, 1, tzinfo=timezone.utc), 10, 5) + + async def test_fetch_envelopes_explicit_sources_override_the_default(self): + conn = _FakeConn(envelopes=[]) + await fetch_envelopes(conn, sources=("gmail",)) + assert conn.queries[-1][1][0] == ["gmail"] + + async def test_only_unchunked_adds_the_queue_predicate(self): + conn = _FakeConn(envelopes=[]) + await fetch_envelopes(conn, only_unchunked=True) + query, _ = conn.queries[-1] + assert "NOT EXISTS" in query and "document_chunk c" in query + + async def test_only_unchunked_is_off_by_default(self): + conn = _FakeConn(envelopes=[]) + await fetch_envelopes(conn) + assert "NOT EXISTS" not in conn.queries[-1][0] async def test_fetch_existing_chunk_keys(self): conn = _FakeConn(existing_keys=[("m1@x", 0), ("m1@x", 1)]) keys = await fetch_existing_chunk_keys(conn, model="bge-m3") assert keys == {("m1@x", 0), ("m1@x", 1)} + assert "envelope_id = ANY" not in conn.queries[-1][0] + + async def test_fetch_existing_chunk_keys_narrows_to_the_working_set(self): + # Unbounded this reads all 389k chunk keys (~26 MB, ~1.0 s) every run — fine once for + # a backfill slice, wasteful every two hours on the cyclic path (recon §2.5 iv). + conn = _FakeConn(existing_keys=[("m1@x", 0)]) + await fetch_existing_chunk_keys(conn, model="bge-m3", envelope_ids=["m1@x"]) + query, params = conn.queries[-1] + assert "envelope_id = ANY($2)" in query + assert params == ("bge-m3", ["m1@x"]) + + async def test_fetch_existing_chunk_keys_short_circuits_on_an_empty_working_set(self): + conn = _FakeConn(existing_keys=[("m1@x", 0)]) + assert await fetch_existing_chunk_keys(conn, model="bge-m3", envelope_ids=[]) == set() + assert conn.queries == [] def _env_row(envelope_id: str, subject: str, ts: datetime = datetime(2025, 8, 1, tzinfo=timezone.utc)): diff --git a/jobs/mail-imap-sync/env.example b/jobs/mail-imap-sync/env.example new file mode 100644 index 0000000..ae5c8be --- /dev/null +++ b/jobs/mail-imap-sync/env.example @@ -0,0 +1,71 @@ +# mail-imap-sync configuration — placeholders only. NEVER commit real values. +# +# On PIHA these keys are APPENDED to the existing /opt/homelab/kb/.env, which is already +# root-owned 0600 and already holds KB_DSN, PAPERLESS_API_TOKEN and ANTHROPIC_API_KEY +# (recon Decyzja (c)). systemd reads EnvironmentFile as root BEFORE dropping to +# User=oskar, so the secrets reach the process without being readable by `oskar` at rest — +# a better property than a 0600 file owned by the job's own user. Keep it that way. +# +# Rules, all learned the hard way in this repo: +# 1. Never on the command line. This job has no --user/--password flag on purpose; +# argv shows up in `ps` and in shell history. +# 2. Never echoed to a terminal. Two secrets have already had to be rotated after +# leaking into a session transcript (docs/sessions/2026-07-15.md, +# docs/sessions/2026-07-21.md). Install them with an editor, not `echo >>`. +# 3. App passwords, not account passwords — revocable one at a time, no access to +# account settings, no bypassing 2FA. +# +# Full setup, including how to mint the app passwords: kb/runbooks/mail-sync-run.md + +# --- which mailboxes to sync ------------------------------------------------------- +# Required. Comma-separated; each name is also the envelope.source value and the +# mail_sync_state.account key. Start with one account if the other is not set up yet. +MAIL_ACCOUNTS=gmail,fastmail + +# --- gmail ------------------------------------------------------------------------- +# App password requires 2FA on the account. Paste it WITHOUT the spaces Google displays. +MAIL_GMAIL_USER=you@gmail.com +MAIL_GMAIL_APP_PASSWORD=change-me-16-char-app-password +# Optional, defaults shown. The folder is chosen by SPECIAL-USE attribute, never by name: +# with a Polish UI the mailbox is called "[Gmail]/Wszystkie", and a hardcoded +# "[Gmail]/All Mail" would sync nothing while reporting success every hour. +# MAIL_GMAIL_HOST=imap.gmail.com +# MAIL_GMAIL_PORT=993 +# MAIL_GMAIL_SPECIAL_USE=\All +# MAIL_GMAIL_FOLDERS= +# +# What the FIRST tick on a folder does — new-only | since | full. Ignored once a +# mail_sync_state row exists for that folder. +# new-only : start the corpus here, fetch nothing older. +# since : fetch everything the server RECEIVED on/after MAIL_*_INITIAL_SINCE. +# full : sweep the whole folder (dedup absorbs what is already in the DB). +# For gmail, 'since' is the point of the exercise: the corpus stops at 2026-06-19, so a +# 'new-only' first tick would leave that gap unfilled forever. Pick a date a few days +# BEFORE the last envelope in the DB — dedup makes the overlap free. +MAIL_GMAIL_INITIAL_MODE=since +MAIL_GMAIL_INITIAL_SINCE=2026-06-15 + +# --- fastmail ---------------------------------------------------------------------- +# Fastmail app passwords can be scoped to IMAP only — use that, not a full-access one. +MAIL_FASTMAIL_USER=you@fastmail.com +MAIL_FASTMAIL_APP_PASSWORD=change-me-app-password +# Optional, defaults shown. Fastmail has no \All equivalent, so the scope is a literal +# list. Spam/Trash/Drafts are excluded on purpose (defined noise; drafts have no stable +# Message-ID and mutate). +# MAIL_FASTMAIL_HOST=imap.fastmail.com +# MAIL_FASTMAIL_PORT=993 +# MAIL_FASTMAIL_FOLDERS=INBOX,Archive,Sent +# +# Fastmail is greenfield — zero rows, zero archive files. Whether to pull its history is +# a decision the recon deliberately leaves open until measured: +# mail-imap-sync --measure +# prints MESSAGES per folder. Decide 'new-only' vs 'full' from that number, then set it +# here BEFORE the first --apply run (after the first run the cursor exists and this key +# no longer has any effect). +MAIL_FASTMAIL_INITIAL_MODE=new-only + +# --- shared ------------------------------------------------------------------------ +# Already present in /opt/homelab/kb/.env on PIHA — listed here for completeness. +# KB_DSN=postgresql://kb:CHANGE-ME@localhost:5433/kb +# MAIL_ARCHIVE_ROOT=/home/oskar/kb/mail/archive +# KB_MAIL_SYNC_PROM_PATH=/opt/homelab/state/node-exporter/kb-mail-sync.prom diff --git a/jobs/mail-imap-sync/pyproject.toml b/jobs/mail-imap-sync/pyproject.toml new file mode 100644 index 0000000..b98a91e --- /dev/null +++ b/jobs/mail-imap-sync/pyproject.toml @@ -0,0 +1,31 @@ +[build-system] +requires = ["setuptools>=68"] +build-backend = "setuptools.build_meta" + +[project] +name = "mail-imap-sync" +version = "0.1.0" +requires-python = ">=3.11" +# No IMAP dependency: imaplib is stdlib (PIHA runs 3.11.2). Recon Decyzja (a) — a synchronous +# stdlib client is the right size for ~37 messages a day, and adds nothing to keep patched. +dependencies = [ + "asyncpg>=0.29", + "structlog>=24.1", + "kb-mail", +] + +[project.optional-dependencies] +dev = [ + "pytest>=8.1", + "pytest-asyncio>=0.23", +] + +[project.scripts] +mail-imap-sync = "mail_imap_sync.sync:main" + +[tool.setuptools.packages.find] +where = ["src"] + +[tool.pytest.ini_options] +asyncio_mode = "auto" +testpaths = ["tests"] diff --git a/jobs/mail-imap-sync/src/mail_imap_sync/__init__.py b/jobs/mail-imap-sync/src/mail_imap_sync/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/jobs/mail-imap-sync/src/mail_imap_sync/config.py b/jobs/mail-imap-sync/src/mail_imap_sync/config.py new file mode 100644 index 0000000..7db02e8 --- /dev/null +++ b/jobs/mail-imap-sync/src/mail_imap_sync/config.py @@ -0,0 +1,141 @@ +"""Account configuration from the environment — and only from the environment. + +Decyzja (c) of `kb/audits/mail-sync-2026-08-06.md`: credentials come from +`/opt/homelab/kb/.env` (root-owned `600`, read by systemd as root before dropping to +`User=oskar`), never from argv. This job therefore has no `--password` flag and no +`--user` flag at all — `--dsn ` lands in `ps` output and in shell history, and the +repo has already had to rotate two secrets that leaked into a session transcript. + +Defaults encode the approved folder scope (Decyzja (e)) so a working config is short: + + gmail -> SPECIAL-USE \\All (one fetch per message despite many labels; + same population as the Takeout the corpus rests on) + fastmail -> INBOX + Archive + Sent (no \\All equivalent in a classic layout) + +Spam, Trash and Drafts are excluded from both: spam and trash are defined noise and were not +in the Takeout either, and drafts have no stable Message-ID — they mutate. +""" +from __future__ import annotations + +import os +from datetime import date, datetime +from pathlib import Path +from typing import Mapping, Optional + +from kb_mail.imap import DEFAULT_PORT, INITIAL_MODES, ImapAccount + +DEFAULT_ARCHIVE_ROOT = Path("/home/oskar/kb/mail/archive") + +# Only a default. `MAIL__HOST` overrides it, and an account not listed here simply +# has to set its host — nothing in the code special-cases a provider. +KNOWN_HOSTS = { + "gmail": "imap.gmail.com", + "fastmail": "imap.fastmail.com", +} + +DEFAULT_SCOPE = { + "gmail": {"special_use": ("\\All",), "folders": ()}, + "fastmail": {"special_use": (), "folders": ("INBOX", "Archive", "Sent")}, +} + + +class ConfigError(RuntimeError): + """A missing or malformed setting. Always fatal at startup: an unattended hourly job + that silently syncs a subset of what the operator configured is worse than one that + refuses to start.""" + + +def _csv(value: Optional[str]) -> Optional[tuple[str, ...]]: + """Comma-separated list. An explicitly EMPTY value means "none" (a real override); + an unset value means "no opinion" and returns None so a default can win.""" + if value is None: + return None + return tuple(part.strip() for part in value.split(",") if part.strip()) + + +def _parse_date(value: str, key: str) -> date: + try: + return datetime.strptime(value.strip(), "%Y-%m-%d").date() + except ValueError: + raise ConfigError(f"{key}={value!r} is not a date (expected YYYY-MM-DD)") + + +def account_from_env(name: str, env: Mapping[str, str]) -> ImapAccount: + """Build one account from `MAIL__*` keys. `name` doubles as `envelope.source`.""" + prefix = f"MAIL_{name.upper().replace('-', '_')}_" + + def get(key: str) -> Optional[str]: + return env.get(prefix + key) + + user = get("USER") + password = get("APP_PASSWORD") + if not user: + raise ConfigError(f"{prefix}USER is not set") + if not password: + raise ConfigError(f"{prefix}APP_PASSWORD is not set") + + host = get("HOST") or KNOWN_HOSTS.get(name) + if not host: + raise ConfigError(f"{prefix}HOST is not set and {name!r} has no known default") + + port_raw = get("PORT") + try: + port = int(port_raw) if port_raw else DEFAULT_PORT + except ValueError: + raise ConfigError(f"{prefix}PORT={port_raw!r} is not an integer") + + scope = DEFAULT_SCOPE.get(name, {"special_use": (), "folders": ()}) + special_use = _csv(get("SPECIAL_USE")) + folders = _csv(get("FOLDERS")) + + initial_mode = (get("INITIAL_MODE") or "new-only").strip() + if initial_mode not in INITIAL_MODES: + raise ConfigError( + f"{prefix}INITIAL_MODE={initial_mode!r} must be one of {list(INITIAL_MODES)}" + ) + since_raw = get("INITIAL_SINCE") + initial_since = _parse_date(since_raw, prefix + "INITIAL_SINCE") if since_raw else None + + try: + return ImapAccount( + name=name, + host=host, + port=port, + user=user, + password=password, + special_use=special_use if special_use is not None else scope["special_use"], + folders=folders if folders is not None else scope["folders"], + initial_mode=initial_mode, + initial_since=initial_since, + ) + except ValueError as exc: + raise ConfigError(str(exc)) + + +def accounts_from_env(env: Optional[Mapping[str, str]] = None) -> list[ImapAccount]: + """Every account named in `MAIL_ACCOUNTS`, in the order listed. + + `MAIL_ACCOUNTS` is required rather than defaulted to "gmail,fastmail": which mailboxes + are live is a fact about the deployment, and inferring it from which credentials happen + to be present would turn a typo'd variable name into a silently half-synced corpus. + """ + env = env if env is not None else os.environ + names = _csv(env.get("MAIL_ACCOUNTS")) + if not names: + raise ConfigError( + "MAIL_ACCOUNTS is not set — list the accounts to sync, e.g. 'gmail,fastmail'" + ) + + seen: set[str] = set() + accounts = [] + for name in names: + if name in seen: + raise ConfigError(f"MAIL_ACCOUNTS lists {name!r} twice") + seen.add(name) + accounts.append(account_from_env(name, env)) + return accounts + + +def archive_root_from_env(env: Optional[Mapping[str, str]] = None) -> Path: + env = env if env is not None else os.environ + return Path(env.get("MAIL_ARCHIVE_ROOT") or DEFAULT_ARCHIVE_ROOT) diff --git a/jobs/mail-imap-sync/src/mail_imap_sync/prom.py b/jobs/mail-imap-sync/src/mail_imap_sync/prom.py new file mode 100644 index 0000000..347f1c1 --- /dev/null +++ b/jobs/mail-imap-sync/src/mail_imap_sync/prom.py @@ -0,0 +1,102 @@ +"""Prometheus textfile-collector output for this job. + +Same delivery path as `kb-ingest` (recon §3.4, no new moving parts): the job writes a `.prom` +file atomically into `/opt/homelab/state/node-exporter/`, node_exporter@PIHA picks it up +through its existing `/:/host:ro` mount, and brain-watchdog polls Prometheus' `/api/v1/alerts` +for Telegram delivery. No Alertmanager, deliberately. + +`documents_ingest.cyclic_ingest` has an equivalent unlabelled renderer. It is not imported +here: this job must not depend on the documents-ingest package (aiohttp, anthropic, a +paperless client) to write eight numbers, and these metrics are per-account labelled, which +that renderer cannot express. Folding both into one shared helper is worth doing the next +time either changes — noted in kb/services/job-mail-imap-sync.md. +""" +from __future__ import annotations + +import os +import re +from dataclasses import dataclass, field +from pathlib import Path +from typing import Optional + +DEFAULT_PROM_PATH = Path("/opt/homelab/state/node-exporter/kb-mail-sync.prom") + + +@dataclass(frozen=True) +class Sample: + value: float + labels: dict = field(default_factory=dict) + + +@dataclass(frozen=True) +class MetricFamily: + name: str + type: str + help: str + samples: list = field(default_factory=list) + + +def _escape_label(value: str) -> str: + return value.replace("\\", "\\\\").replace('"', '\\"').replace("\n", "\\n") + + +def _format_value(value: float) -> str: + """Integers render without a trailing `.0` — Prometheus does not care, humans reading + `cat kb-mail-sync.prom` during a first run do.""" + if isinstance(value, bool): + return str(int(value)) + if isinstance(value, int) or (isinstance(value, float) and value.is_integer()): + return str(int(value)) + return repr(float(value)) + + +def render_prom(families: list[MetricFamily]) -> str: + lines: list[str] = [] + for family in families: + if not family.samples: + # A family with no samples is omitted rather than zeroed: "no accounts reported" + # and "accounts reported zero" are different facts and must not look the same. + continue + lines.append(f"# HELP {family.name} {family.help}") + lines.append(f"# TYPE {family.name} {family.type}") + for sample in family.samples: + if sample.labels: + rendered = ",".join( + f'{k}="{_escape_label(str(v))}"' for k, v in sorted(sample.labels.items()) + ) + lines.append(f"{family.name}{{{rendered}}} {_format_value(sample.value)}") + else: + lines.append(f"{family.name} {_format_value(sample.value)}") + return "\n".join(lines) + "\n" + + +def read_prev_value(path: Path, name: str) -> Optional[float]: + """Previous value of an unlabelled metric, or None. + + Used to carry `kb_mail_sync_last_success_timestamp` forward across a failed run — the + staleness signal must measure "time since the last good tick", not get reset to zero by + the first transient error. + """ + if not path.exists(): + return None + try: + text = path.read_text() + except OSError: + return None + pattern = re.compile(rf"^{re.escape(name)}\s+([0-9.eE+-]+)\s*$") + for line in text.splitlines(): + match = pattern.match(line) + if match: + try: + return float(match.group(1)) + except ValueError: + return None + return None + + +def write_prom_atomic(path: Path, content: str) -> None: + """tmp-write + rename, so the textfile collector never reads a half-written file.""" + path.parent.mkdir(parents=True, exist_ok=True) + tmp_path = path.with_name(f"{path.name}.tmp{os.getpid()}") + tmp_path.write_text(content) + tmp_path.replace(path) diff --git a/jobs/mail-imap-sync/src/mail_imap_sync/sync.py b/jobs/mail-imap-sync/src/mail_imap_sync/sync.py new file mode 100644 index 0000000..c21d813 --- /dev/null +++ b/jobs/mail-imap-sync/src/mail_imap_sync/sync.py @@ -0,0 +1,593 @@ +"""IMAP incremental sync — one job, two accounts (gmail + fastmail). + +Krok 7 of the mail phase (`kb/phases/kb-m5-faza-mailowa.md` §10), built to the approved recon +`kb/audits/mail-sync-2026-08-06.md`. Until this job existed, the mail corpus was a photograph: +225 030 envelopes ending on 2026-06-19, with nothing in the repo that talks to a mail server. + +What one tick does, per account and per folder: + + EXAMINE folder (read-only) -> UIDVALIDITY, UIDNEXT + plan the tick -> first contact / incremental / UIDVALIDITY reset + UID SEARCH -> the UIDs this tick owns + per UID, in ascending order: + UID FETCH (BODY.PEEK[]) -> raw bytes, \\Seen untouched + save_eml() -> append-only; FileExistsError just means "already have it" + insert_envelope() -> entities = [headers, attachment...], ON CONFLICT DO NOTHING + UPDATE mail_sync_state -> only over the contiguous prefix of fully durable messages + +**This job does not embed and does not chunk.** That split is deliberate (recon §3.2): +fetching is network-bound and runs hourly on PIHA, which is up 24/7; chunking and embedding +need Ollama on SOLARIA, which is powered off ~16 h a day by design. Coupling them would mean +mail only arrives when the desktop happens to be on. The handoff needs no queue of its own — +an envelope with no `document_chunk` rows *is* the queue, which `mail-body-ingest +--only-unchunked` drains whenever the GPU is reachable. That is self-healing in a way a +timestamp cursor is not: an interrupted run, a mail inserted with a backdated header, a +message skipped while Ollama was down — all of them simply stay in the queue. + +Two things this job inserts that the bulk importer did not, both from recon §2.5: + +* `entities[type=headers]` at INSERT time. Historical envelopes only have headers because a + separate backfill job walked all 225 030 of them later; without them + `mail_body_ingest.build_prefix` silently produces `Temat: (brak tematu) | Od: ?` — no error, + just worse retrieval forever. +* A count of Message-ID collisions ACROSS accounts. `envelope.id` is a bare Message-ID and so + is globally unique: a mail present in both mailboxes is stored once, under whichever account + inserted it first. That is right for the index, but it makes "new fastmail mails" read low, + so the phenomenon is counted (`envelopes_conflict_other_source`) rather than left invisible. + +Credentials come from the environment only (Decyzja (c)) — there is no `--password` flag. +""" +from __future__ import annotations + +import argparse +import asyncio +import email +import email.policy +import os +import sys +import time +from datetime import datetime, timezone +from pathlib import Path +from typing import Callable, Optional + +import asyncpg +import structlog + +from kb_mail.archive import save_eml +from kb_mail.db import envelope_source, insert_envelope, rows_affected +from kb_mail.envelope import Envelope +from kb_mail.headers import parse_headers_resilient +from kb_mail.imap import ImapAccount, ImapClient, ImapError +from kb_mail.message import eml_ref, message_id, parse_attachments, parse_date +from kb_mail.sync_state import ( + SEARCH_ALL, + SEARCH_FROM_UID, + SEARCH_NONE, + SEARCH_SINCE, + contiguous_last_uid, + get_folder_state, + plan_folder_sync, + upsert_folder_state, +) + +from mail_imap_sync.config import DEFAULT_ARCHIVE_ROOT, ConfigError, accounts_from_env +from mail_imap_sync.prom import ( + DEFAULT_PROM_PATH, + MetricFamily, + Sample, + read_prev_value, + render_prom, + write_prom_atomic, +) + +_log = structlog.get_logger(__name__) + +EXIT_CONFIG_ERROR = 2 + +_ACCOUNT_COUNTERS = ( + # Message-level outcomes — these three groups must balance (see `balance_errors`). + "uids_seen", + "processed", + "vanished", + "errors", + "archived", + "archive_exists", + "envelopes_inserted", + "envelopes_skipped_dup", + "envelopes_conflict_other_source", + # Labels and diagnostics — deliberately outside the balance equations. + "headers_fallback", + "uidvalidity_resets", + "folder_errors", + "account_errors", +) + + +def _new_counters() -> dict: + return {name: 0 for name in _ACCOUNT_COUNTERS} + + +def _add(into: dict, other: dict) -> None: + for key, value in other.items(): + into[key] = into.get(key, 0) + value + + +def total_errors(counters: dict) -> int: + """Message-, folder- and account-level failures together. Kept separate in the counters + because only the message-level one participates in the balance equations — a folder that + blew up on EXAMINE never produced a UID to account for.""" + return counters["errors"] + counters["folder_errors"] + counters["account_errors"] + + +def balance_errors(counters: dict) -> list[str]: + """The three identities an apply run must satisfy. Returned rather than logged so both the + summary line and the exit code can use them — a run whose counters do not add up has not + told the truth about what it did, whatever else it reports.""" + problems = [] + if counters["uids_seen"] != counters["processed"] + counters["vanished"] + counters["errors"]: + problems.append("uids_seen != processed + vanished + errors") + if counters["processed"] != counters["archived"] + counters["archive_exists"]: + problems.append("processed != archived + archive_exists") + envelope_outcomes = ( + counters["envelopes_inserted"] + + counters["envelopes_skipped_dup"] + + counters["envelopes_conflict_other_source"] + ) + if counters["processed"] != envelope_outcomes: + problems.append("processed != inserted + skipped_dup + conflict_other_source") + return problems + + +def _parse_message(raw: bytes): + """compat32, matching `gmail-bulk-import`'s mailbox parse exactly. + + Not a stylistic choice: `message_id()` runs over this object, and the ids it produces must + be byte-identical to the 225 030 already in the table or dedup quietly stops working. + """ + return email.message_from_bytes(raw, policy=email.policy.compat32) + + +async def _sync_folder( + conn: asyncpg.Connection, + client: ImapClient, + account: ImapAccount, + folder: str, + archive_root: Path, + apply: bool, + limit: Optional[int], +) -> dict: + """One folder, one tick. Returns a per-folder report; raises only on protocol failures + that make the folder's state meaningless (those are caught one level up, per account).""" + counters = _new_counters() + status = await asyncio.to_thread(client.examine, folder) + + state = await get_folder_state(conn, account.name, folder) + plan = plan_folder_sync(account, status, state) + if plan.uidvalidity_reset: + counters["uidvalidity_resets"] += 1 + + if plan.search == SEARCH_NONE: + uids: list[int] = [] + elif plan.search == SEARCH_ALL: + uids = await asyncio.to_thread(client.search_all) + elif plan.search == SEARCH_SINCE: + uids = await asyncio.to_thread(client.search_since, plan.since) + elif plan.search == SEARCH_FROM_UID: + uids = await asyncio.to_thread(client.search_from_uid, plan.start_uid) + else: # pragma: no cover — plan_folder_sync only emits the four above + raise ImapError(f"unknown search plan {plan.search!r}") + + truncated = False + if limit is not None and len(uids) > limit: + # Take the LOWEST UIDs: the cursor then advances from the oldest end and the rest is + # picked up by the next tick. Taking the newest would strand the older ones behind a + # cursor that had already moved past them. + uids = uids[:limit] + truncated = True + + report = { + "account": account.name, + "folder": folder, + "mode": plan.mode, + "uidvalidity": status.uidvalidity, + "uidnext": status.uidnext, + "state_last_uid": state.last_uid if state else None, + "candidates": len(uids), + "truncated": truncated, + } + + if not apply: + # Dry run stops at the plan: EXAMINE + SEARCH answer "what would this tick do" without + # pulling a single body off the server or writing a byte. `candidates` is the whole + # answer, so no message-level counter moves and the balance check is skipped upstream. + report["counters"] = counters + report["new_last_uid"] = None + _log.info("folder.planned", **report) + return report + + succeeded: set[int] = set() + for uid in uids: + counters["uids_seen"] += 1 + try: + raw = await asyncio.to_thread(client.fetch_message, uid) + except Exception: + _log.warning("skip.fetch_error", account=account.name, folder=folder, + uid=uid, exc_info=True) + counters["errors"] += 1 + continue + + if raw is None: + # Expunged between SEARCH and FETCH. Ordinary on a live mailbox, and the UID is + # gone for good, so it counts as handled and does not block the cursor. + _log.info("skip.vanished", account=account.name, folder=folder, uid=uid) + counters["vanished"] += 1 + succeeded.add(uid) + continue + + try: + envelope_id, inserted_new = await _store_message( + conn, account, archive_root, raw, counters + ) + except Exception: + _log.warning("skip.store_error", account=account.name, folder=folder, + uid=uid, exc_info=True) + counters["errors"] += 1 + continue + + succeeded.add(uid) + _log.debug("message.stored", account=account.name, folder=folder, uid=uid, + envelope_id=envelope_id, new=inserted_new) + + if not uids: + # Nothing to fetch: adopt the plan's high-water mark (or keep the cursor for an + # incremental tick, where baseline_uid IS the stored cursor). + new_last_uid = plan.baseline_uid + else: + # A floor of 0 for any sweep-shaped plan, deliberately NOT plan.baseline_uid: on a + # full sweep that equals uidnext-1, and using it here would step the cursor straight + # over every message that failed. + floor = plan.start_uid - 1 if plan.search == SEARCH_FROM_UID else 0 + new_last_uid = contiguous_last_uid(floor, uids, succeeded) + + await upsert_folder_state(conn, account.name, folder, status.uidvalidity, new_last_uid) + + report["counters"] = counters + report["new_last_uid"] = new_last_uid + report["stalled"] = bool(uids) and len(succeeded) != len(uids) + _log.info("folder.synced", **{k: v for k, v in report.items() if k != "counters"}, + **counters) + return report + + +async def _store_message( + conn: asyncpg.Connection, + account: ImapAccount, + archive_root: Path, + raw: bytes, + counters: dict, +) -> tuple[str, bool]: + """Archive then insert — in that order, and only then may the cursor move past this UID. + + An exception anywhere here leaves the cursor where it was, so the next tick refetches this + message: the archive write is append-only and the insert is `ON CONFLICT DO NOTHING`, so + the repeat costs one `FileExistsError` and one no-op statement. + """ + msg = _parse_message(raw) + envelope_id = message_id(msg) + ts = parse_date(msg) + + headers_entity, used_fallback = parse_headers_resilient(raw) + if used_fallback: + counters["headers_fallback"] += 1 + entities = [headers_entity] + parse_attachments(msg) + + try: + raw_ref = await save_eml(archive_root, envelope_id, account.name, ts, raw) + counters["archived"] += 1 + except FileExistsError: + raw_ref = eml_ref(envelope_id, account.name, ts) + counters["archive_exists"] += 1 + + envelope = Envelope(id=envelope_id, source=account.name, ts=ts, + raw_ref=raw_ref, entities=entities) + tag = await insert_envelope(conn, envelope) + counters["processed"] += 1 + + if rows_affected(tag) == 1: + counters["envelopes_inserted"] += 1 + return envelope_id, True + + existing = await envelope_source(conn, envelope_id) + if existing is not None and existing != account.name: + _log.info("envelope.conflict_other_source", envelope_id=envelope_id, + account=account.name, existing_source=existing) + counters["envelopes_conflict_other_source"] += 1 + else: + counters["envelopes_skipped_dup"] += 1 + return envelope_id, False + + +async def _sync_account( + conn: asyncpg.Connection, + account: ImapAccount, + archive_root: Path, + apply: bool, + limit: Optional[int], + client_factory: Callable[[ImapAccount], ImapClient], +) -> dict: + """All of one account's folders. Failures are contained here: a Gmail login that stopped + working must not also stop Fastmail from syncing (same isolation as `cyclic_ingest`).""" + counters = _new_counters() + folders_report: list[dict] = [] + error: Optional[str] = None + client = client_factory(account) + + try: + await asyncio.to_thread(client.connect) + folders = await asyncio.to_thread(client.resolve_folders) + for folder in folders: + try: + report = await _sync_folder( + conn, client, account, folder, archive_root, apply, limit + ) + folders_report.append(report) + _add(counters, report["counters"]) + except Exception as exc: + _log.error("folder_failed", account=account.name, folder=folder, + error=str(exc), exc_info=True) + counters["folder_errors"] += 1 + folders_report.append({"account": account.name, "folder": folder, + "error": str(exc)}) + except Exception as exc: + _log.error("account_failed", account=account.name, error=str(exc), exc_info=True) + error = str(exc) + counters["account_errors"] += 1 + finally: + await asyncio.to_thread(client.logout) + + return { + "account": account.name, + "counters": counters, + "folders": folders_report, + "error": error, + } + + +async def fetch_last_message_ts( + conn: asyncpg.Connection, accounts: list[str] +) -> dict[str, float]: + """`max(envelope.ts)` per account — the freshness gauge from recon §3.4. + + Purely observational, so a failure here degrades to "metric absent" rather than failing + the run: the DB being down is already loudly reported by every other counter. + """ + try: + rows = await conn.fetch( + "SELECT source, max(ts) AS newest FROM envelope " + "WHERE source = ANY($1) GROUP BY source", + accounts, + ) + except Exception: + _log.warning("last_message_ts_failed", exc_info=True) + return {} + return { + r["source"]: r["newest"].timestamp() + for r in rows + if r["newest"] is not None + } + + +async def run( + dsn: str, + accounts: list[ImapAccount], + archive_root: Path = DEFAULT_ARCHIVE_ROOT, + apply: bool = False, + limit: Optional[int] = None, + client_factory: Optional[Callable[[ImapAccount], ImapClient]] = None, +) -> dict: + """One tick over every configured account. + + dry-run (`apply=False`) plans and counts: it connects, resolves folders, reads + UIDVALIDITY/UIDNEXT and runs the SEARCH, then stops — no FETCH, no archive write, no DB + write, no state update. That is enough to answer "how much would the first real run pull", + which is exactly the question before a first sync. + """ + factory = client_factory or (lambda account: ImapClient(account)) + totals = _new_counters() + per_account: dict[str, dict] = {} + reports: list[dict] = [] + + conn = await asyncpg.connect(dsn) + try: + for account in accounts: + report = await _sync_account(conn, account, archive_root, apply, limit, factory) + reports.append(report) + per_account[account.name] = report["counters"] + _add(totals, report["counters"]) + + last_message_ts = await fetch_last_message_ts(conn, [a.name for a in accounts]) + finally: + await conn.close() + + # Only an apply run has message-level outcomes to balance; a dry run never fetched. + problems = balance_errors(totals) if apply else [] + if problems: + _log.error("stats_unbalanced", problems=problems, **totals) + + failed = total_errors(totals) > 0 or bool(problems) + result = { + "totals": totals, + "accounts": per_account, + "reports": reports, + "last_message_ts": last_message_ts, + "balance_problems": problems, + "failed": failed, + } + _log.info("run_complete", apply=apply, failed=failed, **totals) + return result + + +async def measure( + accounts: list[ImapAccount], + client_factory: Optional[Callable[[ImapAccount], ImapClient]] = None, +) -> list[dict]: + """`STATUS (MESSAGES UIDNEXT UIDVALIDITY)` for every configured folder. Read-only, no DB. + + This is the measurement Decyzja (e) leaves open: whether to pull Fastmail's history is a + decision about a number nobody has yet, and the recon explicitly refuses to guess it. + Run this first, then choose `initial_mode` from what it prints. + """ + factory = client_factory or (lambda account: ImapClient(account)) + rows: list[dict] = [] + for account in accounts: + client = factory(account) + try: + await asyncio.to_thread(client.connect) + for folder in await asyncio.to_thread(client.resolve_folders): + status = await asyncio.to_thread(client.status, folder) + row = { + "account": account.name, + "folder": folder, + "messages": status.messages, + "uidnext": status.uidnext, + "uidvalidity": status.uidvalidity, + } + rows.append(row) + _log.info("folder.measured", **row) + except Exception as exc: + _log.error("measure_failed", account=account.name, error=str(exc), exc_info=True) + rows.append({"account": account.name, "error": str(exc)}) + finally: + await asyncio.to_thread(client.logout) + return rows + + +def build_metrics(result: dict, now_ts: float, prom_path: Path) -> list[MetricFamily]: + """`run()` result -> the metric families of recon §3.4. + + `kb_mail_sync_last_success_timestamp` carries the previous file's value forward on a + failed run. Resetting it to 0 (or to now) would make the staleness alert either fire + forever after one blip or never fire at all. + """ + failed = result["failed"] + prev_success = read_prev_value(prom_path, "kb_mail_sync_last_success_timestamp") + last_success = now_ts if not failed else (prev_success if prev_success is not None else 0.0) + + def per_account(counter: str) -> list[Sample]: + return [ + Sample(value=counters[counter], labels={"account": name}) + for name, counters in sorted(result["accounts"].items()) + ] + + return [ + MetricFamily("kb_mail_sync_last_run_timestamp", "gauge", + "Unix timestamp of the last kb-mail-sync run (success or failure).", + [Sample(now_ts)]), + MetricFamily("kb_mail_sync_last_success_timestamp", "gauge", + "Unix timestamp of the last kb-mail-sync run with no hard failure.", + [Sample(last_success)]), + MetricFamily("kb_mail_sync_last_exit_code", "gauge", + "Exit code of the last kb-mail-sync run.", + [Sample(1 if failed else 0)]), + MetricFamily("kb_mail_sync_envelopes_inserted", "gauge", + "New envelope rows inserted in the last run, per account.", + per_account("envelopes_inserted")), + MetricFamily("kb_mail_sync_envelopes_skipped_dup", "gauge", + "Messages already present under the same account (Message-ID dedup).", + per_account("envelopes_skipped_dup")), + MetricFamily("kb_mail_sync_conflict_other_source", "gauge", + "Messages whose Message-ID already belongs to a DIFFERENT account — " + "stored once, counted here so the overlap is visible.", + per_account("envelopes_conflict_other_source")), + MetricFamily("kb_mail_sync_uidvalidity_resets", "gauge", + "Folders whose UIDVALIDITY changed in the last run, forcing a full sweep.", + per_account("uidvalidity_resets")), + MetricFamily("kb_mail_sync_errors", "gauge", + "Messages, folders and accounts that failed in the last run.", + [Sample(value=total_errors(counters), labels={"account": name}) + for name, counters in sorted(result["accounts"].items())]), + MetricFamily("kb_mail_sync_last_message_ts", "gauge", + "Timestamp of the newest envelope in the DB, per account.", + [Sample(value=ts, labels={"account": name}) + for name, ts in sorted(result["last_message_ts"].items())]), + ] + + +def main() -> None: + parser = argparse.ArgumentParser( + description="Incremental IMAP sync for the KB mail corpus (gmail + fastmail). " + "Credentials come from the environment only — see jobs/mail-imap-sync/env.example." + ) + parser.add_argument("--dsn", default=os.environ.get("KB_DSN"), + help="asyncpg DSN for kb-postgres (or set KB_DSN env var)") + parser.add_argument("--archive-root", type=Path, + default=Path(os.environ.get("MAIL_ARCHIVE_ROOT", str(DEFAULT_ARCHIVE_ROOT))), + help=f"Mail .eml archive root (default: {DEFAULT_ARCHIVE_ROOT})") + parser.add_argument("--prom-path", type=Path, + default=Path(os.environ.get("KB_MAIL_SYNC_PROM_PATH", str(DEFAULT_PROM_PATH))), + help=f"Textfile-collector output path (default: {DEFAULT_PROM_PATH})") + parser.add_argument("--limit", type=int, default=None, metavar="N", + help="Process at most N messages per folder per tick — a brake for " + "the first big sweep; the remainder is picked up next tick") + parser.add_argument("--measure", action="store_true", + help="Print STATUS (MESSAGES/UIDNEXT/UIDVALIDITY) per folder and exit. " + "Read-only, touches neither the DB nor the archive.") + parser.add_argument("--apply", action="store_true", + help="Actually fetch, archive, insert and advance the sync cursor. " + "Default is dry-run: plan + SEARCH counts only.") + args = parser.parse_args() + + try: + accounts = accounts_from_env() + except ConfigError as exc: + _log.error("config_error", error=str(exc)) + sys.exit(EXIT_CONFIG_ERROR) + + if args.measure: + rows = asyncio.run(measure(accounts)) + _log.info("measure_complete", folders=rows) + sys.exit(1 if any("error" in r for r in rows) else 0) + + if not args.dsn: + _log.error("missing_dsn", hint="pass --dsn or set KB_DSN") + sys.exit(EXIT_CONFIG_ERROR) + if args.apply and not args.archive_root.is_dir(): + _log.error("archive_root_not_found", path=str(args.archive_root)) + sys.exit(EXIT_CONFIG_ERROR) + + result = asyncio.run( + run( + dsn=args.dsn, + accounts=accounts, + archive_root=args.archive_root, + apply=args.apply, + limit=args.limit, + ) + ) + + if args.apply: + # Only a real run publishes metrics: a dry-run must not touch the success clock the + # staleness alert reads. + now_ts = time.time() + write_prom_atomic(args.prom_path, render_prom(build_metrics(result, now_ts, args.prom_path))) + + _log.info( + "summary", + mode="APPLY" if args.apply else "DRY-RUN", + accounts={name: counters for name, counters in result["accounts"].items()}, + last_message=_readable_last_message(result["last_message_ts"]), + balance_problems=result["balance_problems"], + **result["totals"], + ) + sys.exit(1 if result["failed"] else 0) + + +def _readable_last_message(last_message_ts: dict[str, float]) -> dict[str, str]: + return { + name: datetime.fromtimestamp(ts, tz=timezone.utc).isoformat() + for name, ts in sorted(last_message_ts.items()) + } + + +if __name__ == "__main__": + main() diff --git a/jobs/mail-imap-sync/tests/conftest.py b/jobs/mail-imap-sync/tests/conftest.py new file mode 100644 index 0000000..566e421 --- /dev/null +++ b/jobs/mail-imap-sync/tests/conftest.py @@ -0,0 +1,199 @@ +"""Fakes shared by the job's tests — no network, no DB, no live mailbox. + +The IMAP fake sits at the `imaplib` level rather than at `ImapClient`, so every test runs the +adapter's REAL protocol parsing (LIST attributes, EXAMINE untagged responses, `n:*` range +trailers, `(prefix, literal)` FETCH tuples). The DB fake models the two tables the job touches +closely enough that dedup, cross-account collisions and cursor persistence are exercised as +behaviour rather than asserted as SQL strings. +""" +from __future__ import annotations + +from datetime import datetime, timezone + + +class FakeIMAP: + """Multi-folder `imaplib.IMAP4_SSL` stand-in. + + `folders` maps mailbox name -> {uid: raw bytes}. `uidvalidity` may be a per-folder dict so + a test can flip one folder's value to simulate a server-side renumbering. + """ + + def __init__(self, folders, *, uidvalidity=42, list_attrs=None, login_error=None, + fetch_errors=(), vanished=()): + self.folders = folders + self._uidvalidity = uidvalidity + self.list_attrs = list_attrs or {} + self.login_error = login_error + self.fetch_errors = set(fetch_errors) + self.vanished = set(vanished) + self.commands: list[tuple] = [] + self.selected = None + self.logged_out = False + + # -- helpers ----------------------------------------------------------------- + + def uidvalidity_of(self, folder: str) -> int: + if isinstance(self._uidvalidity, dict): + return self._uidvalidity.get(folder, 42) + return self._uidvalidity + + def uidnext_of(self, folder: str) -> int: + uids = self.folders.get(folder) or {} + return (max(uids) + 1) if uids else 1 + + def fetch_commands(self): + return [c for c in self.commands if c[:2] == ("UID", "FETCH")] + + # -- imaplib surface --------------------------------------------------------- + + def login(self, user, password): + self.commands.append(("LOGIN", user)) + if self.login_error: + raise RuntimeError(self.login_error) + return ("OK", [b"ok"]) + + def logout(self): + self.logged_out = True + return ("BYE", [b"bye"]) + + def list(self, directory='""', pattern="*"): + self.commands.append(("LIST",)) + lines = [] + for name in self.folders: + attrs = " ".join(self.list_attrs.get(name, ["\\HasNoChildren"])) + lines.append(f'({attrs}) "/" "{name}"'.encode()) + return ("OK", lines) + + def select(self, mailbox, readonly=False): + name = mailbox.strip('"') + self.commands.append(("SELECT", name, readonly)) + if name not in self.folders: + return ("NO", [b"no such mailbox"]) + self.selected = name + return ("OK", [str(len(self.folders[name])).encode()]) + + def status(self, mailbox, names): + name = mailbox.strip('"') + self.commands.append(("STATUS", name, names)) + line = ( + f'"{name}" (MESSAGES {len(self.folders.get(name) or {})} ' + f"UIDNEXT {self.uidnext_of(name)} UIDVALIDITY {self.uidvalidity_of(name)})" + ).encode() + return ("OK", [line]) + + def response(self, key): + if self.selected is None: + return (key, [None]) + if key == "UIDVALIDITY": + return (key, [str(self.uidvalidity_of(self.selected)).encode()]) + if key == "UIDNEXT": + return (key, [str(self.uidnext_of(self.selected)).encode()]) + return (key, [None]) + + def uid(self, command, *args): + self.commands.append(("UID", command) + args) + uids = sorted(self.folders.get(self.selected) or {}) + if command == "SEARCH": + criteria = args[1:] + if criteria and criteria[0] == "UID": + start = int(criteria[1].split(":")[0]) + hits = [u for u in uids if u >= start] + if not hits and uids: + hits = [uids[-1]] # real servers do this; the client must filter it out + elif criteria and criteria[0] == "SINCE": + hits = [u for u in uids if u >= getattr(self, "since_uid_floor", 0)] + else: + hits = uids + return ("OK", [" ".join(str(u) for u in hits).encode()]) + if command == "FETCH": + uid = int(args[0]) + if uid in self.fetch_errors: + raise RuntimeError(f"fetch of {uid} exploded") + if uid in self.vanished: + return ("OK", [None]) + raw = (self.folders.get(self.selected) or {}).get(uid) + if raw is None: + return ("OK", [None]) + return ("OK", [(f"1 (UID {uid} BODY[] {{{len(raw)}}}".encode(), raw), b")"]) + raise AssertionError(f"unexpected UID command {command}") + + +class FakeConn: + """In-memory `envelope` + `mail_sync_state`, dispatched on SQL keywords.""" + + def __init__(self, envelopes=None, state=None, insert_errors=()): + self.envelopes = dict(envelopes or {}) # id -> {"source", "ts", "entities"} + self.state = dict(state or {}) # (account, folder) -> (uidvalidity, last_uid) + self.insert_errors = set(insert_errors) # envelope ids whose INSERT raises + self.queries: list[tuple] = [] + self.closed = False + + async def fetchrow(self, query, *params): + self.queries.append((query, params)) + if "FROM mail_sync_state" in query: + row = self.state.get((params[0], params[1])) + if row is None: + return None + return {"account": params[0], "folder": params[1], "uidvalidity": row[0], + "last_uid": row[1], "last_sync_ts": None} + raise AssertionError(f"unexpected fetchrow: {query}") + + async def fetchval(self, query, *params): + self.queries.append((query, params)) + if "SELECT source FROM envelope" in query: + row = self.envelopes.get(params[0]) + return row["source"] if row else None + raise AssertionError(f"unexpected fetchval: {query}") + + async def fetch(self, query, *params): + self.queries.append((query, params)) + if "max(ts)" in query: + newest: dict = {} + for row in self.envelopes.values(): + if row["source"] in params[0]: + current = newest.get(row["source"]) + if current is None or row["ts"] > current: + newest[row["source"]] = row["ts"] + return [{"source": s, "newest": ts} for s, ts in newest.items()] + if "FROM mail_sync_state" in query: + return [ + {"account": a, "folder": f, "uidvalidity": v[0], "last_uid": v[1], + "last_sync_ts": None} + for (a, f), v in sorted(self.state.items()) + ] + raise AssertionError(f"unexpected fetch: {query}") + + async def execute(self, query, *params): + self.queries.append((query, params)) + if "INSERT INTO envelope" in query: + envelope_id = params[0] + if envelope_id in self.insert_errors: + raise RuntimeError(f"insert of {envelope_id} exploded") + if envelope_id in self.envelopes: + return "INSERT 0 0" + self.envelopes[envelope_id] = {"source": params[1], "ts": params[2], + "entities": params[5]} + return "INSERT 0 1" + if "INSERT INTO mail_sync_state" in query: + self.state[(params[0], params[1])] = (params[2], params[3]) + return "INSERT 0 1" + raise AssertionError(f"unexpected execute: {query}") + + async def close(self): + self.closed = True + + +def eml(message_id: str, *, subject="Temat", sender="Alice ", + date="Mon, 15 Jun 2026 14:00:00 +0200", body="tresc wiadomosci") -> bytes: + return ( + f"Message-ID: <{message_id}>\r\n" + f"From: {sender}\r\n" + f"To: Oskar \r\n" + f"Subject: {subject}\r\n" + f"Date: {date}\r\n" + f"Content-Type: text/plain; charset=utf-8\r\n" + f"\r\n{body}\r\n" + ).encode("utf-8") + + +JUNE_15 = datetime(2026, 6, 15, 12, 0, tzinfo=timezone.utc) diff --git a/jobs/mail-imap-sync/tests/test_config.py b/jobs/mail-imap-sync/tests/test_config.py new file mode 100644 index 0000000..af8cb0f --- /dev/null +++ b/jobs/mail-imap-sync/tests/test_config.py @@ -0,0 +1,139 @@ +"""Tests for env-only account configuration. + +Two properties matter more than the parsing details: a malformed or missing setting is a +startup failure rather than a quiet half-configuration, and the approved folder scope +(Decyzja (e)) is the default so a working config is short. +""" +from __future__ import annotations + +from datetime import date + +import pytest + +from mail_imap_sync.config import ( + ConfigError, + account_from_env, + accounts_from_env, + archive_root_from_env, +) + +BASE = { + "MAIL_ACCOUNTS": "gmail,fastmail", + "MAIL_GMAIL_USER": "u@gmail.com", + "MAIL_GMAIL_APP_PASSWORD": "pw1", + "MAIL_FASTMAIL_USER": "u@fastmail.com", + "MAIL_FASTMAIL_APP_PASSWORD": "pw2", +} + + +def env(**overrides) -> dict: + merged = dict(BASE) + merged.update(overrides) + return {k: v for k, v in merged.items() if v is not None} + + +class TestDefaults: + def test_gmail_defaults_to_the_special_use_all_folder(self): + account = account_from_env("gmail", env()) + assert account.special_use == ("\\All",) + assert account.folders == () + + def test_fastmail_defaults_to_inbox_archive_sent(self): + account = account_from_env("fastmail", env()) + assert account.folders == ("INBOX", "Archive", "Sent") + assert account.special_use == () + + def test_spam_trash_and_drafts_are_not_in_any_default_scope(self): + for name in ("gmail", "fastmail"): + scope = account_from_env(name, env()).folders + assert not {"Spam", "Trash", "Drafts"} & set(scope) + + def test_known_hosts_are_defaults_not_requirements(self): + assert account_from_env("gmail", env()).host == "imap.gmail.com" + assert account_from_env("gmail", env(MAIL_GMAIL_HOST="mail.example.com")).host \ + == "mail.example.com" + + def test_first_tick_defaults_to_new_only(self): + assert account_from_env("fastmail", env()).initial_mode == "new-only" + + +class TestOverrides: + def test_folders_override_replaces_the_default_list(self): + account = account_from_env("fastmail", env(MAIL_FASTMAIL_FOLDERS="INBOX,Projects")) + assert account.folders == ("INBOX", "Projects") + + def test_empty_override_means_none_not_fall_back_to_default(self): + account = account_from_env("gmail", env(MAIL_GMAIL_SPECIAL_USE="", + MAIL_GMAIL_FOLDERS="INBOX")) + assert account.special_use == () + assert account.folders == ("INBOX",) + + def test_since_mode_parses_the_date(self): + account = account_from_env("gmail", env(MAIL_GMAIL_INITIAL_MODE="since", + MAIL_GMAIL_INITIAL_SINCE="2026-06-15")) + assert account.initial_since == date(2026, 6, 15) + + def test_port_override(self): + assert account_from_env("gmail", env(MAIL_GMAIL_PORT="1993")).port == 1993 + + def test_archive_root_override(self): + assert str(archive_root_from_env({"MAIL_ARCHIVE_ROOT": "/srv/mail"})) == "/srv/mail" + + +class TestFailsLoudly: + def test_missing_accounts_list(self): + with pytest.raises(ConfigError, match="MAIL_ACCOUNTS"): + accounts_from_env({}) + + def test_missing_password(self): + with pytest.raises(ConfigError, match="APP_PASSWORD"): + account_from_env("gmail", env(MAIL_GMAIL_APP_PASSWORD=None)) + + def test_missing_user(self): + with pytest.raises(ConfigError, match="USER"): + account_from_env("gmail", env(MAIL_GMAIL_USER=None)) + + def test_unknown_account_without_a_host(self): + with pytest.raises(ConfigError, match="HOST"): + account_from_env("mailbox-org", {"MAIL_MAILBOX_ORG_USER": "u", + "MAIL_MAILBOX_ORG_APP_PASSWORD": "p"}) + + def test_unknown_account_with_a_host_but_no_scope(self): + with pytest.raises(ConfigError, match="no scope"): + account_from_env("mailbox-org", {"MAIL_MAILBOX_ORG_USER": "u", + "MAIL_MAILBOX_ORG_APP_PASSWORD": "p", + "MAIL_MAILBOX_ORG_HOST": "imap.mailbox.org"}) + + def test_bad_port(self): + with pytest.raises(ConfigError, match="PORT"): + account_from_env("gmail", env(MAIL_GMAIL_PORT="993x")) + + def test_bad_initial_mode(self): + with pytest.raises(ConfigError, match="INITIAL_MODE"): + account_from_env("gmail", env(MAIL_GMAIL_INITIAL_MODE="yesterday")) + + def test_since_mode_without_a_date(self): + with pytest.raises(ConfigError, match="initial_since"): + account_from_env("gmail", env(MAIL_GMAIL_INITIAL_MODE="since")) + + def test_bad_since_date(self): + with pytest.raises(ConfigError, match="not a date"): + account_from_env("gmail", env(MAIL_GMAIL_INITIAL_MODE="since", + MAIL_GMAIL_INITIAL_SINCE="15/06/2026")) + + def test_duplicate_account_name(self): + with pytest.raises(ConfigError, match="twice"): + accounts_from_env(env(MAIL_ACCOUNTS="gmail,gmail")) + + +class TestAccountsFromEnv: + def test_returns_accounts_in_the_listed_order(self): + accounts = accounts_from_env(env(MAIL_ACCOUNTS="fastmail,gmail")) + assert [a.name for a in accounts] == ["fastmail", "gmail"] + + def test_a_single_account_is_a_valid_config(self): + accounts = accounts_from_env(env(MAIL_ACCOUNTS="gmail")) + assert [a.name for a in accounts] == ["gmail"] + + def test_account_name_is_the_envelope_source(self): + assert accounts_from_env(env())[0].name == "gmail" diff --git a/jobs/mail-imap-sync/tests/test_prom.py b/jobs/mail-imap-sync/tests/test_prom.py new file mode 100644 index 0000000..6ecd7a7 --- /dev/null +++ b/jobs/mail-imap-sync/tests/test_prom.py @@ -0,0 +1,85 @@ +"""Tests for the textfile-collector renderer.""" +from __future__ import annotations + +from pathlib import Path + +from mail_imap_sync.prom import ( + MetricFamily, + Sample, + read_prev_value, + render_prom, + write_prom_atomic, +) + + +def _family(samples, name="kb_mail_sync_test", mtype="gauge"): + return MetricFamily(name, mtype, "a help string", samples) + + +class TestRender: + def test_emits_help_and_type_before_samples(self): + text = render_prom([_family([Sample(3)])]) + assert text.splitlines()[:3] == [ + "# HELP kb_mail_sync_test a help string", + "# TYPE kb_mail_sync_test gauge", + "kb_mail_sync_test 3", + ] + + def test_labels_are_sorted_and_quoted(self): + text = render_prom([_family([Sample(1, {"folder": "INBOX", "account": "gmail"})])]) + assert 'kb_mail_sync_test{account="gmail",folder="INBOX"} 1' in text + + def test_label_values_are_escaped(self): + text = render_prom([_family([Sample(1, {"folder": '[Gmail]/"All"\\Mail'})])]) + assert 'folder="[Gmail]/\\"All\\"\\\\Mail"' in text + + def test_integers_render_without_a_trailing_zero(self): + assert "kb_mail_sync_test 3\n" in render_prom([_family([Sample(3.0)])]) + + def test_floats_keep_their_precision(self): + assert "kb_mail_sync_test 1.5" in render_prom([_family([Sample(1.5)])]) + + def test_family_with_no_samples_is_omitted_entirely(self): + # "no accounts reported" and "accounts reported zero" are different facts. + assert render_prom([_family([])]) == "\n" + + def test_output_ends_with_a_newline(self): + assert render_prom([_family([Sample(1)])]).endswith("\n") + + +class TestReadPrevValue: + def test_absent_file_returns_none(self, tmp_path: Path): + assert read_prev_value(tmp_path / "nope.prom", "kb_mail_sync_x") is None + + def test_absent_metric_returns_none(self, tmp_path: Path): + path = tmp_path / "x.prom" + path.write_text("kb_mail_sync_other 5\n") + assert read_prev_value(path, "kb_mail_sync_x") is None + + def test_reads_a_scientific_notation_value(self, tmp_path: Path): + path = tmp_path / "x.prom" + path.write_text("kb_mail_sync_x 1.786e9\n") + assert read_prev_value(path, "kb_mail_sync_x") == 1.786e9 + + def test_help_lines_are_not_mistaken_for_samples(self, tmp_path: Path): + path = tmp_path / "x.prom" + path.write_text("# HELP kb_mail_sync_x 42 things\nkb_mail_sync_x 7\n") + assert read_prev_value(path, "kb_mail_sync_x") == 7.0 + + +class TestWriteAtomic: + def test_creates_the_directory_and_the_file(self, tmp_path: Path): + path = tmp_path / "node-exporter" / "kb-mail-sync.prom" + write_prom_atomic(path, "kb_mail_sync_x 1\n") + assert path.read_text() == "kb_mail_sync_x 1\n" + + def test_leaves_no_temp_file_behind(self, tmp_path: Path): + path = tmp_path / "kb-mail-sync.prom" + write_prom_atomic(path, "kb_mail_sync_x 1\n") + assert [p.name for p in tmp_path.iterdir()] == ["kb-mail-sync.prom"] + + def test_overwrites_in_place(self, tmp_path: Path): + path = tmp_path / "kb-mail-sync.prom" + write_prom_atomic(path, "kb_mail_sync_x 1\n") + write_prom_atomic(path, "kb_mail_sync_x 2\n") + assert path.read_text() == "kb_mail_sync_x 2\n" diff --git a/jobs/mail-imap-sync/tests/test_sync.py b/jobs/mail-imap-sync/tests/test_sync.py new file mode 100644 index 0000000..8017405 --- /dev/null +++ b/jobs/mail-imap-sync/tests/test_sync.py @@ -0,0 +1,509 @@ +"""Unit tests for the IMAP sync job — no network, no DB, no live mailbox. + +Every test drives the real `ImapClient` over a fake `imaplib` connection and the real +`run()` over an in-memory `envelope` / `mail_sync_state` pair, so what is being tested is the +job's actual behaviour: what it fetches, what it archives, where the cursor ends up. The +scenarios mirror the failure modes the recon named — new mail, UIDVALIDITY invalidation, +dedup, resumption after an interrupted run — plus the traps it warned would be silent. +""" +from __future__ import annotations + +import json +from datetime import datetime, timezone +from pathlib import Path + +import pytest +from conftest import FakeConn, FakeIMAP, eml + +from kb_mail.imap import ImapAccount, ImapClient +from mail_imap_sync.prom import read_prev_value, render_prom +from mail_imap_sync.sync import ( + balance_errors, + build_metrics, + measure, + run, + total_errors, +) + +GMAIL_FOLDER = "[Gmail]/Wszystkie" + + +def gmail_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 fastmail_account(**kwargs) -> ImapAccount: + base = dict(name="fastmail", host="imap.example.net", user="u@example.net", + password="secret", special_use=(), folders=("INBOX",)) + base.update(kwargs) + return ImapAccount(**base) + + +def gmail_server(messages, **kwargs) -> FakeIMAP: + return FakeIMAP({GMAIL_FOLDER: messages}, + list_attrs={GMAIL_FOLDER: ["\\HasNoChildren", "\\All"]}, **kwargs) + + +def factory_for(servers: dict): + """Client factory returning a real ImapClient bound to the named account's fake server.""" + def _factory(account: ImapAccount) -> ImapClient: + return ImapClient(account, connection_factory=lambda acc, timeout: servers[acc.name]) + return _factory + + +async def do_run(servers, accounts, conn, archive, *, apply=True, limit=None, monkeypatch=None): + async def _connect(_dsn): + return conn + import asyncpg + monkeypatch.setattr(asyncpg, "connect", _connect) + return await run( + dsn="postgresql://fake", + accounts=accounts, + archive_root=archive, + apply=apply, + limit=limit, + client_factory=factory_for(servers), + ) + + +class TestNewMessages: + async def test_fetches_archives_and_inserts(self, tmp_path: Path, monkeypatch): + server = gmail_server({1: eml("a@x"), 2: eml("b@x")}) + conn = FakeConn() + result = await do_run({"gmail": server}, [gmail_account(initial_mode="full")], + conn, tmp_path, monkeypatch=monkeypatch) + + assert result["totals"]["envelopes_inserted"] == 2 + assert set(conn.envelopes) == {"a@x", "b@x"} + assert sorted(p.name for p in tmp_path.rglob("*.eml")) == ["a@x.eml", "b@x.eml"] + assert not result["failed"] + + async def test_archive_path_is_source_year_month(self, tmp_path: Path, monkeypatch): + server = gmail_server({1: eml("a@x")}) + await do_run({"gmail": server}, [gmail_account(initial_mode="full")], + FakeConn(), tmp_path, monkeypatch=monkeypatch) + assert (tmp_path / "gmail" / "2026" / "06" / "a@x.eml").exists() + + async def test_cursor_advances_to_the_highest_processed_uid(self, tmp_path: Path, monkeypatch): + server = gmail_server({1: eml("a@x"), 2: eml("b@x")}) + conn = FakeConn() + await do_run({"gmail": server}, [gmail_account(initial_mode="full")], + conn, tmp_path, monkeypatch=monkeypatch) + assert conn.state[("gmail", GMAIL_FOLDER)] == (42, 2) + + async def test_envelope_carries_headers_at_insert_time(self, tmp_path: Path, monkeypatch): + # Recon §2.5 (i): without this, mail_body_ingest.build_prefix silently produces + # "Temat: (brak tematu) | Od: ?" for every new mail — no error, just worse retrieval. + server = gmail_server({1: eml("a@x", subject="Faktura 07/2026")}) + conn = FakeConn() + await do_run({"gmail": server}, [gmail_account(initial_mode="full")], + conn, tmp_path, monkeypatch=monkeypatch) + entities = json.loads(conn.envelopes["a@x"]["entities"]) + headers = next(e for e in entities if e["type"] == "headers") + assert headers["subject"] == "Faktura 07/2026" + assert headers["from"]["address"] == "a@example.com" + + async def test_attachment_manifest_is_appended_next_to_headers(self, tmp_path: Path, monkeypatch): + raw = ( + b"Message-ID: \r\nFrom: a@example.com\r\nSubject: s\r\n" + b"Date: Mon, 15 Jun 2026 14:00:00 +0200\r\n" + 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\r\n%PDF\r\n--b--\r\n' + ) + conn = FakeConn() + await do_run({"gmail": gmail_server({1: raw})}, [gmail_account(initial_mode="full")], + conn, tmp_path, monkeypatch=monkeypatch) + types = [e["type"] for e in json.loads(conn.envelopes["att@x"]["entities"])] + assert types == ["headers", "attachment"] + + async def test_never_marks_mail_as_read(self, tmp_path: Path, monkeypatch): + # A plain SELECT + FETCH RFC822 would set \Seen on the operator's own inbox. + server = gmail_server({1: eml("a@x")}) + await do_run({"gmail": server}, [gmail_account(initial_mode="full")], + FakeConn(), tmp_path, monkeypatch=monkeypatch) + assert all(cmd[2] is True for cmd in server.commands if cmd[0] == "SELECT") + assert all(cmd[-1] == "(BODY.PEEK[])" for cmd in server.fetch_commands()) + + +class TestIncremental: + async def test_second_run_fetches_only_what_arrived_since(self, tmp_path: Path, monkeypatch): + server = gmail_server({1: eml("a@x")}) + conn = FakeConn() + account = gmail_account(initial_mode="full") + await do_run({"gmail": server}, [account], conn, tmp_path, monkeypatch=monkeypatch) + + server.folders[GMAIL_FOLDER][2] = eml("b@x") + server.commands.clear() + result = await do_run({"gmail": server}, [account], conn, tmp_path, monkeypatch=monkeypatch) + + assert [int(c[2]) for c in server.fetch_commands()] == [2] + assert result["totals"]["envelopes_inserted"] == 1 + + async def test_idle_mailbox_reports_zero_new(self, tmp_path: Path, monkeypatch): + # The `n:*` range trailer means an unfiltered implementation would report one "new" + # mail on every idle tick, forever. + server = gmail_server({1: eml("a@x")}) + conn = FakeConn() + account = gmail_account(initial_mode="full") + await do_run({"gmail": server}, [account], conn, tmp_path, monkeypatch=monkeypatch) + server.commands.clear() + + result = await do_run({"gmail": server}, [account], conn, tmp_path, monkeypatch=monkeypatch) + assert result["totals"]["uids_seen"] == 0 + assert result["totals"]["envelopes_inserted"] == 0 + assert server.fetch_commands() == [] + + async def test_rerun_of_the_same_uids_is_a_no_op_at_every_layer(self, tmp_path: Path, monkeypatch): + # Force a re-fetch by rewinding the cursor: archive says FileExistsError, DB says + # ON CONFLICT DO NOTHING, and nothing is double-counted as new. + server = gmail_server({1: eml("a@x")}) + conn = FakeConn() + account = gmail_account(initial_mode="full") + await do_run({"gmail": server}, [account], conn, tmp_path, monkeypatch=monkeypatch) + conn.state[("gmail", GMAIL_FOLDER)] = (42, 0) + + result = await do_run({"gmail": server}, [account], conn, tmp_path, monkeypatch=monkeypatch) + totals = result["totals"] + assert totals["envelopes_inserted"] == 0 + assert totals["envelopes_skipped_dup"] == 1 + assert totals["archive_exists"] == 1 + assert totals["archived"] == 0 + assert len(list(tmp_path.rglob("*.eml"))) == 1 + + +class TestUidvalidityInvalidation: + async def test_changed_uidvalidity_triggers_a_full_sweep(self, tmp_path: Path, monkeypatch): + server = gmail_server({1: eml("a@x"), 2: eml("b@x")}) + conn = FakeConn() + account = gmail_account(initial_mode="full") + await do_run({"gmail": server}, [account], conn, tmp_path, monkeypatch=monkeypatch) + + server._uidvalidity = 99 + server.commands.clear() + result = await do_run({"gmail": server}, [account], conn, tmp_path, monkeypatch=monkeypatch) + + assert result["totals"]["uidvalidity_resets"] == 1 + assert sorted(int(c[2]) for c in server.fetch_commands()) == [1, 2] + + async def test_resweep_inserts_nothing_new_thanks_to_message_id_dedup(self, tmp_path, monkeypatch): + server = gmail_server({1: eml("a@x"), 2: eml("b@x")}) + conn = FakeConn() + account = gmail_account(initial_mode="full") + await do_run({"gmail": server}, [account], conn, tmp_path, monkeypatch=monkeypatch) + + server._uidvalidity = 99 + result = await do_run({"gmail": server}, [account], conn, tmp_path, monkeypatch=monkeypatch) + assert result["totals"]["envelopes_inserted"] == 0 + assert result["totals"]["envelopes_skipped_dup"] == 2 + + async def test_new_uidvalidity_is_stored_with_the_recomputed_cursor(self, tmp_path, monkeypatch): + server = gmail_server({1: eml("a@x"), 2: eml("b@x")}) + conn = FakeConn() + account = gmail_account(initial_mode="full") + await do_run({"gmail": server}, [account], conn, tmp_path, monkeypatch=monkeypatch) + + server._uidvalidity = 99 + await do_run({"gmail": server}, [account], conn, tmp_path, monkeypatch=monkeypatch) + assert conn.state[("gmail", GMAIL_FOLDER)] == (99, 2) + + async def test_renumbered_mailbox_with_lower_uids_is_not_skipped(self, tmp_path, monkeypatch): + # The stored cursor (90) is meaningless under a new UIDVALIDITY; treating it as a + # floor would skip every renumbered message below it. + conn = FakeConn(state={("gmail", GMAIL_FOLDER): (42, 90)}) + server = gmail_server({1: eml("a@x"), 2: eml("b@x")}, uidvalidity=99) + result = await do_run({"gmail": server}, [gmail_account()], conn, tmp_path, + monkeypatch=monkeypatch) + assert result["totals"]["envelopes_inserted"] == 2 + assert conn.state[("gmail", GMAIL_FOLDER)] == (99, 2) + + +class TestResumeAfterInterruption: + async def test_cursor_stops_before_a_failed_message(self, tmp_path: Path, monkeypatch): + server = gmail_server({1: eml("a@x"), 2: eml("b@x"), 3: eml("c@x")}, + fetch_errors={2}) + conn = FakeConn() + result = await do_run({"gmail": server}, [gmail_account(initial_mode="full")], + conn, tmp_path, monkeypatch=monkeypatch) + + assert conn.state[("gmail", GMAIL_FOLDER)] == (42, 1) + assert result["totals"]["errors"] == 1 + assert result["failed"] is True + + async def test_messages_after_the_failure_are_still_stored(self, tmp_path: Path, monkeypatch): + # Not skipped, not lost: they land in the archive and DB now, and the re-fetch that + # the un-advanced cursor forces next tick is absorbed by dedup. + server = gmail_server({1: eml("a@x"), 2: eml("b@x"), 3: eml("c@x")}, fetch_errors={2}) + conn = FakeConn() + await do_run({"gmail": server}, [gmail_account(initial_mode="full")], + conn, tmp_path, monkeypatch=monkeypatch) + assert set(conn.envelopes) == {"a@x", "c@x"} + + async def test_next_run_retries_from_the_stalled_uid(self, tmp_path: Path, monkeypatch): + server = gmail_server({1: eml("a@x"), 2: eml("b@x"), 3: eml("c@x")}, fetch_errors={2}) + conn = FakeConn() + account = gmail_account(initial_mode="full") + await do_run({"gmail": server}, [account], conn, tmp_path, monkeypatch=monkeypatch) + + server.fetch_errors.clear() + server.commands.clear() + result = await do_run({"gmail": server}, [account], conn, tmp_path, monkeypatch=monkeypatch) + + assert sorted(int(c[2]) for c in server.fetch_commands()) == [2, 3] + assert result["totals"]["envelopes_inserted"] == 1 # only b@x was actually missing + assert conn.state[("gmail", GMAIL_FOLDER)] == (42, 3) + + async def test_a_failed_db_insert_leaves_the_cursor_behind(self, tmp_path: Path, monkeypatch): + # The archive write already happened; the cursor must not move until the row exists. + server = gmail_server({1: eml("a@x"), 2: eml("b@x")}) + conn = FakeConn(insert_errors={"a@x"}) + result = await do_run({"gmail": server}, [gmail_account(initial_mode="full")], + conn, tmp_path, monkeypatch=monkeypatch) + assert conn.state[("gmail", GMAIL_FOLDER)] == (42, 0) + assert result["totals"]["errors"] == 1 + + async def test_limit_caps_a_tick_and_the_rest_follows_next_time(self, tmp_path, monkeypatch): + server = gmail_server({1: eml("a@x"), 2: eml("b@x"), 3: eml("c@x")}) + conn = FakeConn() + account = gmail_account(initial_mode="full") + await do_run({"gmail": server}, [account], conn, tmp_path, limit=2, monkeypatch=monkeypatch) + assert conn.state[("gmail", GMAIL_FOLDER)] == (42, 2) + + await do_run({"gmail": server}, [account], conn, tmp_path, limit=2, monkeypatch=monkeypatch) + assert set(conn.envelopes) == {"a@x", "b@x", "c@x"} + + +class TestFirstTickModes: + async def test_new_only_records_the_high_water_mark_without_fetching(self, tmp_path, monkeypatch): + server = gmail_server({1: eml("a@x"), 2: eml("b@x")}) + conn = FakeConn() + result = await do_run({"gmail": server}, [gmail_account(initial_mode="new-only")], + conn, tmp_path, monkeypatch=monkeypatch) + assert server.fetch_commands() == [] + assert conn.envelopes == {} + assert conn.state[("gmail", GMAIL_FOLDER)] == (42, 2) + assert not result["failed"] + + async def test_since_mode_issues_an_imap_date_search(self, tmp_path, monkeypatch): + from datetime import date + server = gmail_server({1: eml("a@x")}) + await do_run({"gmail": server}, + [gmail_account(initial_mode="since", initial_since=date(2026, 6, 15))], + FakeConn(), tmp_path, monkeypatch=monkeypatch) + search = [c for c in server.commands if c[:2] == ("UID", "SEARCH")][0] + assert search[-2:] == ("SINCE", "15-Jun-2026") + + async def test_initial_mode_no_longer_applies_once_a_cursor_exists(self, tmp_path, monkeypatch): + conn = FakeConn(state={("gmail", GMAIL_FOLDER): (42, 1)}) + server = gmail_server({1: eml("a@x"), 2: eml("b@x")}) + await do_run({"gmail": server}, [gmail_account(initial_mode="full")], + conn, tmp_path, monkeypatch=monkeypatch) + assert [int(c[2]) for c in server.fetch_commands()] == [2] + + async def test_empty_mailbox_first_tick_stores_a_zero_cursor(self, tmp_path, monkeypatch): + server = gmail_server({}) + conn = FakeConn() + await do_run({"gmail": server}, [gmail_account(initial_mode="new-only")], + conn, tmp_path, monkeypatch=monkeypatch) + assert conn.state[("gmail", GMAIL_FOLDER)] == (42, 0) + + +class TestCrossAccount: + async def test_shared_message_id_counts_as_a_cross_source_conflict(self, tmp_path, monkeypatch): + # Recon §2.4: one row, whichever account got there first — but the overlap must be + # visible, or "new fastmail mails" reads low for no discoverable reason. + servers = { + "gmail": gmail_server({1: eml("shared@list")}), + "fastmail": FakeIMAP({"INBOX": {1: eml("shared@list")}}), + } + conn = FakeConn() + result = await do_run(servers, + [gmail_account(initial_mode="full"), + fastmail_account(initial_mode="full")], + conn, tmp_path, monkeypatch=monkeypatch) + + assert result["accounts"]["gmail"]["envelopes_inserted"] == 1 + assert result["accounts"]["fastmail"]["envelopes_inserted"] == 0 + assert result["accounts"]["fastmail"]["envelopes_conflict_other_source"] == 1 + assert conn.envelopes["shared@list"]["source"] == "gmail" + + async def test_both_accounts_keep_their_own_archive_copy(self, tmp_path, monkeypatch): + servers = { + "gmail": gmail_server({1: eml("shared@list")}), + "fastmail": FakeIMAP({"INBOX": {1: eml("shared@list")}}), + } + await do_run(servers, + [gmail_account(initial_mode="full"), fastmail_account(initial_mode="full")], + FakeConn(), tmp_path, monkeypatch=monkeypatch) + assert (tmp_path / "gmail" / "2026" / "06" / "shared@list.eml").exists() + assert (tmp_path / "fastmail" / "2026" / "06" / "shared@list.eml").exists() + + async def test_a_broken_account_does_not_stop_the_other(self, tmp_path, monkeypatch): + servers = { + "gmail": gmail_server({1: eml("a@x")}, login_error="AUTHENTICATIONFAILED"), + "fastmail": FakeIMAP({"INBOX": {1: eml("b@x")}}), + } + conn = FakeConn() + result = await do_run(servers, + [gmail_account(initial_mode="full"), + fastmail_account(initial_mode="full")], + conn, tmp_path, monkeypatch=monkeypatch) + + assert set(conn.envelopes) == {"b@x"} + assert result["accounts"]["gmail"]["account_errors"] == 1 + assert result["failed"] is True + + async def test_fastmail_syncs_its_three_configured_folders(self, tmp_path, monkeypatch): + server = FakeIMAP({"INBOX": {1: eml("i@x")}, "Archive": {1: eml("ar@x")}, + "Sent": {1: eml("s@x")}}) + conn = FakeConn() + account = fastmail_account(folders=("INBOX", "Archive", "Sent"), initial_mode="full") + await do_run({"fastmail": server}, [account], conn, tmp_path, monkeypatch=monkeypatch) + assert set(conn.envelopes) == {"i@x", "ar@x", "s@x"} + assert len(conn.state) == 3 + + +class TestVanishedAndErrors: + async def test_expunged_uid_is_not_an_error_and_does_not_block_the_cursor(self, tmp_path, monkeypatch): + server = gmail_server({1: eml("a@x"), 2: eml("b@x")}, vanished={1}) + conn = FakeConn() + result = await do_run({"gmail": server}, [gmail_account(initial_mode="full")], + conn, tmp_path, monkeypatch=monkeypatch) + assert result["totals"]["vanished"] == 1 + assert result["totals"]["errors"] == 0 + assert conn.state[("gmail", GMAIL_FOLDER)] == (42, 2) + assert not result["failed"] + + async def test_missing_special_use_folder_fails_the_account_loudly(self, tmp_path, monkeypatch): + # Rather than "synced 0 mails, all good" every hour. + server = FakeIMAP({"INBOX": {}}) + result = await do_run({"gmail": server}, [gmail_account()], FakeConn(), tmp_path, + monkeypatch=monkeypatch) + assert result["failed"] is True + assert result["accounts"]["gmail"]["account_errors"] == 1 + + +class TestBalance: + async def test_counters_balance_on_a_mixed_run(self, tmp_path, monkeypatch): + server = gmail_server({1: eml("a@x"), 2: eml("b@x"), 3: eml("c@x"), 4: eml("d@x")}, + fetch_errors={3}, vanished={2}) + result = await do_run({"gmail": server}, [gmail_account(initial_mode="full")], + FakeConn(), tmp_path, monkeypatch=monkeypatch) + assert balance_errors(result["totals"]) == [] + assert result["balance_problems"] == [] + + def test_balance_errors_names_each_broken_identity(self): + broken = {"uids_seen": 5, "processed": 1, "vanished": 0, "errors": 0, + "archived": 0, "archive_exists": 0, "envelopes_inserted": 0, + "envelopes_skipped_dup": 0, "envelopes_conflict_other_source": 0} + assert len(balance_errors(broken)) == 3 + + def test_total_errors_sums_all_three_levels(self): + assert total_errors({"errors": 1, "folder_errors": 2, "account_errors": 3}) == 6 + + +class TestDryRun: + async def test_dry_run_fetches_nothing_and_writes_nothing(self, tmp_path, monkeypatch): + server = gmail_server({1: eml("a@x"), 2: eml("b@x")}) + conn = FakeConn() + result = await do_run({"gmail": server}, [gmail_account(initial_mode="full")], + conn, tmp_path, apply=False, monkeypatch=monkeypatch) + + assert server.fetch_commands() == [] + assert conn.envelopes == {} + assert conn.state == {} + assert list(tmp_path.rglob("*.eml")) == [] + assert not result["failed"] + + async def test_dry_run_reports_how_much_the_real_run_would_pull(self, tmp_path, monkeypatch): + server = gmail_server({1: eml("a@x"), 2: eml("b@x")}) + result = await do_run({"gmail": server}, [gmail_account(initial_mode="full")], + FakeConn(), tmp_path, apply=False, monkeypatch=monkeypatch) + folder = result["reports"][0]["folders"][0] + assert folder["candidates"] == 2 + assert folder["mode"] == "initial-full" + + async def test_dry_run_is_not_flagged_unbalanced(self, tmp_path, monkeypatch): + # It never fetched, so there are no message-level outcomes to balance. + server = gmail_server({1: eml("a@x")}) + result = await do_run({"gmail": server}, [gmail_account(initial_mode="full")], + FakeConn(), tmp_path, apply=False, monkeypatch=monkeypatch) + assert result["balance_problems"] == [] + + +class TestMeasure: + async def test_reports_message_counts_per_folder(self): + server = FakeIMAP({"INBOX": {1: eml("a@x"), 2: eml("b@x")}, "Archive": {1: eml("c@x")}}) + rows = await measure([fastmail_account(folders=("INBOX", "Archive"))], + client_factory=factory_for({"fastmail": server})) + assert [(r["folder"], r["messages"]) for r in rows] == [("INBOX", 2), ("Archive", 1)] + + async def test_touches_neither_archive_nor_mail_flags(self): + server = FakeIMAP({"INBOX": {1: eml("a@x")}}) + await measure([fastmail_account()], client_factory=factory_for({"fastmail": server})) + assert server.fetch_commands() == [] + + async def test_a_failing_account_is_reported_not_raised(self): + server = FakeIMAP({"INBOX": {}}, login_error="nope") + rows = await measure([fastmail_account()], client_factory=factory_for({"fastmail": server})) + assert "error" in rows[0] + + +class TestMetrics: + def _result(self, *, failed=False, inserted=3): + return { + "totals": {}, + "accounts": { + "gmail": {"envelopes_inserted": inserted, "envelopes_skipped_dup": 1, + "envelopes_conflict_other_source": 0, "uidvalidity_resets": 0, + "errors": 0, "folder_errors": 0, "account_errors": 0}, + "fastmail": {"envelopes_inserted": 0, "envelopes_skipped_dup": 0, + "envelopes_conflict_other_source": 2, "uidvalidity_resets": 1, + "errors": 0, "folder_errors": 0, "account_errors": 0}, + }, + "last_message_ts": {"gmail": 1786000000.0}, + "failed": failed, + } + + def test_counters_are_labelled_per_account(self, tmp_path: Path): + text = render_prom(build_metrics(self._result(), 1786000123.0, + tmp_path / "kb-mail-sync.prom")) + assert 'kb_mail_sync_envelopes_inserted{account="gmail"} 3' in text + assert 'kb_mail_sync_conflict_other_source{account="fastmail"} 2' in text + assert 'kb_mail_sync_uidvalidity_resets{account="fastmail"} 1' in text + + def test_success_timestamp_is_set_on_a_clean_run(self, tmp_path: Path): + text = render_prom(build_metrics(self._result(), 1786000123.0, tmp_path / "x.prom")) + assert "kb_mail_sync_last_success_timestamp 1786000123" in text + assert "kb_mail_sync_last_exit_code 0" in text + + def test_success_timestamp_carries_forward_across_a_failure(self, tmp_path: Path): + path = tmp_path / "kb-mail-sync.prom" + path.write_text(render_prom(build_metrics(self._result(), 1786000000.0, path))) + text = render_prom(build_metrics(self._result(failed=True), 1786009999.0, path)) + # The staleness alert measures time since the last GOOD tick — a transient failure + # must not reset its clock, and must not pretend this run succeeded either. + assert "kb_mail_sync_last_success_timestamp 1786000000" in text + assert "kb_mail_sync_last_run_timestamp 1786009999" in text + assert "kb_mail_sync_last_exit_code 1" in text + + def test_first_ever_run_failing_leaves_the_clock_at_zero(self, tmp_path: Path): + text = render_prom(build_metrics(self._result(failed=True), 1786000123.0, + tmp_path / "absent.prom")) + assert "kb_mail_sync_last_success_timestamp 0" in text + + def test_last_message_ts_is_omitted_for_an_account_with_no_rows(self, tmp_path: Path): + text = render_prom(build_metrics(self._result(), 1786000123.0, tmp_path / "x.prom")) + assert 'kb_mail_sync_last_message_ts{account="gmail"}' in text + assert 'kb_mail_sync_last_message_ts{account="fastmail"}' not in text + + def test_read_prev_value_ignores_labelled_series(self, tmp_path: Path): + path = tmp_path / "x.prom" + path.write_text('kb_mail_sync_envelopes_inserted{account="gmail"} 3\n' + "kb_mail_sync_last_success_timestamp 1786000000\n") + assert read_prev_value(path, "kb_mail_sync_last_success_timestamp") == 1786000000.0 + assert read_prev_value(path, "kb_mail_sync_envelopes_inserted") is None diff --git a/packages/kb-retrieval/src/kb_retrieval/retrieval.py b/packages/kb-retrieval/src/kb_retrieval/retrieval.py index 8b8f816..436b071 100644 --- a/packages/kb-retrieval/src/kb_retrieval/retrieval.py +++ b/packages/kb-retrieval/src/kb_retrieval/retrieval.py @@ -40,7 +40,12 @@ DEFAULT_SUMMARY_MODEL = "claude-haiku-4-5" # plan §2 D3 resolution 2026-07-17: DEFAULT_EMBED_MODEL = "bge-m3" DEFAULT_N = 10 # plan §6.1 start value DEFAULT_K = 5 # plan §6.1 start value -DEFAULT_SUMMARYLESS_SOURCES = ("gmail",) # faza mailowa plan §6 Krok 3: sources with no summaries +# faza mailowa plan §6 Krok 3: sources with no summaries. `fastmail` added 2026-08-06 in the +# same commit that introduced the source (recon Decyzja (g)) — mail reaches /search results +# ONLY through the hybrid path's summaryless leg, which filters `WHERE e.source = ANY($2)`. +# Left out, fastmail envelopes would archive, chunk and embed correctly and then be invisible +# to every query, with nothing anywhere reporting an error. +DEFAULT_SUMMARYLESS_SOURCES = ("gmail", "fastmail") async def flat_retrieve(conn: asyncpg.Connection, query_vector: str, k: int = DEFAULT_K) -> list[dict]: