diff --git a/src/cli/gmail/sync.rs b/src/cli/gmail/sync.rs
index c0152d56..55a69bb0 100644
--- a/src/cli/gmail/sync.rs
+++ b/src/cli/gmail/sync.rs
@@ -107,6 +107,7 @@ impl SyncCommand {
             },
             self.quiet,
             &self.output,
+            std::io::stderr().is_terminal(),
         )
         .await
     }
@@ -120,13 +121,19 @@ impl SyncCommand {
 /// render it, and only *then* decide the process exit condition — a
 /// non-empty `errors` becomes a failing exit code after everything already
 /// ran and printed, never silently.
+///
+/// Takes `stderr_is_terminal` as a value for the same reason
+/// [`should_show_progress`] does — so tests can force the live-progress
+/// branch without depending on the test runner's own (never-a-terminal)
+/// stderr.
 async fn run_sync_command(
     client: &GmailClient,
     opts: SyncOptions,
     quiet: bool,
     output: &OutputFormat,
+    stderr_is_terminal: bool,
 ) -> Result<()> {
-    let show_progress = should_show_progress(quiet, output, std::io::stderr().is_terminal());
+    let show_progress = should_show_progress(quiet, output, stderr_is_terminal);
     let report = if show_progress {
         let (tx, rx) = mpsc::unbounded_channel();
         let bars = SyncProgressBars::new();
@@ -548,6 +555,7 @@ mod tests {
             },
             false,
             &OutputFormat::Table,
+            false,
         )
         .await
         .unwrap();
@@ -594,12 +602,76 @@ mod tests {
             },
             false,
             &OutputFormat::Table,
+            false,
         )
         .await
         .unwrap_err();
         assert!(err.to_string().contains("1 message(s) failed"));
     }
 
+    #[tokio::test]
+    async fn run_sync_command_with_stderr_is_terminal_takes_the_live_progress_path() {
+        // `stderr_is_terminal: true` is what `should_show_progress` needs to
+        // pick the live-progress branch — the test runner's own stderr is
+        // never a terminal, hence injecting it here rather than relying on
+        // `std::io::IsTerminal` (see `run_sync_command`'s doc comment).
+        let server = wiremock::MockServer::start().await;
+        let client = client_with_bootstrapped_token(&server).await;
+        wiremock::Mock::given(wiremock::matchers::method("GET"))
+            .and(wiremock::matchers::path("/gmail/v1/users/me/profile"))
+            .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
+                "emailAddress": "user@example.com", "messagesTotal": 1, "threadsTotal": 1, "historyId": "1"
+            })))
+            .mount(&server)
+            .await;
+        wiremock::Mock::given(wiremock::matchers::method("GET"))
+            .and(wiremock::matchers::path("/gmail/v1/users/me/messages"))
+            .respond_with(
+                wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
+                    "messages": [{"id": "m1", "threadId": "t1"}]
+                })),
+            )
+            .mount(&server)
+            .await;
+        wiremock::Mock::given(wiremock::matchers::method("GET"))
+            .and(wiremock::matchers::path("/gmail/v1/users/me/messages/m1"))
+            .and(wiremock::matchers::query_param("format", "raw"))
+            .respond_with(
+                wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
+                    "id": "m1", "threadId": "t1", "labelIds": ["INBOX"],
+                    "internalDate": "1700000000000", "historyId": "500",
+                    "raw": base64::engine::general_purpose::URL_SAFE_NO_PAD
+                        .encode("From: a@example.com\r\n\r\nBody of m1."),
+                })),
+            )
+            .mount(&server)
+            .await;
+
+        let dir = tempfile::tempdir().unwrap();
+        let output_dir = dir.path().join("archive");
+        run_sync_command(
+            &client,
+            SyncOptions {
+                output_dir: output_dir.clone(),
+                query: None,
+                full: false,
+                concurrency: 4,
+                dry_run: false,
+                extract_attachments: false,
+            },
+            false,
+            &OutputFormat::Table,
+            true,
+        )
+        .await
+        .unwrap();
+
+        assert!(
+            output_dir.exists(),
+            "the live-progress run must still archive the message"
+        );
+    }
+
     #[tokio::test]
     async fn execute_passes_flags_through() {
         let server = wiremock::MockServer::start().await;
@@ -722,6 +794,7 @@ Content-Disposition: attachment; filename=\"report.pdf\"\r\n\
             },
             false,
             &OutputFormat::Table,
+            false,
         )
         .await
         .unwrap();
@@ -750,6 +823,7 @@ Content-Disposition: attachment; filename=\"report.pdf\"\r\n\
             },
             false,
             &OutputFormat::Table,
+            false,
         )
         .await
         .unwrap();
diff --git a/src/cli/gmail/sync/engine.rs b/src/cli/gmail/sync/engine.rs
index fd2bcaaf..a47e9a1d 100644
--- a/src/cli/gmail/sync/engine.rs
+++ b/src/cli/gmail/sync/engine.rs
@@ -1588,6 +1588,81 @@ not-really-a-pdf\r\n\
         assert!(err.to_string().contains("Failed to parse manifest"));
     }
 
