coulomb/inbound-library
5
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Build the inbound-library core: store, IMAP intake, registries, notices.

Browse source 
ILIB-WP-0002-T01..T06: append-only item store, read-only IMAP intake with
quarantine, file-backed registries, idempotent hub/pull/digest notices,
run-due with health and the systemd user timer. 14 tests pass. Registers
the intelligence topics and the intelligence-radar subscription.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 1170398@bnt-lap001
Assistant-Session: 622cd705-cb31-4322-ba08-dd42e44fa583
This commit is contained in:
tegwick 2026-10-05 11:21:13 +02:00
parent fe84482730
commit d05d6ed35b
18 changed files with 1168 additions and 6 deletions
Download patch file Download diff file Expand all files Collapse all files

178
inbound_library/notify.py Normal file
View file

@ -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 <item> Original bytes: {cli} get <item>')
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)}