feat(rsvp-fold): own the fold sequence in a shared foldRsvp() helper

The library exported the predicates for folding an attendee's response onto
its event's thread but stopped short of the sequence around them, so each mail
connector restated the same ~40 lines: build the marker key, read it, check
for a repeat, decide whether the response earns a note, compose it, save it,
record it. The order is load-bearing and none of it was enforced — checking
the repeat second re-emits a re-delivered commented acceptance, and writing
the marker on the suppressed path leaves it describing something the event
thread does not carry. Both mistakes surface the same way: a thread people
had already read goes unread again.

`foldRsvp()` owns that order. Genuinely per-connector concerns stay injected,
because they differ for real reasons: reading and writing the marker (write
through immediately, or collect and flush once per pass to stay inside the
request budget), saving the note, and the note's key, timestamp and unread
flag. Every outcome — emitted, suppressed, already folded — means the response
was dealt with, so each caller records "this message was folded" unconditionally
after the call, in whatever form is its own: an in-pass set, durable per-thread
metadata, or both. The returned outcome names which path ran.

Gmail, Outlook and iCloud mail now call it. No behaviour change: their test
suites pass unchanged.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01GcnBUEdW86ovpv1pT2d3b3

Author: KrisBraun

connectors/apple/src/mail/sync.ts

@@ -1,12 +1,4 @@
-import {
-  alreadyFolded,
-  composeRsvpNote,
-  icsProp,
-  isNonAcceptance,
-  parseIcsReply,
-  priorRsvpKey,
-  shouldEmitRsvpNote,
-} from "@plotday/rsvp-fold";
+import { foldRsvp, icsProp, parseIcsReply } from "@plotday/rsvp-fold";
 import type { ActorId } from "@plotday/twister";
 import type { ImapMailboxStatus, ImapSession } from "@plotday/twister/tools/imap";
 
@@ -90,10 +82,11 @@ export type ThreadMeta = {
    * must never flip: this is per-MESSAGE, and exists so a root whose bundling
    * question is already settled still looks at messages that arrived since.
    *
-   * Required, not an optimisation. `alreadyFolded` stops a duplicate note but
-   * not the IMAP fetch, and inside the 30-day rescan window an RSVP would
-   * otherwise be re-fetched on every pass — roughly 2,900 times, at up to two
-   * round-trips each, against a ~1,000-request execution budget.
+   * Required, not an optimisation. `foldRsvp`'s repeat check stops a
+   * duplicate note but not the IMAP fetch, and inside the 30-day rescan
+   * window an RSVP would otherwise be re-fetched on every pass — roughly
+   * 2,900 times, at up to two round-trips each, against a ~1,000-request
+   * execution budget.
    *
    * Capped at SEEN_ICS_MAX, oldest dropped. Growth is bounded in practice by
    * how many responses one meeting draws, but this document is rewritten every
@@ -121,9 +114,10 @@ export type ThreadMeta = {
    * key can only be dropped together with its `seenIcs` entry, which puts the
    * message back among the unexamined and has the next pass re-fetch and
    * re-fold it. (The response's own `rsvp:` marker is a separate, uncapped
-   * key, so `alreadyFolded` still suppresses a second note.) An independent
-   * cap would have no such backstop — it would drop a folded key while the
-   * message stayed un-fetched, and the note would resurface permanently.
+   * key, so `foldRsvp` still recognises the repeat and writes no second
+   * note.) An independent cap would have no such backstop — it would drop a
+   * folded key while the message stayed un-fetched, and the note would
+   * resurface permanently.
    */
   foldedIcs?: string[];
 };