+    // ── `--full` forces reconciliation even with a valid watermark ──────
+
+    #[tokio::test]
+    async fn run_sync_full_flag_forces_reconciliation_even_with_a_valid_watermark() {
+        let server = wiremock::MockServer::start().await;
+        let client = client_with_bootstrapped_token(&server).await;
+        let dir = tempfile::tempdir().unwrap();
+        let output_dir = dir.path().join("archive");
+        std::fs::create_dir_all(&output_dir).unwrap();
+        state::save(
+            &ArchiveState {
+                history_id: "100".to_string(),
+                email_address: "user@example.com".to_string(),
+                last_sync: Utc::now(),
+                query: None,
+            },
+            &state_path(&output_dir),
+        )
+        .unwrap();
+
+        mount_profile(&server, "user@example.com", "999").await;
+        mount_message_list(&server, &["m1"]).await;
+        mount_raw_get(&server, "m1", "Hello").await;
+        // Deliberately no mock for GET .../history — `--full` must take the
+        // full-listing path even though a valid watermark exists, never
+        // fall through to `run_incremental`.
+
+        let mut o = opts(output_dir.clone());
+        o.full = true;
+        let report = run_sync(&client, &o).await.unwrap();
+
+        assert!(report.errors.is_empty());
+        assert!(report
+            .actions
+            .iter()
+            .any(|a| matches!(a, SyncAction::Fetched { id, .. } if id == "m1")));
+        match state::load(&state_path(&output_dir)) {
+            LoadOutcome::Present(s) => assert_eq!(
+                s.history_id, "999",
+                "the reconciliation's historyId, not the pre-existing watermark, must be persisted"
+            ),
+            _ => panic!("expected state.json to still be present after --full"),
+        }
+    }
+
+    // ── a corrupt state.json is a reconciliation trigger, not a hard error ──
+
+    #[tokio::test]
+    async fn run_sync_reconciles_when_state_json_is_corrupt() {
+        let server = wiremock::MockServer::start().await;
+        let client = client_with_bootstrapped_token(&server).await;
+        let dir = tempfile::tempdir().unwrap();
+        let output_dir = dir.path().join("archive");
+        std::fs::create_dir_all(&output_dir).unwrap();
+        std::fs::write(state_path(&output_dir), "not json\n").unwrap();
+
+        mount_profile(&server, "user@example.com", "999").await;
+        mount_message_list(&server, &["m1"]).await;
+        mount_raw_get(&server, "m1", "Hello").await;
+
+        let report = run_sync(&client, &opts(output_dir.clone())).await.unwrap();
+
+        assert!(report.errors.is_empty());
+        assert!(
+            report.actions.iter().any(
+                |a| matches!(a, SyncAction::Note { message } if message.contains("state.json unreadable"))
+            ),
+            "a corrupt watermark should be reported as a reconciliation note"
+        );
+        match state::load(&state_path(&output_dir)) {
+            LoadOutcome::Present(s) => assert_eq!(s.history_id, "999"),
+            _ => panic!("expected a fresh state.json to replace the corrupt one"),
+        }
+    }
+
     // ── reconciliation must not truncate at HARD_CAP (#1467) ────────────
 
     #[tokio::test]
@@ -1984,6 +2059,63 @@ not-really-a-pdf\r\n\
         assert_eq!(completed_failed, 1);
     }
 
+    #[tokio::test]
+    async fn fetch_and_archive_messages_streaming_dedupes_a_repeated_id() {
+        // A page-boundary race in the live-streamed listing can re-observe
+        // the same id twice — the second occurrence must be a no-op, not a
+        // second fetch, and `listed_ids` must still only contain it once.
+        let server = wiremock::MockServer::start().await;
+        let client = client_with_bootstrapped_token(&server).await;
+        let dir = tempfile::tempdir().unwrap();
+        let output_dir = dir.path().join("archive");
+        std::fs::create_dir_all(&output_dir).unwrap();
+
+        wiremock::Mock::given(wiremock::matchers::method("GET"))
+            .and(wiremock::matchers::path("/gmail/v1/users/me/messages/m1"))
+            .and(wiremock::matchers::query_param("format", "raw"))
+            .respond_with(
+                wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
+                    "id": "m1", "threadId": "t1", "labelIds": ["INBOX"],
+                    "internalDate": "1700000000000", "historyId": "500",
+                    "raw": raw_message_body("m1", "Hello"),
+                })),
+            )
+            .expect(1)
+            .mount(&server)
+            .await;
+
+        let mut manifest = Manifest::default();
+        let limiter = TokenBucket::new(1_000_000, 1_000_000);
+        let mut report = SyncReport::default();
+        let o = opts(output_dir.clone());
+
+        let (ids_tx, ids_rx) = mpsc::unbounded_channel();
+        for id in ["m1", "m1"] {
+            ids_tx.send(id.to_string()).unwrap();
+        }
+        drop(ids_tx);
+
+        let listed_ids = fetch_and_archive_messages_streaming(
+            &client,
+            &mut manifest,
+            ids_rx,
+            &limiter,
+            &o,
+            &mut report,
+            None,
+        )
+        .await
+        .unwrap();
+
+        assert_eq!(listed_ids.len(), 1, "listed_ids must dedupe too");
+        assert_eq!(
+            report.actions.len(),
+            1,
+            "the duplicate id must not be fetched a second time"
+        );
+        assert!(matches!(&report.actions[0], SyncAction::Fetched { id, .. } if id == "m1"));
+    }
+
     #[tokio::test]
     async fn run_sync_with_progress_emits_listing_events_ending_in_listing_done() {
         let server = wiremock::MockServer::start().await;