diff --git a/.gitignore b/.gitignore index e4e0199..e2c6386 100644 --- a/.gitignore +++ b/.gitignore @@ -3,3 +3,7 @@ .claude/* !.claude/rules/ !.claude/rules/*.md + +# Retained mail and library state. Material can be private; it is not versioned. +/data/ +__pycache__/ diff --git a/AGENTS.md b/AGENTS.md index f4df499..eb29458 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -93,6 +93,29 @@ curl -s -X PATCH "http://127.0.0.1:8000/tasks/" \ --- +## Stack and commands + +Python 3.11+ standard library only; no install step. + +| Command | Purpose | +| --- | --- | +| `make check` | unit tests, registry validation, rehash of every retained item | +| `make demo` | intake → match → notices → digest on synthetic mail in a temp root | +| `make sync` | push committed work and reconcile State Hub (`rmgr sync --push`) | +| `make install-timer` | install the 15-minute systemd user timer | +| `python3 -m inbound_library run-due` | one timer wake; safe to repeat | +| `python3 -m inbound_library new --repo [--topic t] [--since iso]` | pull notices | +| `python3 -m inbound_library show ` / `get ` | derived text / original bytes | +| `python3 -m inbound_library health` | last scan, silence, quarantine, pending notices | + +Retained mail lives in the gitignored `data/` (see `docs/data-contracts.md`). +IMAP credentials come only from `INBOUND_LIBRARY_IMAP_USER` and +`INBOUND_LIBRARY_IMAP_PASSWORD`, which the timer reads from +`~/.config/inbound-library/imap.env` (mode 0600). The source of truth is OpenBao +`platform/workloads/inbound-library/imap`. Coding agents do not read that file +or the KV entry. Verify a live scan through `systemctl --user start +inbound-library.service` and its journal. + ## Session Protocol **Start:** diff --git a/Makefile b/Makefile new file mode 100644 index 0000000..565561b --- /dev/null +++ b/Makefile @@ -0,0 +1,23 @@ +.DEFAULT_GOAL := help +.PHONY: help check test demo sync install-timer + +help: ## List available targets + @awk 'BEGIN { print "Available targets:" } /^[a-zA-Z0-9_-]+:.*## / { split($$0, parts, "## "); sub(/:.*/, "", parts[1]); printf " %-16s %s\n", parts[1], parts[2] }' $(MAKEFILE_LIST) + +check: test ## Run tests, validate registries and rehash retained items + python3 -m inbound_library check + +test: ## Run the test suite + python3 -m unittest discover -s tests -v + +demo: ## Run intake, matching and digests on isolated synthetic mail + python3 tools/demo.py + +sync: ## Push committed work and synchronize Statehub + uv run --project /home/worsch/repo-manager rmgr sync --path . --push + +install-timer: ## Install and enable the local user timer + install -D -m 644 ops/systemd/inbound-library.service $(HOME)/.config/systemd/user/inbound-library.service + install -D -m 644 ops/systemd/inbound-library.timer $(HOME)/.config/systemd/user/inbound-library.timer + systemctl --user daemon-reload + systemctl --user enable --now inbound-library.timer diff --git a/config/library.json b/config/library.json new file mode 100644 index 0000000..194486f --- /dev/null +++ b/config/library.json @@ -0,0 +1,13 @@ +{ + "mailbox": { + "host": "imap.ionos.de", + "port": 993, + "folder": "INBOX", + "user_env": "INBOUND_LIBRARY_IMAP_USER", + "password_env": "INBOUND_LIBRARY_IMAP_PASSWORD" + }, + "hub": { + "api_base": "http://127.0.0.1:8000", + "from_agent": "inbound-library" + } +} diff --git a/docs/data-contracts.md b/docs/data-contracts.md new file mode 100644 index 0000000..8a264da --- /dev/null +++ b/docs/data-contracts.md @@ -0,0 +1,93 @@ +# Data contracts — schema `inbound-library/v1` + +Consumers may rely on every field listed here. Any change that breaks one +of them bumps the schema version. + +## Layout + +| Path | Mutability | Versioned in git | +| --- | --- | --- | +| `registry/topics.json` | edited by commit | yes | +| `registry/sources/.json` | edited by commit | yes | +| `registry/subscriptions/.json` | edited by commit | yes | +| `config/library.json` | edited by commit, no secrets | yes | +| `data/items/` | immutable original bytes | no | +| `data/deliveries/.json` | immutable | no | +| `data/notices/.json` | immutable | no | +| `data/derived/.json` | rebuildable, replaceable | no | +| `data/digests///.md` | rebuildable, replaceable | no | +| `data/state/*.json` | cursor and hub-sent state | no | + +`data/` is gitignored because mail can be private. A record's id is the SHA-256 +of its canonical JSON (sorted keys, no whitespace). An item's id is the SHA-256 +of its bytes. `check` rehashes both. + +## Item + +The original bytes, exactly as delivered. For mail, the item is the full +RFC 822 message. Each attachment is also an item of its own. + +## Delivery (provenance) + +One record per delivery of an item. + +| Field | Meaning | +| --- | --- | +| `schema` | `inbound-library/v1` | +| `item` | item id | +| `channel` | `imap` or `attachment` | +| `status` | `routed` (a source matched) or `quarantined` (no source) | +| `source` | source id, or `null` | +| `received_at` | IMAP INTERNALDATE (UTC ISO 8601) | +| `mailbox` | `/`; for `imap` only | +| `message_id`, `sender`, `subject`, `list_id` | for `imap` only; `sender` is lowercased | +| `authentication_results` | raw headers, recorded as provenance, not as trust | +| `parent_item`, `filename`, `content_type` | for `attachment` only | + +IMAP UIDs are kept out of the delivery record on purpose. After a UIDVALIDITY +change, the rescan reproduces identical records. When a source is registered +later, a quarantined mail gets an additional `routed` delivery. The quarantined +record stays as history. + +## Derived + +`subject`, `sender`, `date`, `text` (text/plain, otherwise stripped HTML) and +`attachments` (`item`, `filename`, `content_type`). It can be rebuilt from the +item at any time. + +## Topic + +In `topics.json`: `{"version": n, "topics": {"": ""}}`. +Names use dots, from domain to specific, for example +`intelligence.daily-brief`. Add new topics here first; sources and +subscriptions may only use registered topics. + +## Source + +| Field | Meaning | +| --- | --- | +| `id` | the file name without `.json` | +| `owner_repo` | the repository that registered it | +| `channel` | `email` or `url` | +| `match` | for `email`: `from` (required), `subject` (substring), `list_id` | +| `url` | for `url` | +| `topics` | list of registered topics | +| `cadence` | `daily`, `weekly` or `irregular`; daily and weekly email sources are watched for silence (36 h or 9 days) | +| `visibility` | `shared` (any matching repo) or `restricted` (owner only) | + +## Subscription + +| Field | Meaning | +| --- | --- | +| `repo` | the file name without `.json` | +| `agent` | State Hub agent name that receives notices | +| `topics`, `sources` | matched by either | +| `channels` | a subset of `hub`, `pull`, `digest` | +| `cadence` | `daily` or `weekly`; required for `digest` | + +## Notice + +`{schema, repo, delivery, item, source, received_at}`. It is deterministic, so +each (repo, delivery) pair gets exactly one notice. Hub delivery sends one +message per subscriber per run. A failed send stays pending and is retried on +the next run. diff --git a/inbound_library/__init__.py b/inbound_library/__init__.py new file mode 100644 index 0000000..85aca9b --- /dev/null +++ b/inbound_library/__init__.py @@ -0,0 +1 @@ +"""inbound-library: keep delivered material and route it by topic.""" diff --git a/inbound_library/__main__.py b/inbound_library/__main__.py new file mode 100644 index 0000000..2910fc8 --- /dev/null +++ b/inbound_library/__main__.py @@ -0,0 +1,124 @@ +"""Command line for inbound-library. Run with python3 -m inbound_library.""" +import argparse +import fcntl +import json +import sys +from pathlib import Path + +from . import intake, notify, store +from . import registry as reg + +DEFAULT_CONFIG = Path('config') / 'library.json' + + +def _config(root, path): + path = Path(path) if path else Path(root) / DEFAULT_CONFIG + return json.loads(path.read_text()) if path.exists() else {} + + +def _print(value): + print(json.dumps(value, indent=2, ensure_ascii=False)) + + +def run_due(root, config, mailbox=None, post=notify.post_message): + """One timer wake: intake, re-route quarantine, match, notify, digests.""" + state = Path(root) / 'data' / 'state' + state.mkdir(parents=True, exist_ok=True) + with (state / 'run.lock').open('a') as lock: + try: + fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) + except BlockingIOError: + return {'ok': True, 'status': 'already_running'} + registry = reg.load(root) + problems = reg.validate(registry) + if problems: + return {'ok': False, 'status': 'invalid_registry', 'problems': problems} + report = {'ok': True, 'status': 'checked'} + mailbox = mailbox or intake.mailbox_from_config(config) + if mailbox is None: + report['intake'] = {'ok': False, 'status': 'needs_credentials'} + report['ok'] = False + else: + try: + retained = intake.scan(root, mailbox, registry) + report['intake'] = {'ok': True, 'retained': len(retained)} + except (OSError, ValueError) as exc: + report['intake'] = {'ok': False, 'status': 'error', 'error': str(exc)} + report['ok'] = False + report['requeued'] = len(intake.requeue_quarantine(root, registry)) + report['notices'] = len(notify.match(root, registry)) + hub = config.get('hub', {}) + report['hub'] = notify.deliver_hub(root, hub.get('api_base', 'http://127.0.0.1:8000'), + hub.get('from_agent', 'inbound-library'), + registry, post) + if any(not r['ok'] for r in report['hub'].values()): + report['ok'] = False + report['digests'] = notify.write_digests(root, registry=registry) + return report + + +def main(argv=None): + parser = argparse.ArgumentParser( + prog='python3 -m inbound_library', + description='Keep delivered material and route it by topic to subscribing repos.') + parser.add_argument('--root', default='.', help='repository root (default: .)') + parser.add_argument('--config', help='config file (default: config/library.json)') + sub = parser.add_subparsers(dest='command', required=True) + sub.add_parser('check', help='validate registries and rehash every retained item') + scan = sub.add_parser('scan', help='read the mailbox read-only and retain new mail') + scan.add_argument('--fixture', help='scan a directory of .eml files instead of IMAP') + sub.add_parser('run-due', help='intake, match, notify and write digests; safe to repeat') + new = sub.add_parser('new', help='list items routed to a repo') + new.add_argument('--repo', required=True) + new.add_argument('--topic') + new.add_argument('--since', help='ISO date or timestamp') + show = sub.add_parser('show', help='show an item: derived text and deliveries') + show.add_argument('item') + get = sub.add_parser('get', help='write the original bytes of an item to stdout') + get.add_argument('item') + sub.add_parser('health', help='last scan, per-source counts, silence, quarantine') + args = parser.parse_args(argv) + root = Path(args.root) + config = _config(root, args.config) + + if args.command == 'check': + problems = reg.validate(reg.load(root)) + store.verify(root) + for problem in problems: + print(problem, file=sys.stderr) + if problems: + return 1 + print(f"Registry and store valid; {len(store.records(root, 'deliveries'))} deliveries") + return 0 + if args.command == 'scan': + mailbox = (intake.DirectoryMailbox(args.fixture) if args.fixture + else intake.mailbox_from_config(config)) + if mailbox is None: + print('needs_credentials: set INBOUND_LIBRARY_IMAP_USER and ' + 'INBOUND_LIBRARY_IMAP_PASSWORD', file=sys.stderr) + return 2 + _print({'retained': intake.scan(root, mailbox)}) + return 0 + if args.command == 'run-due': + report = run_due(root, config) + _print(report) + return 0 if report['ok'] else 1 + if args.command == 'new': + _print(notify.notices_for(root, args.repo, args.topic, args.since)) + return 0 + if args.command == 'show': + deliveries = [d for d in store.records(root, 'deliveries') if d['item'] == args.item] + _print({'item': args.item, 'derived': store.get_derived(root, args.item), + 'deliveries': deliveries}) + return 0 + if args.command == 'get': + sys.stdout.buffer.write(store.get_item(root, args.item)) + return 0 + if args.command == 'health': + report = notify.health(root) + _print(report) + return 0 if not report['integrity_problems'] else 1 + return 2 + + +if __name__ == '__main__': + sys.exit(main()) diff --git a/inbound_library/intake.py b/inbound_library/intake.py new file mode 100644 index 0000000..2701a1e --- /dev/null +++ b/inbound_library/intake.py @@ -0,0 +1,157 @@ +"""Read-only mail intake. + +The mailbox is selected read-only and bodies are fetched with BODY.PEEK[], so no +flag changes. The UIDVALIDITY/UID cursor lives in the library. Delivery records +leave out UIDs, so a rescan after a UIDVALIDITY change reproduces the same +records instead of duplicating them. +""" +import email +import email.policy +import email.utils +import html +import imaplib +import json +import os +import re +import ssl +from datetime import datetime, timezone +from pathlib import Path + +from . import registry as reg +from . import store + +CURSOR = Path('data') / 'state' / 'imap-cursor.json' + + +class ImapMailbox: + def __init__(self, host, port, folder, user, password): + self.host, self.port, self.folder = host, int(port), folder + self.user, self.password = user, password + self.name = f'{user}/{folder}' + + def fetch(self, cursor): + """Yield (uidvalidity, uid, internaldate, raw) for UIDs past the cursor.""" + context = ssl.create_default_context() + with imaplib.IMAP4_SSL(self.host, self.port, ssl_context=context) as imap: + imap.login(self.user, self.password) + status, _ = imap.select(self.folder, readonly=True) + if status != 'OK': + raise OSError(f'cannot select {self.folder}') + uidvalidity = int(imap.response('UIDVALIDITY')[1][0]) + last = cursor.get('last_uid', 0) if cursor.get('uidvalidity') == uidvalidity else 0 + status, data = imap.uid('SEARCH', None, f'UID {last + 1}:*') + if status != 'OK': + raise OSError('UID SEARCH failed') + uids = sorted(int(u) for u in data[0].split() if int(u) > last) + for uid in uids: + status, data = imap.uid('FETCH', str(uid), '(INTERNALDATE BODY.PEEK[])') + if status != 'OK': + raise OSError(f'UID FETCH {uid} failed') + meta, raw = next((part[0], part[1]) for part in data if isinstance(part, tuple)) + when = imaplib.Internaldate2tuple(meta) + stamp = datetime(*when[:6], tzinfo=timezone.utc).isoformat() if when else None + yield uidvalidity, uid, stamp, raw + + +class DirectoryMailbox: + """A fixture mailbox: *.eml files in name order, UID = position + 1.""" + + def __init__(self, path, uidvalidity=1): + self.path, self.uidvalidity = Path(path), uidvalidity + self.name = f'dir:{self.path.name}' + + def fetch(self, cursor): + last = cursor.get('last_uid', 0) if cursor.get('uidvalidity') == self.uidvalidity else 0 + for uid, path in enumerate(sorted(self.path.glob('*.eml')), start=1): + if uid > last: + stamp = datetime.fromtimestamp(path.stat().st_mtime, timezone.utc).isoformat() + yield self.uidvalidity, uid, stamp, path.read_bytes() + + +def mailbox_from_config(config, environ=None): + """Build the live mailbox. Credentials come only from the environment.""" + environ = os.environ if environ is None else environ + spec = config.get('mailbox') or {} + user = environ.get(spec.get('user_env', 'INBOUND_LIBRARY_IMAP_USER')) + password = environ.get(spec.get('password_env', 'INBOUND_LIBRARY_IMAP_PASSWORD')) + if not user or not password: + return None + return ImapMailbox(spec.get('host', 'imap.ionos.de'), spec.get('port', 993), + spec.get('folder', 'INBOX'), user, password) + + +def _text(message): + plain = message.get_body(preferencelist=('plain',)) + if plain is not None: + return plain.get_content() + rich = message.get_body(preferencelist=('html',)) + if rich is None: + return '' + stripped = re.sub(r'(?is)<(script|style).*?', ' ', rich.get_content()) + stripped = re.sub(r'(?s)|

