Deploy follow-ups: classify-call thinking pin + regression test, inbound runbook fix, publications backfill

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/<messageId>
+                                  └─► 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+<token>@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).

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()))