feat: preserve truthful cross-provider sync success and backlog evidence, retry durable pending-arrival updates without reading unread mail, and render factual sync status

internal/app/consumer/event_flags_update.go

@@ -38,7 +38,7 @@ func (s *JobsService) HandleFlagsAdd(ctx context.Context, e *models.JobEventFlag
 		}
 	}
 
-	email, err := s.emailForSyncUpdate(ctx, e.UserID, e.ID, func(message *models.EmailMessageStoreData) {
+	email, err := s.emailForSyncUpdate(ctx, e.UserID, e.EmailID, e.ID, func(message *models.EmailMessageStoreData) {
 		for _, flag := range e.Flags {
 			if !slices.Contains(message.Flags, flag) {
 				message.Flags = append(message.Flags, flag)
@@ -155,7 +155,7 @@ func warmupTokenFromFlags(flags []string) string {
 }
 
 func (s *JobsService) HandleFlagsRemove(ctx context.Context, e *models.JobEventFlags) error {
-	email, err := s.emailForSyncUpdate(ctx, e.UserID, e.ID, func(message *models.EmailMessageStoreData) {
+	email, err := s.emailForSyncUpdate(ctx, e.UserID, e.EmailID, e.ID, func(message *models.EmailMessageStoreData) {
 		message.Flags = slices.DeleteFunc(message.Flags, func(flag string) bool { return slices.Contains(e.Flags, flag) })
 		if models.SeenFromFlags(e.Flags) {
 			message.Seen = false

internal/app/consumer/event_sync_state.go

@@ -34,12 +34,11 @@ func (s *JobsService) HandleSyncState(ctx context.Context, e *models.JobEventSyn
 		return err
 	}
 
-	// A relayed state means a pass completed: the worker reached the server,
-	// listed its folders and finished the tick. That is the only signal we
-	// get that an outage is over, and without it a five-minute blip left a
-	// red "needs attention" on the mailbox for good, because nothing but a
-	// credential reconnect ever resolved an error row.
-	s.resolveTransientMailErrors(ctx, e.EmailID, e.State.LastSyncedAt)
+	// Progress relays are not recovery evidence; require a newer successful check.
+	if e.State.LastSyncedAt != nil && e.State.Deferred == 0 && e.State.BackfillCursor.GoogleRecovery == nil &&
+		(prev == nil || prev.LastSyncedAt == nil || e.State.LastSyncedAt.After(*prev.LastSyncedAt)) {
+		s.resolveTransientMailErrors(ctx, e.EmailID, e.State.LastSyncedAt)
+	}
 
 	if s.StreamingPublisher == nil {
 		return nil

internal/app/consumer/event_sync_state_test.go

@@ -14,11 +14,12 @@ import (
 // stubSyncStateRepo accepts what the worker relays.
 type stubSyncStateRepo struct {
 	repository.EmailSyncStateRepository
-	put *models.SyncState
+	put   *models.SyncState
+	saved *models.SyncState
 }
 
 func (s *stubSyncStateRepo) Get(context.Context, uuid.UUID) (*models.SyncState, error) {
-	return nil, nil
+	return s.saved, nil
 }
 
 func (s *stubSyncStateRepo) Put(_ context.Context, _, _ uuid.UUID, st *models.SyncState) error {

internal/app/consumer/event_update_email.go

@@ -13,7 +13,7 @@ import (
 )
 
 func (s *JobsService) HandleUpdateEmail(ctx context.Context, e *models.JobEventEmailUpdate) error {
-	email, err := s.emailForSyncUpdate(ctx, e.UserID, e.ID, func(message *models.EmailMessageStoreData) {
+	email, err := s.emailForSyncUpdate(ctx, e.UserID, e.EmailID, e.ID, func(message *models.EmailMessageStoreData) {
 		token := warmupTokenFromMessage(message)
 		message.Flags = append([]string{}, e.Flags...)
 		if token != "" {
@@ -98,7 +98,7 @@ func (s *JobsService) HandleFolderUpdate(ctx context.Context, e *models.JobEvent
 	if e.Relayed {
 		return s.applyRelayedFolder(ctx, e)
 	}
-	email, err := s.emailForSyncUpdate(ctx, e.UserID, e.ID, func(message *models.EmailMessageStoreData) {
+	email, err := s.emailForSyncUpdate(ctx, e.UserID, e.EmailID, e.ID, func(message *models.EmailMessageStoreData) {
 		// A pending row is not visible yet, so nothing has filed it locally.
 		message.Folder = e.Folder
 	})
@@ -144,16 +144,33 @@ func (s *JobsService) applyRelayedFolder(ctx context.Context, e *models.JobEvent
 }
 
 // emailForSyncUpdate rechecks visible mail if verification won the pending-row lock.
-func (s *JobsService) emailForSyncUpdate(ctx context.Context, userID, id uuid.UUID, updatePending func(*models.EmailMessageStoreData)) (*models.EmailMessageStoreData, error) {
-	message, err := s.UniboxRepository.GetByID(ctx, userID, id)
+var ErrSyncArrivalPending = errors.New("sync arrival pending delivery")
+
+func (s *JobsService) emailForSyncUpdate(ctx context.Context, userID, emailID, id uuid.UUID, updatePending func(*models.EmailMessageStoreData)) (*models.EmailMessageStoreData, error) {
+	message, err := s.UniboxRepository.GetForSync(ctx, userID, emailID, id)
 	if !errors.Is(err, repository.ErrEmailNotFound) {
 		return message, err
 	}
-	updated, err := s.UniboxRepository.UpdatePendingEmail(ctx, userID, id, updatePending)
+	updated, err := s.UniboxRepository.UpdatePendingEmail(ctx, userID, id, func(message *models.EmailMessageStoreData) {
+		if message.EmailID == emailID {
+			updatePending(message)
+		}
+	})
 	if err != nil || updated {
 		return nil, err
 	}
-	message, err = s.UniboxRepository.GetByID(ctx, userID, id)
+	message, err = s.UniboxRepository.GetForSync(ctx, userID, emailID, id)
+	if errors.Is(err, repository.ErrEmailNotFound) && s.ArrivalOutbox != nil {
+		pending, lookupErr := s.ArrivalOutbox.HasPendingArrival(ctx, userID, emailID, id)
+		if lookupErr != nil {
+			return nil, lookupErr
+		}
+		if pending {
+			return nil, ErrSyncArrivalPending
+		}
+		// Delivery may have finished between the visible-row read and the marker read.
+		message, err = s.UniboxRepository.GetForSync(ctx, userID, emailID, id)
+	}
 	if errors.Is(err, repository.ErrEmailNotFound) {
 		return nil, nil
 	}

internal/app/consumer/sync_truth_live_test.go

@@ -0,0 +1,193 @@
+package jobs
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/google/uuid"
+	"github.com/warmbly/warmbly/internal/app/cipher"
+	"github.com/warmbly/warmbly/internal/infrastructure/encryptedkeys"
+	"github.com/warmbly/warmbly/internal/infrastructure/kms"
+	"github.com/warmbly/warmbly/internal/models"
+	"github.com/warmbly/warmbly/internal/repository"
+)
+
+type failSyncArrivalCreate struct {
+	repository.UniboxRepository
+	fail bool
+}
+
+func (r *failSyncArrivalCreate) CreateEntry(ctx context.Context, user uuid.UUID, message *models.EmailMessageStoreData) error {
+	if r.fail {
+		r.fail = false
+		return errors.New("transient arrival write failure")
+	}
+	return r.UniboxRepository.CreateEntry(ctx, user, message)
+}
+
+func TestLiveSyncTruthPendingArrivalRetainsProviderUpdatesAcrossReload(t *testing.T) {
+	for _, kind := range []string{"seen", "folder", "full update"} {
+		t.Run(kind, func(t *testing.T) {
+			s, d := liveWarmupService(t)
+			f := newWarmupFixture(t, d)
+			ctx := t.Context()
+			key, err := kms.NewLocal(make([]byte, 32))
+			if err != nil {
+				t.Fatal(err)
+			}
+			cipherService := cipher.NewService(key, nil, encryptedkeys.NewPostgres(d))
+			r := repository.NewDurableEmailMessageMapRepository(d, cipherService)
+			s.ArrivalOutbox = r
+			s.UniboxRepository = &failSyncArrivalCreate{UniboxRepository: s.UniboxRepository, fail: true}
+			e := f.arrival("<"+uuid.NewString()+"@test.local>", nil)
+			e.Message.Subject = "ordinary customer question"
+			mapping := repository.EmailMessageData{UserID: e.UserID.String(), EmailID: e.Message.EmailID.String(), ID: e.Message.ID.String(), MessageID: e.Message.MessageID}
+			if err := r.AdmitArrival(ctx, mapping, &repository.PendingArrival{Arrival: e}); err != nil {
+				t.Fatal(err)
+			}
+			s.InitEvents()
+			if err := r.DeliverArrivals(ctx, s.deliverArrivalEvent); err == nil {
+				t.Fatal("expected transient arrival write failure")
+			}
+			apply := func(service *JobsService) error {
+				switch kind {
+				case "seen":
+					return service.HandleFlagsAdd(ctx, &models.JobEventFlags{UserID: e.UserID, EmailID: e.Message.EmailID, ID: e.Message.ID, Flags: []string{models.FlagSeen}})
+				case "folder":
+					return service.HandleFolderUpdate(ctx, &models.JobEventFolderUpdate{UserID: e.UserID, EmailID: e.Message.EmailID, ID: e.Message.ID, Folder: models.FolderArchive})
+				default:
+					return service.HandleUpdateEmail(ctx, &models.JobEventEmailUpdate{UserID: e.UserID, EmailID: e.Message.EmailID, ID: e.Message.ID, Flags: []string{models.FlagSeen}, Folder: models.FolderArchive, FolderPath: "Archive", UID: 42, Mailbox: 7})
+				}
+			}
+			if err := apply(s); !errors.Is(err, ErrSyncArrivalPending) {
+				t.Fatalf("update was acknowledged ahead of failed arrival: %v", err)
+			}
+			if _, err := d.Exec(ctx, `UPDATE sync_arrival_outbox SET retry_at=now() WHERE id=$1`, e.Message.ID); err != nil {
+				t.Fatal(err)
+			}
+			fresh, _ := liveWarmupService(t)
+			fresh.ArrivalOutbox = repository.NewDurableEmailMessageMapRepository(d, cipherService)
+			fresh.InitEvents()
+			if err := fresh.ArrivalOutbox.DeliverArrivals(ctx, fresh.deliverArrivalEvent); err != nil {
+				t.Fatal(err)
+			}
+			if err := apply(fresh); err != nil {
+				t.Fatal(err)
+			}
+			message, err := fresh.UniboxRepository.GetForSync(ctx, e.UserID, e.Message.EmailID, e.Message.ID)
+			if err != nil {
+				t.Fatal(err)
+			}
+			if kind != "folder" && !message.Seen {
+				t.Fatal("provider Seen update lost after original-ID arrival replay")
+			}
+			if kind != "seen" && message.Folder != models.FolderArchive {
+				t.Fatalf("provider folder update lost: %q", message.Folder)
+			}
+			if kind == "full update" && (message.UID != 42 || message.Mailbox != 7 || message.FolderPath != "Archive") {
+				t.Fatalf("provider handles lost: %+v", message)
+			}
+			if _, err := d.Exec(ctx, `DELETE FROM email_accounts WHERE id=$1`, e.Message.EmailID); err != nil {
+				t.Fatal(err)
+			}
+			if err := apply(fresh); err != nil {
+				t.Fatalf("deleted mailbox update retried forever: %v", err)
+			}
+		})
+	}
+}
+
+func TestLiveSyncTruthBackgroundReadsKeepUnreadAndStayOwnerScoped(t *testing.T) {
+	s, d := liveWarmupService(t)
+	f := newWarmupFixture(t, d)
+	ctx := t.Context()
+	e := f.arrival("<"+uuid.NewString()+"@test.local>", nil)
+	if err := s.UniboxRepository.CreateEntry(ctx, e.UserID, e.Message); err != nil {
+		t.Fatal(err)
+	}
+	assertUnread := func() {
+		t.Helper()
+		var seen bool
+		if err := d.QueryRow(ctx, `SELECT seen FROM unibox_emails WHERE id=$1`, e.Message.ID).Scan(&seen); err != nil || seen {
+			t.Fatalf("background work marked unread mail seen: %v %v", seen, err)
+		}
+	}
+	for _, owner := range []uuid.UUID{f.senderUser, uuid.New()} {
+		if _, err := s.UniboxRepository.GetForSync(ctx, owner, e.Message.EmailID, e.Message.ID); !errors.Is(err, repository.ErrEmailNotFound) {
+			t.Fatalf("unowned sync lookup returned a row: %v", err)
+		}
+	}
+	for _, mailbox := range []uuid.UUID{f.sender, uuid.New(), uuid.Nil} {
+		if _, err := s.UniboxRepository.GetForSync(ctx, e.UserID, mailbox, e.Message.ID); !errors.Is(err, repository.ErrEmailNotFound) {
+			t.Fatalf("wrong-mailbox sync lookup returned a row: %v", err)
+		}
+	}
+	assertUnread()
+	if err := s.HandleFlagsAdd(ctx, &models.JobEventFlags{UserID: e.UserID, EmailID: e.Message.EmailID, ID: e.Message.ID, Flags: []string{models.FlagFlagged}}); err != nil {
+		t.Fatal(err)
+	}
+	assertUnread()
+	for _, folder := range []string{models.FolderInbox, models.FolderArchive, models.FolderArchive} {
+		if err := s.HandleFolderUpdate(ctx, &models.JobEventFolderUpdate{UserID: e.UserID, EmailID: e.Message.EmailID, ID: e.Message.ID, Folder: folder}); err != nil {
+			t.Fatal(err)
+		}
+		assertUnread()
+	}
+	if err := s.HandleFlagsAdd(ctx, &models.JobEventFlags{UserID: e.UserID, EmailID: e.Message.EmailID, ID: e.Message.ID, Flags: []string{models.FlagSeen}}); err != nil {
+		t.Fatal(err)
+	}
+	message, err := s.UniboxRepository.GetForSync(ctx, e.UserID, e.Message.EmailID, e.Message.ID)
+	if err != nil || message == nil || !message.Seen {
+		t.Fatalf("real provider Seen flag did not mark read: %v", err)
+	}
+	if err := s.UniboxRepository.MarkSeen(ctx, e.UserID, e.Message.ID, false); err != nil {
+		t.Fatal(err)
+	}
+	assertUnread()
+	message, err = s.UniboxRepository.GetByID(ctx, e.UserID, e.Message.ID)
+	if err != nil || message == nil || !message.Seen {
+		t.Fatalf("explicit user view no longer marks read: %v", err)
+	}
+}
+
+func TestLiveSyncTruthLegacyArrivalHasNoPendingGuard(t *testing.T) {
+	s, d := liveWarmupService(t)
+	f := newWarmupFixture(t, d)
+	ctx := t.Context()
+	key, err := kms.NewLocal(make([]byte, 32))
+	if err != nil {
+		t.Fatal(err)
+	}
+	r := repository.NewDurableEmailMessageMapRepository(d, cipher.NewService(key, nil, encryptedkeys.NewPostgres(d)))
+	s.ArrivalOutbox = r
+	s.UniboxRepository = &failSyncArrivalCreate{UniboxRepository: s.UniboxRepository, fail: true}
+	e := f.arrival("<"+uuid.NewString()+"@test.local>", nil)
+	e.Message.Subject = "ordinary legacy customer question"
+	// Legacy workers admit only the map, then publish NEW_EMAIL separately.
+	if err := r.Add(ctx, repository.EmailMessageData{UserID: e.UserID.String(), EmailID: e.Message.EmailID.String(), ID: e.Message.ID.String(), MessageID: e.Message.MessageID}); err != nil {
+		t.Fatal(err)
+	}
+	if err := s.HandleNewEmail(ctx, e); err == nil {
+		t.Fatal("expected legacy arrival creation failure")
+	}
+	if pending, err := r.HasPendingArrival(ctx, e.UserID, e.Message.EmailID, e.Message.ID); err != nil || pending {
+		t.Fatalf("legacy map unexpectedly acquired a durable arrival marker: %v %v", pending, err)
+	}
+	if err := s.HandleFlagsAdd(ctx, &models.JobEventFlags{UserID: e.UserID, EmailID: e.Message.EmailID, ID: e.Message.ID, Flags: []string{models.FlagSeen}}); err != nil {
+		t.Fatalf("legacy map alone inferred a pending arrival: %v", err)
+	}
+	if err := s.HandleFolderUpdate(ctx, &models.JobEventFolderUpdate{UserID: e.UserID, EmailID: e.Message.EmailID, ID: e.Message.ID, Folder: models.FolderArchive}); err != nil {
+		t.Fatalf("legacy map alone inferred a pending arrival: %v", err)
+	}
+	if err := s.HandleNewEmail(ctx, e); err != nil {
+		t.Fatal(err)
+	}
+	message, err := s.UniboxRepository.GetForSync(ctx, e.UserID, e.Message.EmailID, e.Message.ID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if message.Seen || message.Folder != models.FolderInbox {
+		t.Fatalf("legacy update ordering must not be reported as protected: %+v", message)
+	}
+}

internal/app/consumer/sync_truth_test.go

@@ -0,0 +1,167 @@
+package jobs
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/warmbly/warmbly/internal/models"
+	"github.com/warmbly/warmbly/internal/repository"
+)
+
+func TestSyncTruthProgressRelaysDoNotResolveTransportWarnings(t *testing.T) {
+	previous := time.Now().Add(-time.Hour)
+	fresh := previous.Add(time.Minute)
+	for _, tc := range []struct {
+		name     string
+		stamp    *time.Time
+		deferred int
+		recovery *models.GoogleHistoryRecovery
+		resolved int
+	}{
+		{"heartbeat", &previous, 0, nil, 0},
+		{"older redelivery", nil, 0, nil, 0},
+		{"deferred", &fresh, 5, nil, 0},
+		{"recovering", &fresh, 0, &models.GoogleHistoryRecovery{HistoryID: 100}, 0},
+		{"successful catch-up", &fresh, 0, nil, 1},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			errRepo := &stubErrorRepo{}
+			repo := &stubSyncStateRepo{saved: &models.SyncState{LastSyncedAt: &previous}}
+			s := &JobsService{EmailSyncStateRepository: repo, EmailAccountErrorRepository: errRepo}
+			if err := s.HandleSyncState(t.Context(), &models.JobEventSyncState{UserID: uuid.New(), EmailID: uuid.New(), State: models.SyncState{BackfillStatus: models.SyncBackfillComplete, LastSyncedAt: tc.stamp, Deferred: tc.deferred, BackfillCursor: models.SyncCursor{GoogleRecovery: tc.recovery}}}); err != nil {
+				t.Fatal(err)
+			}
+			if repo.put == nil || errRepo.calls != tc.resolved {
+				t.Fatalf("state publication/recovery outcome: saved=%v resolved=%d", repo.put != nil, errRepo.calls)
+			}
+		})
+	}
+}
+
+type syncTruthLookup struct {
+	repository.UniboxRepository
+	message  *models.EmailMessageStoreData
+	reads    int
+	lookups  int
+	appearOn int
+}
+
+func (r *syncTruthLookup) GetByID(context.Context, uuid.UUID, uuid.UUID) (*models.EmailMessageStoreData, error) {
+	r.reads++
+	if r.message == nil {
+		return nil, repository.ErrEmailNotFound
+	}
+	r.message.Seen = true
+	return r.message, nil
+}
+
+func (r *syncTruthLookup) GetForSync(context.Context, uuid.UUID, uuid.UUID, uuid.UUID) (*models.EmailMessageStoreData, error) {
+	r.lookups++
+	if r.message == nil || (r.appearOn > 0 && r.lookups < r.appearOn) {
+		return nil, repository.ErrEmailNotFound
+	}
+	return r.message, nil
+}
+
+func (*syncTruthLookup) UpdatePendingEmail(context.Context, uuid.UUID, uuid.UUID, func(*models.EmailMessageStoreData)) (bool, error) {
+	return false, nil
+}
+
+func (r *syncTruthLookup) UpdateEntry(_ context.Context, _, _, _ uuid.UUID, update *repository.UpdateUniboxEntry) error {
+	if update.Seen != nil {
+		r.message.Seen = *update.Seen
+	}
+	return nil
+}
+
+type syncTruthOutbox struct {
+	repository.ArrivalOutbox
+	user, mailbox, id uuid.UUID
+	pending           bool
+	err               error
+}
+
+func (r *syncTruthOutbox) HasPendingArrival(_ context.Context, user, mailbox, id uuid.UUID) (bool, error) {
+	if user != r.user || mailbox != r.mailbox || id != r.id {
+		return false, nil
+	}
+	return r.pending, r.err
+}
+
+func TestSyncTruthBackgroundUpdatesNeverViewUnreadMail(t *testing.T) {
+	for _, kind := range []string{"flagged", "folder", "same folder"} {
+		t.Run(kind, func(t *testing.T) {
+			r := &syncTruthLookup{message: &models.EmailMessageStoreData{Seen: false, Folder: models.FolderInbox, ProviderFolder: models.FolderInbox}}
+			s := &JobsService{UniboxRepository: r, EmailRepository: warmupInboxEmailRepo{}}
+			user, mailbox, id := uuid.New(), uuid.New(), uuid.New()
+			var err error
+			if kind == "flagged" {
+				err = s.HandleFlagsAdd(t.Context(), &models.JobEventFlags{UserID: user, EmailID: mailbox, ID: id, Flags: []string{models.FlagFlagged}})
+			} else {
+				folder := models.FolderInbox
+				if kind == "folder" {
+					folder = models.FolderArchive
+				}
+				err = s.HandleFolderUpdate(t.Context(), &models.JobEventFolderUpdate{UserID: user, EmailID: mailbox, ID: id, Folder: folder})
+			}
+			if err != nil || r.message.Seen || r.reads != 0 {
+				t.Fatalf("background update treated as view: seen=%v reads=%d err=%v", r.message.Seen, r.reads, err)
+			}
+		})
+	}
+}
+
+func TestSyncTruthPendingArrivalUpdatesRetryOnlyForOwnedUndeliveredMail(t *testing.T) {
+	user, mailbox, id := uuid.New(), uuid.New(), uuid.New()
+	for _, kind := range []string{"flags add", "flags remove", "folder", "full update"} {
+		t.Run(kind, func(t *testing.T) {
+			r := &syncTruthLookup{}
+			outbox := &syncTruthOutbox{user: user, mailbox: mailbox, id: id, pending: true}
+			s := &JobsService{UniboxRepository: r, ArrivalOutbox: outbox}
+			apply := func(owner, box uuid.UUID) error {
+				switch kind {
+				case "flags add":
+					return s.HandleFlagsAdd(t.Context(), &models.JobEventFlags{UserID: owner, EmailID: box, ID: id, Flags: []string{models.FlagSeen}})
+				case "flags remove":
+					return s.HandleFlagsRemove(t.Context(), &models.JobEventFlags{UserID: owner, EmailID: box, ID: id, Flags: []string{models.FlagSeen}})
+				case "folder":
+					return s.HandleFolderUpdate(t.Context(), &models.JobEventFolderUpdate{UserID: owner, EmailID: box, ID: id, Folder: models.FolderArchive})
+				default:
+					return s.HandleUpdateEmail(t.Context(), &models.JobEventEmailUpdate{UserID: owner, EmailID: box, ID: id, Flags: []string{models.FlagSeen}})
+				}
+			}
+			if err := apply(user, mailbox); !errors.Is(err, ErrSyncArrivalPending) {
+				t.Fatalf("pending arrival was acknowledged: %v", err)
+			}
+			if err := apply(uuid.New(), mailbox); err != nil {
+				t.Fatalf("unowned update retried: %v", err)
+			}
+			if err := apply(user, uuid.New()); err != nil {
+				t.Fatalf("other-mailbox update retried: %v", err)
+			}
+			outbox.pending = false
+			if err := apply(user, mailbox); err != nil {
+				t.Fatalf("deliberately hidden/deleted mail retried: %v", err)
+			}
+			outbox.err = errors.New("marker unavailable")
+			if err := apply(user, mailbox); !errors.Is(err, outbox.err) {
+				t.Fatalf("marker failure swallowed: %v", err)
+			}
+		})
+	}
+}
+
+func TestSyncTruthArrivalFinishingDuringLookupStillGetsUpdate(t *testing.T) {
+	user, mailbox, id := uuid.New(), uuid.New(), uuid.New()
+	r := &syncTruthLookup{message: &models.EmailMessageStoreData{EmailID: mailbox}, appearOn: 3}
+	s := &JobsService{UniboxRepository: r, EmailRepository: warmupInboxEmailRepo{}, ArrivalOutbox: &syncTruthOutbox{user: user, mailbox: mailbox, id: id}}
+	if err := s.HandleFlagsAdd(t.Context(), &models.JobEventFlags{UserID: user, EmailID: mailbox, ID: id, Flags: []string{models.FlagSeen}}); err != nil {
+		t.Fatal(err)
+	}
+	if !r.message.Seen {
+		t.Fatal("arrival committed between reads lost the later Seen update")
+	}
+}

internal/app/consumer/unibox_seen_sync_test.go

@@ -21,6 +21,10 @@ func (r *seenSyncRepo) GetByID(context.Context, uuid.UUID, uuid.UUID) (*models.E
 	return r.stored, nil
 }
 
+func (r *seenSyncRepo) GetForSync(context.Context, uuid.UUID, uuid.UUID, uuid.UUID) (*models.EmailMessageStoreData, error) {
+	return r.stored, nil
+}
+
 func (r *seenSyncRepo) UpdateEntry(_ context.Context, _, _, _ uuid.UUID, e *repository.UpdateUniboxEntry) error {
 	r.updates = append(r.updates, *e)
 	return nil

internal/app/worker/wmail/admit.go

@@ -134,15 +134,21 @@ func (w *WMail) laneOf(ctx context.Context, key string, msg *models.EmailMessage
 // beginTick releases an expired hold so the state reports "within budget"
 // before the pass looks at anything.
 func (w *WMail) beginTick() {
+	w.tracker.tickComplete = false
 	w.tracker.release(time.Now())
 }
 
-// endTick folds the pass into the relayed state. deferred is the number of
-// live messages still waiting on the server; it is reported as-is (not
-// accumulated) so a pass that admits everything zeroes it.
-func (w *WMail) endTick(stats *tickStats) {
-	w.tracker.setDeferred(stats.deferred)
-	w.tracker.touch(time.Now())
+// endTick retains unmeasured backlog until a complete pass confirms catch-up.
+func (w *WMail) endTick(ctx context.Context, stats *tickStats, complete bool) {
+	complete = complete && ctx.Err() == nil && !stats.aborted && stats.deferred == 0 && w.tracker.state.BackfillCursor.GoogleRecovery == nil
+	w.tracker.tickComplete = complete
+	if complete {
+		w.tracker.setDeferred(0)
+		w.tracker.touch(time.Now())
+		return
+	}
+	w.tracker.setDeferred(max(w.tracker.state.Deferred, stats.deferred))
+	w.tracker.flush(time.Now())
 }
 
 // storeNew is the shared tail of every provider's new-message path: record

internal/app/worker/wmail/start_sync.go

@@ -78,11 +78,18 @@ func (w *WMail) nextSyncDelay(base time.Duration, last *errx.MailError) time.Dur
 // so one mailbox's bad server response must not take down every other
 // account's sync and send loops.
 func (w *WMail) syncOnce(ctx context.Context) (result *errx.MailError) {
+	if w.tracker != nil {
+		w.tracker.tickComplete = false
+	}
 	defer func() {
 		if r := recover(); r != nil {
+			if w.tracker != nil {
+				w.tracker.tickComplete = false
+			}
 			err := fmt.Errorf("mail sync panic: %v", r)
 			w.CaptureError(err)
 			log.Error().Err(err).Str("email_id", w.ID.String()).Msg("mail sync panicked")
+			result = &errx.MailError{Code: errx.MailErrorCodeImapUnknown, Type: errx.MailErrorWarning, Message: "mail sync pass interrupted"}
 		}
 	}()
 	if err := w.SyncMail(ctx); err != nil {
@@ -100,7 +107,9 @@ func (w *WMail) syncOnce(ctx context.Context) (result *errx.MailError) {
 		log.Warn().Err(err).Str("email_id", w.ID.String()).Msg("mail sync error")
 		return err
 	}
-	w.transportFailures = 0
+	if w.tracker != nil && w.tracker.tickComplete {
+		w.transportFailures = 0
+	}
 	return nil
 }
 

internal/app/worker/wmail/sync_google.go

@@ -29,7 +29,8 @@ func (w *WMail) SyncGoogle(ctx context.Context) *errx.MailError {
 	stats := &tickStats{}
 	w.googleTick = stats
 	w.googleFolders = nil
-	defer w.endTick(stats)
+	complete := false
+	defer func() { w.endTick(ctx, stats, complete) }()
 	if !w.retryUnmap(ctx) {
 		return nil
 	}
@@ -40,7 +41,7 @@ func (w *WMail) SyncGoogle(ctx context.Context) *errx.MailError {
 		return nil
 	}
 
-	newHistoryID, err := w.GoogleData.Client.FetchHistory(ctx, w.GoogleData.LastHistoryID)
+	newHistoryID, caughtUp, err := w.GoogleData.Client.FetchHistoryPass(ctx, w.GoogleData.LastHistoryID)
 	if errors.Is(err, goog.ErrHistoryExpired) {
 		baseline, berr := w.GoogleData.Client.HistoryBaseline(ctx)
 		if berr != nil {
@@ -81,10 +82,11 @@ func (w *WMail) SyncGoogle(ctx context.Context) *errx.MailError {
 	if !stats.aborted {
 		// The pass's own work is done; only a refusal the owner has to act
 		// on is worth failing it for.
-		if merr := w.googleReconcileFolders(ctx, time.Now(), stats); merr != nil && merr.Type == errx.MailErrorCritical {
+		if merr := w.googleReconcileFolders(ctx, time.Now(), stats); merr != nil {
 			return merr
 		}
 	}
+	complete = caughtUp
 	return nil
 }
 
@@ -338,6 +340,7 @@ func (w *WMail) googleBackfill(ctx context.Context, stats *tickStats) *errx.Mail
 		}
 		ids, next, err := w.GoogleData.Client.ListMessages(ctx, q, st.BackfillCursor.PageToken, googleBackfillPage)
 		if err != nil {
+			stats.aborted = true
 			var errMail *errx.MailError
 			if errors.As(err, &errMail) {
 				return errMail
@@ -364,6 +367,7 @@ func (w *WMail) googleBackfill(ctx context.Context, stats *tickStats) *errx.Mail
 			}
 			msg, err := w.GoogleData.Client.GetMessage(ctx, id)
 			if err != nil {
+				stats.aborted = true
 				var errMail *errx.MailError
 				if errors.As(err, &errMail) {
 					return errMail
@@ -436,6 +440,7 @@ func (w *WMail) googleReconcileFolders(ctx context.Context, now time.Time, stats
 	for page := 0; page < config.GmailFolderReconcilePages; page++ {
 		ids, next, err := w.GoogleData.Client.ListLabelMessages(ctx, goog.Inbox, q, token, 500)
 		if err != nil {
+			stats.aborted = true
 			return w.googleReconcileError(err)
 		}
 		for _, id := range ids {
@@ -479,6 +484,7 @@ func (w *WMail) googleReconcileFolders(ctx context.Context, now time.Time, stats
 		labels, found, err := w.GoogleData.Client.MessageLabels(ctx, m.ProviderID)
 		if err != nil {
 			if gmailRetryable(err) {
+				stats.aborted = true
 				return w.googleReconcileError(err)
 			}
 			// One message Gmail refuses must not hold up every row after it.

internal/app/worker/wmail/sync_graph.go

@@ -25,11 +25,14 @@ func (w *WMail) SyncGraph(ctx context.Context) *errx.MailError {
 	w.beginTick()
 	stats := &tickStats{}
 	w.graphTick = stats
+	complete := false
+	defer func() { w.endTick(ctx, stats, complete) }()
 	if !w.retryUnmap(ctx) {
 		return nil
 	}
 
-	if err := w.GraphData.Client.Sync(ctx); err != nil {
+	caughtUp, err := w.GraphData.Client.SyncPass(ctx)
+	if err != nil {
 		var mailErr *errx.MailError
 		if errors.As(err, &mailErr) {
 			return mailErr
@@ -42,7 +45,7 @@ func (w *WMail) SyncGraph(ctx context.Context) *errx.MailError {
 			return merr
 		}
 	}
-	w.endTick(stats)
+	complete = caughtUp
 	return nil
 }
 
@@ -186,6 +189,7 @@ func (w *WMail) graphBackfill(ctx context.Context, stats *tickStats) *errx.MailE
 					return mailErr
 				}
 				w.CaptureError(err)
+				stats.aborted = true
 				return nil
 			}
 			for _, full := range msgs {

internal/app/worker/wmail/sync_imap.go

@@ -32,6 +32,8 @@ func (w *WMail) Sync(ctx context.Context) *errx.MailError {
 	}
 	w.beginTick()
 	stats := &tickStats{}
+	complete := false
+	defer func() { w.endTick(ctx, stats, complete) }()
 	if !w.retryUnmap(ctx) {
 		return nil
 	}
@@ -78,6 +80,7 @@ func (w *WMail) Sync(ctx context.Context) *errx.MailError {
 	// mod-sequences where the server has CONDSTORE, UIDNEXT where it does not
 	// (Outlook.com, Microsoft 365 over IMAP, Yahoo, many hosted servers).
 	condStore := client.HasCondStore()
+	caughtUp := true
 
 	for i := range folders {
 		box := &folders[i]
@@ -136,6 +139,9 @@ func (w *WMail) Sync(ctx context.Context) *errx.MailError {
 			// The pass was aborted before this folder; hold its cursor too.
 			fullyProcessed = false
 		}
+		if !fullyProcessed {
+			caughtUp = false
+		}
 
 		if changed || !slices.Equal(befBox.Attrs, box.Attrs) {
 			// The stored cursor only moves once every change up to it was
@@ -238,7 +244,7 @@ outer:
 		}
 	}
 
-	w.endTick(stats)
+	complete = caughtUp
 	return nil
 }
 

internal/app/worker/wmail/sync_imap_flags.go

@@ -73,7 +73,8 @@ func (w *WMail) imapScanFlags(ctx context.Context, box *models.Mailbox, stats *t
 				return err
 			}
 			if stats.aborted {
-				break
+				scan.at = now
+				return nil
 			}
 		}
 	}

internal/app/worker/wmail/sync_state.go

@@ -16,10 +16,11 @@ const syncStateHeartbeat = 10 * time.Minute
 // of the tick; the consumer persists it and the loader hands it back on the
 // next assignment.
 type syncTracker struct {
-	state    models.SyncState
-	dirty    bool
-	lastSent time.Time
-	emit     func(models.SyncState) error
+	state        models.SyncState
+	dirty        bool
+	lastSent     time.Time
+	emit         func(models.SyncState) error
+	tickComplete bool
 }
 
 func newSyncTracker(seed *models.SyncState, emit func(models.SyncState) error) *syncTracker {
@@ -85,6 +86,11 @@ func (t *syncTracker) setFoldersSkipped(cap, conflict int) {
 func (t *syncTracker) touch(now time.Time) {
 	stamp := now
 	t.state.LastSyncedAt = &stamp
+	t.flush(now)
+}
+
+// flush publishes safe progress without implying a successful provider check.
+func (t *syncTracker) flush(now time.Time) {
 	if !t.dirty && now.Sub(t.lastSent) < syncStateHeartbeat {
 		return
 	}

internal/app/worker/wmail/sync_truth_test.go

@@ -0,0 +1,355 @@
+package wmail
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/warmbly/warmbly/internal/client/goog"
+	"github.com/warmbly/warmbly/internal/client/msgraph"
+	"github.com/warmbly/warmbly/internal/client/smtpimap/imap"
+	"github.com/warmbly/warmbly/internal/config"
+	"github.com/warmbly/warmbly/internal/errx"
+	"github.com/warmbly/warmbly/internal/models"
+	"github.com/warmbly/warmbly/internal/repository"
+)
+
+func seedSyncTruth(w *WMail, relayed *[]models.SyncState) time.Time {
+	previous := time.Now().Add(-time.Hour)
+	w.tracker = newSyncTracker(&models.SyncState{BackfillStatus: models.SyncBackfillComplete, LastSyncedAt: &previous, Deferred: 5}, func(st models.SyncState) error {
+		*relayed = append(*relayed, st)
+		return nil
+	})
+	return previous
+}
+
+func assertNoSyncSuccess(t *testing.T, w *WMail, previous time.Time) {
+	t.Helper()
+	if w.tracker.state.LastSyncedAt == nil || !w.tracker.state.LastSyncedAt.Equal(previous) || w.tracker.state.Deferred != 5 {
+		t.Fatalf("incomplete pass replaced successful evidence/backlog: %+v", w.tracker.state)
+	}
+}
+
+func assertSyncCatchUp(t *testing.T, w *WMail, previous time.Time) {
+	t.Helper()
+	if w.tracker.state.LastSyncedAt == nil || !w.tracker.state.LastSyncedAt.After(previous) || w.tracker.state.Deferred != 0 {
+		t.Fatalf("successful catch-up did not replace old evidence: %+v", w.tracker.state)
+	}
+}
+
+func TestSyncTruthGoogleFailureHeartbeatAndRecovery(t *testing.T) {
+	failing := true
+	srv := httptest.NewServer(http.HandlerFunc(func(out http.ResponseWriter, _ *http.Request) {
+		out.Header().Set("Content-Type", "application/json")
+		if failing {
+			out.WriteHeader(http.StatusServiceUnavailable)
+			_, _ = out.Write([]byte(`{"error":{"code":503,"message":"retry later"}}`))
+			return
+		}
+		_, _ = out.Write([]byte(`{"historyId":"200"}`))
+	}))
+	defer srv.Close()
+	var events []captured
+	w := newGoogleTestMail(t, srv, &events)
+	w.GoogleData.LastHistoryID = 100
+	var relayed []models.SyncState
+	previous := seedSyncTruth(w, &relayed)
+	for range 2 {
+		w.tracker.lastSent = time.Now().Add(-2 * syncStateHeartbeat)
+		if err := w.SyncGoogle(t.Context()); err == nil {
+			t.Fatal("expected provider error")
+		}
+		assertNoSyncSuccess(t, w, previous)
+	}
+	if len(relayed) != 2 {
+		t.Fatalf("failed passes must relay safe heartbeat state: %d", len(relayed))
+	}
+	for _, st := range relayed {
+		if !st.LastSyncedAt.Equal(previous) || st.Deferred != 5 {
+			t.Fatalf("false success escaped via heartbeat: %+v", st)
+		}
+	}
+	failing = false
+	if err := w.SyncGoogle(t.Context()); err != nil {
+		t.Fatal(err)
+	}
+	assertSyncCatchUp(t, w, previous)
+}
+
+type syncTruthMap struct {
+	repository.EmailMessageMapRepository
+	failGet bool
+	failDel bool
+}
+
+func (r *syncTruthMap) Get(context.Context, uuid.UUID, uuid.UUID, string) (*repository.EmailMessageData, error) {
+	if r.failGet {
+		return nil, errors.New("map unavailable")
+	}
+	return nil, nil
+}
+
+func (r *syncTruthMap) Del(context.Context, uuid.UUID, uuid.UUID, string, uuid.UUID) error {
+	if r.failDel {
+		return errors.New("unmap unavailable")
+	}
+	return nil
+}
+
+func TestSyncTruthGraphAbortedAdmissionAndCheckpoint(t *testing.T) {
+	for _, failure := range []string{"map", "checkpoint"} {
+		t.Run(failure, func(t *testing.T) {
+			g := newFakeGraph()
+			var events []captured
+			var relayed []models.SyncState
+			w := newGraphTestMail(t, g.serve(t), &events, &relayed)
+			previous := seedSyncTruth(w, &relayed)
+			before := w.GraphData.Client.DeltaLinks[msgraph.FolderInbox]
+			r := &syncTruthMap{failGet: failure == "map"}
+			w.EmailMessageMapRepository = r
+			failing := true
+			w.onEvent = func(kind models.JobEventType, _ any) error {
+				if failing && failure == "checkpoint" && kind == models.JobEventTypeGraphDeltaUpdate {
+					return errors.New("checkpoint unavailable")
+				}
+				return nil
+			}
+			for range 2 {
+				if failure == "map" {
+					g.live = []string{"new-arrival"}
+				}
+				w.tracker.lastSent = time.Now().Add(-2 * syncStateHeartbeat)
+				_ = w.SyncGraph(t.Context())
+				assertNoSyncSuccess(t, w, previous)
+				if w.GraphData.Client.DeltaLinks[msgraph.FolderInbox] != before {
+					t.Fatal("failed/aborted page advanced local checkpoint")
+				}
+			}
+			failing, r.failGet = false, false
+			if err := w.SyncGraph(t.Context()); err != nil {
+				t.Fatal(err)
+			}
+			assertSyncCatchUp(t, w, previous)
+		})
+	}
+}
+
+func TestSyncTruthRetryUnmapAndCancellation(t *testing.T) {
+	for _, provider := range []string{"google", "graph", "imap"} {
+		t.Run(provider, func(t *testing.T) {
+			w, _ := newIMAPTestMail(&fakeImapConn{}, &fixedBudget{}, &models.Mailbox{})
+			var relayed []models.SyncState
+			previous := seedSyncTruth(w, &relayed)
+			w.unmapPending = map[string]uuid.UUID{"failed-arrival": uuid.New()}
+			w.EmailMessageMapRepository = &syncTruthMap{failDel: true}
+			switch provider {
+			case "google":
+				_ = w.SyncGoogle(t.Context())
+			case "graph":
+				_ = w.SyncGraph(t.Context())
+			case "imap":
+				_ = w.Sync(t.Context())
+			}
+			assertNoSyncSuccess(t, w, previous)
+			if len(relayed) != 1 {
+				t.Fatal("retry-unmap blockage did not relay safe state")
+			}
+		})
+	}
+	w, _ := newIMAPTestMail(&fakeImapConn{}, &fixedBudget{}, &models.Mailbox{})
+	var relayed []models.SyncState
+	previous := seedSyncTruth(w, &relayed)
+	ctx, cancel := context.WithCancel(t.Context())
+	cancel()
+	_ = w.Sync(ctx)
+	assertNoSyncSuccess(t, w, previous)
+}
+
+func TestSyncTruthImapFlagFailureRetainsBothUpdatesAndSuccessEvidence(t *testing.T) {
+	box := &models.Mailbox{Name: "INBOX", UIDValidity: 7, UIDNext: 101}
+	conn := &fakeImapConn{noCondStore: true, folders: []models.Mailbox{*box}, flags: map[uint32]imap.FlagState{
+		1: {MessageID: "one", Flags: nil}, 2: {MessageID: "two", Flags: nil},
+	}}
+	w, events := newIMAPTestMail(conn, &fixedBudget{allow: 10}, box)
+	w.EmailMessageMapRepository = knownMessageMap{id: uuid.NewString()}
+	if err := w.Sync(t.Context()); err != nil {
+		t.Fatal(err)
+	}
+	var relayed []models.SyncState
+	previous := seedSyncTruth(w, &relayed)
+	conn.flags[1], conn.flags[2] = imap.FlagState{MessageID: "one", Flags: []string{models.FlagSeen}}, imap.FlagState{MessageID: "two", Flags: []string{models.FlagSeen}}
+	w.flagScan[box.Name].at = time.Now().Add(-2 * config.ImapFlagScanInterval)
+	failing := true
+	w.onEvent = func(kind models.JobEventType, body any) error {
+		if kind == models.JobEventTypeEmailUpdate && failing {
+			return errors.New("publisher unavailable")
+		}
+		*events = append(*events, captured{eventType: kind, body: body})
+		return nil
+	}
+	if err := w.Sync(t.Context()); err != nil {
+		t.Fatal(err)
+	}
+	assertNoSyncSuccess(t, w, previous)
+	for uid := range conn.flags {
+		if w.flagScan[box.Name].flags[uid] != flagFingerprint(nil) {
+			t.Fatal("failed/unvisited flags were absorbed into baseline")
+		}
+	}
+	failing = false
+	w.flagScan[box.Name].at = time.Now().Add(-2 * config.ImapFlagScanInterval)
+	if err := w.Sync(t.Context()); err != nil {
+		t.Fatal(err)
+	}
+	updates := 0
+	for _, e := range *events {
+		if e.eventType == models.JobEventTypeEmailUpdate {
+			updates++
+		}
+	}
+	if updates != 2 {
+		t.Fatalf("unchanged provider flags did not retry both updates: %d", updates)
+	}
+	assertSyncCatchUp(t, w, previous)
+}
+
+func TestSyncTruthPanicDoesNotResetTransportFailure(t *testing.T) {
+	w, _ := newIMAPTestMail(&fakeImapConn{}, &fixedBudget{}, &models.Mailbox{})
+	w.EmailType = models.InboxProviderGoogle
+	w.GoogleData = nil // Fault inside the provider pass, after beginTick.
+	w.transportFailures = 2
+	var relayed []models.SyncState
+	previous := seedSyncTruth(w, &relayed)
+	if err := w.syncOnce(t.Context()); err == nil {
+		t.Fatal("recovered panic returned successful outcome")
+	}
+	assertNoSyncSuccess(t, w, previous)
+	if w.transportFailures != 2 {
+		t.Fatal("panic reset existing transport warning/backoff")
+	}
+}
+
+func TestSyncTruthBoundedAndPinnedProviderPasses(t *testing.T) {
+	for _, provider := range []string{"google", "graph"} {
+		for _, outcome := range []string{"bounded", "pinned", "stop"} {
+			t.Run(provider+"/"+outcome, func(t *testing.T) {
+				caughtUp := false
+				srv := httptest.NewServer(http.HandlerFunc(func(out http.ResponseWriter, r *http.Request) {
+					out.Header().Set("Content-Type", "application/json")
+					if provider == "google" {
+						if caughtUp {
+							_, _ = out.Write([]byte(`{"historyId":"200"}`))
+							return
+						}
+						if outcome == "bounded" {
+							_, _ = out.Write([]byte(`{"historyId":"200","nextPageToken":"next"}`))
+							return
+						}
+						_, _ = out.Write([]byte(`{"historyId":"200","history":[{"id":"150","messagesAdded":[{"message":{"id":"new-arrival","threadId":"thread"}}]}]}`))
+						return
+					}
+					folder := msgraph.FolderInbox
+					if caughtUp {
+						_, _ = fmt.Fprintf(out, `{"value":[],"@odata.deltaLink":"https://graph.microsoft.com/v1.0/me/mailFolders/%s/messages/delta?final=1"}`, folder)
+						return
+					}
+					if outcome == "bounded" {
+						_, _ = fmt.Fprintf(out, `{"value":[],"@odata.nextLink":"https://graph.microsoft.com%s?next=1"}`, r.URL.Path)
+						return
+					}
+					_, _ = fmt.Fprintf(out, `{"value":[{"id":"new-arrival","isRead":false}],"@odata.deltaLink":"https://graph.microsoft.com/v1.0/me/mailFolders/%s/messages/delta?final=1"}`, folder)
+				}))
+				defer srv.Close()
+				var events []captured
+				var relayed []models.SyncState
+				var w *WMail
+				var sync func(context.Context) *errx.MailError
+				if provider == "google" {
+					w = newGoogleTestMail(t, srv, &events)
+					w.GoogleData.LastHistoryID = 100
+					w.GoogleData.Client.OnMessageAdded = func(context.Context, string, string) (bool, error) {
+						if outcome == "stop" {
+							return false, goog.ErrStop
+						}
+						return false, nil
+					}
+					sync = w.SyncGoogle
+				} else {
+					w = newGraphTestMail(t, srv, &events, &relayed)
+					w.GraphData.Client.OnMessageSeen = func(context.Context, string, string, bool) (bool, error) {
+						if outcome == "stop" {
+							return false, msgraph.ErrStop
+						}
+						return false, nil
+					}
+					sync = w.SyncGraph
+				}
+				previous := seedSyncTruth(w, &relayed)
+				w.transportFailures = 2
+				if err := sync(t.Context()); err != nil {
+					t.Fatal(err)
+				}
+				assertNoSyncSuccess(t, w, previous)
+				caughtUp = true
+				if err := sync(t.Context()); err != nil {
+					t.Fatal(err)
+				}
+				assertSyncCatchUp(t, w, previous)
+			})
+		}
+	}
+}
+
+func TestSyncTruthGoogleCheckpointFailureAndDirtyRelayRetry(t *testing.T) {
+	failingProvider := false
+	srv := httptest.NewServer(http.HandlerFunc(func(out http.ResponseWriter, _ *http.Request) {
+		out.Header().Set("Content-Type", "application/json")
+		if failingProvider {
+			out.WriteHeader(503)
+			_, _ = out.Write([]byte(`{"error":{"code":503,"message":"injected"}}`))
+			return
+		}
+		_, _ = out.Write([]byte(`{"historyId":"200"}`))
+	}))
+	defer srv.Close()
+	var events []captured
+	w := newGoogleTestMail(t, srv, &events)
+	w.GoogleData.LastHistoryID = 100
+	var relayed []models.SyncState
+	previous := seedSyncTruth(w, &relayed)
+	w.transportFailures = 2
+	w.onEvent = func(kind models.JobEventType, _ any) error {
+		if kind == models.JobEventTypeHistoryIDUpdate {
+			return errors.New("checkpoint unavailable")
+		}
+		return nil
+	}
+	if err := w.syncOnce(t.Context()); err != nil {
+		t.Fatal(err)
+	}
+	assertNoSyncSuccess(t, w, previous)
+	if w.GoogleData.LastHistoryID != 100 || w.transportFailures != 2 {
+		t.Fatal("failed checkpoint was installed or claimed recovery")
+	}
+	failingProvider = true
+	expired := time.Now().Add(-time.Minute)
+	w.tracker.state.ThrottledUntil = &expired
+	w.tracker.lastSent = time.Now().Add(-2 * syncStateHeartbeat)
+	w.tracker.emit = func(st models.SyncState) error { return errors.New("state relay unavailable") }
+	_ = w.SyncGoogle(t.Context())
+	assertNoSyncSuccess(t, w, previous)
+	if !w.tracker.dirty || w.tracker.state.ThrottledUntil != nil {
+		t.Fatal("failed relay forgot dirty throttle progress")
+	}
+	w.tracker.emit = func(st models.SyncState) error { relayed = append(relayed, st); return nil }
+	_ = w.SyncGoogle(t.Context())
+	assertNoSyncSuccess(t, w, previous)
+	if w.tracker.dirty || relayed[len(relayed)-1].ThrottledUntil != nil {
+		t.Fatal("safe dirty-state relay did not retry")
+	}
+}

internal/client/goog/history.go

@@ -14,6 +14,8 @@ import (
 // returns the checkpoint reached so far and a nil error.
 var ErrStop = errors.New("goog: stop")
 
+var errHistoryIncomplete = errors.New("goog: history pass incomplete")
+
 var ErrHistoryExpired = errors.New("goog: history expired; resynchronization required")
 
 var ErrRecoveryPageExpired = errors.New("goog: recovery page token expired")
@@ -35,6 +37,20 @@ const historyPagesPerPass = 10
 // continues, so later records that are admitted (a reply, say) still land and
 // the deferred ones are re-offered next tick.
 func (c *Client) FetchHistory(ctx context.Context, lastHistoryID uint64) (uint64, error) {
+	checkpoint, _, err := c.FetchHistoryPass(ctx, lastHistoryID)
+	return checkpoint, err
+}
+
+// FetchHistoryPass distinguishes a bounded or pinned walk from full live catch-up.
+func (c *Client) FetchHistoryPass(ctx context.Context, lastHistoryID uint64) (uint64, bool, error) {
+	checkpoint, err := c.fetchHistory(ctx, lastHistoryID)
+	if errors.Is(err, errHistoryIncomplete) {
+		return checkpoint, false, nil
+	}
+	return checkpoint, err == nil, err
+}
+
+func (c *Client) fetchHistory(ctx context.Context, lastHistoryID uint64) (uint64, error) {
 	if lastHistoryID == 0 {
 		return c.HistoryBaseline(ctx)
 	}
@@ -62,7 +78,7 @@ func (c *Client) FetchHistory(ctx context.Context, lastHistoryID uint64) (uint64
 				}
 				added, err := c.OnMessageAdded(ctx, m.Message.Id, m.Message.ThreadId)
 				if errors.Is(err, ErrStop) {
-					return checkpoint, nil
+					return checkpoint, errHistoryIncomplete
 				}
 				if err != nil {
 					return checkpoint, err
@@ -96,13 +112,14 @@ func (c *Client) FetchHistory(ctx context.Context, lastHistoryID uint64) (uint64
 		if resp.NextPageToken == "" {
 			if !pinned {
 				checkpoint = resp.HistoryId
+				return checkpoint, nil
 			}
-			break
+			return checkpoint, errHistoryIncomplete
 		}
 		call.PageToken(resp.NextPageToken)
 	}
 
-	return checkpoint, nil
+	return checkpoint, errHistoryIncomplete
 }
 
 func (c *Client) HistoryBaseline(ctx context.Context) (uint64, error) {

internal/client/msgraph/delta.go

@@ -19,6 +19,8 @@ func itoa(n int) string { return strconv.Itoa(n) }
 // where it was last persisted.
 var ErrStop = errors.New("msgraph: stop")
 
+var errDeltaIncomplete = errors.New("msgraph: delta pass incomplete")
+
 // deltaSelect keeps delta pages light: we only need the id, read state, and the
 // @removed marker to decide add/update vs remove. Full envelope + body + headers
 // are hydrated per admitted message via FetchMessage.
@@ -64,15 +66,26 @@ var BackfillFolders = []string{FolderInbox, FolderSent, FolderArchive, FolderDra
 // fits the disposable worker natively: no webhook endpoint, no subscription
 // lifecycle, just a cursor the control plane persists via OnDelta.
 func (c *Client) Sync(ctx context.Context) error {
+	_, err := c.SyncPass(ctx)
+	return err
+}
+
+// SyncPass reports catch-up only when every tracked folder reaches an unpinned delta link.
+func (c *Client) SyncPass(ctx context.Context) (bool, error) {
+	complete := true
 	for _, folder := range TrackedFolders {
 		if err := c.syncFolder(ctx, folder); err != nil {
 			if errors.Is(err, ErrStop) {
-				return nil
+				return false, nil
 			}
-			return err
+			if errors.Is(err, errDeltaIncomplete) {
+				complete = false
+				continue
+			}
+			return false, err
 		}
 	}
-	return nil
+	return complete, nil
 }
 
 func (c *Client) syncFolder(ctx context.Context, folder string) error {
@@ -97,7 +110,7 @@ func (c *Client) syncFolder(ctx context.Context, folder string) error {
 			case pg.DeltaLink != "":
 				return c.saveCursor(ctx, folder, pg.DeltaLink)
 			default:
-				return nil
+				return errDeltaIncomplete
 			}
 		}
 	}
@@ -139,19 +152,21 @@ func (c *Client) syncFolder(ctx context.Context, folder string) error {
 			if !pinned {
 				return c.saveCursor(ctx, folder, pg.DeltaLink)
 			}
-			return nil
+			return errDeltaIncomplete
 		default:
-			return nil
+			return errDeltaIncomplete
 		}
 	}
-	return nil
+	return errDeltaIncomplete
 }
 
 func (c *Client) saveCursor(ctx context.Context, folder, link string) error {
-	c.DeltaLinks[folder] = link
 	if c.OnDelta != nil {
-		return c.OnDelta(ctx, folder, link)
+		if err := c.OnDelta(ctx, folder, link); err != nil {
+			return err
+		}
 	}
+	c.DeltaLinks[folder] = link
 	return nil
 }
 

internal/repository/pg_unibox.go

@@ -39,6 +39,7 @@ type UniboxRepository interface {
 	UpdateEntry(ctx context.Context, userID, emailID, id uuid.UUID, e *UpdateUniboxEntry) error
 	GetIncoming(ctx context.Context, userID uuid.UUID, limit int, cursor string) (*models.MailSearchResult, error)
 	GetByID(ctx context.Context, userID, id uuid.UUID) (*models.EmailMessageStoreData, error)
+	GetForSync(ctx context.Context, userID, emailID, id uuid.UUID) (*models.EmailMessageStoreData, error)
 	// GetByIDForOrg is the org-scoped read for the unibox detail view: any
 	// member with unibox access can open a message in the org-wide list. It
 	// returns the row plus the mailbox OWNER's user_id, which the S3 body key
@@ -455,14 +456,35 @@ func (r *uniboxRepository) GetIncoming(ctx context.Context, userID uuid.UUID, li
 var ErrEmailNotFound = errors.New("email not found")
 
 func (r *uniboxRepository) GetByID(ctx context.Context, userID, id uuid.UUID) (*models.EmailMessageStoreData, error) {
+	e, err := r.getMessage(ctx, userID, uuid.Nil, id)
+	if err != nil {
+		return nil, err
+	}
+	if !e.Seen {
+		_ = r.MarkSeen(ctx, userID, id, true)
+		e.Seen = true
+	}
+	return e, nil
+}
+
+// GetForSync reads provider state without treating background work as a user view.
+func (r *uniboxRepository) GetForSync(ctx context.Context, userID, emailID, id uuid.UUID) (*models.EmailMessageStoreData, error) {
+	if emailID == uuid.Nil {
+		return nil, ErrEmailNotFound
+	}
+	return r.getMessage(ctx, userID, emailID, id)
+}
+
+func (r *uniboxRepository) getMessage(ctx context.Context, userID, emailID, id uuid.UUID) (*models.EmailMessageStoreData, error) {
 	query := fmt.Sprintf(`
 		SELECT %s
 		FROM unibox_emails
 		WHERE user_id = $1 AND id = $2
+		  AND ($3::uuid = '00000000-0000-0000-0000-000000000000' OR email_id = $3)
 	`, strings.Join(mailFieldsFull, ", "))
 
 	var e models.EmailMessageStoreData
-	err := r.db.QueryRow(ctx, query, userID, id).Scan(
+	err := r.db.QueryRow(ctx, query, userID, id, emailID).Scan(
 		&e.ID, &e.EmailID, &e.Mailbox, &e.FolderPath, &e.ThreadID, &e.MessageID,
 		&e.GmailID, &e.ParentID, &e.UID, &e.ModSeq,
 		&e.Flags, &e.BCC, &e.CC, &e.FromAddr, &e.InReplyTo, &e.ReplyTo,
@@ -476,12 +498,6 @@ func (r *uniboxRepository) GetByID(ctx context.Context, userID, id uuid.UUID) (*
 		return nil, err
 	}
 
-	// Auto-mark as seen
-	if !e.Seen {
-		_ = r.MarkSeen(ctx, userID, id, true)
-		e.Seen = true
-	}
-
 	return &e, nil
 }
 

web/src/components/app/emails/SyncStatusCard.test.tsx

@@ -0,0 +1,70 @@
+import { act, cleanup, render, screen } from "@testing-library/react";
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+import type EmailSync from "@/lib/api/models/app/emails/SyncState";
+import SyncStatusCard from "./SyncStatusCard";
+
+const fixtures = vi.hoisted(() => ({ data: undefined as EmailSync | undefined }));
+vi.mock("@/lib/api/hooks/app/emails/useSync", () => ({ default: () => ({ data: fixtures.data, isPending: false }) }));
+vi.mock("@/lib/api/hooks/app/emails/useUpdateSyncSkipFolders", () => ({ default: () => ({}) }));
+vi.mock("@/hooks/context/confirm", () => ({ useConfirm: () => ({}) }));
+
+describe("truthful mailbox sync status", () => {
+    beforeEach(() => {
+        vi.useFakeTimers();
+        vi.setSystemTime(new Date("2026-10-09T12:00:00Z"));
+        fixtures.data = {
+            state: { backfill_status: "complete", backfill_synced: 42, deferred: 0 },
+            policy: { backfill_days: 30, backfill_messages: 2000, daily_messages: 500, org_daily_messages: 2000 },
+            folders: [], skip_folders: [],
+        };
+    });
+    afterEach(() => { cleanup(); vi.useRealTimers(); });
+    const draw = () => render(<SyncStatusCard mailboxId="mailbox" />);
+
+    it("does not treat a completed historical import as proof of live health", () => {
+        draw();
+        expect(screen.getByText("Recent-mail import complete")).toBeInTheDocument();
+        expect(screen.getByText("No successful sync check reported yet.")).toBeInTheDocument();
+        expect(screen.getByText(/Import completion does not confirm live sync is current/)).toBeInTheDocument();
+        expect(screen.queryByText(/Up to date/i)).not.toBeInTheDocument();
+    });
+
+    it("ages a recent successful check into stale evidence without a new server payload", () => {
+        fixtures.data!.state!.last_synced_at = new Date("2026-10-09T11:59:00Z");
+        draw();
+        expect(screen.getByText("Last successful sync 1 min ago.")).toBeInTheDocument();
+        act(() => { vi.advanceTimersByTime(21 * 60_000); });
+        expect(screen.getByText("No recent successful sync check")).toBeInTheDocument();
+        expect(screen.getByText(/Recovery is not yet confirmed by a recent successful check/)).toBeInTheDocument();
+        expect(screen.queryByText(/Up to date/i)).not.toBeInTheDocument();
+    });
+
+    it("shows ongoing recovery from the existing cursor instead of historical completion", () => {
+        fixtures.data!.state!.backfill_cursor = { google_recovery: { history_id: 100, since: new Date(), messages_done: false } };
+        draw();
+        expect(screen.getByText("Recovering mailbox sync")).toBeInTheDocument();
+        expect(screen.queryByText("Recent-mail import complete")).not.toBeInTheDocument();
+    });
+
+    it("keeps deferred evidence visible after the budget hold expires", () => {
+        fixtures.data!.state!.deferred = 5;
+        fixtures.data!.state!.throttled_until = new Date("2026-10-09T12:01:00Z");
+        const view = draw();
+        expect(screen.getByText(/Waiting on the sync budget until/)).toBeInTheDocument();
+        act(() => { vi.advanceTimersByTime(2 * 60_000); });
+        expect(screen.getByText("New mail was waiting on sync at the last check")).toBeInTheDocument();
+        expect(screen.getByText(/Last observed backlog: 5 messages/)).toBeInTheDocument();
+        fixtures.data!.state!.deferred = 0;
+        fixtures.data!.state!.last_synced_at = new Date();
+        view.rerender(<SyncStatusCard mailboxId="mailbox" />);
+        expect(screen.getByText("Last successful sync just now.")).toBeInTheDocument();
+        expect(screen.queryByText(/Last observed backlog/)).not.toBeInTheDocument();
+    });
+
+    it("describes import progress separately from the last successful live check", () => {
+        fixtures.data!.state!.backfill_status = "running";
+        draw();
+        expect(screen.getByText("Importing recent mail: 42 messages so far")).toBeInTheDocument();
+        expect(screen.getByText("No successful sync check reported yet.")).toBeInTheDocument();
+    });
+});

web/src/components/app/emails/SyncStatusCard.tsx

@@ -1,4 +1,4 @@
-import { useState } from "react";
+import { useEffect, useState } from "react";
 import { motion } from "framer-motion";
 import { CheckCircle2Icon, DownloadIcon, HourglassIcon, PlusIcon, RefreshCwIcon } from "lucide-react";
 import toast from "react-hot-toast/headless";
@@ -25,8 +25,8 @@ const REASON_COPY: Record<SyncThrottleReason, string> = {
     priority_daily: "the daily limit for replies was reached",
 };
 
-function relative(iso: string | Date): string {
-    const diff = Date.now() - new Date(iso).getTime();
+function relative(iso: string | Date, now: number): string {
+	const diff = now - new Date(iso).getTime();
     const m = Math.round(diff / 60_000);
     if (m < 1) return "just now";
     if (m < 60) return `${m} min ago`;
@@ -148,6 +148,11 @@ function SkipFoldersSection({ mailboxId, listed, skipped }: { mailboxId: string;
 }
 
 export default function SyncStatusCard({ mailboxId, provider }: { mailboxId: string; provider?: string }) {
+	const [now, setNow] = useState(Date.now);
+	useEffect(() => {
+		const timer = setInterval(() => setNow(Date.now()), 60_000);
+		return () => clearInterval(timer);
+	}, []);
     const sync = useSync(mailboxId);
     const state = sync.data?.state ?? null;
     const policy = sync.data?.policy;
@@ -162,15 +167,19 @@ export default function SyncStatusCard({ mailboxId, provider }: { mailboxId: str
     }
     if (!sync.data) return null;
 
-    const throttled = !!state?.throttled_until && new Date(state.throttled_until).getTime() > Date.now();
+    const throttled = !!state?.throttled_until && new Date(state.throttled_until).getTime() > now;
+    const lastSuccess = state?.last_synced_at && new Date(state.last_synced_at).getTime();
+    // The worker relays successful checks at least every ten minutes.
+    const stale = !!lastSuccess && now - lastSuccess > 20 * 60_000;
+    const recovering = !!state?.backfill_cursor?.google_recovery;
     const status = state?.backfill_status ?? "pending";
     const cap = policy?.backfill_messages ?? 0;
     const synced = state?.backfill_synced ?? 0;
     const pct = cap > 0 ? Math.min(100, Math.round((synced / cap) * 100)) : 0;
 
     let headline: React.ReactNode;
-    let Icon = CheckCircle2Icon;
-    let tone = "text-emerald-600";
+    let Icon = RefreshCwIcon;
+    let tone = "text-slate-500";
     if (throttled && state?.throttled_until) {
         Icon = HourglassIcon;
         tone = "text-amber-600";
@@ -181,8 +190,19 @@ export default function SyncStatusCard({ mailboxId, provider }: { mailboxId: str
                 {reason ? <span className="text-slate-500"> ({reason})</span> : null}
             </>
         );
+    } else if (recovering) {
+        tone = "text-sky-600";
+        headline = "Recovering mailbox sync";
+    } else if ((state?.deferred ?? 0) > 0) {
+        Icon = HourglassIcon;
+        tone = "text-amber-600";
+        headline = "New mail was waiting on sync at the last check";
+    } else if (stale) {
+        tone = "text-amber-600";
+        headline = "No recent successful sync check";
     } else if (status === "complete") {
-        headline = state?.last_synced_at ? `Up to date, last checked ${relative(state.last_synced_at)}` : "Up to date";
+        Icon = CheckCircle2Icon;
+        headline = "Recent-mail import complete";
     } else if (status === "running") {
         Icon = DownloadIcon;
         tone = "text-sky-600";
@@ -214,16 +234,23 @@ export default function SyncStatusCard({ mailboxId, provider }: { mailboxId: str
 
             <p className="mt-2 text-[11.5px] leading-relaxed text-slate-500">
                 {status === "complete" && policy
-                    ? `Imported ${synced.toLocaleString()} message${synced === 1 ? "" : "s"} from the last ${policy.backfill_days} days. New mail syncs as it arrives.`
+                    ? `Imported ${synced.toLocaleString()} message${synced === 1 ? "" : "s"} from the configured ${policy.backfill_days}-day window. Import completion does not confirm live sync is current.`
                     : policy
                         ? `The last ${policy.backfill_days} days come in newest first, up to ${cap.toLocaleString()} messages. New mail syncs alongside.`
                         : null}
-                {throttled ? " Replies to your outreach keep syncing; the rest resumes automatically." : null}
+                {throttled ? " Mail is retried automatically within its sync budget." : null}
+            </p>
+
+            <p className="mt-1 text-[11.5px] text-slate-500">
+                {state?.last_synced_at
+                    ? `Last successful sync ${relative(state.last_synced_at, now)}.`
+                    : "No successful sync check reported yet."}
+                {stale ? " Recovery is not yet confirmed by a recent successful check." : null}
             </p>
 
             {(state?.deferred ?? 0) > 0 && (
                 <p className="mt-1 text-[11.5px] text-amber-700">
-                    {state!.deferred.toLocaleString()} message{state!.deferred === 1 ? "" : "s"} waiting on the server.
+                    Last observed backlog: {state!.deferred.toLocaleString()} message{state!.deferred === 1 ? "" : "s"} waiting on the server. The current total may have changed.
                 </p>
             )}
 

web/src/lib/api/hooks/app/emails/useSync.ts

@@ -12,8 +12,8 @@ export default function useSync(id: string, enabled = true) {
         enabled: !!id && enabled,
         staleTime: 15_000,
         refetchInterval: (query) => {
-            const status = query.state.data?.state?.backfill_status;
-            return status === "running" || status === "pending" ? 20_000 : false;
+            const state = query.state.data?.state;
+            return state?.backfill_status === "running" || state?.backfill_status === "pending" || state?.backfill_cursor?.google_recovery ? 20_000 : 60_000;
         },
     });
 }

web/src/lib/api/models/app/emails/SyncState.ts

@@ -12,6 +12,17 @@ export type SyncThrottleReason =
 
 export interface SyncState {
     backfill_status: SyncBackfillStatus;
+    backfill_cursor?: {
+        page_token?: string;
+        folders?: Record<string, { next?: string; uid?: number; done?: boolean }>;
+        google_recovery?: {
+            history_id: number;
+            since: Date | string;
+            page_token?: string;
+            messages_done: boolean;
+            stored_after?: string;
+        };
+    };
     backfill_synced: number;
     backfill_since?: Date;
     backfill_started_at?: Date;