||', '\n', stripped) + return html.unescape(re.sub(r'(?s)<[^>]+>', '', stripped)).strip() + + +def retain_message(root, registry, raw, mailbox, received_at): + """Retain one RFC 822 message and its attachments. Returns the delivery id.""" + message = email.message_from_bytes(raw, policy=email.policy.default) + item, _ = store.put_item(root, raw) + sender = email.utils.parseaddr(str(message.get('From', '')))[1].lower() + subject = str(message.get('Subject', '')) + list_id = str(message.get('List-Id', '')) + source = reg.source_for_mail(registry, sender, subject, list_id) + provenance = { + 'channel': 'imap', 'mailbox': mailbox, 'received_at': received_at, + 'message_id': str(message.get('Message-ID', '')), 'sender': sender, + 'subject': subject, 'list_id': list_id, 'source': source, + 'authentication_results': [str(h) for h in message.get_all('Authentication-Results', [])], + } + delivery = store.record_delivery(root, item, provenance) + attachments = [] + for part in message.iter_attachments(): + payload = part.get_payload(decode=True) + if payload is None: + continue + child, _ = store.put_item(root, payload) + store.record_delivery(root, child, { + 'channel': 'attachment', 'parent_item': item, 'received_at': received_at, + 'filename': part.get_filename() or '', 'content_type': part.get_content_type(), + 'source': source}) + attachments.append({'item': child, 'filename': part.get_filename() or '', + 'content_type': part.get_content_type()}) + store.put_derived(root, item, { + 'subject': subject, 'sender': sender, 'date': str(message.get('Date', '')), + 'text': _text(message), 'attachments': attachments}) + return delivery + + +def scan(root, mailbox, registry=None): + """Retain every new message. The cursor advances only after each retain.""" + root = Path(root) + registry = registry or reg.load(root) + cursor_path = root / CURSOR + cursor = json.loads(cursor_path.read_text()) if cursor_path.exists() else {} + retained = [] + for uidvalidity, uid, received_at, raw in mailbox.fetch(cursor): + retained.append(retain_message(root, registry, raw, mailbox.name, received_at)) + cursor = {'mailbox': mailbox.name, 'uidvalidity': uidvalidity, 'last_uid': uid} + store.atomic(cursor_path, json.dumps(cursor, indent=2).encode() + b'\n') + state = {**cursor, 'scanned_at': datetime.now(timezone.utc).isoformat()} + store.atomic(cursor_path, json.dumps(state, indent=2).encode() + b'\n') + return retained + + +def requeue_quarantine(root, registry=None): + """Re-route quarantined mail whose sender now matches a registered source.""" + registry = registry or reg.load(root) + routed = [] + for delivery in store.records(root, 'deliveries'): + if delivery['status'] != 'quarantined' or delivery.get('channel') != 'imap': + continue + source = reg.source_for_mail(registry, delivery.get('sender'), delivery.get('subject'), + delivery.get('list_id')) + if source: + raw = store.get_item(root, delivery['item']) + routed.append(retain_message(root, registry, raw, delivery['mailbox'], + delivery['received_at'])) + return routed diff --git a/inbound_library/notify.py b/inbound_library/notify.py new file mode 100644 index 0000000..02fe42f --- /dev/null +++ b/inbound_library/notify.py @@ -0,0 +1,178 @@ +"""Matching, idempotent notices and their three channels. + +A notice is an immutable record {repo, delivery, item, source}. Its content is +deterministic, so matching the same delivery twice yields the same notice id. +Hub delivery state is kept separately, so a hub outage leaves notices pending. +""" +import json +import urllib.error +import urllib.request +from datetime import date, datetime, timedelta, timezone +from pathlib import Path + +from . import registry as reg +from . import store + +HUB_STATE = Path('data') / 'state' / 'hub-sent.json' + + +def match(root, registry=None): + """Create notices for every routed delivery and matching subscriber.""" + registry = registry or reg.load(root) + existing = {(n['repo'], n['delivery']) for n in store.records(root, 'notices')} + created = [] + directory = store.data_dir(root) / 'deliveries' + for path in sorted(directory.glob('*.json')) if directory.is_dir() else []: + delivery = store.get_record(root, 'deliveries', path.stem) + if delivery['status'] != 'routed' or delivery.get('channel') == 'attachment': + continue + for repo in reg.subscribers_for(registry, delivery['source']): + if (repo, path.stem) in existing: + continue + created.append(store.put_record(root, 'notices', { + 'schema': store.SCHEMA, 'repo': repo, 'delivery': path.stem, + 'item': delivery['item'], 'source': delivery['source'], + 'received_at': delivery.get('received_at')})) + return created + + +def notices_for(root, repo, topic=None, since=None, registry=None): + registry = registry or reg.load(root) + found = [] + for notice in store.records(root, 'notices'): + if notice['repo'] != repo: + continue + source = registry['sources'].get(notice['source'], {}) + if topic and topic not in source.get('topics', []): + continue + if since and (notice.get('received_at') or '') < since: + continue + derived = store.get_derived(root, notice['item']) or {} + found.append({**notice, 'topics': source.get('topics', []), + 'subject': derived.get('subject', '')}) + return sorted(found, key=lambda n: n.get('received_at') or '') + + +def _notice_ids(root): + directory = store.data_dir(root) / 'notices' + return {p.stem: store.get_record(root, 'notices', p.stem) + for p in sorted(directory.glob('*.json'))} if directory.is_dir() else {} + + +def pending_hub(root, registry=None): + registry = registry or reg.load(root) + path = Path(root) / HUB_STATE + sent = set(json.loads(path.read_text())) if path.exists() else set() + pending = {} + for ident, notice in _notice_ids(root).items(): + subscription = registry['subscriptions'].get(notice['repo'], {}) + if 'hub' in subscription.get('channels', []) and ident not in sent: + pending.setdefault(notice['repo'], []).append(ident) + return pending, sent + + +def post_message(api_base, payload, timeout=15): + request = urllib.request.Request( + api_base.rstrip('/') + '/messages/', data=json.dumps(payload).encode(), + headers={'Content-Type': 'application/json'}, method='POST') + with urllib.request.urlopen(request, timeout=timeout) as response: + if response.status >= 300: + raise OSError(f'hub returned {response.status}') + + +def deliver_hub(root, api_base, from_agent='inbound-library', registry=None, post=post_message): + """Send one message per subscriber with pending notices. Failures stay pending.""" + registry = registry or reg.load(root) + pending, sent = pending_hub(root, registry) + notices = _notice_ids(root) + report = {} + for repo, idents in sorted(pending.items()): + subscription = registry['subscriptions'][repo] + lines = [] + for ident in idents: + notice = notices[ident] + subject = (store.get_derived(root, notice['item']) or {}).get('subject', '') + lines.append(f"- {notice['source']}: {subject or '(no subject)'} item {notice['item'][:12]}") + cli = f'cd {Path(root).resolve()} && python3 -m inbound_library' + body = (f'{len(idents)} new item(s) for {repo}:\n' + '\n'.join(lines) + + f'\n\nList: {cli} new --repo {repo}\n' + f'Read: {cli} show Original bytes: {cli} get ') + try: + post(api_base, {'from_agent': from_agent, 'to_agent': subscription['agent'], + 'subject': f'inbound-library: {len(idents)} new item(s)', + 'body': body}) + except (OSError, urllib.error.URLError, ValueError) as exc: + report[repo] = {'ok': False, 'pending': len(idents), 'error': str(exc)} + continue + sent.update(idents) + store.atomic(Path(root) / HUB_STATE, json.dumps(sorted(sent), indent=2).encode() + b'\n') + report[repo] = {'ok': True, 'sent': len(idents)} + return report + + +def period_of(moment, cadence): + day = moment.date() if isinstance(moment, datetime) else moment + if cadence == 'weekly': + year, week, _ = day.isocalendar() + return f'{year}-W{week:02}' + return day.isoformat() + + +def write_digests(root, now=None, registry=None): + """Write the current-period digest per digest subscriber. Digests are derived.""" + registry = registry or reg.load(root) + now = now or datetime.now(timezone.utc) + written = [] + for repo, subscription in registry['subscriptions'].items(): + if 'digest' not in subscription.get('channels', []): + continue + cadence = subscription['cadence'] + period = period_of(now, cadence) + start = now.date() - timedelta(days=now.weekday()) if cadence == 'weekly' else now.date() + notices = notices_for(root, repo, since=start.isoformat(), registry=registry) + lines = [f'# inbound-library digest — {repo} — {period}', '', + 'Unreviewed material routed by topic. Review belongs to the consumer.', ''] + if not notices: + lines.append('No new items in this period.') + for notice in notices: + lines.append(f"- **{notice['subject'] or '(no subject)'}** — {notice['source']} " + f"({', '.join(notice['topics'])}), received {notice.get('received_at')}, " + f"item `{notice['item']}`") + path = Path(root) / 'data' / 'digests' / repo / cadence / f'{period}.md' + store.atomic(path, ('\n'.join(lines) + '\n').encode()) + written.append(str(path.relative_to(root))) + return written + + +SILENCE = {'daily': timedelta(hours=36), 'weekly': timedelta(days=9)} + + +def health(root, now=None, registry=None): + registry = registry or reg.load(root) + now = now or datetime.now(timezone.utc) + deliveries = store.records(root, 'deliveries') + last_seen, counts = {}, {} + routed_items = {d['item'] for d in deliveries if d['status'] == 'routed'} + quarantine = {d['item'] for d in deliveries + if d['status'] == 'quarantined' and d['item'] not in routed_items} + for delivery in deliveries: + if delivery['status'] != 'routed' or delivery.get('channel') == 'attachment': + continue + counts[delivery['source']] = counts.get(delivery['source'], 0) + 1 + stamp = delivery.get('received_at') or '' + last_seen[delivery['source']] = max(last_seen.get(delivery['source'], ''), stamp) + silent = [] + for ident, source in registry['sources'].items(): + window = SILENCE.get(source.get('cadence')) + if source.get('channel') != 'email' or window is None: + continue + seen = last_seen.get(ident) + if not seen or datetime.fromisoformat(seen) < now - window: + silent.append({'source': ident, 'last_seen': seen}) + cursor_path = Path(root) / 'data' / 'state' / 'imap-cursor.json' + cursor = json.loads(cursor_path.read_text()) if cursor_path.exists() else {} + pending, _ = pending_hub(root, registry) + return {'last_scan': cursor.get('scanned_at'), 'items_per_source': counts, + 'silent_sources': silent, 'quarantined_items': len(quarantine), + 'pending_hub_notices': sum(len(v) for v in pending.values()), + 'integrity_problems': store.verify(root)} diff --git a/inbound_library/registry.py b/inbound_library/registry.py new file mode 100644 index 0000000..5c050f1 --- /dev/null +++ b/inbound_library/registry.py @@ -0,0 +1,118 @@ +"""File-backed registries: topics, sources and subscriptions. + +Consumers change these with a commit. `validate` is the gate that check runs. +""" +import json +from pathlib import Path + +CHANNELS = {'hub', 'pull', 'digest'} +CADENCES = {'daily', 'weekly', 'irregular'} +VISIBILITY = {'shared', 'restricted'} + + +def _load(path): + return json.loads(Path(path).read_text()) + + +def load(root): + base = Path(root) / 'registry' + topics_file = base / 'topics.json' + topics = _load(topics_file) if topics_file.exists() else {'version': 0, 'topics': {}} + sources = {} + for path in sorted((base / 'sources').glob('*.json')): + source = _load(path) + sources[source.get('id', path.stem)] = {**source, '_file': path.name} + subscriptions = {} + for path in sorted((base / 'subscriptions').glob('*.json')): + subscription = _load(path) + subscriptions[subscription.get('repo', path.stem)] = {**subscription, '_file': path.name} + return {'topics': topics, 'sources': sources, 'subscriptions': subscriptions} + + +def validate(registry): + problems = [] + known = set(registry['topics'].get('topics', {})) + rules = {} + for ident, source in registry['sources'].items(): + where = f"sources/{source['_file']}" + if source['_file'] != f'{ident}.json': + problems.append(f'{where}: file name must be .json') + for field in ('id', 'owner_repo', 'channel', 'topics', 'cadence', 'visibility'): + if field not in source: + problems.append(f'{where}: missing {field}') + if source.get('cadence') not in CADENCES: + problems.append(f'{where}: cadence must be one of {sorted(CADENCES)}') + if source.get('visibility') not in VISIBILITY: + problems.append(f'{where}: visibility must be shared or restricted') + for topic in source.get('topics', []): + if topic not in known: + problems.append(f'{where}: unknown topic {topic}') + if source.get('channel') == 'email': + match = source.get('match', {}) + if not match.get('from'): + problems.append(f'{where}: email source needs match.from') + key = (match.get('from', '').lower(), match.get('subject', ''), match.get('list_id', '')) + if key in rules: + problems.append(f'{where}: duplicate match rule with {rules[key]}') + rules[key] = ident + elif source.get('channel') == 'url' and not source.get('url'): + problems.append(f'{where}: url source needs url') + for repo, subscription in registry['subscriptions'].items(): + where = f"subscriptions/{subscription['_file']}" + if subscription['_file'] != f'{repo}.json': + problems.append(f'{where}: file name must be .json') + if not subscription.get('agent'): + problems.append(f'{where}: missing agent') + channels = set(subscription.get('channels', [])) + if not channels or not channels <= CHANNELS: + problems.append(f'{where}: channels must be a non-empty subset of {sorted(CHANNELS)}') + if 'digest' in channels and subscription.get('cadence') not in ('daily', 'weekly'): + problems.append(f'{where}: digest channel needs cadence daily or weekly') + for topic in subscription.get('topics', []): + if topic not in known: + problems.append(f'{where}: unknown topic {topic}') + for ident in subscription.get('sources', []): + source = registry['sources'].get(ident) + if source is None: + problems.append(f'{where}: unknown source {ident}') + elif source.get('visibility') == 'restricted' and source.get('owner_repo') != repo: + problems.append(f'{where}: restricted source {ident} belongs to {source["owner_repo"]}') + return problems + + +def source_for_mail(registry, sender, subject='', list_id=''): + """Return the email source whose match rule fits, or None (quarantine).""" + sender = (sender or '').lower() + for ident, source in registry['sources'].items(): + if source.get('channel') != 'email': + continue + match = source.get('match', {}) + if match.get('from', '').lower() != sender: + continue + if match.get('subject') and match['subject'].lower() not in (subject or '').lower(): + continue + if match.get('list_id') and match['list_id'] != list_id: + continue + return ident + return None + + +def visible_sources(registry, repo): + """Sources a repo may see: shared ones, plus restricted ones it owns.""" + return {ident: source for ident, source in registry['sources'].items() + if source.get('visibility') == 'shared' or source.get('owner_repo') == repo} + + +def subscribers_for(registry, source_id): + """Repos whose subscription matches a source by id or by topic.""" + source = registry['sources'].get(source_id) + if source is None: + return [] + matched = [] + for repo, subscription in registry['subscriptions'].items(): + if source_id not in visible_sources(registry, repo): + continue + if source_id in subscription.get('sources', []) or \ + set(source.get('topics', [])) & set(subscription.get('topics', [])): + matched.append(repo) + return matched diff --git a/inbound_library/store.py b/inbound_library/store.py new file mode 100644 index 0000000..9b0bd9b --- /dev/null +++ b/inbound_library/store.py @@ -0,0 +1,141 @@ +"""Append-only, content-addressed item store. + +Original bytes live under data/items/ and are never overwritten or +deleted. Each delivery of those bytes is its own immutable record. Derived text +is rebuildable and may be replaced. +""" +import hashlib +import json +import os +import tempfile +from pathlib import Path + +SCHEMA = 'inbound-library/v1' +HEX = set('0123456789abcdef') + + +def canonical(value): + return json.dumps(value, sort_keys=True, separators=(',', ':'), ensure_ascii=False).encode() + + +def sha256(data): + return hashlib.sha256(data).hexdigest() + + +def is_id(ident): + return isinstance(ident, str) and len(ident) == 64 and set(ident) <= HEX + + +def atomic(path, data, immutable=False): + """Write via a temp file. With immutable, an existing path is never replaced.""" + path = Path(path) + path.parent.mkdir(parents=True, exist_ok=True) + fd, name = tempfile.mkstemp(dir=path.parent, prefix='.write-') + try: + with os.fdopen(fd, 'wb') as stream: + stream.write(data) + stream.flush() + os.fsync(stream.fileno()) + if immutable: + try: + os.link(name, path) + except FileExistsError: + if path.read_bytes() != data: + raise ValueError(f'refusing to overwrite {path}') + return + os.replace(name, path) + finally: + if os.path.exists(name): + os.unlink(name) + + +def data_dir(root): + return Path(root) / 'data' + + +def put_record(root, kind, value): + """Store an immutable JSON record keyed by the hash of its canonical form.""" + content = canonical(value) + ident = sha256(content) + atomic(data_dir(root) / kind / f'{ident}.json', content + b'\n', immutable=True) + return ident + + +def get_record(root, kind, ident): + if not is_id(ident): + raise ValueError('invalid content ID') + value = json.loads((data_dir(root) / kind / f'{ident}.json').read_text()) + if sha256(canonical(value)) != ident: + raise ValueError(f'integrity mismatch: {kind}/{ident}') + return value + + +def records(root, kind): + directory = data_dir(root) / kind + if not directory.is_dir(): + return [] + return [get_record(root, kind, p.stem) for p in sorted(directory.glob('*.json'))] + + +def put_item(root, data): + """Retain original bytes. Returns (item_id, newly_stored).""" + ident = sha256(data) + path = data_dir(root) / 'items' / ident + existed = path.exists() + atomic(path, data, immutable=True) + return ident, not existed + + +def get_item(root, ident): + if not is_id(ident): + raise ValueError('invalid item ID') + data = (data_dir(root) / 'items' / ident).read_bytes() + if sha256(data) != ident: + raise ValueError(f'integrity mismatch: items/{ident}') + return data + + +def record_delivery(root, item, provenance): + """Record one delivery of an item. Without a source it is quarantined.""" + if not is_id(item): + raise ValueError('invalid item ID') + delivery = {'schema': SCHEMA, 'item': item, **provenance} + delivery['status'] = 'routed' if provenance.get('source') else 'quarantined' + return put_record(root, 'deliveries', delivery) + + +def put_derived(root, item, derived): + """Derived text and metadata are rebuildable, so they may be replaced.""" + atomic(data_dir(root) / 'derived' / f'{item}.json', + canonical({'schema': SCHEMA, 'item': item, **derived}) + b'\n') + + +def get_derived(root, item): + path = data_dir(root) / 'derived' / f'{item}.json' + return json.loads(path.read_text()) if path.exists() else None + + +def verify(root): + """Rehash every item and record. Returns a list of problems.""" + problems = [] + base = data_dir(root) + items = base / 'items' + if items.is_dir(): + for path in sorted(items.iterdir()): + if path.name.startswith('.'): + continue + if not is_id(path.name) or sha256(path.read_bytes()) != path.name: + problems.append(f'item integrity mismatch: {path.name}') + for kind in ('deliveries', 'notices'): + directory = base / kind + if not directory.is_dir(): + continue + for path in sorted(directory.glob('*.json')): + try: + value = get_record(root, kind, path.stem) + except (ValueError, json.JSONDecodeError): + problems.append(f'record integrity mismatch: {kind}/{path.stem}') + continue + if kind == 'deliveries' and not (items / value.get('item', '')).is_file(): + problems.append(f'delivery {path.stem} refers to missing item') + return problems diff --git a/ops/systemd/inbound-library.service b/ops/systemd/inbound-library.service new file mode 100644 index 0000000..e0666e2 --- /dev/null +++ b/ops/systemd/inbound-library.service @@ -0,0 +1,10 @@ +[Unit] +Description=inbound-library intake, matching and notices +After=network-online.target + +[Service] +Type=oneshot +WorkingDirectory=%h/inbound-library +# INBOUND_LIBRARY_IMAP_USER / INBOUND_LIBRARY_IMAP_PASSWORD, mode 0600, never committed. +EnvironmentFile=-%h/.config/inbound-library/imap.env +ExecStart=/usr/bin/python3 -m inbound_library run-due diff --git a/ops/systemd/inbound-library.timer b/ops/systemd/inbound-library.timer new file mode 100644 index 0000000..eaf633f --- /dev/null +++ b/ops/systemd/inbound-library.timer @@ -0,0 +1,10 @@ +[Unit] +Description=Check the inbound-library mailbox + +[Timer] +OnBootSec=3min +OnUnitActiveSec=15min +Persistent=true + +[Install] +WantedBy=timers.target diff --git a/registry/subscriptions/intelligence-radar.json b/registry/subscriptions/intelligence-radar.json new file mode 100644 index 0000000..d14acec --- /dev/null +++ b/registry/subscriptions/intelligence-radar.json @@ -0,0 +1,8 @@ +{ + "repo": "intelligence-radar", + "agent": "intelligence-radar", + "topics": ["intelligence.daily-brief", "intelligence.weekly-brief"], + "sources": [], + "channels": ["hub", "pull", "digest"], + "cadence": "daily" +} diff --git a/registry/topics.json b/registry/topics.json new file mode 100644 index 0000000..4b25614 --- /dev/null +++ b/registry/topics.json @@ -0,0 +1,7 @@ +{ + "version": 2, + "topics": { + "intelligence.daily-brief": "Daily editorial brief on intelligence and AI (automation-generated, unreviewed)", + "intelligence.weekly-brief": "Weekly editorial brief on intelligence and AI (automation-generated, unreviewed)" + } +} diff --git a/tests/test_library.py b/tests/test_library.py new file mode 100644 index 0000000..08fddab --- /dev/null +++ b/tests/test_library.py @@ -0,0 +1,202 @@ +import json +import tempfile +import unittest +from datetime import datetime, timedelta, timezone +from email.message import EmailMessage +from pathlib import Path + +from inbound_library import intake, notify, store +from inbound_library import registry as reg +from inbound_library.__main__ import main, run_due + + +def mail(sender, subject, body, attachment=None, message_id=None): + message = EmailMessage() + message['From'] = sender + message['To'] = 'library@coulomb.social' + message['Subject'] = subject + message['Message-ID'] = message_id or f'<{abs(hash((sender, subject, body)))}@test>' + message['Authentication-Results'] = 'mx.test; dkim=pass' + message.set_content(body) + if attachment: + message.add_attachment(attachment, maintype='text', subtype='plain', filename='brief.txt') + return message.as_bytes() + + +def write(path, value): + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(value)) + + +class Library(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.root = Path(self.tmp.name) + self.mailbox_dir = self.root / 'fixture-mail' + self.mailbox_dir.mkdir() + write(self.root / 'registry' / 'topics.json', + {'version': 1, 'topics': {'intelligence.daily-brief': 'Daily intelligence brief', + 'research.papers': 'Research papers'}}) + write(self.root / 'registry' / 'sources' / 'grok-daily.json', { + 'id': 'grok-daily', 'owner_repo': 'intelligence-radar', 'channel': 'email', + 'match': {'from': 'noreply@x.ai'}, 'topics': ['intelligence.daily-brief'], + 'cadence': 'daily', 'visibility': 'shared'}) + write(self.root / 'registry' / 'subscriptions' / 'intelligence-radar.json', { + 'repo': 'intelligence-radar', 'agent': 'intelligence-radar', + 'topics': ['intelligence.daily-brief'], 'channels': ['hub', 'pull', 'digest'], + 'cadence': 'daily'}) + self.posts = [] + + def tearDown(self): + self.tmp.cleanup() + + def drop(self, name, raw): + (self.mailbox_dir / name).write_bytes(raw) + + def mailbox(self, uidvalidity=1): + return intake.DirectoryMailbox(self.mailbox_dir, uidvalidity) + + def post(self, api_base, payload): + self.posts.append(payload) + + # Store + def test_item_is_never_overwritten_and_duplicates_collapse(self): + first, new = store.put_item(self.root, b'brief') + again, new_again = store.put_item(self.root, b'brief') + self.assertEqual(first, again) + self.assertTrue(new) + self.assertFalse(new_again) + with self.assertRaises(ValueError): + store.atomic(self.root / 'data' / 'items' / first, b'other', immutable=True) + + def test_tampering_is_detected(self): + ident, _ = store.put_item(self.root, b'brief') + (self.root / 'data' / 'items' / ident).write_bytes(b'changed') + self.assertEqual(store.verify(self.root), [f'item integrity mismatch: {ident}']) + + # Registry + def test_registry_rejects_unknown_topic_and_restricted_leak(self): + write(self.root / 'registry' / 'sources' / 'private.json', { + 'id': 'private', 'owner_repo': 'markitect', 'channel': 'url', 'url': 'https://x', + 'topics': ['nope'], 'cadence': 'irregular', 'visibility': 'restricted'}) + write(self.root / 'registry' / 'subscriptions' / 'other.json', { + 'repo': 'other', 'agent': 'other', 'topics': [], 'sources': ['private'], + 'channels': ['pull']}) + problems = reg.validate(reg.load(self.root)) + self.assertIn('sources/private.json: unknown topic nope', problems) + self.assertTrue(any('restricted source private belongs to markitect' in p for p in problems)) + + def test_source_from_one_repo_reaches_another_by_topic(self): + write(self.root / 'registry' / 'sources' / 'arxiv-agents.json', { + 'id': 'arxiv-agents', 'owner_repo': 'intelligence-radar', 'channel': 'url', + 'url': 'https://arxiv.org/list/cs.MA/recent', 'topics': ['research.papers'], + 'cadence': 'irregular', 'visibility': 'shared'}) + write(self.root / 'registry' / 'subscriptions' / 'markitect.json', { + 'repo': 'markitect', 'agent': 'markitect', 'topics': ['research.papers'], + 'channels': ['pull']}) + registry = reg.load(self.root) + self.assertEqual(reg.validate(registry), []) + self.assertEqual(reg.subscribers_for(registry, 'arxiv-agents'), ['markitect']) + + # Intake + def test_scan_retains_message_attachment_and_provenance(self): + self.drop('1.eml', mail('Grok ', 'Daily brief', 'Body text', b'attached')) + [delivery_id] = intake.scan(self.root, self.mailbox()) + delivery = store.get_record(self.root, 'deliveries', delivery_id) + self.assertEqual(delivery['source'], 'grok-daily') + self.assertEqual(delivery['status'], 'routed') + self.assertEqual(delivery['authentication_results'], ['mx.test; dkim=pass']) + raw = store.get_item(self.root, delivery['item']) + self.assertEqual(raw, (self.mailbox_dir / '1.eml').read_bytes()) + derived = store.get_derived(self.root, delivery['item']) + self.assertIn('Body text', derived['text']) + [attachment] = derived['attachments'] + self.assertEqual(store.get_item(self.root, attachment['item']), b'attached') + + def test_cursor_skips_seen_mail_and_uidvalidity_change_does_not_duplicate(self): + self.drop('1.eml', mail('noreply@x.ai', 'Daily brief', 'one')) + self.assertEqual(len(intake.scan(self.root, self.mailbox())), 1) + self.assertEqual(intake.scan(self.root, self.mailbox()), []) + rescanned = intake.scan(self.root, self.mailbox(uidvalidity=2)) + self.assertEqual(len(rescanned), 1) + self.assertEqual(len(store.records(self.root, 'deliveries')), 1) + + def test_unknown_sender_is_quarantined_then_requeued(self): + self.drop('1.eml', mail('someone@example.org', 'Hello', 'spam?')) + [delivery_id] = intake.scan(self.root, self.mailbox()) + self.assertEqual(store.get_record(self.root, 'deliveries', delivery_id)['status'], 'quarantined') + self.assertEqual(notify.match(self.root), []) + self.assertEqual(notify.health(self.root)['quarantined_items'], 1) + write(self.root / 'registry' / 'sources' / 'someone.json', { + 'id': 'someone', 'owner_repo': 'intelligence-radar', 'channel': 'email', + 'match': {'from': 'someone@example.org'}, 'topics': ['intelligence.daily-brief'], + 'cadence': 'irregular', 'visibility': 'shared'}) + self.assertEqual(len(intake.requeue_quarantine(self.root)), 1) + self.assertEqual(notify.health(self.root)['quarantined_items'], 0) + self.assertEqual(len(notify.match(self.root)), 1) + + def test_live_mailbox_needs_credentials_from_environment(self): + self.assertIsNone(intake.mailbox_from_config({}, environ={})) + box = intake.mailbox_from_config({}, environ={'INBOUND_LIBRARY_IMAP_USER': 'u', + 'INBOUND_LIBRARY_IMAP_PASSWORD': 'p'}) + self.assertEqual((box.host, box.port, box.folder), ('imap.ionos.de', 993, 'INBOX')) + + # Notices and channels + def test_notices_are_idempotent_and_hub_outage_stays_pending(self): + self.drop('1.eml', mail('noreply@x.ai', 'Daily brief', 'one')) + intake.scan(self.root, self.mailbox()) + self.assertEqual(len(notify.match(self.root)), 1) + self.assertEqual(notify.match(self.root), []) + + def down(api_base, payload): + raise OSError('connection refused') + report = notify.deliver_hub(self.root, 'http://hub', post=down) + self.assertEqual(report['intelligence-radar']['pending'], 1) + report = notify.deliver_hub(self.root, 'http://hub', post=self.post) + self.assertEqual(report['intelligence-radar'], {'ok': True, 'sent': 1}) + self.assertEqual(notify.deliver_hub(self.root, 'http://hub', post=self.post), {}) + [message] = self.posts + self.assertEqual(message['to_agent'], 'intelligence-radar') + self.assertIn('Daily brief', message['body']) + + def test_pull_and_digest(self): + self.drop('1.eml', mail('noreply@x.ai', 'Daily brief', 'one')) + intake.scan(self.root, self.mailbox()) + notify.match(self.root) + [notice] = notify.notices_for(self.root, 'intelligence-radar', + topic='intelligence.daily-brief') + self.assertEqual(notice['subject'], 'Daily brief') + self.assertEqual(notify.notices_for(self.root, 'intelligence-radar', topic='other'), []) + now = datetime.fromisoformat(notice['received_at']) + [path] = notify.write_digests(self.root, now=now) + self.assertIn('Daily brief', (self.root / path).read_text()) + + def test_silent_source_is_reported(self): + later = datetime.now(timezone.utc) + timedelta(days=3) + self.assertEqual(notify.health(self.root, now=later)['silent_sources'], + [{'source': 'grok-daily', 'last_seen': None}]) + self.drop('1.eml', mail('noreply@x.ai', 'Daily brief', 'one')) + intake.scan(self.root, self.mailbox()) + self.assertEqual(notify.health(self.root)['silent_sources'], []) + self.assertEqual(len(notify.health(self.root, now=later)['silent_sources']), 1) + + def test_run_due_end_to_end_and_repeatable(self): + self.drop('1.eml', mail('noreply@x.ai', 'Daily brief', 'one')) + report = run_due(self.root, {}, mailbox=self.mailbox(), post=self.post) + self.assertTrue(report['ok'], report) + self.assertEqual((report['intake']['retained'], report['notices']), (1, 1)) + report = run_due(self.root, {}, mailbox=self.mailbox(), post=self.post) + self.assertEqual((report['intake']['retained'], report['notices'], report['hub']), (0, 0, {})) + self.assertEqual(len(self.posts), 1) + + def test_run_due_without_credentials_still_notifies(self): + report = run_due(self.root, {}, post=self.post) + self.assertFalse(report['ok']) + self.assertEqual(report['intake']['status'], 'needs_credentials') + + def test_cli_check(self): + self.assertEqual(main(['--root', str(self.root), 'check']), 0) + + +if __name__ == '__main__': + unittest.main() diff --git a/tools/demo.py b/tools/demo.py new file mode 100644 index 0000000..fa7b506 --- /dev/null +++ b/tools/demo.py @@ -0,0 +1,38 @@ +"""Run intake, matching, notices and digests on synthetic mail in a temp root.""" +import json +import sys +import tempfile +from email.message import EmailMessage +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) +from inbound_library import intake, notify # noqa: E402 +from inbound_library.__main__ import run_due # noqa: E402 + +with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + (root / 'registry' / 'sources').mkdir(parents=True) + (root / 'registry' / 'subscriptions').mkdir() + (root / 'registry' / 'topics.json').write_text(json.dumps( + {'version': 1, 'topics': {'demo.brief': 'Synthetic brief'}})) + (root / 'registry' / 'sources' / 'demo-sender.json').write_text(json.dumps({ + 'id': 'demo-sender', 'owner_repo': 'demo', 'channel': 'email', + 'match': {'from': 'brief@demo.test'}, 'topics': ['demo.brief'], + 'cadence': 'daily', 'visibility': 'shared'})) + (root / 'registry' / 'subscriptions' / 'demo.json').write_text(json.dumps({ + 'repo': 'demo', 'agent': 'demo', 'topics': ['demo.brief'], + 'channels': ['hub', 'pull', 'digest'], 'cadence': 'daily'})) + mail = root / 'mail' + mail.mkdir() + for n, sender in enumerate(['brief@demo.test', 'stranger@demo.test'], 1): + message = EmailMessage() + message['From'], message['Subject'] = sender, f'Synthetic message {n}' + message['Message-ID'] = f'' + message.set_content(f'Synthetic body {n}.') + (mail / f'{n}.eml').write_bytes(message.as_bytes()) + sent = [] + report = run_due(root, {}, mailbox=intake.DirectoryMailbox(mail), + post=lambda base, payload: sent.append(payload)) + print(json.dumps(report, indent=2)) + print('\nHub message (not sent):\n' + sent[0]['body']) + print('\nHealth:\n' + json.dumps(notify.health(root), indent=2)) diff --git a/workplans/ILIB-WP-0002-library-infrastructure.md b/workplans/ILIB-WP-0002-library-infrastructure.md index 2af251e..1ec4f76 100644 --- a/workplans/ILIB-WP-0002-library-infrastructure.md +++ b/workplans/ILIB-WP-0002-library-infrastructure.md @@ -29,7 +29,7 @@ Python 3.11+ standard library only. The first consumer is intelligence-radar ```task id: ILIB-WP-0002-T01 -status: todo +status: done priority: high state_hub_task_id: "36f19bb9-045f-59fd-9b7c-509cb7dd00f8" ``` @@ -43,11 +43,13 @@ with schema versions. Record the install, test and run commands in AGENTS.md Acceptance: `make check` runs the unit tests on synthetic fixtures. The contracts document every field that a consumer may rely on. +Done 2026-10-05: package `inbound_library`, CLI, Makefile, `docs/data-contracts.md` and the AGENTS.md commands are in place. `data/` is gitignored because mail can be private. + ## Append-only item store ```task id: ILIB-WP-0002-T02 -status: todo +status: done priority: high state_hub_task_id: "4bbe176a-b210-5c20-9510-c90029bf2425" ``` @@ -61,11 +63,13 @@ never delete. Items without provenance go to quarantine. Acceptance: the tests prove there is no overwrite, that deduplication works, that tampering is detected and that quarantine routing works. +Done 2026-10-05: `store.py` is in place. Tests cover no-overwrite, deduplication, tamper detection and quarantine. + ## Read-only IMAP intake ```task id: ILIB-WP-0002-T03 -status: todo +status: done priority: high state_hub_task_id: "dd06da59-5773-53fa-aba3-dc79dff23c28" ``` @@ -81,11 +85,13 @@ read-only IMAP approach of email-connect EMAIL-WP-0002 where that helps. Acceptance: the tests run against a fixture mailbox. A live run is part of T07. +Done 2026-10-05: `intake.py` is in place. It selects read-only with BODY.PEEK[] and keeps a UIDVALIDITY/UID cursor. UIDs are left out of delivery records so rescans do not duplicate. Attachments are retained as items. Unknown senders are quarantined and re-routed (`requeue_quarantine`) once a source is registered. The tests use a fixture mailbox. + ## File-backed registries ```task id: ILIB-WP-0002-T04 -status: todo +status: done priority: high state_hub_task_id: "4a2cbba2-55d8-5a0e-bd23-74dbfdb7ae61" ``` @@ -101,11 +107,13 @@ other repos. Acceptance: `check` rejects invalid registries. A source registered by repo A is matched for repo B through a shared topic. +Done 2026-10-05: `registry.py` validates topics, sources and subscriptions, including the restricted-source leak check. A test shows a source from repo A matched for repo B through a shared topic. + ## Matching and notification ```task id: ILIB-WP-0002-T05 -status: todo +status: done priority: high state_hub_task_id: "5da65f55-7608-55b6-9378-7bef3ea8e4ca" ``` @@ -122,11 +130,13 @@ hub is unreachable, keep the notice pending and retry on the next run. Acceptance: the tests prove idempotent delivery and recovery when the hub is unreachable. +Done 2026-10-05: `notify.py` is in place. Notices are deterministic. Hub messages give a pull command that works from any repo. Tests cover the `new`/`show`/`get` commands, digest files, idempotent delivery and a hub outage that stays pending. + ## Scheduling, health and source silence ```task id: ILIB-WP-0002-T06 -status: todo +status: done priority: medium state_hub_task_id: "01144f1e-b3ea-50b6-bf7c-32e9a29d7213" ``` @@ -142,6 +152,8 @@ Health reports: Acceptance: a silent source is reported, and repeated timer wakes are safe. +Done 2026-10-05: `run-due` runs under a lock. The systemd units are in `ops/systemd/`, and the service reads `~/.config/inbound-library/imap.env`. Tests cover `health`, including silent-source detection. 14 tests pass. + ## Provision the dedicated mailbox ```task