feat: fix the Warmbly Cloud pool link across both roles (#262): take an enrolled mailbox out of this instance's own warmup pool so local partners stop writing to it and their unverifiable warmup stops landing in the owner's unibox, recognise the cloud's warmup mail whose verify header did not survive delivery through a new warmup-deliveries lookup that ignores consumed_at because instance and cloud read the same mailbox, move the managed-mailbox access token route behind NODE_BROKER_TOKEN so the internet-facing tracking and forms services can no longer mint a live provider token, scope pause and resume to the caller's workspace, keep an enrolled mailbox listed once it goes inactive, release the cloud copy when the local mirror row cannot be written, refuse the one-time handshake when CREDENTIALS_ENCRYPTION_KEY is missing, blank an expired code's plaintext instance token, and stop errx answering 200 for a status outside its table

cmd/worker/main.go

@@ -109,8 +109,14 @@ func main() {
 		log.Fatal(err)
 	}
 	// Mailboxes managed by Warmbly Cloud send with access tokens the backend
-	// brokers; the refresh grant never reaches the worker.
-	tokenBroker, err := repository.NewHTTPBrokeredTokenClient(internalBaseURL, internalToken)
+	// brokers; the refresh grant never reaches the worker. Minting one is a
+	// broker operation, so it carries NODE_BROKER_TOKEN where the instance
+	// issues a separate one and the shared internal token otherwise.
+	brokerToken := os.Getenv("NODE_BROKER_TOKEN")
+	if brokerToken == "" {
+		brokerToken = internalToken
+	}
+	tokenBroker, err := repository.NewHTTPBrokeredTokenClient(internalBaseURL, brokerToken)
 	if err != nil {
 		log.Fatal(err)
 	}

deploy/split-cloud/control-plane.env.example

@@ -39,10 +39,11 @@ CREDENTIALS_ENCRYPTION_KEY=
 INTERNAL_API_TOKEN=
 SECRET_KEY_BASE=
 
-# The credential for the two routes that open a sealed key and sign a blob
-# operation. Falls back to INTERNAL_API_TOKEN when unset; set it to its own
-# value here, because the tracking and forms services are internet-facing and
-# carry the shared one. Nodes receive this, the edge services do not.
+# The credential for the routes that open a sealed key, sign a blob operation
+# and mint a managed mailbox's access token. Falls back to INTERNAL_API_TOKEN
+# when unset; set it to its own value here, because the tracking and forms
+# services are internet-facing and carry the shared one. Nodes receive this,
+# the edge services do not.
 NODE_BROKER_TOKEN=
 
 # --- storage ----------------------------------------------------------------

deploy/split-cloud/node/worker.env.example

@@ -24,7 +24,8 @@ ENCRYPTED_KEYS_PROVIDER=http
 ENCRYPTED_KEYS_BACKEND_URL=https://api.example.com
 ENCRYPTED_KEYS_WORKER_TOKEN=
 INTERNAL_API_TOKEN=
-# Only when the control plane sets one. Used for the brokered providers below.
+# Only when the control plane sets one. Used for the brokered providers below
+# and for the access token a mailbox Warmbly Cloud manages sends with.
 NODE_BROKER_TOKEN=
 
 # --- the bus and cache ------------------------------------------------------

docker-compose.yml

@@ -193,6 +193,10 @@ x-selfhost-env: &selfhost-env
   REDIS: ${REDIS:-redis://redis:6379}
   AUTH_SECRET: ${AUTH_SECRET:-local-dev-auth-secret-minimum-32-characters-long}
   INTERNAL_API_TOKEN: ${INTERNAL_API_TOKEN:-local-dev-internal-token}
+  # The broker routes (open a data key, sign a blob operation, mint a managed
+  # mailbox's access token). Optional; empty falls back to the token above, and
+  # every service that needs it reads it from here so the two never disagree.
+  NODE_BROKER_TOKEN: ${NODE_BROKER_TOKEN:-}
 
   # AI assistant. Omit AI_PROVIDER to run with AI off (endpoints return 503).
   AI_PROVIDER: ${AI_PROVIDER:-}

docs/content/docs/development/configuration.mdx

@@ -57,7 +57,7 @@ Five values protect the whole instance. Compose ships a working default for each
 |---|---|---|---|
 | `AUTH_SECRET` | 32 characters or more | JWT and session signing. The realtime service reads the same value as `JWT_SECRET` | yes |
 | `INTERNAL_API_TOKEN` | any random string | The backend's `/api/v1/internal/` routes, which workers and the tracking service authenticate against | yes |
-| `NODE_BROKER_TOKEN` | any random string | The two routes that perform a privileged operation for the caller: opening a sealed data key and signing a blob operation. Optional, and falls back to `INTERNAL_API_TOKEN`. Worth setting in a split deployment, where the tracking and forms services are internet-facing and hold the shared token | yes |
+| `NODE_BROKER_TOKEN` | any random string | The routes that perform a privileged operation for the caller: opening a sealed data key, signing a blob operation, and minting a provider access token for a mailbox Warmbly Cloud manages. Optional, and falls back to `INTERNAL_API_TOKEN`. The same value has to be set on the control plane and on every node that calls those routes, or they answer 401. Worth setting in a split deployment, where the tracking and forms services are internet-facing and hold the shared token | yes |
 | `SECRET_KEY_BASE` | 64 characters or more | Phoenix session signing in the realtime service | yes |
 | `KMS_LOCAL_MASTER_KEY` | base64, exactly 32 bytes | The root key that seals every per-organization data key | yes |
 | `CREDENTIALS_ENCRYPTION_KEY` | exactly 64 hex characters | Mailbox credentials at rest: SMTP and IMAP passwords, and Gmail and Outlook OAuth access and refresh tokens | yes |

docs/content/docs/development/split-deployment.mdx

@@ -75,7 +75,7 @@ The cost is one HTTPS call per DEK open, which Redis caches, and one per blob op
 
 Two limits are worth knowing. A signed URL covers one verb on one key, and only for the prefixes a node reaches: transport bodies, campaign attachments, and the mailbox bodies a sync stores. Avatars, form assets and workspace export archives cannot be signed for at all, so the credential a node holds is not a credential for the bucket.
 
-That credential is `NODE_BROKER_TOKEN`. It falls back to `INTERNAL_API_TOKEN`, which is what a single-machine install uses, but set it to its own value here: the tracking and forms services are internet-facing and carry the shared token, and there is no reason for what they hold to be enough to open a data key.
+That credential is `NODE_BROKER_TOKEN`. It falls back to `INTERNAL_API_TOKEN`, which is what a single-machine install uses, but set it to its own value here: the tracking and forms services are internet-facing and carry the shared token, and there is no reason for what they hold to be enough to open a data key. The same token covers the third broker route, which mints a provider access token for a mailbox Warmbly Cloud manages.
 
 ## Building it
 

docs/content/docs/guides/warmbly-cloud.mdx

@@ -17,7 +17,8 @@ Your instance keeps a single link to a Warmbly Cloud workspace. When you enroll
 
 - Warmbly Cloud sends the mailbox's warmup mail, replies to partners, and rescues, opens and stars the warmup mail it receives, on Warmbly's own workers
 - the cloud reads only verified warmup mail from the mailbox. Everything else that arrives is dropped unread; no history is imported and nothing is stored
-- your instance stops its own warmup for that mailbox and keeps sending campaigns from it as usual
+- your instance stops its own warmup for that mailbox and takes it out of the local warmup pool, so no mailbox on your server writes to it any more, and keeps sending campaigns from it as usual
+- your instance recognises the cloud's warmup mail arriving in the mailbox and leaves it out of your unibox, asking the cloud to vouch for anything it cannot identify on its own
 - health, daily volume and 7 day totals show up in your instance's settings as the cloud reports them
 
 Unenrolling an SMTP/IMAP mailbox, or disconnecting the instance, deletes its credential on the cloud immediately. A mailbox that was signed in through Warmbly Cloud stays in the cloud workspace when you remove it from the instance; disconnecting the instance removes its mirrors of such mailboxes, since they cannot send without the link.

internal/api/handler/poollink.go

@@ -416,3 +416,28 @@ func (h *Handler) PoolLinkVerifyWarmupToken(c *gin.Context) {
 	}
 	c.JSON(http.StatusOK, gin.H{"valid": valid})
 }
+
+// PoolLinkVerifyWarmupDelivery answers for warmup mail that reached the mailbox
+// without its verify header, which the instance cannot recognise on its own.
+func (h *Handler) PoolLinkVerifyWarmupDelivery(c *gin.Context) {
+	inst := middleware.GetPoolLinkInstance(c)
+	if inst == nil {
+		errx.JSON(c, errx.ErrUnauthorized)
+		return
+	}
+	remoteID, ok := poolLinkRemoteID(c)
+	if !ok {
+		return
+	}
+	var q models.PoolLinkWarmupDeliveryQuery
+	if err := c.ShouldBindJSON(&q); err != nil {
+		errx.JSON(c, errx.New(errx.BadRequest, "invalid request body"))
+		return
+	}
+	valid, xerr := h.PoolLinkService.VerifyWarmupDelivery(c.Request.Context(), inst, remoteID, q)
+	if xerr != nil {
+		errx.JSON(c, xerr)
+		return
+	}
+	c.JSON(http.StatusOK, gin.H{"valid": valid})
+}

internal/api/routes.go

@@ -131,9 +131,9 @@ func Run(
 	// Internal backend-to-backend endpoints. Workers call these instead of
 	// touching Postgres directly, per the no-direct-data-services rule in
 	// CLAUDE.md. Auth: shared bearer token (INTERNAL_API_TOKEN).
-	// The two broker endpoints sit in their own group. They perform a
-	// privileged operation for the caller rather than moving a record, so they
-	// take NODE_BROKER_TOKEN, which falls back to INTERNAL_API_TOKEN but lets a
+	// The broker endpoints sit in their own group. They perform a privileged
+	// operation for the caller rather than moving a record, so they take
+	// NODE_BROKER_TOKEN, which falls back to INTERNAL_API_TOKEN but lets a
 	// split deployment keep the edge services off this credential.
 	broker := r.Group("/api/v1/internal")
 	broker.Use(m.NodeBrokerAuthMiddleware())
@@ -146,6 +146,11 @@ func Run(
 		// The node then transfers directly against the object store, so bodies
 		// and attachments never pass through here.
 		broker.POST("/blobs/presign", h.InternalPresignBlob)
+
+		// Mints a live provider access token for a mailbox Warmbly Cloud
+		// manages, which is worth more than any record the rest of the
+		// internal API moves.
+		broker.GET("/cloud-link/token/:id", h.InternalCloudLinkToken)
 	}
 
 	internal := r.Group("/api/v1/internal")
@@ -174,9 +179,6 @@ func Run(
 		// something the mailbox sent?" (tasks, message map, unibox threads).
 		internal.GET("/sync/own-conversation", h.InternalSyncOwnConversation)
 
-		// Brokered credential for a mailbox managed by Warmbly Cloud.
-		internal.GET("/cloud-link/token/:id", h.InternalCloudLinkToken)
-
 		// Worker bootstrap config + heartbeat. Workers POST their identity
 		// on boot (worker_id + bind_ip + tag) and pull their runtime config
 		// instead of carrying it all in the install-time env file.
@@ -1326,6 +1328,7 @@ func Run(
 				poolLinkInstance.POST("/oauth/finish", h.PoolLinkOAuthFinish)
 				poolLinkInstance.GET("/mailboxes/:remoteId/token", h.PoolLinkAccessToken)
 				poolLinkInstance.GET("/mailboxes/:remoteId/warmup-tokens/:token", h.PoolLinkVerifyWarmupToken)
+				poolLinkInstance.POST("/mailboxes/:remoteId/warmup-deliveries", h.PoolLinkVerifyWarmupDelivery)
 				poolLinkInstance.GET("/workspace-mailboxes", h.PoolLinkWorkspaceMailboxes)
 				poolLinkInstance.POST("/mailboxes/adopt", h.PoolLinkAdopt)
 			}

internal/app/cloudlink/client.go

@@ -69,7 +69,7 @@ func (c *client) do(ctx context.Context, method, path string, body any, out any)
 	if res.StatusCode >= 400 {
 		var re remoteError
 		_ = json.Unmarshal(raw, &re)
-		code := errx.Code(res.StatusCode)
+		code := remoteCode(res.StatusCode)
 		if re.Message == "" {
 			re.Message = fmt.Sprintf("Warmbly Cloud answered %d", res.StatusCode)
 		}
@@ -85,3 +85,21 @@ func (c *client) do(ctx context.Context, method, path string, body any, out any)
 	}
 	return nil
 }
+
+// remoteCode maps the cloud's status onto one errx can answer with. errx.JSON
+// writes status 0 for a code outside its table, which gin turns into a 200, so
+// an unmapped answer (a proxy's 502) would report a failed call as a success.
+func remoteCode(status int) errx.Code {
+	switch status {
+	case http.StatusBadRequest, http.StatusUnauthorized, http.StatusPaymentRequired, http.StatusForbidden,
+		http.StatusNotFound, http.StatusConflict, http.StatusUnprocessableEntity, http.StatusTooManyRequests,
+		http.StatusInternalServerError, http.StatusNotImplemented, http.StatusServiceUnavailable:
+		return errx.Code(status)
+	case http.StatusGone:
+		return errx.NotFound
+	}
+	if status >= 500 {
+		return errx.ServiceUnavailable
+	}
+	return errx.BadRequest
+}

internal/app/cloudlink/managed.go

@@ -107,6 +107,12 @@ func (s *service) mirror(ctx context.Context, l *models.CloudLink, orgID, userID
 		if s.emailSvc != nil {
 			_ = s.emailSvc.Delete(ctx, userID.String(), acc.ID.String())
 		}
+		// Release the cloud side too: a mailbox left linked to this instance
+		// with no mirror here is hidden from the adoptable list and refused on
+		// a second attempt, so it could never be recovered from either side.
+		if rerr := s.clientFor(l).do(ctx, http.MethodDelete, "/instance/mailboxes/"+state.RemoteID.String(), nil, nil); rerr != nil {
+			log.Error().Str("remote_id", state.RemoteID.String()).Str("code", rerr.Identifier).Msg("cloud link: local mirror row failed and the cloud link could not be released")
+		}
 		return nil, errx.InternalError()
 	}
 	if s.emailSvc != nil {
@@ -220,3 +226,29 @@ func (s *service) VerifyWarmupToken(ctx context.Context, accountID uuid.UUID, to
 	}
 	return out.Valid, nil
 }
+
+// IsCloudWarmupDelivery asks the cloud whether a message that arrived without a
+// verify header is its own warmup mail. Every send from a Microsoft mailbox
+// loses the header in transit, so without this the cloud's warmup would be
+// filed as ordinary mail in the owner's inbox.
+func (s *service) IsCloudWarmupDelivery(ctx context.Context, accountID uuid.UUID, sender, messageID, subject string) (bool, error) {
+	if messageID == "" && (sender == "" || subject == "") {
+		return false, nil
+	}
+	m, err := s.repo.GetByAccount(ctx, accountID)
+	if err != nil || m == nil {
+		return false, err
+	}
+	l, xerr := s.link(ctx)
+	if xerr != nil {
+		return false, xerr
+	}
+	var out struct {
+		Valid bool `json:"valid"`
+	}
+	q := models.PoolLinkWarmupDeliveryQuery{Sender: sender, MessageID: messageID, Subject: subject}
+	if xerr := s.clientFor(l).do(ctx, http.MethodPost, "/instance/mailboxes/"+m.RemoteID.String()+"/warmup-deliveries", q, &out); xerr != nil {
+		return false, xerr
+	}
+	return out.Valid, nil
+}

internal/app/cloudlink/service.go

@@ -27,6 +27,7 @@ var (
 	ErrCodeExpired     = errx.NewWithIdentifier(errx.NotFound, "cloud_link_code_expired", "The code expired before it was approved. Start again.")
 	ErrOAuthMailbox    = errx.NewWithIdentifier(errx.Unprocessable, "cloud_link_oauth_mailbox", "Google and Microsoft sign-in mailboxes cannot be warmed by Warmbly Cloud yet, because their refresh grant is bound to this instance's own OAuth app. Connect the mailbox with SMTP/IMAP (an app password) to enroll it.")
 	ErrMailboxInactive = errx.NewWithIdentifier(errx.Unprocessable, "cloud_link_mailbox_inactive", "Only active mailboxes can be enrolled.")
+	ErrNoCredentialKey = errx.NewWithIdentifier(errx.Conflict, "cloud_link_no_key", "This instance has no CREDENTIALS_ENCRYPTION_KEY, so it cannot store the Warmbly Cloud token. Set one and restart before connecting.")
 )
 
 // cloudURLAllowed requires TLS: the token and mailbox passwords travel on
@@ -111,6 +112,8 @@ type Service interface {
 	IsEnrolled(ctx context.Context, accountID uuid.UUID) bool
 	// VerifyWarmupToken is the consumer's check that warmup mail in an enrolled mailbox is the cloud's.
 	VerifyWarmupToken(ctx context.Context, accountID uuid.UUID, token string) (bool, error)
+	// IsCloudWarmupDelivery is the same check for warmup mail whose verify header did not survive.
+	IsCloudWarmupDelivery(ctx context.Context, accountID uuid.UUID, sender, messageID, subject string) (bool, error)
 }
 
 type service struct {
@@ -170,6 +173,11 @@ func (s *service) StartConnect(ctx context.Context, userID uuid.UUID, cloudURL s
 	if l, err := s.repo.Get(ctx); err == nil && l != nil {
 		return nil, ErrAlreadyLinked
 	}
+	// The handshake is one-time: refuse it now rather than lose the token
+	// after a member has already approved the code on the cloud.
+	if err := s.repo.CanStore(); err != nil {
+		return nil, ErrNoCredentialKey
+	}
 	cloudURL = strings.TrimRight(strings.TrimSpace(cloudURL), "/")
 	if cloudURL == "" {
 		cloudURL = CloudURL()
@@ -279,15 +287,25 @@ func (s *service) Disconnect(ctx context.Context) *errx.Error {
 		return xerr
 	}
 	// Managed mirrors have no credential of their own; they end with the link.
-	if rows, err := s.repo.List(ctx); err == nil {
-		for _, m := range rows {
-			if !m.Managed || s.emailSvc == nil {
-				continue
-			}
-			if acc, xerr := s.emails.GetByID(ctx, m.EmailAccountID); xerr == nil {
-				s.forgetToken(m.EmailAccountID)
-				_ = s.emailSvc.Delete(ctx, acc.UserID, acc.ID.String())
-			}
+	// Enumerating them has to succeed before the link row goes: afterwards
+	// there is nothing left to retry from, and the mailboxes would be left
+	// warming nowhere.
+	rows, err := s.repo.List(ctx)
+	if err != nil {
+		return errx.InternalError()
+	}
+	var released []uuid.UUID
+	for _, m := range rows {
+		if !m.Managed {
+			released = append(released, m.EmailAccountID)
+			continue
+		}
+		if s.emailSvc == nil {
+			continue
+		}
+		if acc, xerr := s.emails.GetByID(ctx, m.EmailAccountID); xerr == nil {
+			s.forgetToken(m.EmailAccountID)
+			_ = s.emailSvc.Delete(ctx, acc.UserID, acc.ID.String())
 		}
 	}
 	if err := s.repo.UnenrollAll(ctx); err != nil {
@@ -296,6 +314,10 @@ func (s *service) Disconnect(ctx context.Context) *errx.Error {
 	if err := s.repo.Delete(ctx); err != nil {
 		return errx.InternalError()
 	}
+	// The mailboxes the cloud was warming rejoin this instance's pool.
+	for _, id := range released {
+		s.syncLocalPool(ctx, id)
+	}
 	return nil
 }
 
@@ -332,7 +354,9 @@ func (s *service) ListMailboxes(ctx context.Context, orgID uuid.UUID) ([]models.
 	}
 
 	rows := make([]models.CloudLinkMailboxRow, 0, len(accounts))
-	for _, a := range accounts {
+	seen := make(map[uuid.UUID]bool, len(accounts))
+	add := func(a *models.Email) {
+		seen[a.ID] = true
 		row := models.CloudLinkMailboxRow{ID: a.ID, Email: a.Email, Name: a.Name, Provider: a.Provider, Status: a.Status}
 		if e, ok := byAccount[a.ID]; ok {
 			at := e.EnrolledAt
@@ -343,6 +367,22 @@ func (s *service) ListMailboxes(ctx context.Context, orgID uuid.UUID) ([]models.
 		}
 		rows = append(rows, row)
 	}
+	for i := range accounts {
+		add(&accounts[i])
+	}
+	// An enrolled mailbox the cloud still warms stays listed after it goes
+	// inactive; the scope query above returns active ones only, and dropping it
+	// would leave no way to see, pause or unenroll it.
+	for _, e := range enrolled {
+		if seen[e.EmailAccountID] {
+			continue
+		}
+		acc, xerr := s.emails.GetByID(ctx, e.EmailAccountID)
+		if xerr != nil || acc.OrganizationID == nil || *acc.OrganizationID != orgID {
+			continue
+		}
+		add(acc)
+	}
 	return rows, nil
 }
 
@@ -359,11 +399,8 @@ func (s *service) row(ctx context.Context, orgID, accountID uuid.UUID) (*models.
 	return nil, errx.ErrNotFound
 }
 
-func (s *service) Enroll(ctx context.Context, orgID, accountID uuid.UUID) (*models.CloudLinkMailboxRow, *errx.Error) {
-	l, xerr := s.link(ctx)
-	if xerr != nil {
-		return nil, xerr
-	}
+// ownedAccount refuses a mailbox from another workspace before anything acts on it.
+func (s *service) ownedAccount(ctx context.Context, orgID, accountID uuid.UUID) (*models.Email, *errx.Error) {
 	acc, xerr := s.emails.GetByID(ctx, accountID)
 	if xerr != nil {
 		return nil, xerr
@@ -371,6 +408,27 @@ func (s *service) Enroll(ctx context.Context, orgID, accountID uuid.UUID) (*mode
 	if acc.OrganizationID == nil || *acc.OrganizationID != orgID {
 		return nil, errx.ErrNotFound
 	}
+	return acc, nil
+}
+
+// syncLocalPool re-reads the mailbox's local warmup pool membership. An
+// enrolled mailbox must leave the local pool: warmup sent to it from here
+// carries a token the cloud cannot vouch for, so it lands in the user's inbox.
+func (s *service) syncLocalPool(ctx context.Context, accountID uuid.UUID) {
+	if s.emailSvc != nil {
+		s.emailSvc.SyncWarmupPool(ctx, accountID)
+	}
+}
+
+func (s *service) Enroll(ctx context.Context, orgID, accountID uuid.UUID) (*models.CloudLinkMailboxRow, *errx.Error) {
+	l, xerr := s.link(ctx)
+	if xerr != nil {
+		return nil, xerr
+	}
+	acc, xerr := s.ownedAccount(ctx, orgID, accountID)
+	if xerr != nil {
+		return nil, xerr
+	}
 	if acc.Status != "active" {
 		return nil, ErrMailboxInactive
 	}
@@ -409,17 +467,15 @@ func (s *service) Enroll(ctx context.Context, orgID, accountID uuid.UUID) (*mode
 		}
 		return nil, errx.InternalError()
 	}
+	s.syncLocalPool(ctx, acc.ID)
 	return s.row(ctx, orgID, accountID)
 }
 
 func (s *service) Unenroll(ctx context.Context, orgID, accountID uuid.UUID) *errx.Error {
-	acc, xerr := s.emails.GetByID(ctx, accountID)
+	acc, xerr := s.ownedAccount(ctx, orgID, accountID)
 	if xerr != nil {
 		return xerr
 	}
-	if acc.OrganizationID == nil || *acc.OrganizationID != orgID {
-		return errx.ErrNotFound
-	}
 	m, err := s.repo.GetByAccount(ctx, accountID)
 	if err != nil {
 		return errx.InternalError()
@@ -443,6 +499,7 @@ func (s *service) Unenroll(ctx context.Context, orgID, accountID uuid.UUID) *err
 			return xerr
 		}
 	}
+	s.syncLocalPool(ctx, accountID)
 	return nil
 }
 
@@ -454,6 +511,9 @@ func (s *service) SetLifecycle(ctx context.Context, orgID, accountID uuid.UUID,
 	if xerr != nil {
 		return nil, xerr
 	}
+	if _, xerr := s.ownedAccount(ctx, orgID, accountID); xerr != nil {
+		return nil, xerr
+	}
 	m, err := s.repo.GetByAccount(ctx, accountID)
 	if err != nil {
 		return nil, errx.InternalError()

internal/app/consumer/event_new_email.go

@@ -53,6 +53,10 @@ func (s *JobsService) HandleNewEmail(ctx context.Context, e *models.JobEventNewE
 		// mailbox sends this way, so without this branch its warmup mail is
 		// filed as ordinary inbox mail at every recipient.
 		return nil
+	} else if s.isCloudWarmupDelivery(ctx, e) {
+		// The same message, in a mailbox Warmbly Cloud warms: the token lives
+		// there, so only the cloud can recognise it.
+		return nil
 	}
 
 	// A pool-linked mailbox is warmup-only: everything else is dropped unread.
@@ -215,6 +219,36 @@ func (s *JobsService) handleUnmarkedWarmupEmail(ctx context.Context, e *models.J
 	return true
 }
 
+// cloudWarmupCheckTimeout bounds the one call this handler makes off-box. It
+// runs on every message in an enrolled mailbox, so a slow cloud would otherwise
+// hold up ingest for everything behind it.
+const cloudWarmupCheckTimeout = 5 * time.Second
+
+// isCloudWarmupDelivery asks the cloud whether an unrecognised message in a
+// mailbox it warms is its own warmup mail. Best-effort: an unreachable cloud
+// files the message as ordinary mail rather than dropping the owner's.
+func (s *JobsService) isCloudWarmupDelivery(ctx context.Context, e *models.JobEventNewEmail) bool {
+	if s.CloudLink == nil || e.Message == nil {
+		return false
+	}
+	// Nothing the cloud could match on: skip both lookups.
+	sender := firstSenderAddress(e.Message.FromAddr)
+	if e.Message.MessageID == "" && (sender == "" || e.Message.Subject == "") {
+		return false
+	}
+	if !s.CloudLink.IsEnrolled(ctx, e.Message.EmailID) {
+		return false
+	}
+	ctx, cancel := context.WithTimeout(ctx, cloudWarmupCheckTimeout)
+	defer cancel()
+	ok, err := s.CloudLink.IsCloudWarmupDelivery(ctx, e.Message.EmailID, sender, e.Message.MessageID, e.Message.Subject)
+	if err != nil {
+		log.Warn().Err(err).Str("email_account_id", e.Message.EmailID.String()).Msg("cloud warmup delivery check failed; filing as ordinary mail")
+		return false
+	}
+	return ok
+}
+
 // firstSenderAddress pulls the bare address out of the first From value
 // ("Name <addr>" or a bare address).
 func firstSenderAddress(from []string) string {

internal/app/consumer/service.go

@@ -23,6 +23,7 @@ import (
 type CloudLinkVerifier interface {
 	IsEnrolled(ctx context.Context, accountID uuid.UUID) bool
 	VerifyWarmupToken(ctx context.Context, accountID uuid.UUID, token string) (bool, error)
+	IsCloudWarmupDelivery(ctx context.Context, accountID uuid.UUID, sender, messageID, subject string) (bool, error)
 }
 
 type JobsService struct {

internal/app/email/handler.go

@@ -371,6 +371,23 @@ func (s *emailService) syncWarmupPoolMembership(ctx context.Context, account *mo
 		return
 	}
 
+	// A mailbox Warmbly Cloud warms is no longer a partner here: warmup this
+	// instance sent it carries a token the cloud cannot vouch for, so it would
+	// be filed as ordinary mail in the owner's inbox. Only a definite answer
+	// acts; an unreadable one leaves the membership alone rather than letting a
+	// blip evict a healthy mailbox and discard its health record.
+	if s.cloudLink != nil {
+		enrolled, err := s.cloudLink.GetByAccount(ctx, account.ID)
+		if err != nil {
+			log.Warn().Err(err).Str("account_id", account.ID.String()).Msg("cloud link lookup failed; warmup pool membership left as it is")
+			return
+		}
+		if enrolled != nil {
+			s.removeFromAllWarmupPools(ctx, account)
+			return
+		}
+	}
+
 	if !s.canUseWarmupPool(ctx, account) {
 		s.removeFromAllWarmupPools(ctx, account)
 		return
@@ -397,11 +414,21 @@ func (s *emailService) canUseWarmupPool(ctx context.Context, account *models.Ema
 	if account == nil || account.Status != "active" || account.OrganizationID == nil || s.featureGate == nil {
 		return false
 	}
-
 	canWarmup, err := s.featureGate.CanUseWarmup(ctx, *account.OrganizationID)
 	return err == nil && canWarmup
 }
 
+// SyncWarmupPool re-evaluates one mailbox's local warmup pool membership, for
+// callers that changed something the membership depends on but not the mailbox
+// row itself (enrolling it in, or releasing it from, Warmbly Cloud).
+func (s *emailService) SyncWarmupPool(ctx context.Context, accountID uuid.UUID) {
+	account, xerr := s.emailRepository.GetByID(ctx, accountID)
+	if xerr != nil {
+		return
+	}
+	s.syncWarmupPoolMembership(ctx, account)
+}
+
 // orgSuspendedOrRestricted reports whether the workspace's posture bars the
 // paid warmup pool. Fails open.
 func (s *emailService) orgSuspendedOrRestricted(ctx context.Context, orgID uuid.UUID) bool {

internal/app/email/service.go

@@ -111,6 +111,9 @@ type EmailService interface {
 	// LoadAccountOntoWorker assigns a worker if needed and ships the mailbox
 	// to it (idempotent; the reconciler calls it too).
 	LoadAccountOntoWorker(ctx context.Context, accountID uuid.UUID) error
+	// SyncWarmupPool re-evaluates one mailbox's local warmup pool membership,
+	// for a change outside the mailbox row (Warmbly Cloud enrollment).
+	SyncWarmupPool(ctx context.Context, accountID uuid.UUID)
 	// GetSyncState is the dashboard's view of a mailbox's sync: nil state when
 	// the worker has not reported yet.
 	GetSyncState(ctx context.Context, userID, emailID string) (*models.SyncState, models.SyncPolicy, *errx.Error)

internal/app/poollink/oauth.go

@@ -304,3 +304,23 @@ func (s *service) VerifyWarmupToken(ctx context.Context, inst *models.PoolLinkIn
 	}
 	return t != nil && t.RecipientAccountID == m.EmailAccountID, nil
 }
+
+// VerifyWarmupDelivery answers for warmup mail whose verify header did not
+// survive delivery, which is every send from a Microsoft mailbox.
+func (s *service) VerifyWarmupDelivery(ctx context.Context, inst *models.PoolLinkInstance, remoteID uuid.UUID, q models.PoolLinkWarmupDeliveryQuery) (bool, *errx.Error) {
+	m, err := s.repo.GetMailboxByRemote(ctx, inst.ID, remoteID)
+	if err != nil {
+		return false, errx.InternalError()
+	}
+	if m == nil {
+		return false, ErrMailboxNotFound
+	}
+	if s.warmup == nil {
+		return false, nil
+	}
+	ok, err := s.warmup.IsWarmupDelivery(ctx, m.EmailAccountID, q.Sender, q.MessageID, q.Subject)
+	if err != nil {
+		return false, errx.InternalError()
+	}
+	return ok, nil
+}

internal/app/poollink/service.go

@@ -74,9 +74,9 @@ type Service interface {
 	ListWorkspaceMailboxes(ctx context.Context, inst *models.PoolLinkInstance) ([]models.PoolLinkWorkspaceMailbox, *errx.Error)
 	Adopt(ctx context.Context, inst *models.PoolLinkInstance, req models.PoolLinkAdoptRequest) (*models.PoolLinkMailboxState, *errx.Error)
 	VerifyWarmupToken(ctx context.Context, inst *models.PoolLinkInstance, remoteID, token uuid.UUID) (bool, *errx.Error)
+	// VerifyWarmupDelivery is the same answer for warmup mail that arrived without its verify header.
+	VerifyWarmupDelivery(ctx context.Context, inst *models.PoolLinkInstance, remoteID uuid.UUID, q models.PoolLinkWarmupDeliveryQuery) (bool, *errx.Error)
 
-	// IsLinkedMailbox is the consumer's hot-path warmup-only check.
-	IsLinkedMailbox(ctx context.Context, accountID uuid.UUID) bool
 	// HasActiveLink entitles a workspace to warm its linked mailboxes.
 	HasActiveLink(ctx context.Context, orgID uuid.UUID) bool
 	// Plan is the allowance the workspace currently has.
@@ -659,11 +659,6 @@ func (s *service) Unenroll(ctx context.Context, inst *models.PoolLinkInstance, r
 	return nil
 }
 
-func (s *service) IsLinkedMailbox(ctx context.Context, accountID uuid.UUID) bool {
-	m, err := s.repo.GetMailboxByAccount(ctx, accountID)
-	return err == nil && m != nil
-}
-
 func (s *service) HasActiveLink(ctx context.Context, orgID uuid.UUID) bool {
 	ok, err := s.repo.HasActiveInstance(ctx, orgID)
 	return err == nil && ok

internal/errx/errx.go

@@ -41,7 +41,20 @@ func (e *Error) identifier() string {
 	if e.Identifier != "" {
 		return e.Identifier
 	}
-	return codeToIdentifier[e.Code]
+	if id, ok := codeToIdentifier[e.Code]; ok {
+		return id
+	}
+	return codeToIdentifier[Internal]
+}
+
+// resolve is the HTTP status and title to answer with. A Code outside the
+// table maps to 0, which gin leaves at 200, so an error would be reported as a
+// success; anything unknown is an internal error instead.
+func (e *Error) resolve() (int, string) {
+	if status, ok := codeToHTTP[e.Code]; ok {
+		return status, codeToString[e.Code]
+	}
+	return codeToHTTP[Internal], codeToString[Internal]
 }
 
 // ResponseCode is the machine-readable `code` this error answers with, for
@@ -74,8 +87,7 @@ func Handle(c *gin.Context, err error) {
 	var bizErr *Error
 	if errors.As(err, &bizErr) {
 		// Business error – send clean JSON
-		httpCode := codeToHTTP[bizErr.Code]
-		httpError := codeToString[bizErr.Code]
+		httpCode, httpError := bizErr.resolve()
 		c.JSON(httpCode, response{
 			Error:     httpError,
 			Message:   bizErr.Message,
@@ -91,8 +103,7 @@ func Handle(c *gin.Context, err error) {
 
 // JSON sends a business error as JSON response
 func JSON(c *gin.Context, err *Error) {
-	httpCode := codeToHTTP[err.Code]
-	httpError := codeToString[err.Code]
+	httpCode, httpError := err.resolve()
 	c.JSON(httpCode, response{
 		Error:     httpError,
 		Message:   err.Message,

internal/errx/errx_test.go

@@ -35,3 +35,25 @@ func TestJSONIncludesStableCodeAndRequestID(t *testing.T) {
 		t.Fatalf("unexpected body: %+v", body)
 	}
 }
+
+// A Code outside the table used to map to status 0, which gin leaves at 200.
+// Every error carrying an upstream status (the Warmbly Cloud client is the one
+// that does) would then be reported to the caller as a success.
+func TestJSONAnswersAnUnknownCodeAsAnError(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	rec := httptest.NewRecorder()
+	c, _ := gin.CreateTestContext(rec)
+
+	JSON(c, New(Code(599), "upstream fell over"))
+
+	if rec.Code != http.StatusInternalServerError {
+		t.Fatalf("status = %d, want %d", rec.Code, http.StatusInternalServerError)
+	}
+	var body response
+	if err := json.Unmarshal(rec.Body.Bytes(), &body); err != nil {
+		t.Fatalf("decode response: %v", err)
+	}
+	if body.Code != "internal_error" {
+		t.Fatalf("code = %q", body.Code)
+	}
+}

internal/models/poollink.go

@@ -181,6 +181,15 @@ type PoolLinkWorkspaceMailbox struct {
 	Status   string    `json:"status"`
 }
 
+// PoolLinkWarmupDeliveryQuery asks the cloud whether one message that arrived
+// in a mailbox it warms is its own warmup mail. The verify header does not
+// survive every provider, so the instance cannot always tell on its own.
+type PoolLinkWarmupDeliveryQuery struct {
+	Sender    string `json:"sender"`
+	MessageID string `json:"message_id"`
+	Subject   string `json:"subject"`
+}
+
 // PoolLinkMailboxState is the per-mailbox view shown in both dashboards.
 type PoolLinkMailboxState struct {
 	RemoteID       uuid.UUID              `json:"remote_id"`

internal/repository/pg_cloudlink.go

@@ -15,6 +15,10 @@ import (
 
 // CloudLinkRepository is the self-hosted side of pool link.
 type CloudLinkRepository interface {
+	// CanStore reports whether the instance token can be sealed at all. The
+	// handshake is one-time: a token that cannot be written is gone, and the
+	// link is left standing on the cloud with nobody holding it.
+	CanStore() error
 	Get(ctx context.Context) (*models.CloudLink, error)
 	Put(ctx context.Context, link *models.CloudLink) error
 	Delete(ctx context.Context) error
@@ -41,6 +45,13 @@ func NewCloudLinkRepository(db *pgxpool.Pool, enc *encrypt.Encrypter) CloudLinkR
 	return &cloudLinkRepository{db: db, encrypt: enc}
 }
 
+func (r *cloudLinkRepository) CanStore() error {
+	if r.encrypt == nil {
+		return errNoLinkEncrypter
+	}
+	return nil
+}
+
 func (r *cloudLinkRepository) Get(ctx context.Context) (*models.CloudLink, error) {
 	query := `SELECT cloud_url, instance_id, token, organization_name, connected_by, connected_at, last_synced_at, last_error FROM cloud_link WHERE id = true`
 	var l models.CloudLink

internal/repository/pg_poollink.go

@@ -36,7 +36,6 @@ type PoolLinkRepository interface {
 	GetMailboxByRemote(ctx context.Context, instanceID, remoteID uuid.UUID) (*models.PoolLinkMailbox, error)
 	GetMailboxByAccount(ctx context.Context, accountID uuid.UUID) (*models.PoolLinkMailbox, error)
 	ListMailboxes(ctx context.Context, instanceID uuid.UUID) ([]models.PoolLinkMailbox, error)
-	CountMailboxesForOrganization(ctx context.Context, orgID uuid.UUID) (int, error)
 	DeleteMailbox(ctx context.Context, instanceID, remoteID uuid.UUID) error
 	// TouchMailboxToken records when a managed mailbox last drew an access token.
 	TouchMailboxToken(ctx context.Context, instanceID, remoteID uuid.UUID) error
@@ -150,7 +149,19 @@ func (r *poolLinkRepository) ClaimCode(ctx context.Context, deviceCodeHash strin
 	return &c, token, nil
 }
 
+// DeleteExpiredCodes blanks the plaintext instance token on any expired row and
+// removes rows old enough to be of no interest. The two are separate on
+// purpose: an approved code the instance never came back for would otherwise
+// keep a usable bearer token in plaintext for as long as the row survived,
+// which is what "held only between approval and the next poll" rules out. The
+// instance itself stays, listed under Settings > Linked instances, where it can
+// be revoked.
 func (r *poolLinkRepository) DeleteExpiredCodes(ctx context.Context) error {
+	query := `UPDATE pool_link_codes SET instance_token = NULL WHERE instance_token IS NOT NULL AND expires_at < NOW()`
+	if _, err := r.db.Exec(ctx, query); err != nil {
+		db.CaptureError(err, query, nil, "exec")
+		return err
+	}
 	_, err := r.db.Exec(ctx, `DELETE FROM pool_link_codes WHERE expires_at < NOW() - INTERVAL '1 day'`)
 	return err
 }
@@ -314,20 +325,6 @@ func (r *poolLinkRepository) ListMailboxes(ctx context.Context, instanceID uuid.
 	return out, rows.Err()
 }
 
-func (r *poolLinkRepository) CountMailboxesForOrganization(ctx context.Context, orgID uuid.UUID) (int, error) {
-	var n int
-	query := `
-		SELECT COUNT(*) FROM pool_link_mailboxes m
-		JOIN pool_link_instances i ON i.id = m.instance_id
-		WHERE i.organization_id = $1 AND i.revoked_at IS NULL
-	`
-	if err := r.db.QueryRow(ctx, query, orgID).Scan(&n); err != nil {
-		db.CaptureError(err, query, []any{orgID}, "queryrow")
-		return 0, err
-	}
-	return n, nil
-}
-
 func (r *poolLinkRepository) DeleteMailbox(ctx context.Context, instanceID, remoteID uuid.UUID) error {
 	query := `DELETE FROM pool_link_mailboxes WHERE instance_id = $1 AND remote_id = $2`
 	if _, err := r.db.Exec(ctx, query, instanceID, remoteID); err != nil {

internal/repository/pg_warmup.go

@@ -169,6 +169,9 @@ type WarmupRepository interface {
 	// FindDeliveredWarmupToken resolves the pending token for an inbound
 	// message that carries no verify header.
 	FindDeliveredWarmupToken(ctx context.Context, recipientAccountID uuid.UUID, senderAddress, messageID, subject string) (*models.WarmupToken, error)
+	// IsWarmupDelivery answers the same question for a second reader of the
+	// same mailbox, which must not depend on who consumed the token first.
+	IsWarmupDelivery(ctx context.Context, recipientAccountID uuid.UUID, senderAddress, messageID, subject string) (bool, error)
 	RecordInvalidTokenAttempt(ctx context.Context, accountID uuid.UUID, attemptedToken string) error
 	CountRecentInvalidAttempts(ctx context.Context, accountID uuid.UUID, since time.Time) (int, error)
 
@@ -1200,6 +1203,52 @@ func (r *warmupRepository) FindDeliveredWarmupToken(ctx context.Context, recipie
 	return scanWarmupToken(r.db.QueryRow(ctx, query, recipientAccountID, messageID, senderAddress, subject))
 }
 
+// IsWarmupDelivery reports whether an inbound message is warmup mail this
+// deployment sent to the mailbox. It deliberately ignores consumed_at: the
+// caller is a linked instance reading the same mailbox as this deployment, so
+// a consumed-only match would depend on which of the two synced first.
+func (r *warmupRepository) IsWarmupDelivery(ctx context.Context, recipientAccountID uuid.UUID, senderAddress, messageID, subject string) (bool, error) {
+	messageID = strings.Trim(strings.TrimSpace(messageID), "<>")
+	senderAddress = strings.TrimSpace(senderAddress)
+	subject = strings.TrimSpace(subject)
+	if messageID == "" && (senderAddress == "" || subject == "") {
+		return false, nil
+	}
+	query := `
+		SELECT EXISTS (
+		    SELECT 1 FROM warmup_received wr
+		    WHERE wr.email_account_id = $1
+		      AND $2 <> ''
+		      AND btrim(wr.message_id, '<>') = $2
+		  ) OR EXISTS (
+		    SELECT 1 FROM warmup_tokens wt
+		    WHERE wt.recipient_account_id = $1
+		      AND wt.created_at > NOW() - INTERVAL '7 days'
+		      AND (
+		        ($2 <> '' AND wt.sent_message_id <> '' AND btrim(wt.sent_message_id, '<>') = $2)
+		        OR (
+		          $3 <> '' AND $4 <> ''
+		          -- Only when one side has no Message-ID to compare. Two known
+		          -- ids that differ are a different message, and matching on
+		          -- sender and subject alone would drop the owner's real mail.
+		          AND ($2 = '' OR wt.sent_message_id = '')
+		          AND wt.created_at > NOW() - INTERVAL '2 days'
+		          AND wt.subject <> ''
+		          AND lower(btrim(wt.subject)) = lower($4)
+		          AND EXISTS (
+		              SELECT 1 FROM email_accounts ea
+		              WHERE ea.id = wt.sender_account_id AND lower(ea.email) = lower($3)
+		          )
+		        )
+		      )
+		  )`
+	var ok bool
+	if err := r.db.QueryRow(ctx, query, recipientAccountID, messageID, senderAddress, subject).Scan(&ok); err != nil {
+		return false, err
+	}
+	return ok, nil
+}
+
 // RecordInvalidTokenAttempt records an invalid warmup token attempt
 func (r *warmupRepository) RecordInvalidTokenAttempt(ctx context.Context, accountID uuid.UUID, attemptedToken string) error {
 	query := `

internal/repository/pool_link_live_test.go

@@ -0,0 +1,204 @@
+package repository
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5/pgxpool"
+)
+
+// The pool link's two queries that a second reader of the same mailbox depends
+// on. An enrolled mailbox is synced by the instance AND by Warmbly Cloud, so
+// whichever consumes the token first must not decide what the other sees.
+//
+//	WARMBLY_TEST_DB=postgres://warmbly:warmbly@localhost:15432/warmbly_dev?sslmode=disable \
+//	  go test ./internal/repository/ -run LivePoolLink -v
+
+type poolLinkFixture struct {
+	pool      *pgxpool.Pool
+	user      uuid.UUID
+	org       uuid.UUID
+	recipient uuid.UUID
+	sender    uuid.UUID
+	senderTo  string
+	task      uuid.UUID
+	warmup    WarmupRepository
+}
+
+func newPoolLinkFixture(t *testing.T) *poolLinkFixture {
+	t.Helper()
+	_, pool := liveContactDB(t)
+	ctx := context.Background()
+	f := &poolLinkFixture{
+		pool: pool, user: uuid.New(), org: uuid.New(),
+		recipient: uuid.New(), sender: uuid.New(), task: uuid.New(),
+		warmup: NewWarmupRepository(pool),
+	}
+	exec := func(sql string, args ...any) {
+		t.Helper()
+		if _, err := pool.Exec(ctx, sql, args...); err != nil {
+			t.Fatalf("fixture %q: %v", sql[:min(60, len(sql))], err)
+		}
+	}
+	exec(`INSERT INTO users (id, email, first_name, last_name, password_hash) VALUES ($1, $2, 'Pool', 'Link', 'x')`,
+		f.user, "pl-"+f.user.String()[:8]+"@test.local")
+	exec(`INSERT INTO organizations (id, name, slug, owner_user_id) VALUES ($1, 'Pool Link', $2, $3)`,
+		f.org, "pl-"+f.org.String()[:8], f.user)
+	for _, id := range []uuid.UUID{f.recipient, f.sender} {
+		addr := "pl-" + id.String()[:8] + "@test.local"
+		exec(`INSERT INTO email_accounts (id, user_id, organization_id, email, name, signature_plain,
+		          signature_html, provider, status, campaign_limit, min_wait_time, timezone)
+		      VALUES ($1, $2, $3, $4, 'PL', '', '', 'smtp_imap', 'active', 50, 600, 'UTC')`,
+			id, f.user, f.org, addr)
+		if id == f.sender {
+			f.senderTo = addr
+		}
+	}
+	exec(`INSERT INTO tasks (id, task_type, email_account_id, status, message_id)
+	      VALUES ($1, 'warmup', $2, 'completed', '')`, f.task, f.sender)
+	t.Cleanup(func() {
+		c := context.Background()
+		for _, sql := range []string{
+			`DELETE FROM warmup_received WHERE email_account_id IN (SELECT id FROM email_accounts WHERE organization_id = $1)`,
+			`DELETE FROM warmup_tokens WHERE recipient_account_id IN (SELECT id FROM email_accounts WHERE organization_id = $1)`,
+			`DELETE FROM tasks WHERE email_account_id IN (SELECT id FROM email_accounts WHERE organization_id = $1)`,
+			`DELETE FROM email_accounts WHERE organization_id = $1`,
+			`DELETE FROM organizations WHERE id = $1`,
+			`DELETE FROM users WHERE id = $1`,
+		} {
+			if _, err := pool.Exec(c, sql, f.org); err != nil {
+				t.Errorf("cleanup %q: %v", sql[:min(50, len(sql))], err)
+			}
+		}
+	})
+	return f
+}
+
+// token writes one warmup token addressed to the recipient.
+func (f *poolLinkFixture) token(t *testing.T, messageID, subject string, consumed bool) uuid.UUID {
+	t.Helper()
+	id := uuid.New()
+	var consumedAt any
+	if consumed {
+		consumedAt = time.Now()
+	}
+	_, err := f.pool.Exec(context.Background(),
+		`INSERT INTO warmup_tokens (token, task_id, sender_account_id, recipient_account_id, sent_message_id, subject, consumed_at)
+		 VALUES ($1, $2, $3, $4, $5, $6, $7)`,
+		id, f.task, f.sender, f.recipient, messageID, subject, consumedAt)
+	if err != nil {
+		t.Fatalf("insert token: %v", err)
+	}
+	return id
+}
+
+// A mailbox Warmbly Cloud warms is read by the cloud and by the instance. The
+// cloud consumes the token when it syncs first, and the instance's question
+// ("is this yours?") must still answer yes, or the cloud's warmup mail lands in
+// the owner's unibox.
+func TestLivePoolLinkWarmupDeliverySurvivesAConsumedToken(t *testing.T) {
+	f := newPoolLinkFixture(t)
+	ctx := context.Background()
+	const messageID = "<pool-link-consumed@test.local>"
+	f.token(t, messageID, "Quick question", true)
+
+	found, err := f.warmup.FindDeliveredWarmupToken(ctx, f.recipient, f.senderTo, messageID, "Quick question")
+	if err != nil {
+		t.Fatalf("FindDeliveredWarmupToken: %v", err)
+	}
+	if found != nil {
+		t.Fatal("a consumed token should not be resolvable for local verification")
+	}
+
+	ok, err := f.warmup.IsWarmupDelivery(ctx, f.recipient, f.senderTo, messageID, "Quick question")
+	if err != nil {
+		t.Fatalf("IsWarmupDelivery: %v", err)
+	}
+	if !ok {
+		t.Fatal("consumed warmup mail must still be recognised as warmup for a linked instance")
+	}
+}
+
+func TestLivePoolLinkWarmupDeliveryMatchesAndRefuses(t *testing.T) {
+	f := newPoolLinkFixture(t)
+	ctx := context.Background()
+
+	// Pending token, matched on the sender/subject pair rather than the id.
+	f.token(t, "", "Following up", false)
+	ok, err := f.warmup.IsWarmupDelivery(ctx, f.recipient, f.senderTo, "", "Following up")
+	if err != nil {
+		t.Fatalf("IsWarmupDelivery pair: %v", err)
+	}
+	if !ok {
+		t.Fatal("a pending token for this sender and subject should match")
+	}
+
+	// A delivery already recorded, whose token has since been cleaned up.
+	const receivedID = "<pool-link-received@test.local>"
+	if err := f.warmup.RecordWarmupReceived(ctx, f.recipient, uuid.New(), receivedID, f.sender); err != nil {
+		t.Fatalf("RecordWarmupReceived: %v", err)
+	}
+	ok, err = f.warmup.IsWarmupDelivery(ctx, f.recipient, "someone@elsewhere.test", receivedID, "Anything")
+	if err != nil {
+		t.Fatalf("IsWarmupDelivery received: %v", err)
+	}
+	if !ok {
+		t.Fatal("a recorded warmup delivery should be recognised by its message id")
+	}
+
+	// A pool partner's real email, sharing a subject with a token whose own
+	// Message-ID is known and different. Two known ids that differ are a
+	// different message, so the sender/subject fallback must not claim it.
+	f.token(t, "<warm-2@test.local>", "Quick sync", false)
+	ok, err = f.warmup.IsWarmupDelivery(ctx, f.recipient, f.senderTo, "<genuine-reply@test.local>", "Quick sync")
+	if err != nil {
+		t.Fatalf("IsWarmupDelivery conflicting ids: %v", err)
+	}
+	if ok {
+		t.Fatal("a message whose id differs from the token's must not match on sender and subject")
+	}
+
+	// Real mail from someone else must never be dropped.
+	ok, err = f.warmup.IsWarmupDelivery(ctx, f.recipient, "prospect@elsewhere.test", "<real-mail@elsewhere.test>", "Are you free Thursday?")
+	if err != nil {
+		t.Fatalf("IsWarmupDelivery stranger: %v", err)
+	}
+	if ok {
+		t.Fatal("ordinary mail must not be taken for warmup")
+	}
+}
+
+// An approved code the instance never came back for must not keep a usable
+// bearer token in plaintext once it expires.
+func TestLivePoolLinkExpiredCodeLosesItsToken(t *testing.T) {
+	_, pool := liveContactDB(t)
+	ctx := context.Background()
+	repo := NewPoolLinkRepository(pool)
+	id := uuid.New()
+	suffix := id.String()[:8]
+
+	if _, err := pool.Exec(ctx,
+		`INSERT INTO pool_link_codes (id, device_code_hash, user_code, instance_name, status, instance_token, expires_at)
+		 VALUES ($1, $2, $3, 'Abandoned', 'approved', 'wpl_secret', NOW() - INTERVAL '1 minute')`,
+		id, "hash-"+suffix, "PL"+suffix[:2]+"-"+suffix[2:6]); err != nil {
+		t.Fatalf("insert code: %v", err)
+	}
+	t.Cleanup(func() {
+		if _, err := pool.Exec(context.Background(), `DELETE FROM pool_link_codes WHERE id = $1`, id); err != nil {
+			t.Errorf("cleanup code: %v", err)
+		}
+	})
+
+	if err := repo.DeleteExpiredCodes(ctx); err != nil {
+		t.Fatalf("DeleteExpiredCodes: %v", err)
+	}
+	var token *string
+	if err := pool.QueryRow(ctx, `SELECT instance_token FROM pool_link_codes WHERE id = $1`, id).Scan(&token); err != nil {
+		t.Fatalf("read back: %v", err)
+	}
+	if token != nil {
+		t.Fatalf("expired code kept its plaintext instance token: %q", *token)
+	}
+}

web/src/app/app/settings/warmbly-cloud/ConnectFlow.tsx

@@ -39,6 +39,9 @@ export default function ConnectFlow({
     const [nudged, setNudged] = React.useState(false);
     const [linked, setLinked] = React.useState(status.connected);
     const [enrolledCount, setEnrolledCount] = React.useState(0);
+    // The status prop was read before the handshake, so the workspace name has
+    // to come from the poll that completed it.
+    const [orgName, setOrgName] = React.useState(status.link?.organization_name ?? "");
 
     const issue = step === 0 && !linked ? "Approve the code on Warmbly Cloud first" : null;
     React.useEffect(() => {
@@ -80,14 +83,16 @@ export default function ConnectFlow({
                                 <LinkStep
                                     status={status}
                                     linked={linked}
-                                    onLinked={() => {
+                                    orgName={orgName}
+                                    onLinked={(name) => {
+                                        setOrgName(name);
                                         setLinked(true);
                                         setTimeout(() => goTo(1), 650);
                                     }}
                                 />
                             )}
                             {step === 1 && <MailboxesStep onCountChange={setEnrolledCount} />}
-                            {step === 2 && <DoneStep status={status} enrolledCount={enrolledCount} />}
+                            {step === 2 && <DoneStep orgName={orgName} enrolledCount={enrolledCount} />}
                         </motion.div>
                     </AnimatePresence>
                 </div>
@@ -202,19 +207,8 @@ function Footer({ step, issue, nudged, onBack, onNext }: { step: Step; issue: st
 }
 
 // Step 1: the shared link card.
-function LinkStep({ status, linked, onLinked }: { status: CloudLinkStatus; linked: boolean; onLinked: () => void }) {
-    const [orgName, setOrgName] = React.useState(status.link?.organization_name ?? "");
-    return (
-        <CloudLinkCard
-            linked={linked}
-            orgName={orgName}
-            cloudUrl={status.default_cloud_url}
-            onLinked={(name) => {
-                setOrgName(name);
-                onLinked();
-            }}
-        />
-    );
+function LinkStep({ status, linked, orgName, onLinked }: { status: CloudLinkStatus; linked: boolean; orgName: string; onLinked: (orgName: string) => void }) {
+    return <CloudLinkCard linked={linked} orgName={orgName} cloudUrl={status.default_cloud_url} onLinked={onLinked} />;
 }
 
 // Step 2: pick mailboxes.
@@ -296,7 +290,7 @@ function MailboxesStep({ onCountChange }: { onCountChange: (n: number) => void }
     );
 }
 
-function DoneStep({ status, enrolledCount }: { status: CloudLinkStatus; enrolledCount: number }) {
+function DoneStep({ orgName, enrolledCount }: { orgName: string; enrolledCount: number }) {
     return (
         <div className="flex flex-col items-center justify-center text-center gap-3 py-8">
             <motion.span
@@ -312,7 +306,7 @@ function DoneStep({ status, enrolledCount }: { status: CloudLinkStatus; enrolled
                     {enrolledCount === 0 ? "You are all set" : `${enrolledCount} mailbox${enrolledCount === 1 ? "" : "es"} warming in the pool`}
                 </p>
                 <p className="text-[12.5px] text-slate-500 mt-0.5 max-w-md">
-                    {status.link?.organization_name ? `Linked to ${status.link.organization_name}. ` : ""}
+                    {orgName ? `Linked to ${orgName}. ` : ""}
                     Warmup starts on the first slot of each mailbox's window and ramps daily. Health and volume show up on this page as they arrive.
                 </p>
             </div>