diff --git a/docs/inbound-email.md b/docs/inbound-email.md new file mode 100644 index 0000000..f06108f --- /dev/null +++ b/docs/inbound-email.md @@ -0,0 +1,107 @@ +# Inbound email (reply-to-review) — architecture and runbook + +PIs are emailed collaboration proposals and can answer by replying: a rating +(1–4) files a `ProposalReview`, instructions reopen the proposal for +refinement. This document covers how the pipeline works, why it was dead in +production, and how to bring it up safely. + +## Architecture + +``` +PI hits "reply" ──► DNS MX (reply.copi.science) + └─► SES inbound SMTP (us-east-2) — receipt rule + └─► S3 s3://copi-inbound-email/inbound/ + └─► worker poll_inbound_emails (every 60s, + gated on ENABLE_INBOUND_EMAIL) + └─► process_inbound_email: + SES auth verdicts → auto-reply + filter → token lookup → sender + match → LLM classify → + review / instruction / help email +``` + +Outbound review emails set `Reply-To: review+@reply.copi.science` +(token = 64-char urlsafe secret stored on the `EmailNotification` row). The +worker deletes each S3 object after processing; objects that fail processing +3 times are quarantined under `failed/` for inspection. + +## Why it was dead in production (investigated 2026-08-11) + +Every layer below the outbound send was missing. In order of the mail's path: + +1. **No MX record** on `reply.copi.science` (only an A record to the EC2 + box, which listens on no SMTP port) — PI replies bounced after their mail + server gave up retrying. +2. **No S3 bucket**: `copi-inbound-email` did not exist in account + 215751090072. +3. **No SES receipt rule** delivering the reply domain to S3 (and the reply + domain was not verified for receiving). +4. **Instance role** `copi-ec2-ses-role` has send-only SES perms and no S3 + read/delete on the inbound bucket. +5. **`ENABLE_INBOUND_EMAIL` unset** in the prod `.env`, so the worker never + polled even if 1–4 had existed. + +Meanwhile the outbound emails actively told PIs to reply (129 sent by +2026-08-06; outbound was then paused by disabling notification categories in +the DB). + +## Code changes on the email-fix branch + +- Outbound review/new-proposal/welcome emails only solicit replies (and only + set `Reply-To` to the reply domain) when `ENABLE_INBOUND_EMAIL=true` — + outbound email can be re-enabled safely before inbound is provisioned. +- The SEC-5 anti-spoofing gate trusts only the topmost (SES-stamped) + `Authentication-Results` header; a sender-forged `...pass` header no longer + overrides SES's fail verdicts. +- HTML-only replies fall back to tag-stripped HTML instead of being silently + dropped. +- Auto-submitted mail (RFC 3834, e.g. out-of-office) is ignored — no help + email is sent back, so no mail loops. +- The declared per-token rate limit (10 replies/hour) is enforced. +- A poison message is quarantined to `failed/` after 3 attempts instead of + being retried every 60 seconds forever. + +## Bringing inbound email up + +Run each step with **admin** AWS credentials (the instance role cannot do +this — see finding 4): + +```bash +# 1. See what's missing: +python scripts/setup_inbound_email.py --check + +# 2. Create bucket, bucket policy, receipt rule set/rule; prints DNS + IAM steps: +python scripts/setup_inbound_email.py --provision +``` + +Then, in this order: + +1. Add the printed DNS records at the registrar (Namecheap): + `reply.copi.science. MX 10 inbound-smtp.us-east-2.amazonaws.com.` plus the + `_amazonses` TXT verification record if the domain was newly verified. +2. Attach the printed S3 policy to `copi-ec2-ses-role`. +3. Re-run `--check` until all layers are OK. +4. Set `ENABLE_INBOUND_EMAIL=true` in the prod `.env` and recreate BOTH the + worker (polling + proposal/reminder emails) and the app (the welcome email + reads the same flag for its reply-vs-dashboard copy — recreating only the + worker leaves new signups being told the dashboard is the only way in): + `docker compose -f docker-compose.prod.yml -f docker-compose.override.yml up -d app worker` + (`up -d` recreates on env change; a bare `docker restart` re-runs the OLD + environment — `env_file` is resolved at container creation.) +5. End-to-end test: trigger a proposal notification to a test recipient, + reply with "3 sounds great", and watch + `docker logs -f copi-python-worker-1` for `Email review created`. + Confirm the `proposal_reviews` row and the confirmation email. + +Only after step 5 passes, re-enable the notification categories that were +turned off in the DB (`email_notification_preferences.enabled`) / user +frequencies as desired. + +## Operational notes + +- The reply flow degrades safely: with `ENABLE_INBOUND_EMAIL` unset/false the + worker skips polling AND outbound emails stop soliciting replies. +- Quarantined mail lands in `s3://copi-inbound-email/failed/` — inspect and + delete manually. +- The rate limiter and quarantine counters are in-memory; a worker restart + resets them (by design — worst case is one extra processing round). diff --git a/scripts/backfill_publications.py b/scripts/backfill_publications.py new file mode 100644 index 0000000..80b31bf --- /dev/null +++ b/scripts/backfill_publications.py @@ -0,0 +1,139 @@ +"""Backfill publications rows from a curated agent_id -> PMID mapping. + +Issue #29 rollout prerequisite: eleven active labs have zero publications +rows because the only ingest path (profile pipeline: ORCID works -> PMID) +found nothing for them — their ORCID profiles list no works — so the +fail-closed authorship guard mutes every first-person paper claim they make. +PubMed author search cannot disambiguate names like Wu or Wilson reliably, +so the input here is a human-curated JSON mapping: + + {"good": ["21234567", "31234567"], "cravatt": ["19876543"]} + +Usage (inside the app container, dry run first): + + docker compose exec app python scripts/backfill_publications.py --file data/backfill_pmids.json + docker compose exec app python scripts/backfill_publications.py --file data/backfill_pmids.json --apply + +Rows are visible to the running simulation on its next ~30s roster sync +(_load_publication_records) — no restart needed. +""" + +import argparse +import asyncio +import json +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +# (agent_id, action, pmid) — action is one of: +# would-insert / insert / skip-existing / error-no-agent / error-no-record +ReportEntry = tuple[str, str, str] + + +async def backfill(db, mapping: dict[str, list[str]], fetch=None, apply: bool = False) -> list[ReportEntry]: + """Insert Publication rows for each agent's curated PMIDs. + + Idempotent: PMIDs the user already has are skipped (and not fetched). + Dry run (the default) reports what WOULD be inserted and writes nothing. + """ + from sqlalchemy import select + + from src.models import AgentRegistry, Publication + from src.services.pubmed import fetch_pubmed_records, normalize_doi + + if fetch is None: + fetch = fetch_pubmed_records + + report: list[ReportEntry] = [] + for agent_id, pmids in mapping.items(): + row = ( + await db.execute( + select(AgentRegistry).where(AgentRegistry.agent_id == agent_id) + ) + ).scalar_one_or_none() + if row is None or row.user_id is None: + report.append((agent_id, "error-no-agent", "")) + continue + + existing = { + p + for (p,) in ( + await db.execute( + select(Publication.pmid).where(Publication.user_id == row.user_id) + ) + ).all() + if p + } + wanted = [str(p).strip() for p in pmids if str(p).strip()] + missing: list[str] = [] + for pmid in wanted: + if pmid in existing: + report.append((agent_id, "skip-existing", pmid)) + else: + missing.append(pmid) + if not missing: + continue + + records = {r["pmid"]: r for r in await fetch(missing) if r.get("pmid")} + for pmid in missing: + rec = records.get(pmid) + if rec is None: + report.append((agent_id, "error-no-record", pmid)) + continue + if apply: + db.add( + Publication( + user_id=row.user_id, + pmid=pmid, + pmcid=rec.get("pmcid"), + doi=normalize_doi(rec.get("doi")), + title=rec.get("title", ""), + abstract=rec.get("abstract", ""), + journal=rec.get("journal"), + year=rec.get("year"), + ) + ) + report.append((agent_id, "insert", pmid)) + else: + report.append((agent_id, "would-insert", pmid)) + if apply: + await db.flush() + return report + + +async def _main() -> int: + ap = argparse.ArgumentParser(description=__doc__) + ap.add_argument("--file", required=True, help="JSON file: {agent_id: [pmid, ...]}") + ap.add_argument("--apply", action="store_true", help="Write rows (default: dry run)") + args = ap.parse_args() + + mapping = json.loads(Path(args.file).read_text(encoding="utf-8")) + if not isinstance(mapping, dict): + print("Input must be a JSON object mapping agent_id -> [pmid, ...]") + return 2 + + from src.database import get_session_factory + + factory = get_session_factory() + async with factory() as db: + report = await backfill(db, mapping, apply=args.apply) + if args.apply: + await db.commit() + + for agent_id, action, pmid in report: + print(f"{agent_id}: {action} {pmid}".rstrip()) + inserted = sum(1 for _, a, _ in report if a == "insert") + planned = sum(1 for _, a, _ in report if a == "would-insert") + errors = sum(1 for _, a, _ in report if a.startswith("error")) + if args.apply: + print(f"\nInserted {inserted} rows ({errors} errors). The running simulation") + print("picks them up on its next ~30s roster sync — no restart needed.") + else: + print(f"\nDry run: {planned} rows would be inserted ({errors} errors).") + print("Re-run with --apply to write them.") + return 1 if errors else 0 + + +if __name__ == "__main__": + sys.exit(asyncio.run(_main())) diff --git a/scripts/setup_inbound_email.py b/scripts/setup_inbound_email.py new file mode 100644 index 0000000..957a0e2 --- /dev/null +++ b/scripts/setup_inbound_email.py @@ -0,0 +1,321 @@ +#!/usr/bin/env python3 +""" +Check (and optionally provision) the AWS/DNS infrastructure for inbound email +replies — the review+TOKEN@reply.copi.science flow. + +Background (investigation of 2026-08-11) +----------------------------------------- +The reply-by-email review flow shipped in code but its infrastructure was +never provisioned on prod. Every layer was missing, so PI replies bounced and +nothing was processed: + + 1. DNS: reply.copi.science had NO MX record (replies never reached AWS). + 2. S3: the copi-inbound-email bucket did not exist. + 3. SES: no receipt rule delivered mail for the reply domain to S3. + 4. IAM: copi-ec2-ses-role had send-only perms (no S3 read/delete for polling). + 5. Env: ENABLE_INBOUND_EMAIL was unset, so the worker never polled anyway. + +This script verifies each layer (--check, the default) and can create the AWS +pieces (--provision). DNS records must be added at the registrar by hand; the +script prints exactly what to add. + +Prerequisites +------------- +Run from a machine/profile with ADMIN AWS credentials (SES receipt rules, S3 +bucket creation, IAM read). The EC2 instance role is NOT sufficient — that is +finding #4 above. + +Usage +----- + # Report the state of every layer, change nothing: + python scripts/setup_inbound_email.py --check + + # Create bucket + policy + receipt rule set/rule, then print DNS + IAM steps: + python scripts/setup_inbound_email.py --provision + + # Non-default names: + python scripts/setup_inbound_email.py --check \ + --region us-east-2 --bucket copi-inbound-email \ + --prefix inbound/ --reply-domain reply.copi.science + +After provisioning +------------------ + 1. Add the printed MX (and, if newly verifying the domain, TXT) records. + 2. Attach the printed IAM policy to the instance role (copi-ec2-ses-role). + 3. Set ENABLE_INBOUND_EMAIL=true in the prod .env and recreate the worker: + docker compose -f docker-compose.prod.yml -f docker-compose.override.yml \ + up -d worker + 4. Send a test reply and watch: docker logs -f copi-python-worker-1 +""" + +import argparse +import json +import subprocess +import sys + +RULE_SET_NAME = "copi-inbound" +RULE_NAME = "copi-reply-to-s3" + + +def _print(status: str, layer: str, detail: str) -> None: + print(f" [{status:^4}] {layer}: {detail}") + + +def check_mx(reply_domain: str, region: str) -> bool: + """MX must point at SES inbound SMTP for the region.""" + expected = f"inbound-smtp.{region}.amazonaws.com" + try: + out = subprocess.run( + ["dig", "+short", "MX", reply_domain], + capture_output=True, text=True, timeout=10, + ).stdout.strip() + except (FileNotFoundError, subprocess.TimeoutExpired): + _print("SKIP", "DNS", f"`dig` unavailable — check manually that {reply_domain} " + f"has MX 10 {expected}") + return False + if expected in out: + _print("OK", "DNS", f"MX for {reply_domain} → {expected}") + return True + _print("FAIL", "DNS", f"no MX for {reply_domain} pointing at {expected} " + f"(got: {out or 'no MX record at all'})") + print(f" Add at the registrar: {reply_domain}. MX 10 {expected}.") + return False + + +def check_bucket(s3, bucket: str) -> bool: + try: + s3.head_bucket(Bucket=bucket) + _print("OK", "S3", f"bucket {bucket} exists and is reachable") + return True + except Exception as exc: + _print("FAIL", "S3", f"bucket {bucket}: {exc}") + return False + + +def check_identity(ses, reply_domain: str) -> bool: + try: + attrs = ses.get_identity_verification_attributes(Identities=[reply_domain]) + status = ( + attrs["VerificationAttributes"] + .get(reply_domain, {}) + .get("VerificationStatus", "NotFound") + ) + except Exception as exc: + _print("SKIP", "SES identity", f"cannot query ({exc})") + return False + if status == "Success": + _print("OK", "SES identity", f"{reply_domain} is verified") + return True + _print("FAIL", "SES identity", f"{reply_domain} verification status: {status}") + return False + + +def check_receipt_rule(ses, bucket: str, reply_domain: str) -> bool: + try: + active = ses.describe_active_receipt_rule_set() + except Exception as exc: + _print("SKIP", "SES receipt", f"cannot query receipt rule sets ({exc})") + return False + for rule in active.get("Rules", []): + recipients = rule.get("Recipients", []) + domain_match = not recipients or any( + r == reply_domain or r.endswith("@" + reply_domain) for r in recipients + ) + s3_actions = [a["S3Action"] for a in rule.get("Actions", []) if "S3Action" in a] + if rule.get("Enabled") and domain_match and any( + a["BucketName"] == bucket for a in s3_actions + ): + _print("OK", "SES receipt", + f"active rule '{rule['Name']}' delivers {reply_domain} → s3://{bucket}") + return True + name = (active.get("Metadata") or {}).get("Name") + _print("FAIL", "SES receipt", + f"active rule set {name or '(none)'} has no enabled rule delivering " + f"{reply_domain} to s3://{bucket}") + return False + + +def check_env_flag() -> bool: + """This checks the LOCAL environment only — the flag that matters is the + one in the prod .env consumed by the worker container.""" + import os + + val = os.environ.get("ENABLE_INBOUND_EMAIL", "") + if val.lower() in ("1", "true", "yes"): + _print("OK", "Env", "ENABLE_INBOUND_EMAIL is set here") + else: + _print("WARN", "Env", + "ENABLE_INBOUND_EMAIL not set in this shell — ensure it is " + "true in the prod .env (worker service) once AWS+DNS are ready") + return True + + +def instance_role_policy(bucket: str, prefix: str) -> dict: + """The statements copi-ec2-ses-role needs for the worker's polling loop + (read+delete under the inbound prefix, write for failed/ quarantine).""" + return { + "Version": "2012-10-17", + "Statement": [ + { + "Sid": "CopiInboundEmailList", + "Effect": "Allow", + "Action": ["s3:ListBucket"], + "Resource": f"arn:aws:s3:::{bucket}", + }, + { + "Sid": "CopiInboundEmailReadWrite", + "Effect": "Allow", + "Action": ["s3:GetObject", "s3:DeleteObject", "s3:PutObject"], + "Resource": [ + f"arn:aws:s3:::{bucket}/{prefix}*", + f"arn:aws:s3:::{bucket}/failed/*", + ], + }, + ], + } + + +def ses_bucket_policy(bucket: str, account_id: str, region: str) -> dict: + """Allow SES (this account's receipt rules only) to write into the bucket.""" + return { + "Version": "2012-10-17", + "Statement": [ + { + "Sid": "AllowSESPuts", + "Effect": "Allow", + "Principal": {"Service": "ses.amazonaws.com"}, + "Action": "s3:PutObject", + "Resource": f"arn:aws:s3:::{bucket}/*", + "Condition": { + "StringEquals": {"AWS:SourceAccount": account_id}, + "ArnLike": { + "AWS:SourceArn": f"arn:aws:ses:{region}:{account_id}:receipt-rule-set/*" + }, + }, + } + ], + } + + +def provision(region: str, bucket: str, prefix: str, reply_domain: str) -> None: + import boto3 + + account_id = boto3.client("sts", region_name=region).get_caller_identity()["Account"] + s3 = boto3.client("s3", region_name=region) + ses = boto3.client("ses", region_name=region) + + # 1. Bucket (idempotent) + SES write policy + try: + s3.head_bucket(Bucket=bucket) + print(f"bucket {bucket} already exists") + except Exception: + kwargs = {"Bucket": bucket} + if region != "us-east-1": + kwargs["CreateBucketConfiguration"] = {"LocationConstraint": region} + s3.create_bucket(**kwargs) + s3.put_public_access_block( + Bucket=bucket, + PublicAccessBlockConfiguration={ + "BlockPublicAcls": True, "IgnorePublicAcls": True, + "BlockPublicPolicy": True, "RestrictPublicBuckets": True, + }, + ) + print(f"created bucket {bucket}") + s3.put_bucket_policy( + Bucket=bucket, Policy=json.dumps(ses_bucket_policy(bucket, account_id, region)) + ) + print("attached SES write policy to bucket") + + # 2. Domain identity for receiving (prints the TXT record if new) + attrs = ses.get_identity_verification_attributes(Identities=[reply_domain]) + status = ( + attrs["VerificationAttributes"].get(reply_domain, {}).get("VerificationStatus") + ) + if status != "Success": + token = ses.verify_domain_identity(Domain=reply_domain)["VerificationToken"] + print(f"requested domain verification for {reply_domain}; add DNS record:") + print(f' _amazonses.{reply_domain}. TXT "{token}"') + + # 3. Receipt rule set + rule (idempotent), then activate + try: + ses.create_receipt_rule_set(RuleSetName=RULE_SET_NAME) + print(f"created receipt rule set {RULE_SET_NAME}") + except ses.exceptions.AlreadyExistsException: + print(f"receipt rule set {RULE_SET_NAME} already exists") + rule = { + "Name": RULE_NAME, + "Enabled": True, + "Recipients": [reply_domain], + "Actions": [ + { + "S3Action": { + "BucketName": bucket, + "ObjectKeyPrefix": prefix, + } + } + ], + "ScanEnabled": True, + "TlsPolicy": "Optional", + } + try: + ses.create_receipt_rule(RuleSetName=RULE_SET_NAME, Rule=rule) + print(f"created receipt rule {RULE_NAME}") + except ses.exceptions.AlreadyExistsException: + ses.update_receipt_rule(RuleSetName=RULE_SET_NAME, Rule=rule) + print(f"updated receipt rule {RULE_NAME}") + active = ses.describe_active_receipt_rule_set().get("Metadata") or {} + if active.get("Name") != RULE_SET_NAME: + if active.get("Name"): + print(f"WARNING: replacing active rule set {active['Name']!r} — its rules " + f"stop matching. Merge them into {RULE_SET_NAME} first if needed.") + ses.set_active_receipt_rule_set(RuleSetName=RULE_SET_NAME) + print(f"activated receipt rule set {RULE_SET_NAME}") + + # 4. What cannot be done from here + print("\nRemaining manual steps:") + print(f" 1. Registrar DNS: {reply_domain}. MX 10 " + f"inbound-smtp.{region}.amazonaws.com.") + print(" 2. Attach this policy to the EC2 instance role (copi-ec2-ses-role):") + print(json.dumps(instance_role_policy(bucket, prefix), indent=4)) + print(" 3. Set ENABLE_INBOUND_EMAIL=true in the prod .env and recreate the worker.") + + +def main() -> int: + ap = argparse.ArgumentParser(description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter) + ap.add_argument("--check", action="store_true", default=False) + ap.add_argument("--provision", action="store_true", default=False) + ap.add_argument("--region", default="us-east-2") + ap.add_argument("--bucket", default="copi-inbound-email") + ap.add_argument("--prefix", default="inbound/") + ap.add_argument("--reply-domain", default="reply.copi.science") + args = ap.parse_args() + + if args.provision: + provision(args.region, args.bucket, args.prefix, args.reply_domain) + return 0 + + # Default: --check + import boto3 + + s3 = boto3.client("s3", region_name=args.region) + ses = boto3.client("ses", region_name=args.region) + print(f"Inbound email infrastructure check ({args.reply_domain} → " + f"s3://{args.bucket}/{args.prefix} in {args.region}):") + results = [ + check_mx(args.reply_domain, args.region), + check_identity(ses, args.reply_domain), + check_receipt_rule(ses, args.bucket, args.reply_domain), + check_bucket(s3, args.bucket), + check_env_flag(), + ] + if all(results): + print("All layers OK.") + return 0 + print("\nOne or more layers missing — run with --provision (admin creds) " + "and follow the printed manual steps.") + return 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/src/services/email.py b/src/services/email.py index f446120..871fc8b 100644 --- a/src/services/email.py +++ b/src/services/email.py @@ -198,6 +198,43 @@ def build_welcome_email(to_email: str, name: str | None = None, user_id: str | N greeting_name = (name or "").strip().split(" ")[0] if name else "" greeting = f"Hi {greeting_name}," if greeting_name else "Hi there," + + # Only describe the reply-by-email review flow when the inbound pipeline + # is actually enabled; otherwise point at the web dashboard alone. + reply_enabled = settings.enable_inbound_email + if reply_enabled: + review_how_text = ( + "HOW PROPOSAL REVIEW WORKS\n" + "When your agent and another lab's agent develop a promising idea, we email\n" + "you a short proposal. You can:\n" + " - Reply with a rating from 1 to 4:\n" + " 1 = Not a good idea 2 = Good idea\n" + " 3 = Great idea 4 = Excellent idea\n" + ' - Reply with instructions (e.g. "focus on the mitochondrial angle") and\n' + " your agent will re-engage to refine the idea.\n" + " - Or review it on the web dashboard.\n" + "Note: while you have unreviewed proposals, your agent pauses new\n" + "conversations — reviewing promptly keeps it active." + ) + review_how_html = ( + "
  • Rate it by replying with a number from 1 to 4.
    • \n" + "
  • Give instructions to refine it, and your agent re-engages.
    • \n" + "
  • Review it on the web dashboard.
    • " + ) + else: + review_how_text = ( + "HOW PROPOSAL REVIEW WORKS\n" + "When your agent and another lab's agent develop a promising idea, we email\n" + "you a short proposal. Open your dashboard to rate it from 1 to 4\n" + "(1 = Not a good idea, 2 = Good idea, 3 = Great idea, 4 = Excellent idea)\n" + "or to give your agent instructions to refine the idea.\n" + "Note: while you have unreviewed proposals, your agent pauses new\n" + "conversations — reviewing promptly keeps it active." + ) + review_how_html = ( + "
  • Rate it from 1 to 4 on your dashboard.
    • \n" + "
  • Give instructions to refine it, and your agent re-engages.
    • " + ) # HTML-escaped greeting for the HTML body (the name is the ORCID display # name, i.e. user-controlled) (SEC-13). greeting_html = f"Hi {esc(greeting_name)}," if greeting_name else "Hi there," @@ -226,17 +263,7 @@ def build_welcome_email(to_email: str, name: str | None = None, user_id: str | N - My Agent ({agent_url}) — request your agent and manage it. - Settings ({settings_url}) — choose which emails you receive and how often. -HOW PROPOSAL REVIEW WORKS -When your agent and another lab's agent develop a promising idea, we email -you a short proposal. You can: - - Reply with a rating from 1 to 4: - 1 = Not a good idea 2 = Good idea - 3 = Great idea 4 = Excellent idea - - Reply with instructions (e.g. "focus on the mitochondrial angle") and - your agent will re-engage to refine the idea. - - Or review it on the web dashboard. -Note: while you have unreviewed proposals, your agent pauses new -conversations — reviewing promptly keeps it active. +{review_how_text} Welcome aboard, The CoPI team — Scripps Research @@ -321,9 +348,7 @@ def build_welcome_email(to_email: str, name: str | None = None, user_id: str | N you a short proposal. You can:

    1 = Not a good idea • 2 = Good idea • 3 = Great idea • 4 = Excellent idea diff --git a/src/services/email_inbound.py b/src/services/email_inbound.py index a1c0fc6..77f8f90 100644 --- a/src/services/email_inbound.py +++ b/src/services/email_inbound.py @@ -25,6 +25,39 @@ # Rate limit: max replies per token per hour MAX_REPLIES_PER_TOKEN_PER_HOUR = 10 +# Processing attempts per S3 object before it is quarantined under failed/. +MAX_S3_PROCESS_ATTEMPTS = 3 + +# token -> recent reply timestamps (monotonic-ish epoch seconds). +_RECENT_REPLY_TIMES: dict[str, list[float]] = {} + +# s3 key -> consecutive processing failures (in-memory; resets on restart). +_S3_FAILURE_COUNTS: dict[str, int] = {} + + +def _reply_rate_ok(token: str, now: float | None = None) -> bool: + """Sliding one-hour window per reply token, capped at + MAX_REPLIES_PER_TOKEN_PER_HOUR. In-memory: the worker is a single + long-lived process, and a restart merely resets the window.""" + import time + + ts = time.time() if now is None else now + window = [t for t in _RECENT_REPLY_TIMES.get(token, []) if ts - t < 3600] + if len(window) >= MAX_REPLIES_PER_TOKEN_PER_HOUR: + _RECENT_REPLY_TIMES[token] = window + return False + window.append(ts) + _RECENT_REPLY_TIMES[token] = window + return True + + +def _is_auto_submitted(msg: email.message.Message) -> bool: + """RFC 3834: any Auto-Submitted value other than "no" marks auto-generated + mail (out-of-office replies, list expansions). Processing those — and + answering them with a help email — is how mail loops start.""" + auto = (msg.get("Auto-Submitted") or "").strip().lower() + return bool(auto) and auto != "no" and not auto.startswith("no ") + # Auth verdicts (from the SES-stamped Authentication-Results header) that mean # the message failed a check — any of these on spf/dkim/dmarc rejects the reply. # ("none" is intentionally excluded: it means the sender domain publishes no @@ -50,13 +83,26 @@ def _authentication_results_ok(msg: email.message.Message) -> bool: logger.warning("Rejecting inbound reply: no Authentication-Results header") return False + # Trust ONLY the topmost header. SES prepends its own Authentication- + # Results on receipt, so a sender-forged header always sits below it — + # merging verdicts across all headers ("a pass wins") let a self-stamped + # spf=pass override SES's spf=fail. The topmost header must also carry + # SES's authserv-id: anything else did not transit our SES receipt path. + header = headers[0] + authserv_id = header.split(";", 1)[0].strip().lower() + if authserv_id != "amazonses.com": + logger.warning( + "Rejecting inbound reply: topmost Authentication-Results is from %r, " + "not amazonses.com", + authserv_id, + ) + return False + verdicts: dict[str, str] = {} - for header in headers: - for mech, result in _AUTH_VERDICT_RE.findall(header): - mech_l, result_l = mech.lower(), result.lower() - # Keep the strongest verdict seen for each mechanism (a pass wins). - if mech_l not in verdicts or result_l == "pass": - verdicts[mech_l] = result_l + for mech, result in _AUTH_VERDICT_RE.findall(header): + # First occurrence wins: the leading verdict is the mechanism's result; + # later matches can come from propagated or commented values. + verdicts.setdefault(mech.lower(), result.lower()) for mech in ("spf", "dkim", "dmarc"): if verdicts.get(mech) in _AUTH_FAIL_VERDICTS: @@ -108,10 +154,33 @@ async def poll_inbound_emails(session_factory: async_sessionmaker) -> int: # Delete processed email from S3 s3.delete_object(Bucket=bucket, Key=key) + _S3_FAILURE_COUNTS.pop(key, None) processed += 1 except Exception as exc: logger.error("Error processing inbound email %s: %s", key, exc, exc_info=True) + # A poison message would otherwise be retried every poll + # forever. After MAX_S3_PROCESS_ATTEMPTS consecutive failures, + # quarantine it under failed/ (outside the polled prefix) for + # manual inspection. The counter is in-memory, so a restart + # grants a fresh round of attempts — acceptable. + _S3_FAILURE_COUNTS[key] = _S3_FAILURE_COUNTS.get(key, 0) + 1 + if _S3_FAILURE_COUNTS[key] >= MAX_S3_PROCESS_ATTEMPTS: + try: + failed_key = "failed/" + key.removeprefix(prefix) + s3.copy_object( + Bucket=bucket, + CopySource={"Bucket": bucket, "Key": key}, + Key=failed_key, + ) + s3.delete_object(Bucket=bucket, Key=key) + _S3_FAILURE_COUNTS.pop(key, None) + logger.error( + "Quarantined inbound email %s to %s after %d failed attempts", + key, failed_key, MAX_S3_PROCESS_ATTEMPTS, + ) + except Exception: + logger.error("Failed to quarantine %s", key, exc_info=True) except Exception as exc: logger.error("Error polling inbound emails: %s", exc, exc_info=True) @@ -130,6 +199,12 @@ async def process_inbound_email(raw_email: bytes, db: AsyncSession) -> None: if not _authentication_results_ok(msg): return + # Auto-generated mail (OOO replies, etc.) must never be answered — our + # help email replying to an auto-responder is a mail loop. + if _is_auto_submitted(msg): + logger.info("Ignoring auto-submitted inbound mail (Auto-Submitted header)") + return + # Extract reply token from To header to_addr = msg.get("To", "") token = _extract_reply_token(to_addr) @@ -137,6 +212,12 @@ async def process_inbound_email(raw_email: bytes, db: AsyncSession) -> None: logger.warning("No reply token found in To address: %s", to_addr) return + if not _reply_rate_ok(token): + logger.warning( + "Rate limit exceeded for reply token %s... — dropping reply", token[:8] + ) + return + # Look up notification by token result = await db.execute( select(EmailNotification).where(EmailNotification.reply_token == token) @@ -259,19 +340,50 @@ def _extract_email_address(from_header: str) -> str | None: return None +def _decode_part(part: email.message.Message) -> str: + charset = part.get_content_charset() or "utf-8" + payload = part.get_payload(decode=True) or b"" + return payload.decode(charset, errors="replace") + + +def _html_to_text(html_body: str) -> str: + """Best-effort text extraction for HTML-only replies. + + Quoted history is dropped structurally (

    /gmail_quote) because + the '>' line-prefix convention below only exists in plain text.""" + import html as html_mod + + text = re.sub(r"(?is)<(script|style)\b.*?", "", html_body) + text = re.sub(r'(?is)]*class="[^"]*gmail_quote[^"]*".*', "", text) + text = re.sub(r"(?is)", "", text) + text = re.sub(r"(?i)|

    |", "\n", text) + text = re.sub(r"(?s)<[^>]+>", "", text) + return html_mod.unescape(text) + + def _extract_reply_body(msg: email.message.Message) -> str: - """Extract the reply body, stripping quoted content and signatures.""" + """Extract the reply body, stripping quoted content and signatures. + + Prefers text/plain; falls back to tag-stripped text/html so an HTML-only + reply (some corporate clients) is not silently dropped.""" body = "" + html_body = "" if msg.is_multipart(): for part in msg.walk(): - if part.get_content_type() == "text/plain": - charset = part.get_content_charset() or "utf-8" - body = part.get_payload(decode=True).decode(charset, errors="replace") + ctype = part.get_content_type() + if ctype == "text/plain": + body = _decode_part(part) break + if ctype == "text/html" and not html_body: + html_body = _decode_part(part) + elif msg.get_content_type() == "text/html": + html_body = _decode_part(msg) else: - charset = msg.get_content_charset() or "utf-8" - body = msg.get_payload(decode=True).decode(charset, errors="replace") + body = _decode_part(msg) + + if not body.strip() and html_body: + body = _html_to_text(html_body) # Strip quoted content (lines starting with >) lines = body.split("\n") @@ -343,6 +455,11 @@ async def classify_reply(body: str, proposal_summary: str) -> dict: message = client.messages.create( model=settings.llm_agent_model_sonnet, max_tokens=500, + # Sonnet 5 thinks by default and max_tokens caps thinking + text + # together, so without this pin content[0] is a thinking block and + # the .text read below raises — every inbound reply would classify + # as a failure. 500 tokens leaves no room to share with reasoning. + thinking={"type": "disabled"}, system=system_prompt, messages=[{"role": "user", "content": user_message}], ) diff --git a/src/services/email_notifications.py b/src/services/email_notifications.py index 9e01e82..bdf26c6 100644 --- a/src/services/email_notifications.py +++ b/src/services/email_notifications.py @@ -331,8 +331,13 @@ async def send_proposal_notification( db.add(notification) await db.flush() - # Build email - reply_to = f"review+{reply_token}@{settings.ses_reply_domain}" + # Build email. Soliciting a reply is only honest when the inbound pipeline + # is actually on — otherwise PIs answer a dead reply domain and get + # silence (this is exactly what happened on prod through 2026-08). + reply_enabled = settings.enable_inbound_email + reply_to = ( + f"review+{reply_token}@{settings.ses_reply_domain}" if reply_enabled else None + ) dashboard_url = f"{settings.base_url}/agent/{agent.agent_id}/dashboard" unsubscribe_token = _generate_unsubscribe_token(str(user.id)) unsubscribe_url = f"{settings.base_url}/settings/unsubscribe/{unsubscribe_token}" @@ -373,25 +378,58 @@ async def send_proposal_notification( f"Review all proposals.

    " ) + if reply_enabled: + review_options_text = ( + f"To review this proposal, you can:\n\n" + f"1. Reply to this email with a rating (1-4) and any comments:\n" + f" 1 = Not a good idea (not interesting, or multiple major weaknesses)\n" + f" 2 = Good idea (medium interest, or one major weakness)\n" + f" 3 = Great idea (high interest, minor weaknesses only)\n" + f" 4 = Excellent idea (high interest, no notable weaknesses)\n\n" + f"2. Reply with instructions for your agent (e.g., \"focus on the\n" + f' mitochondrial angle instead") and it will re-engage to refine\n' + f" the proposal.\n\n" + f"3. Review on the web: {dashboard_url}\n" + ) + else: + review_options_text = ( + f"To review this proposal, rate it on your dashboard: {dashboard_url}\n" + f" 1 = Not a good idea (not interesting, or multiple major weaknesses)\n" + f" 2 = Good idea (medium interest, or one major weakness)\n" + f" 3 = Great idea (high interest, minor weaknesses only)\n" + f" 4 = Excellent idea (high interest, no notable weaknesses)\n" + ) + text_body = ( f"{agent.bot_name} and {other_bot_name} developed a collaboration proposal in #{channel}:\n\n" f"---\n{summary}\n---\n\n" - f"To review this proposal, you can:\n\n" - f"1. Reply to this email with a rating (1-4) and any comments:\n" - f" 1 = Not a good idea (not interesting, or multiple major weaknesses)\n" - f" 2 = Good idea (medium interest, or one major weakness)\n" - f" 3 = Great idea (high interest, minor weaknesses only)\n" - f" 4 = Excellent idea (high interest, no notable weaknesses)\n\n" - f"2. Reply with instructions for your agent (e.g., \"focus on the\n" - f' mitochondrial angle instead") and it will re-engage to refine\n' - f" the proposal.\n\n" - f"3. Review on the web: {dashboard_url}\n" + f"{review_options_text}" f"{backlog_text}\n" f"---\n" f"Unsubscribe: {unsubscribe_url}\n" f"Manage preferences: {settings_url}\n" ) + rating_legend_html = ( + '

    \n' + " 1 = Not a good idea • 2 = Good idea • 3 = Great idea • 4 = Excellent idea\n" + "

    " + ) + if reply_enabled: + review_options_html = ( + '

    Reply to this email to review:

    \n' + '
      \n' + "
    • Rate it with a number 1-4 and any comments
      • \n" + "
    • Give instructions to refine the proposal
      • \n" + "
    \n" + f" {rating_legend_html}" + ) + else: + review_options_html = ( + '

    Rate it on your dashboard:

    \n' + f" {rating_legend_html}" + ) + html_body = email_shell_open() + f"""

    New collaboration proposal

    @@ -402,14 +440,7 @@ async def send_proposal_notification(

    {summary_html}

    -

    Reply to this email to review:

    -
      -
    • Rate it with a number 1-4 and any comments
      • -
    • Give instructions to refine the proposal
      • -
    -

    - 1 = Not a good idea • 2 = Good idea • 3 = Great idea • 4 = Excellent idea -

    + {review_options_html}
    " raw_msg["List-Unsubscribe-Post"] = "List-Unsubscribe=One-Click" @@ -967,7 +999,12 @@ async def _send_new_proposal_email( summary = td.summary_text or "(No summary available)" channel = td.channel or "unknown" - reply_to = f"review+{reply_token}@{settings.ses_reply_domain}" + # Same gating as send_proposal_notification: never solicit a reply while + # the inbound pipeline is off. + reply_enabled = settings.enable_inbound_email + reply_to = ( + f"review+{reply_token}@{settings.ses_reply_domain}" if reply_enabled else None + ) dashboard_url = f"{settings.base_url}/agent/{agent.agent_id}/dashboard" unsubscribe_token = _generate_unsubscribe_token(str(user.id)) unsubscribe_url = f"{settings.base_url}/settings/unsubscribe/{unsubscribe_token}" @@ -982,11 +1019,28 @@ async def _send_new_proposal_email( subject = f"{clean_subject(agent.bot_name)} proposed a collaboration with {clean_subject(other_bot_name)}" + if reply_enabled: + review_line = ( + f"Reply to this email to rate it (1-4) or give your agent instructions, " + f"or review on the web: {dashboard_url}" + ) + review_html = ( + '

    \n' + " Reply to this email to rate it (1–4) or give instructions.\n" + "

    " + ) + else: + review_line = f"Rate it (1-4) on the web: {dashboard_url}" + review_html = ( + '

    \n' + " Rate it (1–4) or give instructions on your dashboard.\n" + "

    " + ) + text_body = ( f"{agent.bot_name} just proposed a collaboration with {other_bot_name} in #{channel}:\n\n" f"---\n{summary}\n---\n\n" - f"Reply to this email to rate it (1-4) or give your agent instructions, " - f"or review on the web: {dashboard_url}\n\n" + f"{review_line}\n\n" f"---\n" f"Unsubscribe: {unsubscribe_url}\n" f"Manage preferences: {settings_url}\n" @@ -999,9 +1053,7 @@ async def _send_new_proposal_email(

    {summary_html}

    -

    - Reply to this email to rate it (1–4) or give instructions. -

    + {review_html}     diff --git a/tests/unit/test_backfill_publications.py b/tests/unit/test_backfill_publications.py new file mode 100644 index 0000000..af0b980 --- /dev/null +++ b/tests/unit/test_backfill_publications.py @@ -0,0 +1,125 @@ +"""Curated publications backfill for labs whose ORCID works are empty. + +Eleven active labs have zero publications rows because the only ingest path +(profile pipeline: ORCID works -> PMID) found nothing for them, which mutes +them under the issue-29 fail-closed authorship guard. The backfill takes a +curated agent_id -> PMID mapping, fetches the PubMed records, and inserts +Publication rows; the running simulation picks them up on the next ~30s +roster sync with no restart. +""" + +import uuid + +import pytest +from sqlalchemy import func, select + +from scripts.backfill_publications import backfill +from src.models import AgentRegistry, Publication, User + +pytestmark = pytest.mark.integration + + +async def _seed_agent(db, agent_id: str) -> User: + user = User(id=uuid.uuid4(), name=f"PI {agent_id}", orcid=f"0000-{agent_id}") + db.add(user) + await db.flush() + db.add( + AgentRegistry( + agent_id=agent_id, + bot_name=f"{agent_id.title()}Bot", + pi_name=f"PI {agent_id}", + status="active", + user_id=user.id, + ) + ) + await db.flush() + return user + + +def _fake_fetch(records: list[dict]): + """Async fetch stub that records which PMIDs were requested.""" + requested: list[list[str]] = [] + + async def fetch(pmids: list[str]) -> list[dict]: + requested.append(list(pmids)) + return [r for r in records if r["pmid"] in pmids] + + fetch.requested = requested # type: ignore[attr-defined] + return fetch + + +async def _count_pubs(db, user_id) -> int: + return ( + await db.execute( + select(func.count()).select_from(Publication).where( + Publication.user_id == user_id + ) + ) + ).scalar() + + +REC = { + "pmid": "111", + "title": "A real paper", + "abstract": "An abstract.", + "journal": "J. Test", + "year": 2024, + "doi": "doi:10.1000/xyz.1", # prefixed on purpose: must land canonicalized +} + + +async def test_dry_run_reports_but_writes_nothing(db_session): + user = await _seed_agent(db_session, "good") + fetch = _fake_fetch([REC]) + + report = await backfill(db_session, {"good": ["111"]}, fetch=fetch, apply=False) + + assert ("good", "would-insert", "111") in report + assert await _count_pubs(db_session, user.id) == 0 + + +async def test_apply_inserts_row_with_canonical_doi(db_session): + user = await _seed_agent(db_session, "good") + fetch = _fake_fetch([REC]) + + report = await backfill(db_session, {"good": ["111"]}, fetch=fetch, apply=True) + + assert ("good", "insert", "111") in report + row = ( + await db_session.execute( + select(Publication).where(Publication.user_id == user.id) + ) + ).scalar_one() + assert row.pmid == "111" + assert row.doi == "10.1000/xyz.1" + assert row.title == "A real paper" + + +async def test_pmids_the_user_already_has_are_skipped_and_not_fetched(db_session): + user = await _seed_agent(db_session, "good") + db_session.add(Publication(user_id=user.id, pmid="111", title="Already here")) + await db_session.flush() + fetch = _fake_fetch([REC]) + + report = await backfill(db_session, {"good": ["111"]}, fetch=fetch, apply=True) + + assert ("good", "skip-existing", "111") in report + assert await _count_pubs(db_session, user.id) == 1 + assert fetch.requested in ([], [[]]) # nothing left to fetch + + +async def test_unknown_agent_is_reported_not_fatal(db_session): + fetch = _fake_fetch([]) + + report = await backfill(db_session, {"ghost": ["111"]}, fetch=fetch, apply=True) + + assert ("ghost", "error-no-agent", "") in report + + +async def test_pmid_with_no_pubmed_record_is_reported(db_session): + await _seed_agent(db_session, "good") + fetch = _fake_fetch([]) # PubMed returns nothing for the pmid + + report = await backfill(db_session, {"good": ["999"]}, fetch=fetch, apply=True) + + assert ("good", "error-no-record", "999") in report diff --git a/tests/unit/test_email_inbound_hardening.py b/tests/unit/test_email_inbound_hardening.py new file mode 100644 index 0000000..273aef4 --- /dev/null +++ b/tests/unit/test_email_inbound_hardening.py @@ -0,0 +1,255 @@ +"""Hardening for inbound email reply processing. + +These pin the defects found while investigating the dead prod reply flow +(2026-08-11): a sender-forged ``Authentication-Results: ... pass`` header +defeated the SEC-5 anti-spoofing gate, HTML-only replies were silently +dropped, auto-responders could loop with the help email, the declared +per-token rate limit was never enforced, and a poison message in the inbound +bucket was retried forever. +""" + +import email + +import pytest + +import src.services.email_inbound as inbound +from src.services.email_inbound import ( + MAX_REPLIES_PER_TOKEN_PER_HOUR, + _authentication_results_ok, + _extract_reply_body, + _reply_rate_ok, + poll_inbound_emails, + process_inbound_email, +) + + +def _msg(raw: str) -> email.message.Message: + return email.message_from_string(raw) + + +# --- Authentication-Results: only SES's own (topmost) header is trusted ----- + + +def test_forged_pass_header_below_ses_fail_is_rejected(): + """SES prepends its header on receipt, so a sender-supplied pass sits below + it. Merging verdicts across headers let the forged pass win (SEC-5).""" + raw = ( + "Authentication-Results: amazonses.com; spf=fail smtp.mailfrom=evil.com; " + "dkim=none; dmarc=fail header.from=scripps.edu\n" + "Authentication-Results: amazonses.com; spf=pass; dkim=pass; dmarc=pass\n" + "From: pi@scripps.edu\n\nbody" + ) + assert _authentication_results_ok(_msg(raw)) is False + + +def test_verdicts_below_the_topmost_header_are_ignored_entirely(): + raw = ( + "Authentication-Results: amazonses.com; spf=pass smtp.mailfrom=scripps.edu; " + "dkim=pass; dmarc=pass header.from=scripps.edu\n" + "Authentication-Results: evil.example; spf=fail; dkim=fail; dmarc=fail\n" + "From: pi@scripps.edu\n\nbody" + ) + assert _authentication_results_ok(_msg(raw)) is True + + +def test_topmost_header_with_foreign_authserv_id_is_rejected(): + """Everything on our receipt path is stamped by amazonses.com; anything + else means the message did not transit SES receiving.""" + raw = ( + "Authentication-Results: mx.evil.example; spf=pass; dkim=pass; dmarc=pass\n" + "From: pi@scripps.edu\n\nbody" + ) + assert _authentication_results_ok(_msg(raw)) is False + + +# --- HTML-only replies are not silently dropped ------------------------------ + + +def test_html_only_reply_body_falls_back_to_stripped_html(): + raw = ( + "From: pi@scripps.edu\n" + "MIME-Version: 1.0\n" + 'Content-Type: multipart/alternative; boundary="xyz"\n' + "\n" + "--xyz\n" + 'Content-Type: text/html; charset="UTF-8"\n' + "\n" + "
    4 — excellent, go ahead!
    \n" + '
    quoted proposal text ' + "1 = Not a good idea
    \n" + "\n" + "--xyz--\n" + ) + body = _extract_reply_body(_msg(raw)) + assert "4" in body and "excellent" in body + assert "Not a good idea" not in body # quoted HTML must not leak through + + +def test_singlepart_html_reply_body_is_extracted(): + raw = ( + "From: pi@scripps.edu\n" + 'Content-Type: text/html; charset="UTF-8"\n' + "\n" + "

    2 & please focus on assay development

    \n" + ) + body = _extract_reply_body(_msg(raw)) + assert "2 & please focus on assay development" in body + + +def test_plain_text_part_still_wins_over_html(): + raw = ( + "From: pi@scripps.edu\n" + "MIME-Version: 1.0\n" + 'Content-Type: multipart/alternative; boundary="qq"\n' + "\n" + "--qq\n" + 'Content-Type: text/plain; charset="UTF-8"\n' + "\n" + "3 sounds great\n" + "\n" + "--qq\n" + 'Content-Type: text/html; charset="UTF-8"\n' + "\n" + "
    3 sounds great
    \n" + "\n" + "--qq--\n" + ) + assert _extract_reply_body(_msg(raw)) == "3 sounds great" + + +# --- Auto-submitted mail is dropped before any processing -------------------- + + +_SES_PASS = "Authentication-Results: amazonses.com; spf=pass; dkim=pass; dmarc=pass\n" + + +async def test_auto_submitted_reply_is_ignored_before_touching_the_db(): + """RFC 3834: an OOO auto-reply answering our help email must not trigger + another help email (mail loop). db=None proves the early return.""" + raw = ( + _SES_PASS + + "Auto-Submitted: auto-replied\n" + "From: pi@scripps.edu\n" + "To: review+sometoken@reply.copi.science\n" + "\n" + "I am out of the office.\n" + ).encode() + await process_inbound_email(raw, db=None) # must not raise + + +async def test_auto_submitted_no_is_not_treated_as_an_auto_reply(): + """``Auto-Submitted: no`` explicitly marks human-generated mail; it must + proceed into normal processing (here: to the token lookup, which needs a + db — the AttributeError on db=None is the evidence it got past the gate).""" + raw = ( + _SES_PASS + + "Auto-Submitted: no\n" + "From: pi@scripps.edu\n" + "To: review+sometoken@reply.copi.science\n" + "\n" + "3 great idea\n" + ).encode() + with pytest.raises(AttributeError): + await process_inbound_email(raw, db=None) + + +# --- The declared per-token rate limit is enforced --------------------------- + + +def test_reply_rate_limit_blocks_the_11th_reply_in_an_hour(monkeypatch): + monkeypatch.setattr(inbound, "_RECENT_REPLY_TIMES", {}) + token = "tok-" + "x" * 60 + base = 1_000_000.0 + for i in range(MAX_REPLIES_PER_TOKEN_PER_HOUR): + assert _reply_rate_ok(token, now=base + i) is True + assert _reply_rate_ok(token, now=base + 60) is False + + +def test_reply_rate_limit_window_slides(monkeypatch): + monkeypatch.setattr(inbound, "_RECENT_REPLY_TIMES", {}) + token = "tok-" + "y" * 60 + base = 2_000_000.0 + for i in range(MAX_REPLIES_PER_TOKEN_PER_HOUR): + assert _reply_rate_ok(token, now=base + i) is True + # An hour later the old entries have aged out. + assert _reply_rate_ok(token, now=base + 3601) is True + + +# --- Poison messages are quarantined, not retried forever -------------------- + + +class _FakeS3: + """Just enough of the S3 client for poll_inbound_emails.""" + + def __init__(self, keys): + self.objects = {k: b"raw email bytes" for k in keys} + self.copied: list[tuple[str, str]] = [] + self.deleted: list[str] = [] + + def list_objects_v2(self, Bucket, Prefix, MaxKeys): + return { + "Contents": [{"Key": k} for k in sorted(self.objects)], + "KeyCount": len(self.objects), + } + + def get_object(self, Bucket, Key): + import io + + return {"Body": io.BytesIO(self.objects[Key])} + + def copy_object(self, Bucket, CopySource, Key): + self.copied.append((CopySource["Key"], Key)) + self.objects[Key] = self.objects[CopySource["Key"]] + + def delete_object(self, Bucket, Key): + self.deleted.append(Key) + self.objects.pop(Key, None) + + +class _NullSessionFactory: + def __call__(self): + return self + + async def __aenter__(self): + return self + + async def __aexit__(self, *exc): + return False + + async def commit(self): + pass + + +async def test_poison_email_is_quarantined_after_repeated_failures(monkeypatch): + fake = _FakeS3(["inbound/poison"]) + monkeypatch.setattr("boto3.client", lambda *a, **k: fake) + monkeypatch.setattr(inbound, "_S3_FAILURE_COUNTS", {}) + + async def _boom(raw, db): + raise RuntimeError("unparseable in a way that always raises") + + monkeypatch.setattr(inbound, "process_inbound_email", _boom) + + for _ in range(inbound.MAX_S3_PROCESS_ATTEMPTS): + assert await poll_inbound_emails(_NullSessionFactory()) == 0 + + assert fake.copied == [("inbound/poison", "failed/poison")] + assert fake.deleted == ["inbound/poison"] + # Quarantined: the next poll sees only failed/ (outside the prefix filter + # in real S3; the fake returns everything, so assert the key is gone). + assert "inbound/poison" not in fake.objects + + +async def test_a_transient_failure_is_retried_not_quarantined(monkeypatch): + fake = _FakeS3(["inbound/flaky"]) + monkeypatch.setattr("boto3.client", lambda *a, **k: fake) + monkeypatch.setattr(inbound, "_S3_FAILURE_COUNTS", {}) + + async def _boom(raw, db): + raise RuntimeError("db briefly down") + + monkeypatch.setattr(inbound, "process_inbound_email", _boom) + await poll_inbound_emails(_NullSessionFactory()) + + assert fake.copied == [] + assert "inbound/flaky" in fake.objects # still there for the next poll diff --git a/tests/unit/test_email_inbound_llm_pin.py b/tests/unit/test_email_inbound_llm_pin.py new file mode 100644 index 0000000..1d244d3 --- /dev/null +++ b/tests/unit/test_email_inbound_llm_pin.py @@ -0,0 +1,50 @@ +"""classify_reply must pin thinking off on its direct messages.create call. + +Sonnet 5 thinks by default and max_tokens caps thinking + text together, so +without an explicit thinking={"type": "disabled"} the first content block is +a thinking block: the .text read raises and EVERY inbound reply classifies as +"unparseable". Pinned on prod as hotfix 0e2ed84; this test keeps a refactor of +classify_reply from silently dropping the pin again. +""" + +import json + +from src.services.email_inbound import classify_reply + + +class _FakeMessages: + def __init__(self): + self.calls: list[dict] = [] + + def create(self, **kwargs): + self.calls.append(kwargs) + + class _Block: + text = json.dumps( + {"category": "review", "rating": 3, "comment": "", "instruction": ""} + ) + + class _Msg: + content = [_Block()] + + return _Msg() + + +class _FakeClient: + def __init__(self): + self.messages = _FakeMessages() + + +async def test_classify_reply_pins_thinking_disabled(monkeypatch): + import src.services.llm as llm + + fake = _FakeClient() + monkeypatch.setattr(llm, "get_anthropic_client", lambda: fake) + + result = await classify_reply("3 — looks great", "proposal summary") + + # The call succeeded through the fake, so the classification came through… + assert result["category"] == "review" + # …and the call itself must carry the pin. + assert len(fake.messages.calls) == 1 + assert fake.messages.calls[0].get("thinking") == {"type": "disabled"} diff --git a/tests/unit/test_email_reply_solicitation.py b/tests/unit/test_email_reply_solicitation.py new file mode 100644 index 0000000..9b3022c --- /dev/null +++ b/tests/unit/test_email_reply_solicitation.py @@ -0,0 +1,161 @@ +"""Reply-soliciting email copy must be gated on inbound email being enabled. + +Prod (2026-08-11): review emails told PIs "reply to this email to rate it" +while ENABLE_INBOUND_EMAIL was off and the reply pipeline (MX record, S3 +bucket, receipt rule) did not exist — every PI who replied got silence plus a +bounce. Until inbound is provisioned AND enabled, outbound mail must direct +PIs to the web dashboard only, and must not carry a Reply-To pointing at the +dead reply domain. +""" + +import email +import uuid +from types import SimpleNamespace + +import pytest + +from src.config import get_settings +from src.services.email import build_welcome_email +from src.services.email_notifications import ( + _send_new_proposal_email, + send_proposal_notification, +) + + +class _SESRecorder: + def __init__(self): + self.raw_messages: list[str] = [] + + def send_raw_email(self, **kwargs): + self.raw_messages.append(kwargs["RawMessage"]["Data"]) + return {"MessageId": "m-1"} + + +class _FakeDb: + def __init__(self): + self.added = [] + + def add(self, obj): + self.added.append(obj) + + async def flush(self): + pass + + +@pytest.fixture +def ses(monkeypatch): + recorder = _SESRecorder() + monkeypatch.setattr("boto3.client", lambda *a, **k: recorder) + monkeypatch.setattr(get_settings(), "outbound_email_allowlist", "") + return recorder + + +def _lab(): + user = SimpleNamespace(id=uuid.uuid4(), email="pi@lab.test", name="Ada Alpha") + agent = SimpleNamespace(id=uuid.uuid4(), agent_id="alpha", bot_name="AlphaBot") + td = SimpleNamespace( + id=uuid.uuid4(), summary_text="A joint proposal.", channel="degrader-chem" + ) + return user, agent, td + + +def _parts(raw: str) -> tuple[email.message.Message, str, str]: + msg = email.message_from_string(raw) + text = html = "" + for part in msg.walk(): + if part.get_content_type() == "text/plain": + text = part.get_payload(decode=True).decode("utf-8") + elif part.get_content_type() == "text/html": + html = part.get_payload(decode=True).decode("utf-8") + return msg, text, html + + +# --- proposal_review reminder ------------------------------------------------ + + +async def test_review_reminder_is_web_only_while_inbound_is_disabled( + ses, monkeypatch +): + monkeypatch.setattr(get_settings(), "enable_inbound_email", False) + user, agent, td = _lab() + + ok = await send_proposal_notification( + user=user, thread_decision=td, agent=agent, + other_bot_name="BetaBot", total_unreviewed=1, db=_FakeDb(), + ) + + assert ok is True + msg, text, html = _parts(ses.raw_messages[0]) + assert msg["Reply-To"] is None + assert "Reply to this email" not in text + assert "Reply to this email" not in html + assert "/agent/alpha/dashboard" in text # the web path remains + + +async def test_review_reminder_solicits_replies_when_inbound_is_enabled( + ses, monkeypatch +): + monkeypatch.setattr(get_settings(), "enable_inbound_email", True) + user, agent, td = _lab() + + await send_proposal_notification( + user=user, thread_decision=td, agent=agent, + other_bot_name="BetaBot", total_unreviewed=1, db=_FakeDb(), + ) + + msg, text, html = _parts(ses.raw_messages[0]) + assert msg["Reply-To"].startswith("review+") + assert msg["Reply-To"].endswith(f"@{get_settings().ses_reply_domain}") + assert "Reply to this email" in text + + +# --- new_proposal alert -------------------------------------------------------- + + +async def test_new_proposal_alert_is_web_only_while_inbound_is_disabled( + ses, monkeypatch +): + monkeypatch.setattr(get_settings(), "enable_inbound_email", False) + user, agent, td = _lab() + + ok = await _send_new_proposal_email(user, td, agent, "BetaBot", _FakeDb()) + + assert ok is True + msg, text, html = _parts(ses.raw_messages[0]) + assert msg["Reply-To"] is None + assert "Reply to this email" not in text + assert "Reply to this email" not in html + assert "/agent/alpha/dashboard" in text + + +async def test_new_proposal_alert_solicits_replies_when_inbound_is_enabled( + ses, monkeypatch +): + monkeypatch.setattr(get_settings(), "enable_inbound_email", True) + user, agent, td = _lab() + + await _send_new_proposal_email(user, td, agent, "BetaBot", _FakeDb()) + + msg, text, _ = _parts(ses.raw_messages[0]) + assert msg["Reply-To"].startswith("review+") + assert "Reply to this email" in text + + +# --- welcome email ------------------------------------------------------------- + + +def test_welcome_email_omits_reply_instructions_while_inbound_is_disabled( + monkeypatch, +): + monkeypatch.setattr(get_settings(), "enable_inbound_email", False) + _, msg = build_welcome_email("pi@lab.test", "Ada") + _, text, html = _parts(msg.as_string()) + assert "Reply with a rating" not in text + assert "review" in text.lower() # web review guidance remains + + +def test_welcome_email_describes_replying_when_inbound_is_enabled(monkeypatch): + monkeypatch.setattr(get_settings(), "enable_inbound_email", True) + _, msg = build_welcome_email("pi@lab.test", "Ada") + _, text, _ = _parts(msg.as_string()) + assert "Reply with a rating" in text