diff --git a/backend/src/alphapulse/api/crawler.py b/backend/src/alphapulse/api/crawler.py
index 3881564..7b319d2 100644
--- a/backend/src/alphapulse/api/crawler.py
+++ b/backend/src/alphapulse/api/crawler.py
@@ -185,17 +185,22 @@ async def list_sources(session: Annotated[AsyncSession, Depends(get_session)]) -
         select(NewsItem.source_code, func.count()).group_by(NewsItem.source_code)
     )).all())
     latest_runs = (await session.scalars(select(CrawlRun).order_by(CrawlRun.source_code, CrawlRun.started_at.desc()).distinct(CrawlRun.source_code))).all()
-    run_stats = {run.source_code: run.stats for run in latest_runs}
+    run_stats = {run.source_code: {**run.stats, 'round_running': int(run.status == 'running')} for run in latest_runs}
     statuses = await redis.hgetall(STATUS_KEY)
     db_size = await session.scalar(select(func.pg_database_size(func.current_database())))
     attachment_stats = {}
     for source in sources:
         retries = (source.history_cursor or {}).get('attachment_retries', {})
         pending = sum(value['attempts'] < 3 for value in retries.values())
+        documents = (source.history_cursor or {}).get('document_retries', {})
+        document_pending = sum(value['attempts'] < 3 for value in documents.values())
         attachment_stats[source.code] = {
             'attachment_gaps': len(retries),
             'attachment_retry_pending': pending,
             'attachment_retry_exhausted': len(retries) - pending,
+            'document_gaps': len(documents),
+            'document_retry_pending': document_pending,
+            'document_retry_exhausted': len(documents) - document_pending,
         }
     return success(SourcesOut(
         manager_alive=bool(statuses),
diff --git a/backend/src/alphapulse/crawler/spiders/base.py b/backend/src/alphapulse/crawler/spiders/base.py
index 1472c75..221425b 100644
--- a/backend/src/alphapulse/crawler/spiders/base.py
+++ b/backend/src/alphapulse/crawler/spiders/base.py
@@ -7,6 +7,7 @@ import uuid
 from collections.abc import Callable, Coroutine, Iterable
 from dataclasses import dataclass, field
 from datetime import UTC, datetime, timedelta
+from email.utils import parsedate_to_datetime
 from typing import Any
 from zoneinfo import ZoneInfo
 
@@ -67,6 +68,9 @@ class RunState:
     history_deadline: float = 0
     official_cursor: dict | None = None
     stats: dict = field(default_factory=dict)
+    completed_documents_safe: bool = False
+    cooldown_until: datetime | None = None
+    document_retries_started: set[str] = field(default_factory=set)
 
     @property
     def full_scan(self) -> bool:
@@ -188,6 +192,7 @@ class NewsSpider(RedisSpider):
 
     def _on_item_error(self, failure: Failure) -> None:
         if self.run:
+            self.run.completed_documents_safe = False
             self.run.errors.append(f"新闻处理或落库失败：{failure.value!r}")
 
     async def _on_closed(self) -> None:
@@ -238,12 +243,21 @@ class NewsSpider(RedisSpider):
                 if history_saved:
                     cursor["history"] = run.official_cursor.get("history", {})
                 source.history_cursor = cursor
+            elif failed and run.official_cursor is not None and run.completed_documents_safe:
+                # Retain completed reads and explicit gaps, never failed lane positions.
+                cursor = dict(source.history_cursor or {})
+                for key in ('checked', 'document_retries', 'attachment_retries', 'bill_retries'):
+                    if key in run.official_cursor:
+                        cursor[key] = run.official_cursor[key]
+                source.history_cursor = cursor
             if history_saved:
                 source.history_position = run.history_position
                 source.history_done = run.history_done
             if failed:
                 source.consecutive_failures += 1
                 delay = min(BACKOFF_BASE_SECONDS * 2 ** (source.consecutive_failures - 1), BACKOFF_MAX_SECONDS)
+                if run.cooldown_until:
+                    delay = max(delay, int((run.cooldown_until - now).total_seconds()) + 1)
                 source.paused_until = now + timedelta(seconds=delay)
                 source.last_error = run.errors[0][:1000]
             else:
@@ -293,10 +307,22 @@ class NewsSpider(RedisSpider):
             headers=getattr(self, "request_headers", None),
         )
 
+    def block_until(self, until: datetime) -> None:
+        self.run.blocked = True
+        self.run.cooldown_until = max(self.run.cooldown_until or until, until)
+
     def on_request_error(self, failure: Failure) -> None:
         status = failure.value.response.status if failure.check(HttpError) else None
         if status in (403, 429):
-            self.run.blocked = True
+            now = datetime.now(UTC)
+            self.block_until(now + timedelta(seconds=BACKOFF_MAX_SECONDS))
+            retry_after = failure.value.response.headers.get('Retry-After', b'').decode('ascii', errors='ignore')
+            if retry_after:
+                try:
+                    until = now + timedelta(seconds=int(retry_after)) if retry_after.isdigit() else parsedate_to_datetime(retry_after)
+                    self.block_until(until)
+                except (ValueError, TypeError, OverflowError):
+                    self.logger.warning('忽略无效 Retry-After 响应头')
         message = f"请求失败 {failure.request.url}：{status or repr(failure.value)}"
         self.run.errors.append(message)
         self.logger.error(message)
diff --git a/backend/src/alphapulse/crawler/spiders/us_official.py b/backend/src/alphapulse/crawler/spiders/us_official.py
index 5224add..1ec3d5b 100644
--- a/backend/src/alphapulse/crawler/spiders/us_official.py
+++ b/backend/src/alphapulse/crawler/spiders/us_official.py
@@ -12,6 +12,7 @@ from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit
 
 from scrapy import Request
 from scrapy.http import Response
+from scrapy.spidermiddlewares.httperror import HttpError
 from sqlalchemy import select
 
 from alphapulse.calendar.umich import pdf_text
@@ -105,6 +106,7 @@ class USOfficialSpider(NewsSpider):
             self._starting = False
 
     def first_requests(self):
+        self.run.completed_documents_safe = True
         if self.name == 'doj' and self.cursor.get('transport') != 'doj-api-v1':
             # Retain retry evidence; HTML discovery cursors cannot use API pagination.
             self.cursor = {k: v for k, v in self.cursor.items() if k not in ('history', 'realtime', 'checked')}
@@ -115,9 +117,18 @@ class USOfficialSpider(NewsSpider):
             self.run.errors.append(error)
             return []
         state = self.cursor.setdefault('realtime', {'entry': 0})
+        retries = self.cursor.get('document_retries', {})
+        due_documents = [{**value['document'], 'document_retry': True} for value in retries.values() if value['attempts'] < 3 and value['next_at'] <= self.run.started_at.timestamp()]
+        if due_documents:
+            state.setdefault('pending', []).extend(copy.deepcopy(d) for d in due_documents if d['url'] not in {p['url'] for p in state['pending']})
         due = [value['document'] for value in self.cursor.get('attachment_retries', {}).values() if value['attempts'] < 3 and value['next_at'] <= self.run.started_at.timestamp()]
-        if due and not state.get('pending'):
-            state['pending'] = [{**doc, 'attachment_retry': True} for doc in due]
+        for doc in due:
+            pending = state.setdefault('pending', [])
+            existing = next((value for value in pending if value['url'] == doc['url']), None)
+            if existing is None:
+                pending.append({**copy.deepcopy(doc), 'attachment_retry': True})
+            else:
+                existing['attachment_retry'] = True
         bill_due = [{**v['document'], 'bill_retry': True} for v in self.cursor.get('bill_retries', {}).values() if v['attempts'] < 3 and v['next_at'] <= self.run.started_at.timestamp()]
         if bill_due:
             state.setdefault('pending', []).extend(d for d in bill_due if d['url'] not in {p['url'] for p in state['pending']})
@@ -136,12 +147,21 @@ class USOfficialSpider(NewsSpider):
 
     def advance(self, lane):
         state = self.cursor[lane]
+        if self.run.blocked:
+            return []
         if configuration_error(self.name):
             return []
         if lane == 'history' and time.monotonic() >= self.run.history_deadline:
             return []
         if state.get('pending'):
             doc = state['pending'][0]
+            gap = self.cursor.get('document_retries', {}).get(doc['url'])
+            if doc['url'] in self.run.document_retries_started or (gap and (gap['attempts'] >= 3 or gap['next_at'] > self.run.started_at.timestamp())):
+                state['pending'].pop(0)
+                return self.advance(lane)
+            if gap:
+                doc['document_retry'] = True
+                self.run.document_retries_started.add(doc['url'])
             if doc['kind'] == 'bill_finish':
                 self.finish_bill(lane)
                 return self.advance(lane)
@@ -213,6 +233,8 @@ class USOfficialSpider(NewsSpider):
         elif self.name in ('congress', 'federal_register'):
             data = response.json()
             key = ('committeeMeetings' if '/committee-meeting' in response.url else 'bills') if self.name == 'congress' else 'results'
+            if self.name == 'federal_register' and key not in data and type(data.get('count')) is int and data['count'] == 0 and isinstance(data.get('description'), str):
+                data[key] = []
             if key not in data or not isinstance(data[key], list):
                 raise ValueError('官方 API 列表结构未识别')
             if self.name == 'federal_register' and data.get('count', 0) > 2000:
@@ -228,6 +250,10 @@ class USOfficialSpider(NewsSpider):
                     docs.append({'url': f"https://www.federalregister.gov/api/v1/documents/{doc['document_number']}.json", 'kind': 'register'})
             next_url = data.get('pagination', {}).get('next') if self.name == 'congress' else data.get('next_page_url')
         else:
+            heading = ' '.join(response.css('title, h1').xpath('normalize-space(.)').getall())
+            if re.search(r'just a moment|access denied|verify (?:you are|your)|captcha', heading, re.I) or response.css('#challenge-form, #cf-challenge-running'):
+                self.block_until(datetime.now(UTC) + timedelta(minutes=15))
+                raise ValueError('官方列表返回验证页面')
             if not response.css('h1'):
                 raise ValueError('官方列表页面未识别')
             pattern = SOURCES[self.name].article_path
@@ -238,6 +264,16 @@ class USOfficialSpider(NewsSpider):
                     continue
                 if re.search(pattern, urlsplit(url).path) and url not in SOURCES[self.name].entries and not re.search(r'/\d{4}-(?:press|speeches|testimony)\.htm$', urlsplit(url).path) and not url.lower().endswith('.pdf'):
                     doc = {'url': url, 'kind': 'article'}
+                    if self.name in ('treasury', 'cftc', 'fincen', 'whitehouse'):
+                        container = {
+                            'treasury': 'ancestor::div[h3][1]',
+                            'cftc': 'ancestor::tr[1]',
+                            'fincen': 'ancestor::div[contains(@class,"fincen-news-article__content")][1]',
+                            'whitehouse': 'ancestor::div[contains(@class,"wp-block-whitehouse-post-template__content")][1]',
+                        }[self.name]
+                        dates = a.xpath(container + '//time/@datetime').getall()
+                        if len(dates) == 1:
+                            doc['listing_date'] = published(dates[0])[0].isoformat()
                     if self.name == 'fed' and '/monetarypolicy/fomcminutes' in url:
                         # The URL identifies the meeting; the calendar gives the release day.
                         match = re.search(r'Released\s+([A-Za-z]+ \d{1,2}, \d{4})', a.xpath('normalize-space(parent::*)').get() or '')
@@ -261,6 +297,12 @@ class USOfficialSpider(NewsSpider):
                     return True
                 return day >= begin.date()
             docs = [doc for doc in docs if in_range(doc)]
+        if self.name in ('treasury', 'cftc', 'fincen', 'whitehouse') and docs:
+            begin = self.run.history_target if lane == 'history' else self.run.cutoff
+            dates = [published(d['listing_date'])[0] for d in docs if d.get('listing_date')]
+            if len(dates) == len(docs) and dates == sorted(dates, reverse=True) and dates[-1].date() < begin.date():
+                next_url = None
+            docs = [d for d in docs if not d.get('listing_date') or published(d['listing_date'])[0].date() >= begin.date()]
         if lane == 'realtime':
             # OFAC's recent-actions entry is a descending day-indexed feed.
             if self.name == 'ofac' and docs:
@@ -271,6 +313,9 @@ class USOfficialSpider(NewsSpider):
             threshold = (self.run.started_at - timedelta(hours=24)).timestamp()
             self.cursor['checked'] = {k: v for k, v in checked.items() if v >= threshold}
             docs = [d for d in docs if self.cursor['checked'].get(d['url'], 0) < threshold]
+        retries = self.cursor.get('document_retries', {})
+        docs = [d for d in docs if d['url'] not in retries or (retries[d['url']]['attempts'] < 3 and retries[d['url']]['next_at'] <= self.run.started_at.timestamp())]
+        docs = [{**d, 'document_retry': True} if d['url'] in retries else d for d in docs]
         if next_url and official_url(self.name, next_url, response.url) == response.url:
             raise ValueError('官方列表翻页重复当前页面，未推进断点')
         visited = state.setdefault('visited_pages', [])
@@ -287,8 +332,54 @@ class USOfficialSpider(NewsSpider):
             state['day_complete'] = True
         yield from self.advance(lane)
 
+    def document_failed(self, request, error):
+        doc = request.meta.get('document')
+        if not doc or doc['kind'] not in ('article', 'doj'):
+            return None
+        state = self.cursor[request.meta['lane']]
+        if not state.get('pending') or state['pending'][0]['url'] != doc['url']:
+            return None
+        retries = self.cursor.setdefault('document_retries', {})
+        previous = retries.get(doc['url'], {})
+        now = datetime.now(UTC)
+        if not previous or previous['next_at'] <= now.timestamp():
+            retries[doc['url']] = {'document': copy.deepcopy(doc), 'attempts': previous.get('attempts', 0) + 1,
+                                  'next_at': (now + timedelta(hours=1)).timestamp(), 'error': error}
+        self.run.document_retries_started.add(doc['url'])
+        self.run.errors.append(error)
+        self.logger.error('文档缺口：%s；%s', doc['url'], error)
+        state['pending'].pop(0)
+        return self.advance(request.meta['lane'])
+
+    def on_request_error(self, failure):
+        status = failure.value.response.status if failure.check(HttpError) else None
+        if status not in (403, 429):
+            requests = self.document_failed(failure.request, f'请求失败 {failure.request.url}：{status or repr(failure.value)}')
+            if requests is not None:
+                return requests
+        return super().on_request_error(failure)
+
     async def parse_document(self, response):
+        try:
+            async for result in self._parse_document(response):
+                yield result
+        except (ValueError, KeyError, TypeError) as exc:
+            if '验证' in str(exc) or '登录页面' in str(exc):
+                self.block_until(datetime.now(UTC) + timedelta(minutes=15))
+            requests = self.document_failed(response.request, f'解析异常 {response.url}：{exc!r}')
+            if requests is None:
+                raise
+            for request in requests:
+                yield request
+
+    async def _parse_document(self, response):
         lane, doc = response.meta['lane'], response.meta['document']
+        begin = self.run.history_target if lane == 'history' else self.run.cutoff
+        if not (doc.get('document_retry') or doc.get('attachment_retry')) and doc.get('listing_date') and published(doc['listing_date'])[0].date() < begin.date():
+            self.run.stats['outside_range'] = self.run.stats.get('outside_range', 0) + 1
+            for result in self.complete(lane):
+                yield result
+            return
         if doc['kind'] == 'register_issue':
             root = etree.fromstring(response.body, etree.XMLParser(resolve_entities=False, no_network=True))
             if root.tag != 'FEDREG' or published(root.findtext('DATE', default=''))[0].date().isoformat() != doc['day']:
@@ -459,7 +550,7 @@ class USOfficialSpider(NewsSpider):
                     context.setdefault('unmatched_texts', []).append(copy.deepcopy(doc))
             parsed['extra']['selection_reasons'] = context['reasons']
         begin = self.run.history_target if lane == 'history' else self.run.cutoff
-        if doc.get('attachment_retry') or (doc.get('bill') and self.cursor[lane]['bill'].get('retry')):
+        if doc.get('attachment_retry') or doc.get('document_retry') or (doc.get('bill') and self.cursor[lane]['bill'].get('retry')):
             begin = datetime(2025, 1, 1, tzinfo=UTC)
         outside_range = parsed['published_at'].date() < begin.date() if parsed['extra'].get('published_at_precision') == 'date' else parsed['published_at'] < begin
         if outside_range:
@@ -487,7 +578,7 @@ class USOfficialSpider(NewsSpider):
         retries = self.cursor.setdefault('bill_retries', {})
         if not context['bodies'] or context.get('missing_material'):
             previous = retries.get(context['url'], {})
-            retries[context['url']] = {'document': {'url': context['url'], 'kind': 'bill'}, 'version': context['version'], 'attempts': previous.get('attempts', 0) + 1, 'next_at': (self.run.started_at + timedelta(hours=1)).timestamp()}
+            retries[context['url']] = {'document': {'url': context['url'], 'kind': 'bill'}, 'version': context['version'], 'attempts': previous.get('attempts', 0) + 1, 'next_at': (datetime.now(UTC) + timedelta(hours=1)).timestamp()}
         else:
             retries.pop(context['url'], None)
             if not context['reasons']:
@@ -511,7 +602,7 @@ class USOfficialSpider(NewsSpider):
         retries = self.cursor.setdefault('attachment_retries', {})
         if incomplete and retry_doc:
             previous = retries.get(parsed['url'], {})
-            retries[parsed['url']] = {'document': retry_doc, 'attempts': previous.get('attempts', 0) + 1, 'next_at': (self.run.started_at + timedelta(hours=1)).timestamp()}
+            retries[parsed['url']] = {'document': retry_doc, 'attempts': previous.get('attempts', 0) + 1, 'next_at': (datetime.now(UTC) + timedelta(hours=1)).timestamp()}
         elif not incomplete:
             retries.pop(parsed['url'], None)
         if reasons:
@@ -546,6 +637,7 @@ class USOfficialSpider(NewsSpider):
 
     def complete(self, lane):
         doc = self.cursor[lane]['pending'].pop(0)
+        self.cursor.get('document_retries', {}).pop(doc['url'], None)
         if lane == 'realtime':
             self.cursor.setdefault('checked', {})[doc['url']] = self.run.started_at.timestamp()
         return self.advance(lane)
@@ -556,6 +648,7 @@ class SecSpider(USOfficialSpider):
 
 
 class CftcSpider(USOfficialSpider):
+    custom_settings = {**USOfficialSpider.custom_settings, 'DOWNLOAD_DELAY': 3.0, 'CONCURRENT_REQUESTS': 1}
     name = 'cftc'
 
 
diff --git a/backend/tests/fixtures/us_news/collection_failures.json.gz b/backend/tests/fixtures/us_news/collection_failures.json.gz
new file mode 100644
index 0000000..fec6ab5
Binary files /dev/null and b/backend/tests/fixtures/us_news/collection_failures.json.gz differ
diff --git a/backend/tests/test_us_official_news.py b/backend/tests/test_us_official_news.py
index f7e05af..b5ab880 100644
--- a/backend/tests/test_us_official_news.py
+++ b/backend/tests/test_us_official_news.py
@@ -9,6 +9,8 @@ from types import SimpleNamespace
 import pytest
 from scrapy import Request
 from scrapy.http import HtmlResponse, Response
+from scrapy.spidermiddlewares.httperror import HttpError
+from twisted.python.failure import Failure
 
 from alphapulse.core.us_news import SOURCES, article, official_url, published, selection
 
@@ -31,6 +33,314 @@ def response(url, body, **meta):
     return HtmlResponse(url, body=body.encode(), encoding='utf-8', request=Request(url, meta={'raw_response_id': 12, **meta}))
 
 
+def collection_sample(source, suffix=''):
+    import gzip
+    from pathlib import Path
+    samples = json.load(gzip.open(Path(__file__).parent / 'fixtures/us_news/collection_failures.json.gz', 'rt'))
+    return next(s for s in samples if s['source'] == source and s['url'].endswith(suffix))
+
+
+def test_real_empty_register_day_continues_to_previous_day(trading):
+    obj = spider(trading, 'federal_register')
+    obj.cursor['realtime']['day'] = '2026-11-05'
+    sample = collection_sample('federal_register')
+    request, = obj.parse_listing(response(sample['url'], sample['body'], lane='realtime'))
+    assert '2026-11-04' in request.url
+    assert not obj.run.errors
+
+
+@pytest.mark.parametrize('data', [{'count': 2}, {'count': '0', 'description': 'x'}, {'count': 0}, {'count': 0, 'description': 'x', 'results': None}])
+def test_register_does_not_hide_malformed_responses(trading, data):
+    obj = spider(trading, 'federal_register')
+    with pytest.raises(ValueError, match='列表结构'):
+        list(obj.parse_listing(response(SOURCES[obj.name].entries[0], json.dumps(data), lane='realtime')))
+
+
+@pytest.mark.parametrize('source', ['treasury', 'cftc', 'fincen', 'whitehouse'])
+def test_real_realtime_lists_filter_dated_old_documents(trading, source):
+    obj = spider(trading, source)
+    sample = collection_sample(source, '/press-releases' if source == 'fincen' else '')
+    # All sample pages precede January 2027; no old body should be requested.
+    obj.run.cutoff = datetime(2027, 1, 1, tzinfo=UTC)
+    requests = list(obj.parse_listing(response(sample['url'], sample['body'], lane='realtime')))
+    pending = obj.cursor['realtime'].get('pending', [])
+    assert not pending
+    assert all(not r.meta.get('document') for r in requests)
+
+
+def test_treasury_real_realtime_stops_at_date_boundary_before_broken_pagination(trading):
+    obj = spider(trading, 'treasury')
+    sample = collection_sample('treasury')
+    obj.cursor['realtime']['page'] = sample['url']
+    requests = list(obj.parse_listing(response(sample['url'], sample['body'], lane='realtime')))
+    assert requests and requests[0].meta['document']['kind'] == 'article'
+    assert obj.cursor['realtime']['next'] is None
+    assert all(published(d['listing_date'])[0].date() >= obj.run.cutoff.date() for d in obj.cursor['realtime']['pending'])
+    # The same response cannot establish historical coverage through 2025.
+    with pytest.raises(ValueError, match='重复当前页面'):
+        list(obj.parse_listing(response(sample['url'], sample['body'], lane='history')))
+
+
+def test_undated_or_unordered_lists_do_not_stop_early(trading):
+    obj = spider(trading, 'treasury')
+    url = SOURCES[obj.name].entries[0]
+    body = '<h1>News</h1><div><time datetime="2025-01-01"></time><h3><a href="/news/press-releases/sb0001">Old</a></h3></div><a href="/news/press-releases/sb0002">Undated</a><a rel="next" href="?page=2">Next</a>'
+    list(obj.parse_listing(response(url, body, lane='realtime')))
+    assert obj.cursor['realtime']['next'].endswith('?page=2')
+    assert obj.cursor['realtime']['pending'][0]['url'].endswith('sb0002')
+
+
+def test_failed_document_does_not_block_next_document_or_retry_immediately(trading):
+    obj = spider(trading, 'doj')
+    docs = [{'url': f'https://www.justice.gov/api/v1/press_releases/{uuid.uuid4()}.json', 'kind': 'doj'} for _ in range(2)]
+    obj.cursor['realtime']['pending'] = copy.deepcopy(docs)
+    request = Request(docs[0]['url'], meta={'document': docs[0], 'lane': 'realtime'})
+    failed = Failure(HttpError(Response(request.url, status=503, request=request)))
+    failed.request = request
+    following, = obj.on_request_error(failed)
+    assert following.url == docs[1]['url']
+    assert obj.run.errors and docs[0]['url'] not in obj.cursor.get('checked', {})
+    gap = obj.cursor['document_retries'][docs[0]['url']]
+    assert gap['document'] == docs[0] and gap['attempts'] == 1
+    assert gap['next_at'] > obj.run.started_at.timestamp()
+
+
+@pytest.mark.parametrize('attempts,due,expected', [(1, False, False), (3, True, False), (1, True, True)])
+def test_document_retry_is_bounded_and_survives_rounds(trading, attempts, due, expected):
+    obj = spider(trading, 'ofac')
+    url = 'https://ofac.treasury.gov/recent-actions/20261104'
+    obj.cursor['document_retries'] = {url: {'document': {'url': url, 'kind': 'article'}, 'attempts': attempts, 'next_at': obj.run.started_at.timestamp() + (-60 if due else 3600), 'error': 'HTTP503'}}
+    listing = '<h1>Actions</h1><a href="'+url+'">Sanctions</a>'
+    requests = list(obj.parse_listing(response(SOURCES[obj.name].entries[0], listing, lane='realtime')))
+    assert any(r.url == url for r in requests) is expected
+    assert url in obj.cursor['document_retries']
+
+
+def test_old_pending_cursor_cannot_bypass_exhausted_document_retry(trading):
+    obj = spider(trading)
+    doc = {'url': 'https://ofac.treasury.gov/recent-actions/20261104', 'kind': 'article'}
+    obj.cursor['realtime']['pending'] = [doc]
+    obj.cursor['document_retries'] = {doc['url']: {'document': doc, 'attempts': 3, 'next_at': 0, 'error': 'unavailable'}}
+    requests = obj.advance('realtime')
+    assert all(r.url != doc['url'] for r in requests)
+    assert doc['url'] in obj.cursor['document_retries']
+
+
+def test_due_retry_in_old_pending_cursor_is_not_discarded_by_moving_window(trading):
+    obj = spider(trading)
+    doc = {'url': 'https://ofac.treasury.gov/recent-actions/20250101', 'kind': 'article', 'listing_date': '2025-01-01'}
+    obj.cursor['realtime']['pending'] = [doc]
+    obj.cursor['document_retries'] = {doc['url']: {'document': copy.deepcopy(doc), 'attempts': 1, 'next_at': 0, 'error': 'unavailable'}}
+    request, = obj.first_requests()
+    assert request.meta['document']['document_retry']
+    body = '<h1>Crypto sanctions</h1><main><article class="node--type-ofac-recent-action"><div class="field--name-field-body"><p>New sanctions against a virtual currency exchange institution.</p></div></article></main>'
+    async def collect():
+        return [item async for item in obj.parse_document(response(request.url, body, **request.meta))]
+    items = asyncio.run(collect())
+    assert any(isinstance(item, dict) and item['url'] == doc['url'] for item in items)
+    assert not obj.cursor['document_retries']
+    assert doc['url'] in obj.cursor['checked']
+
+
+@pytest.mark.parametrize('restored_pending', [False, True])
+def test_attachment_retry_outside_realtime_window_fetches_and_completes_pdf(trading, monkeypatch, restored_pending):
+    from alphapulse.crawler.spiders import us_official
+    obj = spider(trading)
+    doc = {'url': 'https://ofac.treasury.gov/recent-actions/20250101', 'kind': 'article', 'listing_date': '2025-01-01'}
+    obj.cursor['attachment_retries'] = {doc['url']: {'document': doc, 'attempts': 1, 'next_at': 0}}
+    if restored_pending:
+        obj.cursor['realtime']['pending'] = [copy.deepcopy(doc)]
+    request, = obj.first_requests()
+    body = '<h1>Crypto sanctions</h1><main><article class="node--type-ofac-recent-action"><div class="field--name-field-body"><p>New sanctions against a virtual currency exchange institution.</p><a href="/media/123/download">Order</a></div></article></main>'
+    async def collect():
+        return [item async for item in obj.parse_document(response(request.url, body, **request.meta))]
+    attachment, = asyncio.run(collect())
+    assert isinstance(attachment, Request) and attachment.url.endswith('/media/123/download')
+    assert obj.cursor['attachment_retries'][doc['url']]['attempts'] == 1
+    assert doc['url'] not in obj.cursor.get('checked', {})
+    monkeypatch.setattr(us_official, 'pdf_text', lambda body: 'The official sanctions order applies to a virtual currency exchange.')
+    items = list(obj.parse_attachment(Response(attachment.url, body=b'%PDF-test', request=attachment.replace(meta={**attachment.meta, 'raw_response_id': 13}))))
+    item = next(item for item in items if isinstance(item, dict))
+    assert item['extra']['content_scope'] == 'article'
+    assert item['extra']['attachments'][0]['status'] == 'complete'
+    assert not obj.cursor['attachment_retries']
+    assert doc['url'] in obj.cursor['checked']
+
+
+def test_due_attachment_is_queued_alongside_document_retry_without_duplicates(trading):
+    obj = spider(trading)
+    documents = [{'url': f'https://ofac.treasury.gov/recent-actions/2026110{day}', 'kind': 'article'} for day in (3, 4)]
+    obj.cursor['document_retries'] = {documents[0]['url']: {'document': documents[0], 'attempts': 1, 'next_at': 0, 'error': 'HTTP503'}}
+    obj.cursor['attachment_retries'] = {doc['url']: {'document': doc, 'attempts': 1, 'next_at': 0} for doc in documents}
+    request, = obj.first_requests()
+    pending = obj.cursor['realtime']['pending']
+    assert [doc['url'] for doc in pending] == [doc['url'] for doc in documents]
+    assert request.meta['document']['document_retry'] and request.meta['document']['attachment_retry']
+    requests = obj.complete('realtime')
+    assert requests[0].url == documents[1]['url'] and requests[0].meta['document']['attachment_retry']
+
+
+@pytest.mark.parametrize('first_lane', ['realtime', 'history'])
+def test_successful_retry_is_not_requeued_by_other_lane(trading, first_lane):
+    obj = spider(trading)
+    doc = {'url': 'https://ofac.treasury.gov/recent-actions/20261104', 'kind': 'article'}
+    obj.cursor['document_retries'] = {doc['url']: {'document': doc, 'attempts': 1, 'next_at': 0, 'error': 'HTTP503'}}
+    for lane in ('realtime', 'history'):
+        obj.cursor[lane]['pending'] = [copy.deepcopy(doc)]
+    request, = obj.advance(first_lane)
+    body = '<h1>Crypto sanctions</h1><main><article class="node--type-ofac-recent-action"><div class="field--name-field-body"><p>New sanctions against a virtual currency exchange institution.</p></div></article></main>'
+    async def collect():
+        return [item async for item in obj.parse_document(response(request.url, body, **request.meta))]
+    items = asyncio.run(collect())
+    assert any(isinstance(item, dict) for item in items)
+    assert not obj.cursor['document_retries']
+    other_lane = 'history' if first_lane == 'realtime' else 'realtime'
+    assert all(request.url != doc['url'] for request in obj.advance(other_lane))
+
+
+@pytest.mark.parametrize('first_lane', ['realtime', 'history'])
+def test_document_retry_is_requested_once_across_both_lanes(trading, first_lane):
+    obj = spider(trading)
+    doc = {'url': 'https://ofac.treasury.gov/recent-actions/20261104', 'kind': 'article'}
+    obj.cursor['document_retries'] = {doc['url']: {'document': doc, 'attempts': 1, 'next_at': 0, 'error': 'HTTP503'}}
+    for lane in ('realtime', 'history'):
+        obj.cursor[lane]['pending'] = [copy.deepcopy(doc)]
+    request, = obj.advance(first_lane)
+    other_lane = 'history' if first_lane == 'realtime' else 'realtime'
+    following = obj.advance(other_lane)
+    assert all(r.url != doc['url'] for r in following)
+    failure = Failure(HttpError(Response(request.url, status=503, request=request)))
+    failure.request = request
+    obj.on_request_error(failure)
+    assert obj.cursor['document_retries'][doc['url']]['attempts'] == 2
+    assert doc['url'] not in obj.cursor.get('checked', {})
+    # Completing a concurrent listing must not re-queue the same retry.
+    listing = '<h1>Actions</h1><a href="' + doc['url'] + '">Sanctions</a>'
+    requests = list(obj.parse_listing(response(SOURCES[obj.name].entries[0], listing, lane=other_lane)))
+    assert all(r.url != doc['url'] for r in requests)
+
+
+def test_concurrent_initial_document_failures_keep_one_attempt_and_all_errors(trading):
+    obj = spider(trading)
+    doc = {'url': 'https://ofac.treasury.gov/recent-actions/20261104', 'kind': 'article'}
+    for lane in ('realtime', 'history'):
+        obj.cursor[lane]['pending'] = [copy.deepcopy(doc)]
+    requests = [obj.advance(lane)[0] for lane in ('realtime', 'history')]
+    for request in requests:
+        failure = Failure(HttpError(Response(request.url, status=503, request=request)))
+        failure.request = request
+        obj.on_request_error(failure)
+    assert obj.cursor['document_retries'][doc['url']]['attempts'] == 1
+    assert len(obj.run.errors) == 2
+
+
+def test_document_retry_waits_one_hour_after_failure_in_long_round(trading):
+    obj = spider(trading)
+    obj.run.started_at = datetime.now(UTC) - timedelta(hours=2)
+    doc = {'url': 'https://ofac.treasury.gov/recent-actions/20261104', 'kind': 'article'}
+    obj.cursor['realtime']['pending'] = [doc]
+    request, = obj.advance('realtime')
+    failure = Failure(HttpError(Response(request.url, status=503, request=request)))
+    failure.request = request
+    before = datetime.now(UTC)
+    obj.on_request_error(failure)
+    after = datetime.now(UTC)
+    due_at = obj.cursor['document_retries'][doc['url']]['next_at']
+    assert (before + timedelta(hours=1)).timestamp() <= due_at <= (after + timedelta(hours=1)).timestamp()
+
+
+def test_pipeline_error_prevents_completed_read_checkpoint(trading):
+    obj = spider(trading)
+    obj.run.completed_documents_safe = True
+    obj._on_item_error(Failure(ValueError('test persistence failure')))
+    assert not obj.run.completed_documents_safe
+    assert '落库失败' in obj.run.errors[0]
+
+
+def test_completed_document_error_cannot_remove_next_pending_document(trading, monkeypatch):
+    obj = spider(trading)
+    doc = {'url': 'https://ofac.treasury.gov/recent-actions/20261104', 'kind': 'article'}
+    following = {'url': 'https://ofac.treasury.gov/recent-actions/20261103', 'kind': 'article'}
+    obj.cursor['realtime']['pending'] = [following]
+    async def scheduling_failed(res):
+        raise ValueError('next request scheduling failed')
+        yield
+    monkeypatch.setattr(obj, '_parse_document', scheduling_failed)
+    async def collect():
+        return [item async for item in obj.parse_document(response(doc['url'], '', document=doc, lane='realtime'))]
+    with pytest.raises(ValueError, match='scheduling failed'):
+        asyncio.run(collect())
+    assert obj.cursor['realtime']['pending'] == [following]
+    assert not obj.cursor.get('document_retries')
+
+
+@pytest.mark.parametrize('safe', [False, True])
+def test_failed_round_saves_only_safe_completed_reads_and_explicit_gaps(trading, monkeypatch, safe):
+    from alphapulse.crawler.spiders import base
+    obj = spider(trading)
+    source = SimpleNamespace(history_target=obj.run.history_target, history_cursor={'history': {'entry': 0}, 'realtime': {'entry': 0}}, consecutive_failures=0)
+    class Session:
+        async def __aenter__(self): return self
+        async def __aexit__(self, *args): pass
+        async def execute(self, *args): pass
+        async def get(self, *args, **kwargs): return source
+        async def commit(self): pass
+    monkeypatch.setattr(base, 'SessionLocal', Session)
+    obj.run.errors = ['failed document']
+    obj.run.completed_documents_safe = safe
+    obj.run.official_cursor = {'history': {'entry': 5}, 'realtime': {'entry': 8}, 'checked': {'good': 123}, 'document_retries': {'bad': {'attempts': 1}}}
+    asyncio.run(obj._finish_run(obj.run))
+    assert source.history_cursor['history'] == {'entry': 0}
+    assert source.history_cursor['realtime'] == {'entry': 0}
+    assert ('checked' in source.history_cursor) is safe
+    assert ('document_retries' in source.history_cursor) is safe
+    if safe:
+        assert source.history_cursor['checked'] == {'good': 123}
+        assert 'bad' in source.history_cursor['document_retries']
+
+
+@pytest.mark.parametrize('status', [403, 429])
+def test_access_block_stops_round_and_honors_retry_after(trading, status):
+    obj = spider(trading, 'cftc')
+    doc = {'url': 'https://www.cftc.gov/PressRoom/PressReleases/9321-26', 'kind': 'article'}
+    obj.cursor['realtime']['pending'] = [doc]
+    request = Request(doc['url'], meta={'document': doc, 'lane': 'realtime'})
+    failure = Failure(HttpError(Response(request.url, status=status, headers={'Retry-After': '3600'}, request=request)))
+    failure.request = request
+    obj.on_request_error(failure)
+    assert obj.run.blocked
+    assert obj.run.cooldown_until > datetime.now(UTC) + timedelta(minutes=59)
+    assert obj.advance('realtime') == []
+    assert obj.cursor['realtime']['pending'] == [doc]
+    assert not obj.cursor.get('document_retries')
+
+
+@pytest.mark.parametrize('later_block', ['403', '429', 'listing_challenge', 'document_challenge'])
+def test_later_block_does_not_shorten_existing_retry_after(trading, later_block):
+    obj = spider(trading, 'cftc')
+    doc = {'url': 'https://www.cftc.gov/PressRoom/PressReleases/9321-26', 'kind': 'article'}
+    obj.cursor['realtime']['pending'] = [doc]
+    request = Request(doc['url'], meta={'document': doc, 'lane': 'realtime'})
+    failure = Failure(HttpError(Response(request.url, status=429, headers={'Retry-After': '3600'}, request=request)))
+    failure.request = request
+    obj.on_request_error(failure)
+    until = obj.run.cooldown_until
+    if later_block in ('403', '429'):
+        failure = Failure(HttpError(Response(request.url, status=int(later_block), request=request)))
+        failure.request = request
+        obj.on_request_error(failure)
+    else:
+        body = '<h1>Just a moment...</h1><p>Enable JavaScript and cookies to continue.</p>'
+        if later_block == 'listing_challenge':
+            with pytest.raises(ValueError, match='验证页面'):
+                list(obj.parse_listing(response(SOURCES[obj.name].entries[0], body, lane='history')))
+        else:
+            async def collect():
+                return [item async for item in obj.parse_document(response(doc['url'], body, document=doc, lane='realtime'))]
+            assert asyncio.run(collect()) == []
+    assert obj.run.blocked and obj.run.cooldown_until == until
+
+
 @pytest.mark.parametrize('source,title,text,expected', [
     ('doj', 'Man sentenced for bitcoin theft', 'Personal bitcoin theft worth $300', False),
     ('doj', 'Exchange charged', 'Cryptocurrency exchange charged with money laundering', True),
@@ -130,9 +440,9 @@ def test_source_api_keeps_attachment_gaps_visible_across_rounds(trading, monkeyp
     from alphapulse.models import Source
 
     source = Source(code='doj', name='DOJ', list_url=SOURCES['doj'].entries[0], enabled=False, interval_seconds=300, backfill_hours=72,
-                    consecutive_failures=0, last_success_at=None, last_error=None, paused_until=None, history_cursor={'attachment_retries': {'a': {'attempts': 1}, 'b': {'attempts': 2}, 'c': {'attempts': 3}}},
+                    consecutive_failures=0, last_success_at=None, last_error=None, paused_until=None, history_cursor={'attachment_retries': {'a': {'attempts': 1}, 'b': {'attempts': 2}, 'c': {'attempts': 3}}, 'document_retries': {'d': {'attempts': 1}, 'e': {'attempts': 3}}},
                     history_target=None, history_position=None, history_done=False)
-    latest = SimpleNamespace(source_code='doj', stats={'discovered': 2, 'attachment_gaps': 999})
+    latest = SimpleNamespace(source_code='doj', status='success', stats={'discovered': 2, 'attachment_gaps': 999})
 
     class Session:
         async def scalars(self, statement):
@@ -151,6 +461,10 @@ def test_source_api_keeps_attachment_gaps_visible_across_rounds(trading, monkeyp
         assert result.collection_stats['attachment_gaps'] == 3
         assert result.collection_stats['attachment_retry_pending'] == 2
         assert result.collection_stats['attachment_retry_exhausted'] == 1
+        assert result.collection_stats['document_gaps'] == 2
+        assert result.collection_stats['document_retry_pending'] == 1
+        assert result.collection_stats['document_retry_exhausted'] == 1
+        assert result.collection_stats['round_running'] == 0
     source.history_cursor['attachment_retries'].pop('a')
     result = asyncio.run(crawler.list_sources(Session())).data.sources[0]
     assert result.collection_stats['attachment_gaps'] == 2
@@ -160,6 +474,10 @@ def test_source_api_keeps_attachment_gaps_visible_across_rounds(trading, monkeyp
     result = asyncio.run(crawler.list_sources(Session())).data.sources[0]
     assert result.collection_stats['attachment_gaps'] == 0
     assert result.collection_stats['attachment_retry_exhausted'] == 0
+    assert result.collection_stats['document_gaps'] == 0
+    latest.status = 'running'
+    result = asyncio.run(crawler.list_sources(Session())).data.sources[0]
+    assert result.collection_stats['round_running'] == 1
 
 
 def test_challenge_is_failure_not_empty_success(trading):
@@ -367,10 +685,13 @@ def test_attachment_failure_is_visible_and_keeps_a_bounded_retry(trading):
     obj.cursor['history']['pending'] = [doc]
     parsed = {'url': url, 'title': 'Crypto sanctions', 'external_id': 'x', 'content_html': '<p>Virtual currency exchange sanctions.</p>', 'published_at': datetime(2025, 1, 1, tzinfo=UTC), 'raw_response_id': 12, 'retry_document': doc,
               'extra': {'attachments': [{'url': 'https://ofac.treasury.gov/a.pdf', 'title': 'License', 'status': 'unavailable'}]}}
+    before = datetime.now(UTC)
     item, = obj.attach_or_finish(parsed, 'history')
+    after = datetime.now(UTC)
     assert item['extra']['content_scope'] == 'partial'
     assert obj.cursor['attachment_retries'][url]['attempts'] == 1
     assert obj.cursor['attachment_retries'][url]['document'] == doc
+    assert (before + timedelta(hours=1)).timestamp() <= obj.cursor['attachment_retries'][url]['next_at'] <= (after + timedelta(hours=1)).timestamp()
 
 
 @pytest.mark.parametrize('canceled,failed', [(False, False), (True, False), (False, True)])
@@ -566,10 +887,13 @@ def test_missing_bill_material_is_a_retryable_gap(trading, monkeypatch):
     asyncio.run(parse({'bill': {'title': 'GENIUS Act'}}))
     asyncio.run(parse({'subjects': {}}))
     asyncio.run(parse({'summaries': []}))
+    before = datetime.now(UTC)
     asyncio.run(parse({'textVersions': []}))
+    after = datetime.now(UTC)
     assert obj.cursor['bill_retries'][base]['attempts'] == 1
     assert obj.run.stats['unresolved_bills'] == 1
     assert not obj.run.stats.get('excluded')
+    assert (before + timedelta(hours=1)).timestamp() <= obj.cursor['bill_retries'][base]['next_at'] <= (after + timedelta(hours=1)).timestamp()
 
 
 @pytest.mark.parametrize('attempts,version,expected_kind', [(3, 'v1', 'listing'), (1, 'v1', 'listing'), (3, 'v2', 'subjects')])
@@ -761,8 +1085,10 @@ def test_doj_body_identity_date_and_validation(trading, bad_body):
     async def collect():
         return [v async for v in obj.parse_document(response(doc['url'], json.dumps({'metadata': {'responseInfo': {'status': 200}}, 'results': [record]}), lane='history', document=doc))]
     if bad_body:
-        with pytest.raises(ValueError): asyncio.run(collect())
-        assert obj.cursor['history']['pending'] == [doc]
+        assert asyncio.run(collect()) == []
+        assert obj.run.blocked and obj.run.errors
+        assert obj.cursor['document_retries'][doc['url']]['attempts'] == 1
+        assert obj.cursor['history']['pending'] == []
     else:
         item, = asyncio.run(collect())
         assert item['url'] == 'https://www.justice.gov/opa/pr/test'
diff --git a/deploy/README.md b/deploy/README.md
index 3d2416d..e5c052a 100644
--- a/deploy/README.md
+++ b/deploy/README.md
@@ -263,9 +263,26 @@ list traversal does not prove complete bodies or attachments. Attachment failure
 remain visible on the news item, with up to three scheduled attempts one hour apart;
 exhausted gaps require explicit investigation or another traversal.
 
+Article and DOJ document failures also remain explicit gaps, with at most three
+scheduled attempts one hour apart, measured from the actual failure time.
+Real-time and history lanes share the document retry quota; a document retry is
+requested only once per round. Due body and attachment retries remain eligible
+after their document leaves the real-time lookback window.
+A failed round may reuse completed document
+reads and retain retry evidence, but cannot advance its lane or history position;
+a pipeline error prevents saving completed reads. HTTP 403/429 and detected
+challenge pages stop the round for at least fifteen minutes; a longer valid
+`Retry-After` is honored and subsequent blocks cannot shorten that deadline.
+CFTC document requests are serialized with a three-second
+delay. The console distinguishes active collection, scheduled waiting, failure
+backoff, stopped processes and unresolved document gaps.
+
 Real-time discovery follows pagination, including every Federal Register page
 within a day. Only verified descending feeds stop early at a fully dated page
 outside the lookback window; unknown ordering is traversed conservatively.
+Treasury currently returns the first page again for subsequent page requests.
+A dated real-time page can end discovery at the lookback boundary; a repeated
+historical page still fails explicitly and cannot establish historical coverage.
 Congress screening reads subjects, CRS summaries, and available text before
 excluding a bill whose title alone is inconclusive. Missing bill evidence remains
 an unresolved gap, retried at most three times one hour apart per official update;
diff --git a/scripts/smoke_us_news.py b/scripts/smoke_us_news.py
index 82a1169..a708de6 100644
--- a/scripts/smoke_us_news.py
+++ b/scripts/smoke_us_news.py
@@ -178,13 +178,11 @@ async def main():
     assert captured[0]['external_id'] == 'legacy-rss-guid'
     captured.clear()
     invalid = HtmlResponse(sample['url'], body=b'<h1>Just a moment...</h1><p>Enable JavaScript and cookies to continue.</p>', encoding='utf-8', request=request)
-    try:
-        async for item in obj.parse_document(invalid):
-            await pipeline.process_item(item)
-    except ValueError:
-        pass
-    else:
-        raise AssertionError('Challenge must fail before the pipeline')
+    obj.cursor['history']['pending'] = [doc]
+    assert not [item async for item in obj.parse_document(invalid)]
+    assert obj.run.errors and obj.run.blocked
+    assert obj.cursor['document_retries'][doc['url']]['attempts'] == 1
+    assert not captured
     async with SessionLocal() as session:
         news = await session.scalar(select(NewsItem).where(NewsItem.external_id == 'legacy-rss-guid'))
         assert (news.content_hash, news.revision_no) == baseline
@@ -216,6 +214,31 @@ async def main():
         saved = await session.get(CrawlRun, original_id)
         assert saved.status == 'success'
         assert (await session.get(Source, 'ofac')).history_cursor['checkpoint_probe'] == 'committed'
+    # A failed document preserves safe reads and explicit gaps, not lane positions.
+    async with SessionLocal() as session:
+        before_cursor = copy.deepcopy((await session.get(Source, 'ofac')).history_cursor)
+    completed = {'https://ofac.treasury.gov/recent-actions/20261104': obj.run.started_at.timestamp()}
+    gap_url = 'https://ofac.treasury.gov/recent-actions/20261103'
+    gaps = {gap_url: {'document': {'url': gap_url, 'kind': 'article'}, 'attempts': 1, 'next_at': obj.run.started_at.timestamp()+3600, 'error': 'HTTP503'}}
+    failed_run = copy.deepcopy(obj.run)
+    failed_run.id = uuid.uuid4()
+    failed_run.errors = ['document unavailable']
+    failed_run.completed_documents_safe = True
+    failed_run.official_cursor = {'checked': completed, 'document_retries': gaps, 'realtime': {'entry': 99}, 'history': {'entry': 99}}
+    await obj._finish_run(failed_run)
+    async with SessionLocal() as session:
+        cursor = (await session.get(Source, 'ofac')).history_cursor
+        assert cursor['checked'] == completed and cursor['document_retries'] == gaps
+        assert cursor.get('history') == before_cursor.get('history')
+        assert cursor.get('realtime') == before_cursor.get('realtime')
+        assert (await session.get(CrawlRun, failed_run.id)).status == 'failed'
+    unsafe_run = copy.deepcopy(failed_run)
+    unsafe_run.id = uuid.uuid4()
+    unsafe_run.completed_documents_safe = False
+    unsafe_run.official_cursor['checked'] = {'must-not-be-skipped': 1}
+    await obj._finish_run(unsafe_run)
+    async with SessionLocal() as session:
+        assert (await session.get(Source, 'ofac')).history_cursor['checked'] == completed
     # API/PDF packages use the real parser and database revision pipeline too.
     api_items = await public_api_items() + await congress_reference_items() + await congress_api_items()
     with patch.object(pipelines, 'redis', SimpleNamespace(publish=AsyncMock())):
@@ -235,7 +258,7 @@ async def main():
     async with SessionLocal() as session:
         assert await session.scalar(select(func.count()).select_from(NewsRevision)) == baseline_count + 1
     await engine.dispose()
-    print('US source migration, HTML, API/PDF, GovInfo and Congress packages, API/PDF replay and revision, negative screening, daily XML, RSS identity, challenge rejection and checkpoint recovery passed')
+    print('US source migration, HTML, API/PDF, GovInfo and Congress packages, API/PDF replay and revision, negative screening, daily XML, RSS identity, challenge rejection, checkpoint recovery and failed-document persistence passed')
 
 
 if __name__ == '__main__':
diff --git a/web/src/pages/crawler/components/SourcePanel.vue b/web/src/pages/crawler/components/SourcePanel.vue
index 768b859..73b9109 100644
--- a/web/src/pages/crawler/components/SourcePanel.vue
+++ b/web/src/pages/crawler/components/SourcePanel.vue
@@ -7,6 +7,7 @@ import { useSessionScope } from "@@/composables/useSessionScope"
 import { formatTime } from "@@/utils/text"
 import { CircleCheck, RefreshRight } from "@element-plus/icons-vue"
 import dayjs from "dayjs"
+import { sourceStatus } from "../sourceStatus"
 
 /** logs 只用来发现「本轮完成」以刷新最新入库；alert 为连续失败达到告警阈值 */
 const props = defineProps<{
@@ -87,22 +88,7 @@ async function cancelHistory() {
   emit("changed")
 }
 
-const statusTag = computed(() => {
-  if (props.source.configuration_error)
-    return { type: "warning", text: props.source.configuration_error } as const
-  const process = props.source.process
-  if (props.restarting || process?.state === "restarting")
-    return { type: "warning", text: "重启中" } as const
-  if (!process)
-    return { type: "danger", text: "管理进程未运行" } as const
-  if (process.state === "stopped")
-    return { type: "danger", text: "子进程未运行" } as const
-  if (!props.source.enabled)
-    return { type: "info", text: "已停用" } as const
-  if (props.source.paused_until && new Date(props.source.paused_until) > new Date())
-    return { type: "warning", text: "失败退避中" } as const
-  return { type: "success", text: "运行中" } as const
-})
+const statusTag = computed(() => sourceStatus(props.source, props.restarting))
 
 onMounted(loadLatest)
 </script>
@@ -163,6 +149,9 @@ onMounted(loadLatest)
     <div v-if="source.collection_stats?.attachment_gaps" class="text-xs text-amber-600">
       附件待核验文档 {{ source.collection_stats.attachment_gaps }}；待重试 {{ source.collection_stats.attachment_retry_pending }}；重试已耗尽 {{ source.collection_stats.attachment_retry_exhausted }}，请核对原文
     </div>
+    <div v-if="source.collection_stats?.document_gaps" class="text-xs text-amber-600">
+      正文待核验 {{ source.collection_stats.document_gaps }}；待重试 {{ source.collection_stats.document_retry_pending }}；重试已耗尽 {{ source.collection_stats.document_retry_exhausted }}，请核对来源日志
+    </div>
     <div v-if="source.collection_stats?.discovered" class="text-xs text-slate-500">
       本轮发现 {{ source.collection_stats.discovered }}；筛除 {{ source.collection_stats.excluded ?? 0 }}；不完整包 {{ source.collection_stats.partial ?? 0 }}
     </div>
diff --git a/web/src/pages/crawler/index.vue b/web/src/pages/crawler/index.vue
index 86d0f19..ec26d5e 100644
--- a/web/src/pages/crawler/index.vue
+++ b/web/src/pages/crawler/index.vue
@@ -1,5 +1,5 @@
 <script lang="ts" setup>
-import type { SourcesInfo } from "@@/apis/crawler/type"
+import type { SourceInfo, SourcesInfo } from "@@/apis/crawler/type"
 import { createRssApi, listSourcesApi, restartCrawlerApi } from "@@/apis/crawler"
 import { useSessionScope } from "@@/composables/useSessionScope"
 import { formatTime } from "@@/utils/text"
@@ -8,6 +8,7 @@ import { logRequestError } from "@/http/axios"
 import CalendarSourcePanel from "./components/CalendarSourcePanel.vue"
 import SourcePanel from "./components/SourcePanel.vue"
 import { useCrawlerLogs } from "./composables/useCrawlerLogs"
+import { sourceStatus } from "./sourceStatus"
 
 const rssVisible = ref(false)
 const rssSaving = ref(false)
@@ -178,10 +179,10 @@ onMounted(async () => {
             {{ regions[row.region_code as keyof typeof regions] ?? row.region_code }}
           </template>
         </el-table-column>
-        <el-table-column label="状态" width="110">
+        <el-table-column label="状态" min-width="130">
           <template #default="{ row }">
-            <el-tag size="small" :type="row.enabled && row.process?.state === 'running' ? 'success' : 'info'">
-              {{ !row.enabled ? '已停用' : row.process?.state === 'running' ? '运行中' : '未运行' }}
+            <el-tag size="small" :type="sourceStatus(row as SourceInfo, restarting.has(row.code)).type">
+              {{ sourceStatus(row as SourceInfo, restarting.has(row.code)).text }}
             </el-tag>
           </template>
         </el-table-column>
diff --git a/web/src/pages/crawler/sourceStatus.ts b/web/src/pages/crawler/sourceStatus.ts
new file mode 100644
index 0000000..f595fce
--- /dev/null
+++ b/web/src/pages/crawler/sourceStatus.ts
@@ -0,0 +1,21 @@
+import type { SourceInfo } from "@@/apis/crawler/type"
+
+export function sourceStatus(source: SourceInfo, restarting = false, now = Date.now()) {
+  if (source.configuration_error)
+    return { type: "warning", text: source.configuration_error } as const
+  if (!source.enabled)
+    return { type: "info", text: "已停用" } as const
+  if (restarting || source.process?.state === "restarting")
+    return { type: "warning", text: "重启中" } as const
+  if (!source.process || source.process.state === "stopped")
+    return { type: "danger", text: "采集进程未运行" } as const
+  if (source.paused_until && new Date(source.paused_until).getTime() > now)
+    return { type: "warning", text: "失败退避中" } as const
+  if (source.collection_stats?.round_running)
+    return { type: "primary", text: "采集中" } as const
+  if (source.consecutive_failures)
+    return { type: "danger", text: "失败待重试" } as const
+  if (source.collection_stats?.document_gaps)
+    return { type: "warning", text: "存在文档缺口" } as const
+  return { type: "info", text: "等待调度" } as const
+}
diff --git a/web/tests/crawler.test.ts b/web/tests/crawler.test.ts
new file mode 100644
index 0000000..341cda9
--- /dev/null
+++ b/web/tests/crawler.test.ts
@@ -0,0 +1,37 @@
+import type { SourceInfo } from "../src/common/apis/crawler/type.ts"
+import assert from "node:assert/strict"
+import { test } from "node:test"
+import { sourceStatus } from "../src/pages/crawler/sourceStatus.ts"
+
+test("crawler status distinguishes process health from collection results", () => {
+  const source: SourceInfo = {
+    code: "test",
+    name: "Test",
+    region_code: null,
+    list_url: "https://example.invalid",
+    interval_seconds: 60,
+    backfill_hours: 72,
+    last_success_at: null,
+    last_error: null,
+    paused_until: null,
+    supports_history: false,
+    history_target: null,
+    history_position: null,
+    history_done: false,
+    today_new: 0,
+    today_revised: 0,
+    total_items: 0,
+    enabled: true,
+    consecutive_failures: 0,
+    process: { state: "running", pid: 1, started_at: null, next_run_at: "2026-11-05T00:00:00Z" },
+    collection_stats: { round_running: 0 }
+  }
+  assert.equal(sourceStatus(source).text, "等待调度")
+  assert.equal(sourceStatus({ ...source, collection_stats: { round_running: 1 } }).text, "采集中")
+  assert.equal(sourceStatus({ ...source, consecutive_failures: 2 }).text, "失败待重试")
+  assert.equal(sourceStatus({ ...source, paused_until: "2026-11-06T00:00:00Z" }, false, Date.parse("2026-11-05")).text, "失败退避中")
+  assert.equal(sourceStatus({ ...source, collection_stats: { document_gaps: 2 } }).text, "存在文档缺口")
+  assert.equal(sourceStatus({ ...source, enabled: false, consecutive_failures: 3 }).text, "已停用")
+  assert.equal(sourceStatus({ ...source, configuration_error: "待配置 CRAWLER_CONTACT_EMAIL" }).type, "warning")
+  assert.equal(sourceStatus({ ...source, process: null }).text, "采集进程未运行")
+})