diff --git a/crates/utopia-cli/src/main.rs b/crates/utopia-cli/src/main.rs
index f3683aabd..1d93a525b 100644
--- a/crates/utopia-cli/src/main.rs
+++ b/crates/utopia-cli/src/main.rs
@@ -82,7 +82,7 @@ struct ManifestDataDir {
 /// not a side effect of a code change.
 // 是迁移文件的**个数**，不是最大的编号（守卫 `schema_version_policy_compares_against_current`
 // 按个数比）：编号有空缺时两者不同——0071 由一个开放 PR 占着，0072 先落，个数是 71
-const CURRENT_SCHEMA_VERSION: u32 = 106;
+const CURRENT_SCHEMA_VERSION: u32 = 107;
 
 fn main() -> anyhow::Result<()> {
     dotenvy::dotenv().ok();
diff --git a/crates/utopia-server/src/github_issues.rs b/crates/utopia-server/src/github_issues.rs
index 2dee2a118..0d9aa0118 100644
--- a/crates/utopia-server/src/github_issues.rs
+++ b/crates/utopia-server/src/github_issues.rs
@@ -39,13 +39,16 @@
 
 use chrono::{DateTime, Utc};
 use serde::Deserialize;
+use std::collections::BTreeMap;
+use std::time::Duration;
 
 use crate::ticket::stamp;
 
-/// 一次同步最多取多少页（每页 100）。GitHub 的分页没有天然终点，
-/// 一个活跃仓库能翻很久；这里沿用单端点的分页预算。
-const MAX_PAGES: u32 = 10;
+/// 列表预算可在完整时间组边界延长；单张工单的详情不受此预算限制。
+const MAX_LISTING_PAGES: u32 = 10;
 const PER_PAGE: u32 = 100;
+const MAX_RATE_LIMIT_RETRIES: u32 = 3;
+const MAX_RATE_LIMIT_WAIT: Duration = Duration::from_secs(30);
 
 #[derive(Debug, Deserialize)]
 pub struct Issue {
@@ -190,16 +193,39 @@ pub(crate) async fn get_checked(
     auth: Option<&str>,
     accept: Option<&str>,
 ) -> anyhow::Result<reqwest::Response> {
-    let mut req = http.get(url);
-    if let Some(a) = auth {
-        req = req.header(reqwest::header::AUTHORIZATION, a);
-    }
-    if let Some(a) = accept {
-        req = req.header(reqwest::header::ACCEPT, a);
-    }
-    let resp = req.send().await?;
-    let status = resp.status();
-    if !status.is_success() {
+    let mut retries = 0;
+    loop {
+        let mut req = http.get(url.clone());
+        if let Some(a) = auth {
+            req = req.header(reqwest::header::AUTHORIZATION, a);
+        }
+        if let Some(a) = accept {
+            req = req.header(reqwest::header::ACCEPT, a);
+        }
+        let resp = req.send().await?;
+        let status = resp.status();
+        if status.is_success() {
+            return Ok(resp);
+        }
+        if matches!(
+            status,
+            reqwest::StatusCode::FORBIDDEN | reqwest::StatusCode::TOO_MANY_REQUESTS
+        ) && retries < MAX_RATE_LIMIT_RETRIES
+        {
+            if let Some(delay) = rate_limit_delay(resp.headers(), Utc::now()) {
+                // 等待留在当前请求内，否则超大时间组会在每次重试时重新耗尽前缀的配额。
+                let delay = delay.max(Duration::from_secs(1 << retries));
+                retries += 1;
+                tracing::info!(
+                    retries,
+                    wait_seconds = delay.as_secs(),
+                    "GitHub rate limited; waiting before retry"
+                );
+                drop(resp);
+                tokio::time::sleep(delay).await;
+                continue;
+            }
+        }
         // 限流说清楚是限流：未认证时每小时 60 次，一个中等仓库一次同步就能吃光。
         // 报成通用 HTTP 错会让人去查网络，而正确的动作是配一个令牌
         let remaining = resp
@@ -214,10 +240,163 @@ pub(crate) async fn get_checked(
         }
         anyhow::bail!("HTTP {status} from GitHub");
     }
-    Ok(resp)
 }
 
-/// 分页取一个端点，直到空页或触顶。
+fn rate_limit_delay(headers: &reqwest::header::HeaderMap, now: DateTime<Utc>) -> Option<Duration> {
+    let retry_after = headers
+        .get(reqwest::header::RETRY_AFTER)
+        .and_then(|value| value.to_str().ok())
+        .and_then(|value| {
+            value
+                .trim()
+                .parse::<u64>()
+                .ok()
+                .map(Duration::from_secs)
+                .or_else(|| {
+                    DateTime::parse_from_rfc2822(value)
+                        .ok()
+                        .map(|at| at.signed_duration_since(now).to_std().unwrap_or_default())
+                })
+        });
+    let delay = retry_after.or_else(|| {
+        if headers.get("x-ratelimit-remaining")?.to_str().ok()? != "0" {
+            return None;
+        }
+        let reset = headers
+            .get("x-ratelimit-reset")?
+            .to_str()
+            .ok()?
+            .parse::<i64>()
+            .ok()?;
+        let reset = DateTime::from_timestamp(reset, 0)?;
+        Some(
+            reset
+                .signed_duration_since(now)
+                .to_std()
+                .unwrap_or_default(),
+        )
+    })?;
+    // 长等待不能占住 worker；拒绝本次重试，而不是早于服务端允许的时间重发。
+    (delay <= MAX_RATE_LIMIT_WAIT).then_some(delay)
+}
+
+pub struct IssueWindow {
+    pub issues: Vec<Issue>,
+    pub checkpoint: Option<DateTime<Utc>>,
+}
+
+/// offset 分页不是快照；复读同一前缀，变化时让调用者保留旧 checkpoint 后重试。
+pub async fn fetch_window(
+    http: &reqwest::Client,
+    base: &str,
+    since: Option<DateTime<Utc>>,
+    item_limit: usize,
+    include_prs: bool,
+    auth: Option<&str>,
+) -> anyhow::Result<IssueWindow> {
+    let (window, positions) =
+        fetch_window_pass(http, base, since, item_limit, include_prs, auth, None).await?;
+    if let Some(checkpoint) = window
+        .checkpoint
+        .filter(|checkpoint| Some(*checkpoint) != since)
+    {
+        let (verified, verified_positions) = fetch_window_pass(
+            http,
+            base,
+            since,
+            item_limit,
+            include_prs,
+            auth,
+            Some(checkpoint),
+        )
+        .await?;
+        anyhow::ensure!(
+            verified.checkpoint == Some(checkpoint) && positions == verified_positions,
+            "GitHub issues changed during pagination; retry the previous checkpoint"
+        );
+    }
+    Ok(window)
+}
+
+async fn fetch_window_pass(
+    http: &reqwest::Client,
+    base: &str,
+    since: Option<DateTime<Utc>>,
+    item_limit: usize,
+    include_prs: bool,
+    auth: Option<&str>,
+    through: Option<DateTime<Utc>>,
+) -> anyhow::Result<(IssueWindow, BTreeMap<i64, DateTime<Utc>>)> {
+    let mut query = vec![
+        ("state", "all".to_owned()),
+        ("sort", "updated".to_owned()),
+        ("direction", "asc".to_owned()),
+    ];
+    if let Some(since) = since {
+        query.push(("since", since.to_rfc3339()));
+    }
+    let mut window = IssueWindow {
+        issues: Vec::new(),
+        checkpoint: since,
+    };
+    let mut positions = BTreeMap::new();
+    let mut previous_time = None;
+    let mut stop_after = through;
+    let mut listing_pages = 0;
+    let mut page = 1;
+    loop {
+        let mut url = reqwest::Url::parse(base)?;
+        {
+            let mut parameters = url.query_pairs_mut();
+            for (key, value) in &query {
+                parameters.append_pair(key, value);
+            }
+            parameters.append_pair("per_page", &PER_PAGE.to_string());
+            parameters.append_pair("page", &page.to_string());
+        }
+        let batch: Vec<Issue> = get_checked(http, url, auth, None).await?.json().await?;
+        let count = batch.len();
+        let mut page_counted = false;
+        for issue in batch {
+            let updated = issue.updated_at;
+            anyhow::ensure!(
+                previous_time.is_none_or(|previous| updated >= previous),
+                "GitHub issues are not ordered by update time; retry the previous checkpoint"
+            );
+            previous_time = Some(updated);
+            if stop_after.is_some_and(|boundary| updated > boundary) {
+                return Ok((window, positions));
+            }
+            // 已消费条目和排除的 PR 也能移动页边界，必须参加复读核对。
+            anyhow::ensure!(
+                positions.insert(issue.number, updated).is_none(),
+                "GitHub issue repeated during pagination; retry the previous checkpoint"
+            );
+            if since.is_some_and(|checkpoint| updated <= checkpoint) {
+                continue;
+            }
+            if !page_counted {
+                listing_pages += 1;
+                page_counted = true;
+            }
+            window.checkpoint = Some(updated);
+            if include_prs || issue.pull_request.is_none() {
+                window.issues.push(issue);
+            }
+            if through.is_none()
+                && (window.issues.len() >= item_limit || listing_pages >= MAX_LISTING_PAGES)
+            {
+                stop_after = Some(updated);
+            }
+        }
+        if count < PER_PAGE as usize {
+            return Ok((window, positions));
+        }
+        page += 1;
+    }
+}
+
+/// 详情必须读到末页，不能用截断的评论或事件覆盖已有完整正文。
 pub async fn fetch_all<T: for<'de> Deserialize<'de>>(
     http: &reqwest::Client,
     base: &str,
@@ -225,7 +404,8 @@ pub async fn fetch_all<T: for<'de> Deserialize<'de>>(
     auth: Option<&str>,
 ) -> anyhow::Result<Vec<T>> {
     let mut out = Vec::new();
-    for page in 1..=MAX_PAGES {
+    let mut page = 1;
+    loop {
         // 自己拼 query：这套 reqwest 特性组合里没有 RequestBuilder::query，
         // 而 sync_custom 本来也是这么拼的
         let mut url = reqwest::Url::parse(base)?;
@@ -244,6 +424,7 @@ pub async fn fetch_all<T: for<'de> Deserialize<'de>>(
         if n < PER_PAGE as usize {
             break;
         }
+        page += 1;
     }
     Ok(out)
 }
@@ -251,6 +432,355 @@ pub async fn fetch_all<T: for<'de> Deserialize<'de>>(
 #[cfg(test)]
 mod tests {
     use super::*;
+    use serde_json::{json, Value};
+    use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
+    use wiremock::{matchers::method, Mock, MockServer, Request, ResponseTemplate};
+
+    fn listing_row(number: i64, updated: &str) -> Value {
+        json!({
+            "number": number, "title": format!("Issue {number}"), "state": "open",
+            "created_at": "2026-01-01T00:00:00Z", "updated_at": updated,
+        })
+    }
+
+    fn requested_page(request: &Request) -> usize {
+        request
+            .url
+            .query_pairs()
+            .find(|(key, _)| key == "page")
+            .unwrap()
+            .1
+            .parse()
+            .unwrap()
+    }
+
+    #[test]
+    fn rate_limit_headers_supply_a_bounded_wait_without_guessing() {
+        let now: DateTime<Utc> = "2026-01-02T00:00:00Z".parse().unwrap();
+        let mut headers = reqwest::header::HeaderMap::new();
+        for (value, expected) in [
+            ("2", Some(2)),
+            ("0", Some(0)),
+            ("30", Some(30)),
+            ("31", None),
+            ("-1", None),
+            ("broken", None),
+            ("86401", None),
+        ] {
+            headers.insert(reqwest::header::RETRY_AFTER, value.parse().unwrap());
+            assert_eq!(
+                rate_limit_delay(&headers, now),
+                expected.map(Duration::from_secs)
+            );
+        }
+        for (seconds, expected) in [(3, Some(3)), (31, None)] {
+            headers.insert(
+                reqwest::header::RETRY_AFTER,
+                (now + chrono::Duration::seconds(seconds))
+                    .to_rfc2822()
+                    .parse()
+                    .unwrap(),
+            );
+            assert_eq!(
+                rate_limit_delay(&headers, now),
+                expected.map(Duration::from_secs)
+            );
+        }
+        headers.remove(reqwest::header::RETRY_AFTER);
+        headers.insert("x-ratelimit-remaining", "0".parse().unwrap());
+        headers.insert(
+            "x-ratelimit-reset",
+            (now.timestamp() + 30).to_string().parse().unwrap(),
+        );
+        assert_eq!(
+            rate_limit_delay(&headers, now),
+            Some(Duration::from_secs(30))
+        );
+        headers.insert("x-ratelimit-remaining", "1".parse().unwrap());
+        assert_eq!(rate_limit_delay(&headers, now), None);
+        headers.insert("x-ratelimit-remaining", "0".parse().unwrap());
+        headers.insert(
+            "x-ratelimit-reset",
+            (now.timestamp() + 31).to_string().parse().unwrap(),
+        );
+        assert_eq!(rate_limit_delay(&headers, now), None);
+    }
+
+    #[tokio::test]
+    async fn a_rate_limited_request_waits_and_preserves_its_url_and_headers() {
+        let server = MockServer::start().await;
+        let http = reqwest::Client::builder().no_proxy().build().unwrap();
+        for status in [403, 429] {
+            server.reset().await;
+            let attempts = AtomicUsize::new(0);
+            Mock::given(method("GET"))
+                .respond_with(move |_: &Request| {
+                    if attempts.fetch_add(1, Ordering::SeqCst) == 0 {
+                        ResponseTemplate::new(status).insert_header("Retry-After", "0")
+                    } else {
+                        ResponseTemplate::new(200)
+                    }
+                })
+                .mount(&server)
+                .await;
+            let url = reqwest::Url::parse(&format!("{}/resource?page=2", server.uri())).unwrap();
+            let started = std::time::Instant::now();
+            get_checked(
+                &http,
+                url.clone(),
+                Some("Bearer fixture"),
+                Some("application/vnd.github.raw+json"),
+            )
+            .await
+            .unwrap();
+            assert!(started.elapsed() >= Duration::from_secs(1));
+            let requests = server.received_requests().await.unwrap();
+            assert_eq!(requests.len(), 2);
+            for request in &requests {
+                // wiremock 为 origin-form URI 补 localhost；实际目标主机保留在 Host 头中。
+                assert_eq!(request.url.path(), url.path());
+                assert_eq!(request.url.query(), url.query());
+                assert_eq!(
+                    request.headers.get("host").unwrap().to_str().unwrap(),
+                    server.address().to_string()
+                );
+                assert_eq!(
+                    request.headers.get("authorization").unwrap(),
+                    "Bearer fixture"
+                );
+                assert_eq!(
+                    request.headers.get("accept").unwrap(),
+                    "application/vnd.github.raw+json"
+                );
+            }
+        }
+    }
+
+    #[tokio::test]
+    async fn persistent_rate_limits_stop_after_a_bounded_number_of_requests() {
+        let server = MockServer::start().await;
+        Mock::given(method("GET"))
+            .respond_with(ResponseTemplate::new(429).insert_header("Retry-After", "0"))
+            .mount(&server)
+            .await;
+        let http = reqwest::Client::builder().no_proxy().build().unwrap();
+        assert!(
+            get_checked(&http, server.uri().parse().unwrap(), None, None)
+                .await
+                .is_err()
+        );
+        assert_eq!(server.received_requests().await.unwrap().len(), 4);
+    }
+
+    #[tokio::test]
+    async fn long_rate_limit_waits_fail_without_sleeping_or_retrying() {
+        let server = MockServer::start().await;
+        let http = reqwest::Client::builder().no_proxy().build().unwrap();
+        for status in [403, 429] {
+            let responses = [
+                ResponseTemplate::new(status).insert_header("Retry-After", "31"),
+                ResponseTemplate::new(status).insert_header(
+                    "Retry-After",
+                    (Utc::now() + chrono::Duration::seconds(60)).to_rfc2822(),
+                ),
+                ResponseTemplate::new(status)
+                    .insert_header("x-ratelimit-remaining", "0")
+                    .insert_header(
+                        "x-ratelimit-reset",
+                        (Utc::now().timestamp() + 60).to_string(),
+                    ),
+            ];
+            for response in responses {
+                server.reset().await;
+                Mock::given(method("GET"))
+                    .respond_with(response)
+                    .mount(&server)
+                    .await;
+                let result = tokio::time::timeout(
+                    Duration::from_secs(5),
+                    get_checked(&http, server.uri().parse().unwrap(), None, None),
+                )
+                .await
+                .expect("a long rate-limit wait must fail immediately");
+                assert!(result.is_err());
+                assert_eq!(server.received_requests().await.unwrap().len(), 1);
+            }
+        }
+    }
+
+    #[tokio::test]
+    async fn an_empty_window_retains_the_previous_checkpoint_without_verification() {
+        let server = MockServer::start().await;
+        Mock::given(method("GET"))
+            .respond_with(ResponseTemplate::new(200).set_body_json(json!([])))
+            .mount(&server)
+            .await;
+        let http = reqwest::Client::builder().no_proxy().build().unwrap();
+        let since = Some("2026-01-02T00:00:00Z".parse().unwrap());
+        let window = fetch_window(&http, &server.uri(), since, 200, false, None)
+            .await
+            .unwrap();
+        assert!(window.issues.is_empty());
+        assert_eq!(window.checkpoint, since);
+        assert_eq!(server.received_requests().await.unwrap().len(), 1);
+    }
+
+    #[tokio::test]
+    async fn a_large_timestamp_group_finishes_and_the_next_run_passes_its_old_pages() {
+        let server = MockServer::start().await;
+        let http = reqwest::Client::builder().no_proxy().build().unwrap();
+        let group_time = "2026-01-02T00:00:00Z";
+        let later_time = "2026-01-03T00:00:00Z";
+        // PR 多时先耗尽十页预算；全是工单时先耗尽 200 条预算。
+        for issue_interval in [1, 100] {
+            server.reset().await;
+            let mut rows: Vec<_> = (1..=1001)
+                .map(|number| {
+                    let mut row = listing_row(number, group_time);
+                    if (number - 1) % issue_interval as i64 != 0 {
+                        row["pull_request"] = json!({});
+                    }
+                    row
+                })
+                .collect();
+            rows.push(listing_row(1002, later_time));
+            Mock::given(method("GET"))
+                .respond_with(move |request: &Request| {
+                    let page = requested_page(request);
+                    let rows: Vec<_> = rows
+                        .iter()
+                        .skip((page - 1) * 100)
+                        .take(100)
+                        .cloned()
+                        .collect();
+                    ResponseTemplate::new(200).set_body_json(rows)
+                })
+                .mount(&server)
+                .await;
+            let first = fetch_window(&http, &server.uri(), None, 200, false, None)
+                .await
+                .unwrap();
+            let numbers: Vec<_> = first.issues.iter().map(|issue| issue.number).collect();
+            assert_eq!(
+                numbers,
+                (1..=1001).step_by(issue_interval).collect::<Vec<_>>()
+            );
+            assert_eq!(first.checkpoint, Some(group_time.parse().unwrap()));
+
+            // 模拟端点仍返回 inclusive 下界；十页旧记录不应耗尽新一轮的任一预算。
+            let next = fetch_window(&http, &server.uri(), first.checkpoint, 200, false, None)
+                .await
+                .unwrap();
+            assert_eq!(next.issues.len(), 1);
+            assert_eq!(next.issues[0].number, 1002);
+            assert_eq!(next.checkpoint, Some(later_time.parse().unwrap()));
+            let requests = server.received_requests().await.unwrap();
+            assert!(requests.iter().all(|request| {
+                request
+                    .url
+                    .query_pairs()
+                    .any(|(key, value)| key == "sort" && value == "updated")
+                    && request
+                        .url
+                        .query_pairs()
+                        .any(|(key, value)| key == "direction" && value == "asc")
+            }));
+            assert!(requests.iter().any(|request| requested_page(request) == 11));
+        }
+    }
+
+    #[tokio::test]
+    async fn a_moved_issue_rejects_the_window_and_is_read_after_pagination_stabilizes() {
+        let server = MockServer::start().await;
+        let moved = AtomicBool::new(false);
+        Mock::given(method("GET"))
+            .respond_with(move |request: &Request| {
+                let page = requested_page(request);
+                if page == 2 {
+                    moved.store(true, Ordering::SeqCst);
+                }
+                let mut rows: Vec<_> = (1..=201)
+                    .map(|number| listing_row(number, "2026-01-02T00:00:00Z"))
+                    .collect();
+                if moved.load(Ordering::SeqCst) {
+                    rows.remove(0);
+                    rows.push(listing_row(1, "2026-01-03T00:00:00Z"));
+                }
+                let rows: Vec<_> = rows.into_iter().skip((page - 1) * 100).take(100).collect();
+                ResponseTemplate::new(200).set_body_json(rows)
+            })
+            .mount(&server)
+            .await;
+        let http = reqwest::Client::builder().no_proxy().build().unwrap();
+        let result = fetch_window(&http, &server.uri(), None, 200, false, None).await;
+        assert!(result
+            .err()
+            .unwrap()
+            .to_string()
+            .contains("changed during pagination"));
+
+        let recovered = fetch_window(&http, &server.uri(), None, 200, false, None)
+            .await
+            .unwrap();
+        let recovered_numbers: Vec<_> = recovered.issues.iter().map(|issue| issue.number).collect();
+        assert_eq!(recovered_numbers, (2..=201).collect::<Vec<_>>());
+        assert_eq!(
+            recovered.checkpoint,
+            Some("2026-01-02T00:00:00Z".parse().unwrap())
+        );
+        let next = fetch_window(&http, &server.uri(), recovered.checkpoint, 200, false, None)
+            .await
+            .unwrap();
+        assert_eq!(next.issues.len(), 1);
+        assert_eq!(next.issues[0].number, 1);
+        assert_eq!(
+            next.checkpoint,
+            Some("2026-01-03T00:00:00Z".parse().unwrap())
+        );
+    }
+
+    #[tokio::test]
+    async fn a_complete_discussion_is_read_beyond_ten_pages() {
+        let server = MockServer::start().await;
+        for count in [1000, 1001, 1100] {
+            server.reset().await;
+            Mock::given(method("GET"))
+                .respond_with(move |request: &Request| {
+                    let page = requested_page(request);
+                    let rows: Vec<_> = (0..count).skip((page - 1) * 100).take(100).collect();
+                    ResponseTemplate::new(200).set_body_json(rows)
+                })
+                .mount(&server)
+                .await;
+            let http = reqwest::Client::builder().no_proxy().build().unwrap();
+            let result = fetch_all::<u32>(&http, &server.uri(), &[], None)
+                .await
+                .unwrap();
+            assert_eq!(result, (0..count).collect::<Vec<_>>());
+            let requests = server.received_requests().await.unwrap();
+            assert_eq!(requests.len(), count as usize / 100 + 1);
+        }
+    }
+
+    #[tokio::test]
+    async fn a_failed_discussion_page_does_not_return_a_partial_history() {
+        let server = MockServer::start().await;
+        Mock::given(method("GET"))
+            .respond_with(|request: &Request| {
+                let page = requested_page(request);
+                if page == 11 {
+                    return ResponseTemplate::new(503);
+                }
+                let rows: Vec<_> = (0..1001).skip((page - 1) * 100).take(100).collect();
+                ResponseTemplate::new(200).set_body_json(rows)
+            })
+            .mount(&server)
+            .await;
+        let http = reqwest::Client::builder().no_proxy().build().unwrap();
+        let result = fetch_all::<u32>(&http, &server.uri(), &[], None).await;
+        assert!(result.unwrap_err().to_string().contains("503"));
+        assert_eq!(server.received_requests().await.unwrap().len(), 11);
+    }
 
     fn issue(json: serde_json::Value) -> Issue {
         serde_json::from_value(json).unwrap()
diff --git a/crates/utopia-server/src/github_sync_tests.rs b/crates/utopia-server/src/github_sync_tests.rs
index 39b38db2b..025acb57e 100644
--- a/crates/utopia-server/src/github_sync_tests.rs
+++ b/crates/utopia-server/src/github_sync_tests.rs
@@ -260,14 +260,15 @@ async fn incremental_github_sync_preserves_the_complete_discussion() -> anyhow::
             .iter()
             .filter(|request| request.url.path() == "/repos/test/repo/issues")
             .collect();
-        anyhow::ensure!(issue_requests.len() == 6);
+        // 每轮先选窗口，再复读检查分页期间是否发生移动。
+        anyhow::ensure!(issue_requests.len() == 12);
         for (index, request) in issue_requests.iter().enumerate() {
             let lower_bound = request
                 .url
                 .query_pairs()
                 .find(|(key, _)| key == "since")
                 .map(|(_, value)| time(&value));
-            anyhow::ensure!(lower_bound == if index == 0 { None } else { since });
+            anyhow::ensure!(lower_bound == if index < 2 { None } else { since });
         }
         anyhow::ensure!(
             requests
diff --git a/crates/utopia-server/src/ingest_sources.rs b/crates/utopia-server/src/ingest_sources.rs
index a45359802..cb012ae3f 100644
--- a/crates/utopia-server/src/ingest_sources.rs
+++ b/crates/utopia-server/src/ingest_sources.rs
@@ -16,6 +16,10 @@ use uuid::Uuid;
 #[path = "source_checkpoint_tests.rs"]
 mod source_checkpoint_tests;
 
+#[cfg(test)]
+#[path = "timestamp_checkpoint_tests.rs"]
+mod timestamp_checkpoint_tests;
+
 /// 单次同步的新文档上限（防超长 feed/URL 列表拖垮任务）
 const MAX_NEW_PER_SYNC: usize = 200;
 const MAX_FEED_BYTES: usize = 4 * 1024 * 1024;
@@ -39,6 +43,7 @@ pub struct SyncStats {
     pub discovered: usize,
     pub queued_for_content: usize,
     pub content_terminal: usize,
+    pub checkpoint: Option<DateTime<Utc>>,
 }
 
 #[derive(Debug)]
@@ -69,9 +74,12 @@ pub async fn sync_source(state: &AppState, source_id: Uuid) -> anyhow::Result<()
     let source = utopia_store::sources::get(&state.pool, source_id).await?;
     let kind = SourceKind::parse(&source.kind);
     let since = match kind {
-        Some(SourceKind::Custom | SourceKind::GithubIssues | SourceKind::JiraIssues) => {
+        Some(SourceKind::Custom) => {
             utopia_store::sources::last_successful_sync_start(&state.pool, source_id).await?
         }
+        Some(SourceKind::GithubIssues | SourceKind::JiraIssues) => {
+            utopia_store::sources::sync_checkpoint(&state.pool, source_id).await?
+        }
         _ => None,
     };
     utopia_store::sources::mark_running(&state.pool, source_id).await?;
@@ -104,19 +112,51 @@ pub async fn sync_source(state: &AppState, source_id: Uuid) -> anyhow::Result<()
         None => Err(anyhow::anyhow!("unknown source kind `{}`", source.kind)),
     };
 
-    match outcome {
-        Ok(stats) => {
-            utopia_store::sources::finish_run(
+    let outcome = match outcome {
+        Ok(stats)
+            if matches!(
+                kind,
+                Some(SourceKind::GithubIssues | SourceKind::JiraIssues)
+            ) =>
+        {
+            utopia_store::sources::finish_timestamp_sync(
                 &state.pool,
                 run_id,
                 source_id,
-                None,
+                since,
+                stats.checkpoint,
                 stats.created as i32,
                 stats.updated as i32,
             )
-            .await?;
-            utopia_store::sources::finish_sync(&state.pool, source_id, None, stats.total() as i32)
+            .await
+            .map(|()| stats)
+            .map_err(anyhow::Error::from)
+        }
+        other => other,
+    };
+    match outcome {
+        Ok(stats) => {
+            if !matches!(
+                kind,
+                Some(SourceKind::GithubIssues | SourceKind::JiraIssues)
+            ) {
+                utopia_store::sources::finish_run(
+                    &state.pool,
+                    run_id,
+                    source_id,
+                    None,
+                    stats.created as i32,
+                    stats.updated as i32,
+                )
+                .await?;
+                utopia_store::sources::finish_sync(
+                    &state.pool,
+                    source_id,
+                    None,
+                    stats.total() as i32,
+                )
                 .await?;
+            }
             state.emit_source(source.kb_id);
             tracing::info!(
                 %source_id,
@@ -834,20 +874,20 @@ async fn sync_github_issues_with_client(
 
     let base = format!("{api_base}/repos/{repo}");
 
-    // 增量：GitHub 的 since 是"这之后更新过的"
-    let mut issue_q: Vec<(&str, String)> = vec![("state", "all".into())];
-    if let Some(t) = since {
-        issue_q.push(("since", t.to_rfc3339()));
-    }
-
-    let issues: Vec<crate::github_issues::Issue> =
-        crate::github_issues::fetch_all(http, &format!("{base}/issues"), &issue_q, auth).await?;
-    let mut stats = SyncStats::default();
-    for issue in issues
-        .iter()
-        .filter(|i| include_prs || i.pull_request.is_none())
-        .take(MAX_NEW_PER_SYNC)
-    {
+    let window = crate::github_issues::fetch_window(
+        http,
+        &format!("{base}/issues"),
+        since,
+        MAX_NEW_PER_SYNC,
+        include_prs,
+        auth,
+    )
+    .await?;
+    let mut stats = SyncStats {
+        checkpoint: window.checkpoint,
+        ..SyncStats::default()
+    };
+    for issue in &window.issues {
         // 只重建本轮选中的工单，但每张要读完整评论历史。仓库级评论即使
         // 去掉 since，也可能被无关工单和 PR 的评论挤出分页上限。
         let comments = crate::github_issues::fetch_all::<crate::github_issues::Comment>(
@@ -1207,21 +1247,14 @@ async fn sync_jira_issues(
     )
     .await?;
 
-    let jql = crate::jira_issues::jql(project, since);
-    let (issues, total) = crate::jira_issues::fetch_all(&http, base_url, &jql, auth).await?;
-    // **截断了就说出来。** 一个跑了多年的项目动辄上万张工单，翻页上限意味着
-    // 这一轮只覆盖了一段；不报的话界面上"同步完成"就是一句误导
-    if total > issues.len() as i64 {
-        tracing::warn!(
-            source_id = %source.id,
-            fetched = issues.len(),
-            total,
-            "Jira 结果被翻页上限截断，本轮只覆盖了一部分；下一轮的 JQL 窗口会接上"
-        );
-    }
-
-    let mut stats = SyncStats::default();
-    for issue in issues.iter().take(MAX_NEW_PER_SYNC) {
+    let window =
+        crate::jira_issues::fetch_window(&http, base_url, project, since, MAX_NEW_PER_SYNC, auth)
+            .await?;
+    let mut stats = SyncStats {
+        checkpoint: window.checkpoint,
+        ..SyncStats::default()
+    };
+    for issue in &window.issues {
         let body = crate::jira_issues::render(issue);
         // 逻辑身份带上站点：一个知识库接两个 Jira 时，PROJ-1 不会互相覆盖
         let host = reqwest::Url::parse(base_url)
diff --git a/crates/utopia-server/src/jira_issues.rs b/crates/utopia-server/src/jira_issues.rs
index 4ade51412..b01d36b33 100644
--- a/crates/utopia-server/src/jira_issues.rs
+++ b/crates/utopia-server/src/jira_issues.rs
@@ -40,13 +40,15 @@
 
 use chrono::{DateTime, Utc};
 use serde::Deserialize;
+use std::collections::BTreeMap;
+use std::time::Duration;
 
 use crate::ticket::stamp;
 
-/// 一次同步最多翻多少页。Jira 的 `total` 常常是几万，全量拉回来没有意义——
-/// 增量窗口之外的等下一次 JQL 取。
+/// 达到列表预算后仍读完当前时间组，下次从完整组之后继续。
 const MAX_PAGES: u32 = 10;
 const PAGE_SIZE: u32 = 50;
+const MAX_RATE_LIMIT_RETRIES: u32 = 3;
 
 /// 想要的字段。**必须显式列**：不列 `comment` 就不返回评论，
 /// 而默认返回全部字段会把响应撑到几百 KB 一条。
@@ -57,7 +59,7 @@ const FIELDS: &str = "summary,status,issuetype,priority,created,updated,resoluti
 pub struct SearchPage {
     #[serde(default)]
     pub issues: Vec<Issue>,
-    #[serde(default)]
+    // total 用来识别提前结束的分页，缺失时不能把空页当作完整结果。
     pub total: i64,
 }
 
@@ -243,67 +245,494 @@ pub fn jql(project: &str, since: Option<DateTime<Utc>>) -> String {
             t.format("%Y-%m-%d %H:%M")
         ));
     }
-    q.push_str(" ORDER BY updated ASC");
+    q.push_str(" ORDER BY updated ASC, key ASC");
     q
 }
 
-/// 分页取。Jira 用 `startAt`/`maxResults`，`total` 常常是几万——
-/// 靠 MAX_PAGES 封顶，剩下的等下一次增量窗口。
-///
-/// **把 `total` 一起返回**：截断了就得说出来。取回 500 条而服务端有 14506 条时，
-/// 界面上"同步完成"是一句误导——真实情况是"这一轮只覆盖了一小段"。
-pub async fn fetch_all(
+pub struct IssueWindow {
+    pub issues: Vec<Issue>,
+    pub checkpoint: Option<DateTime<Utc>>,
+}
+
+/// 复读固定的已选前缀，检测更新导致的 offset 位移；变化时不确认本轮 checkpoint。
+pub async fn fetch_window(
+    http: &reqwest::Client,
+    base_url: &str,
+    project: &str,
+    since: Option<DateTime<Utc>>,
+    item_limit: usize,
+    auth: Option<&str>,
+) -> anyhow::Result<IssueWindow> {
+    let query = jql(project, since);
+    let (window, positions) =
+        fetch_window_pass(http, base_url, &query, since, item_limit, auth, None).await?;
+    if let Some(checkpoint) = window
+        .checkpoint
+        .filter(|checkpoint| Some(*checkpoint) != since)
+    {
+        let (verified, verified_positions) = fetch_window_pass(
+            http,
+            base_url,
+            &query,
+            since,
+            item_limit,
+            auth,
+            Some(checkpoint),
+        )
+        .await?;
+        anyhow::ensure!(
+            verified.checkpoint == Some(checkpoint) && positions == verified_positions,
+            "Jira issues changed during pagination; retry the previous checkpoint"
+        );
+    }
+    Ok(window)
+}
+
+async fn fetch_window_pass(
     http: &reqwest::Client,
     base_url: &str,
-    jql: &str,
+    query: &str,
+    since: Option<DateTime<Utc>>,
+    item_limit: usize,
     auth: Option<&str>,
-) -> anyhow::Result<(Vec<Issue>, i64)> {
-    let mut out = Vec::new();
-    let mut total = 0i64;
-    for page in 0..MAX_PAGES {
+    through: Option<DateTime<Utc>>,
+) -> anyhow::Result<(IssueWindow, BTreeMap<String, DateTime<Utc>>)> {
+    let mut window = IssueWindow {
+        issues: Vec::new(),
+        checkpoint: since,
+    };
+    let mut positions = BTreeMap::new();
+    let mut previous_time = None;
+    let mut stop_after = through;
+    let mut listing_pages = 0;
+    let mut start_at = 0;
+    loop {
         let mut url = reqwest::Url::parse(&format!(
             "{}/rest/api/2/search",
             base_url.trim_end_matches('/')
         ))?;
         {
             let mut q = url.query_pairs_mut();
-            q.append_pair("jql", jql);
+            q.append_pair("jql", query);
             q.append_pair("expand", "changelog");
             q.append_pair("fields", FIELDS);
             q.append_pair("maxResults", &PAGE_SIZE.to_string());
-            q.append_pair("startAt", &(page * PAGE_SIZE).to_string());
+            q.append_pair("startAt", &start_at.to_string());
+        }
+        let page_data = fetch_page(http, url, auth).await?;
+        let count = page_data.issues.len();
+        anyhow::ensure!(
+            page_data.total >= 0 && (start_at + count) as i64 <= page_data.total,
+            "Jira returned an inconsistent page total; retry the previous checkpoint"
+        );
+        anyhow::ensure!(
+            count != 0 || start_at as i64 >= page_data.total,
+            "Jira returned an empty page before its total; retry the previous checkpoint"
+        );
+        let mut page_counted = false;
+        for issue in page_data.issues {
+            let updated = issue
+                .fields
+                .updated
+                .ok_or_else(|| {
+                    anyhow::anyhow!("Jira issue {} is missing its update time", issue.key)
+                })?
+                .0;
+            anyhow::ensure!(
+                previous_time.is_none_or(|previous| updated >= previous),
+                "Jira issues are not ordered by update time; retry the previous checkpoint"
+            );
+            previous_time = Some(updated);
+            if stop_after.is_some_and(|boundary| updated > boundary) {
+                return Ok((window, positions));
+            }
+            // 分钟取整返回的旧记录也可能改变 offset，复读时不能只比较新记录。
+            anyhow::ensure!(
+                positions.insert(issue.key.clone(), updated).is_none(),
+                "Jira issue repeated during pagination; retry the previous checkpoint"
+            );
+            if since.is_some_and(|checkpoint| updated <= checkpoint) {
+                continue;
+            }
+            if !page_counted {
+                listing_pages += 1;
+                page_counted = true;
+            }
+            window.checkpoint = Some(updated);
+            window.issues.push(issue);
+            if through.is_none()
+                && (window.issues.len() >= item_limit || listing_pages >= MAX_PAGES)
+            {
+                stop_after = Some(updated);
+            }
+        }
+        start_at += count;
+        if count == 0 || start_at as i64 >= page_data.total {
+            return Ok((window, positions));
         }
+    }
+}
+
+async fn fetch_page(
+    http: &reqwest::Client,
+    url: reqwest::Url,
+    auth: Option<&str>,
+) -> anyhow::Result<SearchPage> {
+    let mut retries = 0;
+    loop {
         let mut req = http
-            .get(url)
+            .get(url.clone())
             .header(reqwest::header::ACCEPT, "application/json");
-        if let Some(a) = auth {
-            req = req.header(reqwest::header::AUTHORIZATION, a);
+        if let Some(auth) = auth {
+            req = req.header(reqwest::header::AUTHORIZATION, auth);
         }
         let resp = req.send().await?;
         let status = resp.status();
-        if !status.is_success() {
-            // Jira 把 JQL 语法错误也报成 400，正文里才是原因。
-            // 只说 "HTTP 400" 会让人去查网络，而真正该改的是 project key 或时间格式
-            let detail = resp.text().await.unwrap_or_default();
-            anyhow::bail!(
-                "HTTP {status} from Jira: {}",
-                detail.chars().take(200).collect::<String>()
-            );
+        if status.is_success() {
+            return Ok(resp.json().await?);
         }
-        let page_data: SearchPage = resp.json().await?;
-        total = page_data.total;
-        let n = page_data.issues.len();
-        out.extend(page_data.issues);
-        if n < PAGE_SIZE as usize {
-            break;
+        if status == reqwest::StatusCode::TOO_MANY_REQUESTS && retries < MAX_RATE_LIMIT_RETRIES {
+            if let Some(delay) = rate_limit_delay(resp.headers(), Utc::now()) {
+                // 保留已读取的组和当前页，不能把限流变成从头扫描的循环。
+                let delay = delay.max(Duration::from_secs(1 << retries));
+                retries += 1;
+                tracing::info!(
+                    retries,
+                    wait_seconds = delay.as_secs(),
+                    "Jira rate limited; waiting before retry"
+                );
+                drop(resp);
+                tokio::time::sleep(delay).await;
+                continue;
+            }
         }
+        // Jira 把 JQL 语法错误也报成 400，正文里才是原因。
+        let detail = resp.text().await.unwrap_or_default();
+        anyhow::bail!(
+            "HTTP {status} from Jira: {}",
+            detail.chars().take(200).collect::<String>()
+        );
     }
-    Ok((out, total))
+}
+
+fn rate_limit_delay(headers: &reqwest::header::HeaderMap, now: DateTime<Utc>) -> Option<Duration> {
+    let value = headers
+        .get(reqwest::header::RETRY_AFTER)?
+        .to_str()
+        .ok()?
+        .trim();
+    let delay = value
+        .parse::<u64>()
+        .ok()
+        .map(Duration::from_secs)
+        .or_else(|| {
+            DateTime::parse_from_rfc2822(value)
+                .ok()
+                .map(|at| at.signed_duration_since(now).to_std().unwrap_or_default())
+        })?;
+    // shortcut: 长等待走普通失败重试；严格遵循跨任务的 Retry-After 需要持久化调度时间。
+    (delay <= Duration::from_secs(30)).then_some(delay)
 }
 
 #[cfg(test)]
 mod tests {
     use super::*;
+    use serde_json::{json, Value};
+    use std::sync::atomic::{AtomicBool, Ordering};
+    use wiremock::{matchers::method, Mock, MockServer, Request, ResponseTemplate};
+
+    fn listing_row(number: i64, updated: &str) -> Value {
+        json!({"key": format!("X-{number}"), "fields": {"summary": "Issue", "updated": updated}})
+    }
+
+    fn requested_offset(request: &Request) -> usize {
+        request
+            .url
+            .query_pairs()
+            .find(|(key, _)| key == "startAt")
+            .unwrap()
+            .1
+            .parse()
+            .unwrap()
+    }
+
+    #[test]
+    fn rate_limit_headers_supply_a_bounded_wait_without_guessing() {
+        let now: DateTime<Utc> = "2026-01-02T00:00:00Z".parse().unwrap();
+        let mut headers = reqwest::header::HeaderMap::new();
+        assert_eq!(rate_limit_delay(&headers, now), None);
+        for (value, expected) in [
+            ("2", Some(2)),
+            ("0", Some(0)),
+            ("30", Some(30)),
+            ("31", None),
+            ("-1", None),
+            ("broken", None),
+            ("86401", None),
+        ] {
+            headers.insert(reqwest::header::RETRY_AFTER, value.parse().unwrap());
+            assert_eq!(
+                rate_limit_delay(&headers, now),
+                expected.map(Duration::from_secs)
+            );
+        }
+        for (seconds, expected) in [(3, Some(3)), (30, Some(30)), (31, None)] {
+            headers.insert(
+                reqwest::header::RETRY_AFTER,
+                (now + chrono::Duration::seconds(seconds))
+                    .to_rfc2822()
+                    .parse()
+                    .unwrap(),
+            );
+            assert_eq!(
+                rate_limit_delay(&headers, now),
+                expected.map(Duration::from_secs)
+            );
+        }
+    }
+
+    #[tokio::test]
+    async fn a_long_rate_limit_wait_fails_without_holding_the_sync_slot() {
+        let server = MockServer::start().await;
+        Mock::given(method("GET"))
+            .respond_with(ResponseTemplate::new(429).insert_header("Retry-After", "31"))
+            .mount(&server)
+            .await;
+        let http = reqwest::Client::builder().no_proxy().build().unwrap();
+        let url = reqwest::Url::parse(&server.uri()).unwrap();
+        let result = tokio::time::timeout(Duration::from_secs(1), fetch_page(&http, url, None))
+            .await
+            .expect("a long Retry-After must return an error without waiting");
+        assert!(result.unwrap_err().to_string().contains("HTTP 429"));
+        assert_eq!(server.received_requests().await.unwrap().len(), 1);
+    }
+
+    #[tokio::test]
+    async fn a_rate_limited_page_resumes_without_rereading_the_previous_page() {
+        let server = MockServer::start().await;
+        let limited = AtomicBool::new(false);
+        Mock::given(method("GET"))
+            .respond_with(move |request: &Request| {
+                let offset = requested_offset(request);
+                if offset == 50 && !limited.swap(true, Ordering::SeqCst) {
+                    return ResponseTemplate::new(429).insert_header("Retry-After", "0");
+                }
+                let rows: Vec<_> = (1..=51)
+                    .skip(offset)
+                    .take(50)
+                    .map(|number| listing_row(number, "2026-01-02T00:00:00Z"))
+                    .collect();
+                ResponseTemplate::new(200).set_body_json(json!({"issues": rows, "total": 51}))
+            })
+            .mount(&server)
+            .await;
+        let http = reqwest::Client::builder().no_proxy().build().unwrap();
+        let started = std::time::Instant::now();
+        let window = fetch_window(&http, &server.uri(), "X", None, 200, Some("Bearer fixture"))
+            .await
+            .unwrap();
+        assert_eq!(window.issues.len(), 51);
+        assert!(started.elapsed() >= Duration::from_secs(1));
+        let requests = server.received_requests().await.unwrap();
+        assert_eq!(
+            requests.iter().map(requested_offset).collect::<Vec<_>>(),
+            [0, 50, 50, 0, 50]
+        );
+        assert!(requests
+            .iter()
+            .all(
+                |request| request.headers.get("authorization").unwrap() == "Bearer fixture"
+                    && request.headers.get("accept").unwrap() == "application/json"
+            ));
+    }
+
+    #[tokio::test]
+    async fn an_empty_window_retains_the_previous_checkpoint_without_verification() {
+        let server = MockServer::start().await;
+        Mock::given(method("GET"))
+            .respond_with(
+                ResponseTemplate::new(200).set_body_json(json!({"issues": [], "total": 0})),
+            )
+            .mount(&server)
+            .await;
+        let http = reqwest::Client::builder().no_proxy().build().unwrap();
+        let since = Some("2026-01-02T00:00:00Z".parse().unwrap());
+        let window = fetch_window(&http, &server.uri(), "X", since, 200, None)
+            .await
+            .unwrap();
+        assert!(window.issues.is_empty());
+        assert_eq!(window.checkpoint, since);
+        assert_eq!(server.received_requests().await.unwrap().len(), 1);
+    }
+
+    #[tokio::test]
+    async fn a_large_timestamp_group_finishes_and_the_next_run_passes_its_old_pages() {
+        // Jira may return fewer items per page than the requested maxResults.
+        for (group_size, page_size) in [(501, 50), (101, 10)] {
+            let server = MockServer::start().await;
+            let group_time = "2026-01-02T12:34:56Z";
+            let later_time = "2026-01-02T12:34:57Z";
+            let mut rows: Vec<_> = (1..=group_size)
+                .map(|number| listing_row(number, group_time))
+                .collect();
+            rows.push(listing_row(group_size + 1, later_time));
+            Mock::given(method("GET"))
+                .respond_with(move |request: &Request| {
+                    let offset = requested_offset(request);
+                    let page: Vec<_> = rows.iter().skip(offset).take(page_size).cloned().collect();
+                    ResponseTemplate::new(200).set_body_json(
+                        json!({"issues": page, "total": rows.len(), "maxResults": page_size}),
+                    )
+                })
+                .mount(&server)
+                .await;
+            let http = reqwest::Client::builder().no_proxy().build().unwrap();
+            let first = fetch_window(&http, &server.uri(), "X", None, 200, None)
+                .await
+                .unwrap();
+            assert_eq!(first.issues.len(), group_size as usize);
+            assert_eq!(
+                first
+                    .issues
+                    .iter()
+                    .map(|issue| issue.key.clone())
+                    .collect::<Vec<_>>(),
+                (1..=group_size)
+                    .map(|number| format!("X-{number}"))
+                    .collect::<Vec<_>>()
+            );
+            assert_eq!(first.checkpoint, Some(group_time.parse().unwrap()));
+            let next = fetch_window(&http, &server.uri(), "X", first.checkpoint, 200, None)
+                .await
+                .unwrap();
+            assert_eq!(next.issues.len(), 1);
+            assert_eq!(next.issues[0].key, format!("X-{}", group_size + 1));
+            assert_eq!(next.checkpoint, Some(later_time.parse().unwrap()));
+            let requests = server.received_requests().await.unwrap();
+            assert!(requests.iter().any(|request| {
+                requested_offset(request) == 10 * page_size
+                    && request.url.query_pairs().any(|(key, value)| {
+                        key == "jql"
+                    && value
+                        == "project = X AND updated >= \"2026-01-02 12:34\" ORDER BY updated ASC, key ASC"
+                    })
+            }));
+        }
+    }
+
+    #[tokio::test]
+    async fn a_moved_issue_rejects_the_window_and_a_stable_retry_recovers_every_issue() {
+        let server = MockServer::start().await;
+        let moved = AtomicBool::new(false);
+        Mock::given(method("GET"))
+            .respond_with(move |request: &Request| {
+                let offset = requested_offset(request);
+                if offset == 50 {
+                    moved.store(true, Ordering::SeqCst);
+                }
+                let mut rows: Vec<_> = (1..=201)
+                    .map(|number| listing_row(number, "2026-01-02T00:00:00Z"))
+                    .collect();
+                if moved.load(Ordering::SeqCst) {
+                    rows.remove(0);
+                    rows.push(listing_row(1, "2026-01-03T00:00:00Z"));
+                }
+                let page: Vec<_> = rows.into_iter().skip(offset).take(50).collect();
+                ResponseTemplate::new(200).set_body_json(json!({"issues": page, "total": 201}))
+            })
+            .mount(&server)
+            .await;
+        let http = reqwest::Client::builder().no_proxy().build().unwrap();
+        let result = fetch_window(&http, &server.uri(), "X", None, 200, None).await;
+        assert!(result
+            .err()
+            .unwrap()
+            .to_string()
+            .contains("changed during pagination"));
+
+        let recovered = fetch_window(&http, &server.uri(), "X", None, 200, None)
+            .await
+            .unwrap();
+        let mut keys: Vec<_> = recovered.issues.iter().map(|issue| &issue.key).collect();
+        keys.sort();
+        let mut expected_keys: Vec<_> = (2..=201).map(|number| format!("X-{number}")).collect();
+        expected_keys.sort();
+        assert_eq!(keys, expected_keys.iter().collect::<Vec<_>>());
+        assert_eq!(
+            recovered.checkpoint,
+            Some("2026-01-02T00:00:00Z".parse().unwrap())
+        );
+
+        let next = fetch_window(&http, &server.uri(), "X", recovered.checkpoint, 200, None)
+            .await
+            .unwrap();
+        assert_eq!(next.issues.len(), 1);
+        assert_eq!(next.issues[0].key, "X-1");
+        assert_eq!(
+            next.checkpoint,
+            Some("2026-01-03T00:00:00Z".parse().unwrap())
+        );
+    }
+
+    #[tokio::test]
+    async fn an_issue_without_an_update_time_cannot_advance_a_checkpoint() {
+        let server = MockServer::start().await;
+        Mock::given(method("GET"))
+            .respond_with(ResponseTemplate::new(200).set_body_json(json!({
+                "issues": [{"key": "X-1", "fields": {"summary": "Missing time"}}], "total": 1,
+            })))
+            .mount(&server)
+            .await;
+        let http = reqwest::Client::builder().no_proxy().build().unwrap();
+        let result = fetch_window(&http, &server.uri(), "X", None, 200, None).await;
+        assert!(result
+            .err()
+            .unwrap()
+            .to_string()
+            .contains("missing its update time"));
+    }
+
+    #[tokio::test]
+    async fn an_empty_page_cannot_finish_a_group_when_jira_reports_unread_items() {
+        let server = MockServer::start().await;
+        Mock::given(method("GET"))
+            .respond_with(move |request: &Request| {
+                let rows: Vec<_> = if requested_offset(request) == 0 {
+                    (1..=50)
+                        .map(|number| listing_row(number, "2026-01-02T00:00:00Z"))
+                        .collect()
+                } else {
+                    Vec::new()
+                };
+                ResponseTemplate::new(200).set_body_json(json!({"issues": rows, "total": 51}))
+            })
+            .mount(&server)
+            .await;
+        let http = reqwest::Client::builder().no_proxy().build().unwrap();
+        let result = fetch_window(&http, &server.uri(), "X", None, 200, None).await;
+        assert!(result
+            .err()
+            .unwrap()
+            .to_string()
+            .contains("empty page before its total"));
+        assert!(serde_json::from_value::<SearchPage>(json!({"issues": []})).is_err());
+        for response in [
+            json!({"issues": [], "total": -1}),
+            json!({"issues": [listing_row(1, "2026-01-02T00:00:00Z")], "total": 0}),
+        ] {
+            server.reset().await;
+            Mock::given(method("GET"))
+                .respond_with(ResponseTemplate::new(200).set_body_json(response))
+                .mount(&server)
+                .await;
+            let result = fetch_window(&http, &server.uri(), "X", None, 200, None).await;
+            assert!(result
+                .err()
+                .unwrap()
+                .to_string()
+                .contains("inconsistent page total"));
+        }
+    }
 
     /// **Jira 的时间戳不是 RFC3339。** `+0000` 没有冒号，chrono 的默认实现解不了。
     /// 这一条是整个模块最容易悄悄坏掉的地方：解不出来就是整页工单丢掉。
@@ -322,13 +751,14 @@ mod tests {
         let t = DateTime::parse_from_rfc3339("2026-08-30T12:34:56Z")
             .unwrap()
             .with_timezone(&Utc);
-        let q = jql("KAFKA", Some(t));
-        assert!(q.contains("project = KAFKA"), "{q}");
-        assert!(q.contains("updated >= \"2026-08-30 12:34\""), "{q}");
-        assert!(q.contains("ORDER BY updated ASC"), "{q}");
-        // 没有 since 时不该带上 updated 子句。**查子句不查词**——
-        // ORDER BY updated ASC 里也有 "updated"，第一版断言就栽在这上面
-        assert!(!jql("KAFKA", None).contains("updated >="));
+        assert_eq!(
+            jql("KAFKA", Some(t)),
+            "project = KAFKA AND updated >= \"2026-08-30 12:34\" ORDER BY updated ASC, key ASC"
+        );
+        assert_eq!(
+            jql("KAFKA", None),
+            "project = KAFKA ORDER BY updated ASC, key ASC"
+        );
     }
 
     /// **拿真实响应钉住字段形状。**
diff --git a/crates/utopia-server/src/timestamp_checkpoint_tests.rs b/crates/utopia-server/src/timestamp_checkpoint_tests.rs
new file mode 100644
index 000000000..f97ae4561
--- /dev/null
+++ b/crates/utopia-server/src/timestamp_checkpoint_tests.rs
@@ -0,0 +1,495 @@
+//! ADR 0072: persisted progress describes complete source timestamp groups.
+use super::{sync_github_issues_with_client, sync_source};
+use chrono::{DateTime, Utc};
+use serde_json::{json, Value};
+use std::sync::{Arc, Mutex};
+use utopia_store::sources;
+use uuid::Uuid;
+use wiremock::{matchers::method, Mock, MockServer, Request, ResponseTemplate};
+
+const FIRST: &str = "2026-01-02T12:34:20Z";
+const NEXT: &str = "2026-01-02T12:34:40Z";
+
+fn time(value: &str) -> DateTime<Utc> {
+    value.parse().unwrap()
+}
+
+#[derive(Clone, Copy)]
+enum Provider {
+    Github,
+    Jira,
+}
+
+impl Provider {
+    fn kind(self) -> &'static str {
+        match self {
+            Self::Github => "github_issues",
+            Self::Jira => "jira_issues",
+        }
+    }
+
+    fn row(self, number: usize, updated: &str) -> Value {
+        match self {
+            Self::Github => json!({"number":number, "title":format!("Issue {number}"),
+                "state":"open", "body":format!("Body {number}"),
+                "created_at":"2026-01-01T00:00:00Z", "updated_at":updated}),
+            Self::Jira => json!({"key":format!("TEST-{number}"), "fields":{
+                "summary":format!("Issue {number}"), "description":format!("Body {number}"),
+                "created":"2026-01-01T00:00:00Z", "updated":updated}}),
+        }
+    }
+}
+
+#[derive(Default)]
+struct Feed {
+    rows: Vec<Value>,
+    fail_page: Option<usize>,
+    oversized_discussion: Option<(&'static str, usize)>,
+    fail_detail_page: Option<usize>,
+}
+
+struct Fixture {
+    provider: Provider,
+    state: crate::state::AppState,
+    source_id: Uuid,
+    org_id: Uuid,
+    server: MockServer,
+    feed: Arc<Mutex<Feed>>,
+    _directory: tempfile::TempDir,
+}
+
+impl Fixture {
+    async fn new(provider: Provider) -> anyhow::Result<Option<Self>> {
+        let Some(url) = utopia_store::test_db::url() else {
+            return Ok(None);
+        };
+        let pool = sqlx::PgPool::connect(&url).await?;
+        let (org_id, workspace_id, kb_id, source_id) = (
+            Uuid::now_v7(),
+            Uuid::now_v7(),
+            Uuid::now_v7(),
+            Uuid::now_v7(),
+        );
+        sqlx::query("INSERT INTO organizations(id,name) VALUES($1,'timestamp-test')")
+            .bind(org_id)
+            .execute(&pool)
+            .await?;
+        sqlx::query("INSERT INTO workspaces(id,org_id,name) VALUES($1,$2,'timestamp-test')")
+            .bind(workspace_id)
+            .bind(org_id)
+            .execute(&pool)
+            .await?;
+        sqlx::query(
+            "INSERT INTO knowledge_bases(id,workspace_id,name) VALUES($1,$2,'timestamp-test')",
+        )
+        .bind(kb_id)
+        .bind(workspace_id)
+        .execute(&pool)
+        .await?;
+        let server = MockServer::start().await;
+        let config = match provider {
+            Provider::Github => json!({"repo":"test/repo"}),
+            Provider::Jira => json!({"base_url":server.uri(), "project":"TEST"}),
+        };
+        sqlx::query(
+            "INSERT INTO sources(id,kb_id,kind,name,config) VALUES($1,$2,$3,'timestamp-test',$4)",
+        )
+        .bind(source_id)
+        .bind(kb_id)
+        .bind(provider.kind())
+        .bind(config)
+        .execute(&pool)
+        .await?;
+        let directory = tempfile::tempdir()?;
+        let config = utopia_core::config::AppConfig {
+            data_dir: directory.path().to_string_lossy().into_owned(),
+            ..Default::default()
+        };
+        let search = Arc::new(utopia_search::SearchIndex::open(
+            &directory.path().join("search"),
+        )?);
+        let state = crate::state::AppState::new(pool, &config, search, "test-only".into());
+        let feed = Arc::new(Mutex::new(Feed::default()));
+        let current = feed.clone();
+        Mock::given(method("GET")).respond_with(move |request: &Request| {
+            let feed = current.lock().unwrap();
+            let query_number = |key: &str, default: usize| request.url.query_pairs()
+                .find(|(name, _)| name == key)
+                .map(|(_, value)| value.parse::<usize>().unwrap()).unwrap_or(default);
+            let path = request.url.path();
+            if path.ends_with("/comments") || path.ends_with("/events") {
+                if let Some((kind, count)) = feed.oversized_discussion {
+                    if path.ends_with(kind) && path.contains("/issues/1/") {
+                        let page = query_number("page", 1);
+                        if feed.fail_detail_page == Some(page) {
+                            return ResponseTemplate::new(503);
+                        }
+                        let rows: Vec<_> = (1..=count).skip((page - 1) * 100).take(100)
+                            .map(|number| if kind == "comments" {
+                                json!({"issue_url":"https://api.github.com/repos/test/repo/issues/1",
+                                    "created_at":FIRST, "body":format!("Discussion item {number}")})
+                            } else {
+                                json!({"event":"closed", "created_at":FIRST,
+                                    "actor":{"login":format!("Discussion item {number}")}})
+                            }).collect();
+                        return ResponseTemplate::new(200).set_body_json(rows);
+                    }
+                }
+                return ResponseTemplate::new(200).set_body_json(json!([]));
+            }
+            let (offset, limit) = match provider {
+                Provider::Github => ((query_number("page", 1) - 1) * 100, 100),
+                Provider::Jira => (query_number("startAt", 0), query_number("maxResults", 50)),
+            };
+            if feed.fail_page == Some(offset / limit + 1) {
+                return ResponseTemplate::new(503);
+            }
+            // Return the consumed prefix deliberately: provider bounds are inclusive/rounded.
+            let rows: Vec<_> = feed.rows.iter().skip(offset).take(limit).cloned().collect();
+            let body = match provider {
+                Provider::Github => json!(rows),
+                Provider::Jira => json!({"issues":rows,"total":feed.rows.len(),"startAt":offset,"maxResults":limit}),
+            };
+            ResponseTemplate::new(200).set_body_json(body)
+        }).mount(&server).await;
+        Ok(Some(Self {
+            provider,
+            state,
+            source_id,
+            org_id,
+            server,
+            feed,
+            _directory: directory,
+        }))
+    }
+
+    fn set_rows(&self, first_count: usize, next_count: usize) {
+        self.feed.lock().unwrap().rows = (1..=first_count + next_count)
+            .map(|number| {
+                self.provider
+                    .row(number, if number <= first_count { FIRST } else { NEXT })
+            })
+            .collect();
+    }
+
+    async fn sync(&self) -> anyhow::Result<()> {
+        if matches!(self.provider, Provider::Jira) {
+            sync_source(&self.state, self.source_id).await?;
+            return Ok(());
+        }
+        let previous = sources::sync_checkpoint(&self.state.pool, self.source_id).await?;
+        let source = sources::get(&self.state.pool, self.source_id).await?;
+        let run = sources::start_run(&self.state.pool, self.source_id).await?;
+        let http = reqwest::Client::builder().no_proxy().build()?;
+        let stats = sync_github_issues_with_client(
+            &self.state,
+            &source,
+            previous,
+            &http,
+            &self.server.uri(),
+        )
+        .await?;
+        sources::finish_timestamp_sync(
+            &self.state.pool,
+            run,
+            self.source_id,
+            previous,
+            stats.checkpoint,
+            stats.created as i32,
+            stats.updated as i32,
+        )
+        .await?;
+        Ok(())
+    }
+
+    async fn checkpoint(&self) -> anyhow::Result<Option<DateTime<Utc>>> {
+        Ok(sources::sync_checkpoint(&self.state.pool, self.source_id).await?)
+    }
+
+    async fn counts(&self) -> anyhow::Result<(i64, i64, i64)> {
+        Ok(sqlx::query_as("SELECT
+            (SELECT count(*) FROM documents WHERE source_id=$1),
+            (SELECT count(*) FROM document_versions v JOIN documents d ON d.id=v.document_id WHERE d.source_id=$1),
+            (SELECT count(*) FROM jobs j JOIN documents d ON d.id::text=j.payload->>'document_id' WHERE d.source_id=$1)")
+            .bind(self.source_id).fetch_one(&self.state.pool).await?)
+    }
+
+    async fn cleanup(self) -> anyhow::Result<()> {
+        sqlx::query("DELETE FROM jobs WHERE payload->>'document_id' IN (SELECT id::text FROM documents WHERE source_id=$1)")
+            .bind(self.source_id).execute(&self.state.pool).await?;
+        sqlx::query("DELETE FROM organizations WHERE id=$1")
+            .bind(self.org_id)
+            .execute(&self.state.pool)
+            .await?;
+        Ok(())
+    }
+}
+
+async fn large_group_progress(provider: Provider, group_size: usize) -> anyhow::Result<()> {
+    let Some(fixture) = Fixture::new(provider).await? else {
+        return Ok(());
+    };
+    fixture.set_rows(group_size, 1);
+    fixture.sync().await?;
+    assert_eq!(fixture.counts().await?.0, group_size as i64);
+    assert_eq!(fixture.checkpoint().await?, Some(time(FIRST)));
+    fixture.sync().await?;
+    let counts = fixture.counts().await?;
+    assert_eq!(
+        counts,
+        (
+            (group_size + 1) as i64,
+            (group_size + 1) as i64,
+            (group_size + 1) as i64
+        )
+    );
+    assert_eq!(fixture.checkpoint().await?, Some(time(NEXT)));
+    fixture.sync().await?;
+    assert_eq!(fixture.counts().await?, counts);
+    assert_eq!(fixture.checkpoint().await?, Some(time(NEXT)));
+    let requests = fixture.server.received_requests().await.unwrap();
+    let incremental = requests.iter().any(|request| match provider {
+        Provider::Github => request.url.query_pairs().any(|(key, value)| {
+            key == "since"
+                && (time(&value) == time(FIRST)
+                    || time(&value) == time(FIRST) + chrono::Duration::seconds(1))
+        }),
+        Provider::Jira => request
+            .url
+            .query_pairs()
+            .any(|(key, value)| key == "jql" && value.contains("2026-01-02 12:34")),
+    });
+    assert!(
+        incremental,
+        "the next run must send the persisted source bound"
+    );
+    fixture.cleanup().await
+}
+
+#[tokio::test]
+async fn github_imports_1001_equal_time_items_then_skips_the_old_prefix() -> anyhow::Result<()> {
+    large_group_progress(Provider::Github, 1001).await
+}
+
+#[tokio::test]
+async fn jira_imports_501_equal_time_items_then_skips_the_rounded_prefix() -> anyhow::Result<()> {
+    large_group_progress(Provider::Jira, 501).await
+}
+
+#[tokio::test]
+async fn processing_budget_finishes_item_240_and_leaves_the_next_group() -> anyhow::Result<()> {
+    for provider in [Provider::Github, Provider::Jira] {
+        large_group_progress(provider, 240).await?;
+    }
+    Ok(())
+}
+
+#[tokio::test]
+async fn failed_listing_does_not_advance_and_retry_deduplicates() -> anyhow::Result<()> {
+    for provider in [Provider::Github, Provider::Jira] {
+        let Some(fixture) = Fixture::new(provider).await? else {
+            return Ok(());
+        };
+        fixture.set_rows(240, 0);
+        fixture.feed.lock().unwrap().fail_page = Some(2);
+        assert!(fixture.sync().await.is_err());
+        assert_eq!(fixture.checkpoint().await?, None);
+        fixture.feed.lock().unwrap().fail_page = None;
+        fixture.sync().await?;
+        assert_eq!(fixture.counts().await?, (240, 240, 240));
+        assert_eq!(fixture.checkpoint().await?, Some(time(FIRST)));
+        fixture.cleanup().await?;
+    }
+    Ok(())
+}
+
+#[tokio::test]
+async fn github_failed_discussion_tail_preserves_the_document_and_recovers() -> anyhow::Result<()> {
+    for (endpoint, count) in [("comments", 1001), ("events", 1100)] {
+        let Some(fixture) = Fixture::new(Provider::Github).await? else {
+            return Ok(());
+        };
+        fixture.set_rows(1, 0);
+        fixture.sync().await?;
+        let original: String =
+            sqlx::query_scalar("SELECT sha256 FROM documents WHERE source_id=$1")
+                .bind(fixture.source_id)
+                .fetch_one(&fixture.state.pool)
+                .await?;
+        {
+            let mut feed = fixture.feed.lock().unwrap();
+            feed.rows = vec![Provider::Github.row(1, NEXT), Provider::Github.row(2, NEXT)];
+            feed.oversized_discussion = Some((endpoint, count));
+            feed.fail_detail_page = Some(11);
+        }
+        assert!(
+            fixture.sync().await.is_err(),
+            "failed {endpoint} tail must not replace a document"
+        );
+        assert_eq!(fixture.checkpoint().await?, Some(time(FIRST)));
+        let current: String = sqlx::query_scalar("SELECT sha256 FROM documents WHERE source_id=$1")
+            .bind(fixture.source_id)
+            .fetch_one(&fixture.state.pool)
+            .await?;
+        assert_eq!(current, original);
+        assert_eq!(fixture.counts().await?, (1, 1, 1));
+        fixture.feed.lock().unwrap().fail_detail_page = None;
+        fixture.sync().await?;
+        assert_eq!(fixture.checkpoint().await?, Some(time(NEXT)));
+        assert_eq!(fixture.counts().await?, (2, 3, 3));
+        let updated: String = sqlx::query_scalar(
+            "SELECT sha256 FROM documents WHERE source_id=$1 AND external_key='github:test/repo#1'",
+        )
+        .bind(fixture.source_id)
+        .fetch_one(&fixture.state.pool)
+        .await?;
+        let body = String::from_utf8(fixture.state.blob.get(&updated).await?)?;
+        assert!(body.contains(&format!("Discussion item {count}")));
+        fixture.sync().await?;
+        assert_eq!(fixture.counts().await?, (2, 3, 3));
+        fixture.cleanup().await?;
+    }
+    Ok(())
+}
+
+#[tokio::test]
+async fn recovery_ignores_old_success_and_retains_history_after_audit_pruning() -> anyhow::Result<()>
+{
+    for provider in [Provider::Github, Provider::Jira] {
+        let Some(fixture) = Fixture::new(provider).await? else {
+            return Ok(());
+        };
+        fixture.set_rows(1, 0);
+        fixture.sync().await?;
+        let before = fixture.counts().await?;
+        sqlx::query("UPDATE sources SET sync_checkpoint=NULL WHERE id=$1")
+            .bind(fixture.source_id)
+            .execute(&fixture.state.pool)
+            .await?;
+        assert_eq!(fixture.checkpoint().await?, None);
+        fixture.sync().await?;
+        assert_eq!(
+            fixture.counts().await?,
+            before,
+            "recovery must retain versions and jobs"
+        );
+        assert_eq!(fixture.checkpoint().await?, Some(time(FIRST)));
+        for _ in 0..51 {
+            let run = sources::start_run(&fixture.state.pool, fixture.source_id).await?;
+            sources::finish_run(
+                &fixture.state.pool,
+                run,
+                fixture.source_id,
+                Some("fixture failure"),
+                0,
+                0,
+            )
+            .await?;
+        }
+        assert_eq!(
+            sources::list_runs(&fixture.state.pool, fixture.source_id, 100)
+                .await?
+                .len(),
+            50
+        );
+        assert_eq!(fixture.checkpoint().await?, Some(time(FIRST)));
+        fixture.feed.lock().unwrap().rows.clear();
+        fixture.sync().await?;
+        assert_eq!(
+            fixture.checkpoint().await?,
+            Some(time(FIRST)),
+            "empty reads must not advance to wall clock"
+        );
+        assert_eq!(fixture.counts().await?, before);
+        fixture.cleanup().await?;
+    }
+    Ok(())
+}
+
+#[tokio::test]
+async fn checkpoint_and_run_success_commit_atomically() -> anyhow::Result<()> {
+    let Some(fixture) = Fixture::new(Provider::Github).await? else {
+        return Ok(());
+    };
+    let pool = &fixture.state.pool;
+    assert!(sources::finish_timestamp_sync(
+        pool,
+        Uuid::now_v7(),
+        fixture.source_id,
+        None,
+        Some(time(FIRST)),
+        1,
+        0
+    )
+    .await
+    .is_err());
+    assert_eq!(
+        fixture.checkpoint().await?,
+        None,
+        "missing run must roll back source checkpoint"
+    );
+    let run = sources::start_run(pool, fixture.source_id).await?;
+    sources::finish_timestamp_sync(pool, run, fixture.source_id, None, Some(time(FIRST)), 1, 0)
+        .await?;
+    let stale_run = sources::start_run(pool, fixture.source_id).await?;
+    assert!(sources::finish_timestamp_sync(
+        pool,
+        stale_run,
+        fixture.source_id,
+        None,
+        Some(time(NEXT)),
+        1,
+        0
+    )
+    .await
+    .is_err());
+    assert_eq!(fixture.checkpoint().await?, Some(time(FIRST)));
+    let status: String = sqlx::query_scalar("SELECT status FROM source_sync_runs WHERE id=$1")
+        .bind(stale_run)
+        .fetch_one(pool)
+        .await?;
+    assert_eq!(
+        status, "running",
+        "checkpoint conflict must not mark the stale run successful"
+    );
+    fixture.cleanup().await
+}
+
+#[tokio::test]
+async fn a_document_write_failure_keeps_the_checkpoint_and_retry_deduplicates() -> anyhow::Result<()>
+{
+    for provider in [Provider::Github, Provider::Jira] {
+        let Some(fixture) = Fixture::new(provider).await? else {
+            return Ok(());
+        };
+        fixture.set_rows(3, 0);
+        // The constraint affects only this fixture's second document, including on shared test databases.
+        let constraint = format!("checkpoint_write_failure_{}", fixture.source_id.simple());
+        let second_key = match provider {
+            Provider::Github => "github:test/repo#2".to_owned(),
+            Provider::Jira => "jira:127.0.0.1/TEST-2".to_owned(),
+        };
+        sqlx::query(&format!("ALTER TABLE documents ADD CONSTRAINT {constraint} CHECK \
+            (source_id IS DISTINCT FROM '{}'::uuid OR external_key IS DISTINCT FROM '{}') NOT VALID",
+            fixture.source_id, second_key))
+            .execute(&fixture.state.pool).await?;
+        let failed = fixture.sync().await;
+        sqlx::query(&format!(
+            "ALTER TABLE documents DROP CONSTRAINT {constraint}"
+        ))
+        .execute(&fixture.state.pool)
+        .await?;
+        assert!(failed.is_err(), "the second document write must fail");
+        assert_eq!(fixture.checkpoint().await?, None);
+        assert_eq!(
+            fixture.counts().await?,
+            (1, 1, 1),
+            "the first item was committed before the failure"
+        );
+        fixture.sync().await?;
+        assert_eq!(fixture.counts().await?, (3, 3, 3));
+        assert_eq!(fixture.checkpoint().await?, Some(time(FIRST)));
+        fixture.cleanup().await?;
+    }
+    Ok(())
+}
diff --git a/crates/utopia-store/src/sources.rs b/crates/utopia-store/src/sources.rs
index 6eb3d01c0..29e4d7d9a 100644
--- a/crates/utopia-store/src/sources.rs
+++ b/crates/utopia-store/src/sources.rs
@@ -459,6 +459,76 @@ pub async fn touch_sync_time(pool: &PgPool, id: Uuid, at: DateTime<Utc>) -> AppR
 // 同步运行记录（渠道审计历史）
 // ---------------------------------------------------------------------------
 
+/// 工单来源的进度独立于只保留 50 条的审计历史；NULL 也不能回退到旧的 ok 行。
+pub async fn sync_checkpoint(pool: &PgPool, source_id: Uuid) -> AppResult<Option<DateTime<Utc>>> {
+    sqlx::query_scalar("SELECT sync_checkpoint FROM sources WHERE id = $1")
+        .bind(source_id)
+        .fetch_optional(pool)
+        .await?
+        .ok_or(AppError::NotFound)
+}
+
+/// 完整时间戳组的进度与成功记录一起提交，任一写入失败都保留旧游标。
+pub async fn finish_timestamp_sync(
+    pool: &PgPool,
+    run_id: Uuid,
+    source_id: Uuid,
+    previous_checkpoint: Option<DateTime<Utc>>,
+    checkpoint: Option<DateTime<Utc>>,
+    created_docs: i32,
+    updated_docs: i32,
+) -> AppResult<()> {
+    if checkpoint < previous_checkpoint {
+        return Err(AppError::Validation(
+            "Source checkpoint cannot move backwards".into(),
+        ));
+    }
+    let mut tx = pool.begin().await?;
+    let source = sqlx::query(
+        "UPDATE sources SET sync_checkpoint = $3, last_sync_status = 'ok',
+                last_sync_error = NULL, last_sync_added = $4, last_sync_at = now()
+         WHERE id = $1 AND kind IN ('github_issues', 'jira_issues')
+           AND sync_checkpoint IS NOT DISTINCT FROM $2::timestamptz",
+    )
+    .bind(source_id)
+    .bind(previous_checkpoint)
+    .bind(checkpoint)
+    .bind(created_docs + updated_docs)
+    .execute(&mut *tx)
+    .await?;
+    if source.rows_affected() != 1 {
+        return Err(AppError::Conflict(
+            "Source checkpoint changed during sync".into(),
+        ));
+    }
+    let run = sqlx::query(
+        "UPDATE source_sync_runs SET finished_at = now(), status = 'ok', error = NULL,
+                created_docs = $3, updated_docs = $4
+         WHERE id = $1 AND source_id = $2 AND status = 'running'",
+    )
+    .bind(run_id)
+    .bind(source_id)
+    .bind(created_docs)
+    .bind(updated_docs)
+    .execute(&mut *tx)
+    .await?;
+    if run.rows_affected() != 1 {
+        return Err(AppError::Conflict(
+            "Source sync run is no longer running".into(),
+        ));
+    }
+    sqlx::query(
+        "DELETE FROM source_sync_runs WHERE source_id = $1 AND id NOT IN
+         (SELECT id FROM source_sync_runs WHERE source_id = $1
+          ORDER BY started_at DESC LIMIT 50)",
+    )
+    .bind(source_id)
+    .execute(&mut *tx)
+    .await?;
+    tx.commit().await?;
+    Ok(())
+}
+
 /// 调度时钟包含失败尝试，不能作增量游标。回到上次成功运行的开始，
 /// 让那次拉取期间发生的更新也能在下一轮读到；没有成功记录就重新全量读取。
 pub async fn last_successful_sync_start(
diff --git a/docs/decisions/0072-a-source-checkpoint-ends-at-a-timestamp-boundary.md b/docs/decisions/0072-a-source-checkpoint-ends-at-a-timestamp-boundary.md
index d0f0bbbc9..5f143ac41 100644
--- a/docs/decisions/0072-a-source-checkpoint-ends-at-a-timestamp-boundary.md
+++ b/docs/decisions/0072-a-source-checkpoint-ends-at-a-timestamp-boundary.md
@@ -1,6 +1,6 @@
 # A source checkpoint ends at a timestamp boundary
 
-Proposed 2026-10-19 (#1212, #1213); open: implementation and regression coverage.
+In progress 2026-10-25 (#1359, migration 0119); open: custom batch progression and recovery (#1397).
 
 ## Context
 
@@ -27,8 +27,14 @@ This follows the [maintainer's decision](https://github.com/deeplethe/utopia/iss
 
 ### GitHub and Jira
 
-Read items in ascending update-time order. Skip items at or before the
-checkpoint before counting either budget. When the 200-item processing budget
+GitHub/Jira store their watermark in `sources.sync_checkpoint`, initially NULL even
+when successful audit rows exist. Progress and the successful run commit together.
+The 2026-10-23 revision below records the listing verification, complete GitHub
+discussion reads and bounded waits used by these readers.
+
+Read items in ascending update-time order. Jira breaks ties by key so an unchanged
+result keeps the same page boundaries. Skip items at or before the checkpoint
+before counting either budget. When the 200-item processing budget
 or ten-page listing budget is exhausted inside a group, continue reading and
 processing until that group is complete. Stop before consuming the next
 timestamp group, then save the completed group's time as the checkpoint.
@@ -58,6 +64,45 @@ group cap or a group-size alert; use the existing rate-limit backoff. The
 exception applies to source listing and processing, not permission to replace
 a GitHub document with a truncated comments or events history.
 
+### Revision 2026-10-23: verification, complete discussions and bounded waits
+
+Keep the second listing pass before ingestion. Re-read through the selected
+timestamp boundary and compare the listed identities and update times with the
+first pass, including skipped old rows and excluded GitHub PRs. Those rows can
+still move offsets. For example, after reading issues 1–100, updating issue 1
+can move it to the tail; the next page then starts at issue 102 and silently
+misses issue 101. Advancing the checkpoint without detecting that movement can
+make issue 101 unreachable on later incremental runs. Jira has the same risk.
+
+The second pass roughly doubles requests for the selected listing prefix, not
+GitHub's per-ticket comments and events requests. It does not extend the chosen
+window, and changes only beyond that boundary do not invalidate the prefix.
+Comparison ignores ordering within one timestamp group. A changed identity or
+update time fails the run before ingestion, preserves the previous checkpoint,
+sets `last_sync_status` to `failed` and raises the existing `source.sync_failed`
+alert. A later run starts at the same checkpoint. Keep this failure behavior;
+do not add repeated scans inside the same run. This detects observed movement,
+not an atomic snapshot. A continuously changing prefix can repeatedly fail;
+progress under sustained concurrent updates is not guaranteed.
+
+GitHub comments and events must be read to the end, even beyond ten pages.
+The listing budget is not a limit on a selected ticket's discussion history.
+Do not skip a large discussion, return truncated content or advance progress
+past a ticket whose details could not be read completely. A detail-page failure
+keeps the old document and checkpoint. Large discussions therefore increase
+requests, memory use and run duration. Jira's search `total` remains required:
+without it, a premature empty page could be mistaken for a complete group.
+
+For explicit rate limits, keep at most three retries of the current request,
+preserving its in-memory position. Honor a valid wait hint only when it is at
+most 30 seconds per retry, retaining the existing minimum backoff. An invalid
+or longer hint fails the run immediately without sleeping or shortening the
+provider's requested wait. This prevents one long wait from occupying a sync
+slot for hours; it is not a total run-duration limit. The existing job retry
+and source schedule govern later attempts and do not persist `Retry-After`, so
+they do not guarantee that a later attempt waits until the provider's deadline.
+Strict cross-job rate-limit scheduling remains outside this revision.
+
 ### Custom
 
 The endpoint already returns the whole requested response. Walk its items in
@@ -73,7 +118,7 @@ incomplete runs. No source-update field, pagination token or saved response is
 added to the custom protocol; `doc_time` remains document time.
 
 The custom fix is being reviewed in
-[PR #1246](https://github.com/deeplethe/utopia/pull/1246). It does not resolve
+[PR #1397](https://github.com/deeplethe/utopia/pull/1397). It does not resolve
 the GitHub and Jira parts of #1212.
 
 ## Recovery and tradeoffs
@@ -120,6 +165,12 @@ must demonstrate:
   this progress without omissions or duplicate versions and processing jobs.
 - A failed page fetch, detail fetch or write cannot advance the checkpoint;
   retry recovers the whole group. Updates during pagination remain reachable.
+- A page-moving update rejects the window; once the listing stabilizes, a
+  retry recovers the skipped issue and a later run reads the moved issue.
+- GitHub comments and events beyond ten pages are fully read; failure on a
+  later detail page preserves the old document and checkpoint.
+- A 30-second rate-limit hint is accepted; a 31-second hint returns an error
+  without sleeping or retrying that request. Cover seconds and HTTP-date hints.
 - Custom's 201 changed items finish on the second run; 450 finish over three
   capped runs. Cover an unchanged prefix and no custom update field.
 - The one-time recovery scan preserves existing data and returns to incremental
diff --git a/docs/decisions/README.md b/docs/decisions/README.md
index 64b16381b..3ac9cafb3 100644
--- a/docs/decisions/README.md
+++ b/docs/decisions/README.md
@@ -115,7 +115,7 @@ By domain; the domains are the files of [../design/](../design/README.md). **Sta
 | 0052 | [Document content is a read contract over the retained ledger](0052-document-content-is-a-read-contract.md) | Proposed |  |
 | 0054 | [A source may push statements in the open contract](0054-a-source-may-push-statements-in-the-open-contract.md) | Implemented |  |
 | 0065 | [The OCR reader chooses its provider](0065-readers-choose-their-provider.md) | Implemented |  |
-| 0072 | [A source checkpoint ends at a timestamp boundary](0072-a-source-checkpoint-ends-at-a-timestamp-boundary.md) | Proposed |  |
+| 0072 | [A source checkpoint ends at a timestamp boundary](0072-a-source-checkpoint-ends-at-a-timestamp-boundary.md) | In progress |  |
 | 0074 | [A Google Drive source is connected by a person](0074-a-google-drive-source-is-connected-by-a-person.md) | Implemented |  |
 
 ### [lakehouse-and-actions](../design/lakehouse-and-actions.md)
diff --git a/docs/design/sources.md b/docs/design/sources.md
index 8af6d16c9..5cd6fca9b 100644
--- a/docs/design/sources.md
+++ b/docs/design/sources.md
@@ -28,6 +28,23 @@ from a `HEAD` on the file before anything is downloaded [#1327].
 
 **Google Drive.** A person connects it with OAuth and it reads Drive as them; the refresh token is a sealed source secret and the browser never holds a Google token; the flow is bound to the session that started it; only the person who connected picks folders, subfolders come along, and a sync stops once that person is no longer an editor; a run downloads at most 2000 files and says so when it stops there; Docs, Sheets and Slides are exported to DOCX, XLSX and PPTX; change is Drive's marker (`md5Checksum`, or `modifiedTime` for Google formats) because an export's bytes differ every time; "not in source" only after a complete walk [0074].
 
+**Ticket checkpoints.** GitHub/Jira read by ascending update time and stop only after
+a whole timestamp group, even when that exceeds 200 items or ten listing pages.
+Already consumed timestamps use neither budget. Their private source checkpoint is
+committed with the successful run, independent of retained audit history; empty or
+failed runs do not advance it. Upgrades start these two readers from NULL and reuse
+document/version/job deduplication. A second listing pass checks the selected prefix
+for page movement before ingestion, roughly doubling listing requests. A changed
+prefix fails with the existing sync alert and keeps the old checkpoint; neither
+API provides a snapshot, and a constantly changing prefix may repeatedly fail.
+GitHub reads selected tickets' comments and events to the end, including beyond
+ten pages, before replacing a document. Explicit rate limits retry the current
+request at most three times with waits of at most 30 seconds each; longer hints
+fail immediately. Later ordinary job retries and source schedules do not preserve
+the provider's `Retry-After` deadline. Jira requires the search `total` to detect
+premature pagination endings. Custom batch progression is separate work in #1397
+[0072, revision 2026-10-23].
+
 **Statements.** A `statements` source takes the open extraction contract itself (`e`, `s`, `n`) on `POST /sources/{id}/statements`, with the `api` push's identity, versions, tombstones and run history; the payload is the document and its single chunk, and extraction parses the chunk instead of prompting a model, then runs the unchanged path, so a pushed statement is an open statement with the item as its evidence (null offsets) and reaches the typed graph only through alignment. The contract has no slot for a property or a class and unknown keys are refused; a second push under the same identity marks the earlier statements stale rather than closing them; tables belong on a mount, not here [0054].
 
 **RSS.** Observations are not documents: one table with baseline, candidate and no_source rows;
diff --git a/migrations/0119_a_source_checkpoint_ends_at_a_timestamp_boundary.sql b/migrations/0119_a_source_checkpoint_ends_at_a_timestamp_boundary.sql
new file mode 100644
index 000000000..5659156b0
--- /dev/null
+++ b/migrations/0119_a_source_checkpoint_ends_at_a_timestamp_boundary.sql
@@ -0,0 +1,4 @@
+-- 不从旧的 ok 审计记录回填：旧版本可能只读完前 200 条就记成功。
+-- GitHub/Jira 从 NULL 开始恢复扫描，保留全部文档、版本、任务与审计历史。
+-- Custom 的批次推进由 #1397 单独实现，这里不改变它的游标。
+ALTER TABLE sources ADD COLUMN sync_checkpoint timestamptz;