@@ -423,9 +417,11 @@ export async function detectCalendarBundles(
         // `dedupeCopies` only collapses them later, inside `transformMessages` —
         // `unexamined` was filtered against a snapshot of `seen`, so both copies
         // are in it. The second copy has nothing new to read, and routing it
-        // again would emit its response note twice: the marker written below is
-        // flushed once at the end of the pass, so the second copy would still
-        // read the pre-pass value and look un-folded.
+        // again would emit its response note twice: the marker is collected
+        // rather than written through (see `writeMarker` below) and flushed
+        // once at the end of the pass, so the second copy would still read the
+        // pre-pass value and look un-folded. This guard is the in-pass dedup
+        // `foldRsvp` requires of any connector that batches its markers.
         if (seen.has(noteKeyOf(m))) continue;
 
         const part = (m.attachments ?? []).find((a) => isCalendarAttachment(a.mimeType))!;
@@ -449,50 +445,31 @@ export async function detectCalendarBundles(
           // reply is not an answer to the bundling question.
           const replyUid = icsProp(ics, "UID");
           if (replyUid) {
-            const priorKey = priorRsvpKey(replyUid, reply.attendeeEmail, reply.occurrence);
-            const stored = await host.get<string>(priorKey);
-
-            // Order is the library's documented contract: `alreadyFolded`
-            // FIRST, and only when it is false decide whether to emit.
-            // Re-emitting a note the thread already carries re-applies its
-            // unread intent and drags the thread back to unread for everyone
-            // who had read it — and a response inside the 30-day rescan window
-            // is re-read on every pass.
-            if (
-              !alreadyFolded(stored, reply) &&
-              shouldEmitRsvpNote(reply, isNonAcceptance(stored))
-            ) {
-              // `saveNote` returns null when no thread carries `icaluid:<uid>`
-              // yet (the calendar event has not synced); `deferUntilThread` has
-              // the platform hold the note and attach it once that thread
-              // appears.
-              await host.integrations.saveNote({
-                thread: { source: `icaluid:${replyUid}` },
+            // The decision order (repeat check first, marker written only on
+            // the emitting path) and the reasoning behind it live once, on
+            // `foldRsvp`. Only what is iCloud-specific stays here — including
+            // the batched marker write below, which this connector needs
+            // because a response inside the 30-day rescan window would
+            // otherwise cost a store request per response per pass.
+            await foldRsvp({
+              uid: replyUid,
+              reply,
+              // `created` may be absent here — `foldRsvp` leaves the field
+              // off the note rather than sending an explicit undefined.
+              note: {
                 key: noteKeyOf(m),
-                content: composeRsvpNote(reply),
-                contentType: "markdown",
-                ...(m.date ? { created: m.date } : {}),
-                author: {
-                  email: reply.attendeeEmail,
-                  ...(reply.attendeeName ? { name: reply.attendeeName } : {}),
-                },
-                // Explicit on both paths. An omitted flag does NOT mean "leave
-                // read state alone" — attaching a note already marks the thread
-                // unread for every recipient except its author, so only an
-                // explicit false overrides it.
+                created: m.date,
                 unread: !initialRoots.has(root),
-                deferUntilThread: true,
-              });
-              // Recorded ONLY on the path that emits, and regardless of the
-              // return value: a deferred note returns no id, and gating on it
-              // would leave a deferred non-acceptance unrecorded forever,
-              // wrongly treating a later bare acceptance as reversing nothing.
-              // The marker holds the last response actually folded onto the
-              // thread — for every emitted response, acceptances included,
-              // which is what lets `alreadyFolded` recognise a repeat of ANY
-              // partstat.
-              rsvpMarkers.push([priorKey, reply.partstat]);
-            }
+              },
+              readMarker: (key) => host.get<string>(key),
+              // Collected, not written: flushed in ONE `setMany` in the
+              // `finally` below, so a `saveNote` that throws later in the pass
+              // cannot take the markers of responses already written with it.
+              writeMarker: (key, partstat) => {
+                rsvpMarkers.push([key, partstat]);
+              },
+              saveNote: (note) => host.integrations.saveNote(note),
+            });
 
             // Folded whether or not a note was written — a bare acceptance is
             // dropped from the mail thread rather than left to become an email

connectors/google/src/mail/sync.ts

@@ -17,13 +17,7 @@
  * returns a descriptor and lets the caller own the scheduling.
  */
 import { enrichLinkContactsFromGoogle } from "@plotday/google-contacts";
-import {
-  alreadyFolded,
-  composeRsvpNote,
-  isNonAcceptance,
-  priorRsvpKey,
-  shouldEmitRsvpNote,
-} from "@plotday/rsvp-fold";
+import { foldRsvp } from "@plotday/rsvp-fold";
 import {
   baseEmail,
   canonicalizeEmail,
@@ -1562,65 +1556,30 @@ async function saveTransformedThread(
     const replies = extractCalendarReplies(thread.messages ?? [], icsByMessage);
     if (replies.length > 0) {
       for (const reply of replies) {
-        const priorKey = priorRsvpKey(reply.uid, reply.attendeeEmail, reply.occurrence);
-        // Read on every response, not just a bare acceptance: `alreadyFolded`
-        // needs the stored value on every path, so there is no cheaper way to
-        // skip this round-trip anymore (there used to be one for the
-        // non-acceptance/commented-acceptance cases — traded away below).
-        const stored = await host.get<string>(priorKey);
-
-        // Re-processing a conversation re-runs this loop for a response
-        // already folded onto the event thread — Gmail history replay, a
-        // backfill overlap, at-least-once delivery. The note itself upserts
-        // by key, so re-saving it wouldn't duplicate it, but its `unread`
-        // intent would still be re-applied and drag the thread back to
-        // unread for anyone who already read it. Comparing against the
-        // stored partstat (not just presence) means a genuine change of
-        // response is never caught by this: an attendee who edits only their
-        // comment on an unchanged response gets no updated note, which is
-        // the accepted trade for not re-raising unread on every re-deliver.
-        if (alreadyFolded(stored, reply)) {
-          foldedMessageIds.add(reply.messageId);
-          continue;
-        }
-
-        // A bare acceptance says nothing the event's guest list does not
-        // already show. Drop the message rather than writing a note: a note
-        // is the only thing that could mark the organiser's thread unread,
-        // and marking it folded here keeps the responses-only conversation
-        // from becoming an email thread of its own.
-        if (!shouldEmitRsvpNote(reply, isNonAcceptance(stored))) {
-          foldedMessageIds.add(reply.messageId);
-          continue;
-        }
-
-        // saveNote returns null when no thread carries `icaluid:<uid>` yet —
-        // the calendar event has not synced. deferUntilThread has the
-        // platform hold the note and attach it once that thread appears,
-        // instead of us leaving it in the email thread as a fallback. The
-        // return value is not consulted: null now covers both "parked" and,
-        // in principle, "genuinely rejected", and deferUntilThread means we
-        // no longer need to tell those apart.
-        await host.tools.integrations.saveNote({
-          thread: { source: `icaluid:${reply.uid}` },
-          key: reply.messageId,
-          content: composeRsvpNote(reply),
-          contentType: "markdown",
-          created: reply.sourceCreatedAt,
-          author: {
-            email: reply.attendeeEmail,
-            ...(reply.attendeeName ? { name: reply.attendeeName } : {}),
+        // The decision order (repeat check first, marker written only on the
+        // emitting path) and the reasoning behind it live once, on
+        // `foldRsvp`. Only what is Gmail-specific stays here.
+        await foldRsvp({
+          uid: reply.uid,
+          reply,
+          note: {
+            key: reply.messageId,
+            created: reply.sourceCreatedAt,
+            unread: !initialSync,
           },
-          // Explicit on both paths. An omitted flag does NOT mean "leave read
-          // state alone": attaching a note already surfaces the thread as
-          // unread for every recipient except its author, so only an explicit
-          // false overrides it.
-          unread: !initialSync,
-          deferUntilThread: true,
+          readMarker: (key) => host.get<string>(key),
+          writeMarker: (key, partstat) => host.set(key, partstat),
+          saveNote: (note) => host.tools.integrations.saveNote(note),
         });
-        // Folded either way. A miss is now held by the platform rather than
-        // returned to us, so a responses-only conversation never creates an
-        // email thread even when the event is missing — the note is
+
+        // Folded on every outcome — emitted, suppressed, or already folded.
+        // A response never stays in the email thread: dropping it is what
+        // keeps a responses-only conversation from becoming an email thread
+        // of its own, and a suppressed bare acceptance is folded precisely
+        // because writing no note is the right answer for it.
+        //
+        // A note whose event has not synced is held by the platform rather
+        // than returned to us, so this holds even on a miss — the note is
         // guaranteed to land eventually, but only if the sweep that attaches
         // parked notes keeps working. A regression there loses the response
         // silently instead of leaving it visibly stranded in the inbox.
@@ -1637,20 +1596,6 @@ async function saveTransformedThread(
         // that email thread for responses-only conversations even at the
         // cost of losing the response outright for calendar-less recipients.
         foldedMessageIds.add(reply.messageId);
-        // Recorded regardless of `noteId`: the decision to emit is what this
-        // bookkeeping tracks, and with deferUntilThread the platform now
-        // guarantees the note lands eventually even on a miss. Previously
-        // this only ran on an immediate attach, with the drain's own retry
-        // recording it later on success — with the drain gone, gating on
-        // `noteId` here would leave a deferred non-acceptance unrecorded
-        // forever, wrongly treating a later bare acceptance as reversing
-        // nothing.
-        //
-        // Always set, never cleared: the key now holds the last response
-        // actually folded, for every emitted response (including an
-        // acceptance) — that's what lets `alreadyFolded` recognise a repeat
-        // of ANY partstat, not just an outstanding non-acceptance.
-        await host.set(priorKey, reply.partstat);
       }
 
       if (foldedMessageIds.size > 0) {

connectors/outlook/src/mail/sync.ts

@@ -37,13 +37,7 @@ import type {
 } from "@plotday/twister/plot";
 import type { WebhookRequest } from "@plotday/twister/tools/network";
 import { markdownToHtml } from "@plotday/twister/utils/markdown-html";
-import {
-  alreadyFolded,
-  composeRsvpNote,
-  isNonAcceptance,
-  priorRsvpKey,
-  shouldEmitRsvpNote,
-} from "@plotday/rsvp-fold";
+import { foldRsvp } from "@plotday/rsvp-fold";
 
 import { enrichLinkContactsFromOutlook } from "./enrich";
 import {
@@ -1550,70 +1544,23 @@ export async function processConversationsFn(
         if (!reply) continue;
 
         const noteKey = m.internetMessageId ?? m.id;
-        const priorKey = priorRsvpKey(uid, reply.attendeeEmail, reply.occurrence);
-        // Read on every response, not just a bare acceptance: `alreadyFolded`
-        // needs the stored value on every path, so there is no cheaper way to
-        // skip this round-trip anymore (there used to be one for the
-        // non-acceptance/commented-acceptance cases — traded away below).
-        const stored = await host.get<string>(priorKey);
-
-        // Re-processing a conversation re-runs this loop for a response
-        // already folded onto the event thread — Graph's subscription fires
-        // on `updated` as well as `created`, and the drain re-fetches the
-        // whole conversation for any notified message. The note itself
-        // upserts by key, so re-saving it wouldn't duplicate it, but its
-        // `unread` intent would still be re-applied and drag the thread back
-        // to unread for anyone who already read it. Comparing against the
-        // stored partstat (not just presence) means a genuine change of
-        // response is never caught by this: an attendee who edits only their
-        // comment on an unchanged response gets no updated note, which is
-        // the accepted trade for not re-raising unread on every re-deliver.
-        if (alreadyFolded(stored, reply)) {
-          foldedMessageIds.add(noteKey);
-          continue;
-        }
-
-        // A bare acceptance says nothing the event's guest list does not
-        // already show. Drop the message rather than writing a note: a note
-        // is the only thing that could mark the organiser's thread unread,
-        // and marking it folded here keeps a responses-only conversation
-        // from becoming an email thread of its own.
-        if (!shouldEmitRsvpNote(reply, isNonAcceptance(stored))) {
-          foldedMessageIds.add(noteKey);
-          continue;
-        }
-
-        // saveNote returns null when no thread carries `icaluid:<uid>` yet
-        // (the calendar event hasn't synced). deferUntilThread has the
-        // platform hold the note and attach it once that thread appears.
-        await host.tools.integrations.saveNote({
-          thread: { source: `icaluid:${uid}` },
-          key: noteKey,
-          content: composeRsvpNote(reply),
-          contentType: "markdown",
-          created: messageDate(m),
-          author: {
-            email: reply.attendeeEmail,
-            ...(reply.attendeeName ? { name: reply.attendeeName } : {}),
-          },
-          // Explicit on both paths. An omitted flag does NOT mean "leave read
-          // state alone": attaching a note already surfaces the thread as
-          // unread for every recipient except its author, so only an
-          // explicit false overrides it.
-          unread: !initialSync,
-          deferUntilThread: true,
+        // The decision order (repeat check first, marker written only on the
+        // emitting path) and the reasoning behind it live once, on
+        // `foldRsvp`. Only what is Graph-specific stays here.
+        await foldRsvp({
+          uid,
+          reply,
+          note: { key: noteKey, created: messageDate(m), unread: !initialSync },
+          readMarker: (key) => host.get<string>(key),
+          writeMarker: (key, partstat) => host.set(key, partstat),
+          saveNote: (note) => host.tools.integrations.saveNote(note),
         });
+        // Folded on every outcome — emitted, suppressed, or already folded.
+        // A response never stays in the mail thread: dropping it is what
+        // keeps a responses-only conversation from becoming an email thread
+        // of its own, and a suppressed bare acceptance is folded precisely
+        // because writing no note is the right answer for it.
         foldedMessageIds.add(noteKey);
-        // Recorded regardless of the saveNote return value: a deferred note
-        // returns no id, and gating on it would leave a deferred
-        // non-acceptance unrecorded forever, wrongly treating a later bare
-        // acceptance as reversing nothing.
-        //
-        // Always set, never cleared: the key now holds the last response
-        // actually folded, for every emitted response (including an
-        // acceptance) — that's what lets `alreadyFolded` recognise a repeat
-        // of ANY partstat, not just an outstanding non-acceptance.
-        await host.set(priorKey, reply.partstat);
       }
       // Messages whose note survived the fold — everything below that reads
       // `item.messages` to pick a "parent" (facets, calendar bundling) must

libs/rsvp-fold/README.md

@@ -5,6 +5,14 @@ tentative) onto the event's thread as a note.
 
 ## What it does
 
+- `foldRsvp({ uid, reply, note, readMarker, writeMarker, saveNote })` — folds
+  one response onto its event's thread, owning the order the rules below have
+  to be applied in: check for a repeat first, only then decide whether the
+  response earns a note, and record it only when one was actually written.
+  Getting that order wrong re-raises unread on threads people have already
+  read, so connectors call this rather than sequencing the predicates
+  themselves. Reading and writing the marker, and saving the note, are
+  injected — connectors differ in how they batch those.
 - `composeRsvpNote(reply)` — formats the one-line note (plus the responder's
   personal comment, when they left one) that a connector attaches to the
   event thread.