Fix official news retries, date boundaries and collection status

Bound document retries across realtime and history lanes, preserve safe completed reads on failed rounds, and retain unresolved gaps. Honor the longest access-block cooldown and measure retries from the actual failure time.

Filter dated official listings at the collection boundary, handle empty Federal Register days, and distinguish active rounds from waiting or failed processes in the console. Add captured-response regressions and isolated checkpoint validation.

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),

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)

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'
 
 

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'

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;

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__':

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>

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>

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
+}

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, "采集进程未运行")
+})