Step 4A-6-C-R3.2.8.2 Acronym Boundary Authenticated S2 Global Rate Limit Fix

config.py

@@ -7,10 +7,12 @@ def load_config():
         "lit_target_papers": int(os.environ.get("LIT_TARGET_PAPERS", "80")),
         "lit_per_page": int(os.environ.get("LIT_PER_PAGE", "100")),
         "lit_max_workers": int(os.environ.get("LIT_MAX_WORKERS", "5")),
-        # R3.2.7: low-volume SmartPaper uses the public Semantic Scholar path
-        # by default. OpenAlex remains the primary resilient source.
-        "s2_anonymous_first": os.environ.get("S2_ANONYMOUS_FIRST", "true").strip().lower() in {"1", "true", "yes", "on"},
-        "s2_anonymous_min_interval_seconds": max(0.8, float(os.environ.get("S2_ANONYMOUS_MIN_INTERVAL_SECONDS", "1.10"))),
+        # R3.2.8.2: OpenAlex is primary; Semantic Scholar is a bounded
+        # authenticated auxiliary source. Every S2 HTTP attempt across all
+        # endpoints must share one >=1 RPS gate.
+        "s2_min_interval_seconds": max(1.0, float(os.environ.get("S2_MIN_INTERVAL_SECONDS", "1.10"))),
+        "s2_result_limit": max(1, min(50, int(os.environ.get("S2_RESULT_LIMIT", "20")))),
+        "s2_max_logical_searches": max(1, min(6, int(os.environ.get("S2_MAX_LOGICAL_SEARCHES", "3")))),
         "citation_planner_batch_size": max(8, min(24, int(os.environ.get("CITATION_PLANNER_BATCH_SIZE", "18")))),
         "citation_planner_max_tokens": max(1200, min(3200, int(os.environ.get("CITATION_PLANNER_MAX_TOKENS", "2200")))),
         "chunk_words_default": int(os.environ.get("CHUNK_WORDS_DEFAULT", "1500")),

core/pipeline_agent.py

@@ -66,8 +66,9 @@ class SSciPipelineAgent:
         # exposed to the process as environment variables; Streamlit is not a
         # dependency of the executor or pipeline.
         self.s2_key = str(os.environ.get("S2_API_KEY", s2_key) or "").strip()
-        self.s2_anonymous_first = bool(APP_CONFIG.get("s2_anonymous_first", True))
-        self.s2_anonymous_min_interval_seconds = float(APP_CONFIG.get("s2_anonymous_min_interval_seconds", 1.10))
+        self.s2_min_interval_seconds = float(APP_CONFIG.get("s2_min_interval_seconds", 1.10))
+        self.s2_result_limit = int(APP_CONFIG.get("s2_result_limit", 20))
+        self.s2_max_logical_searches = int(APP_CONFIG.get("s2_max_logical_searches", 3))
         self.openalex_key = str(os.environ.get("OA_API_KEY", "") or "").strip()
         self.openalex_email = str(APP_CONFIG.get("openalex_email") or "").strip()
             
@@ -92,21 +93,25 @@ class SSciPipelineAgent:
                 "key_present": bool(self.s2_key) if source == "s2" else bool(self.openalex_key),
                 "authenticated_status": None,
                 "anonymous_status": None,
-                "access_mode": ("anonymous" if self.s2_anonymous_first else "key") if source == "s2" and self.s2_key else "anonymous" if source == "s2" else "api",
+                "access_mode": "key" if source == "s2" and self.s2_key else "anonymous" if source == "s2" else "api",
             }
             for source in ("s2", "oa")
         }
         self._cancel_checker = None
         self._execution_progress_hook = None
-        # S2 authentication recovery is task-local. Never log or persist the key.
-        # "key" means use the configured key, "anonymous" means a 401/403 key
-        # rejection was recovered by an anonymous probe, and "blocked" means
-        # both authenticated and anonymous access were rejected.
-        self._s2_access_mode = "anonymous" if self.s2_anonymous_first else ("key" if self.s2_key else "anonymous")
-        self._s2_probe_lock = threading.Lock()
-        self._s2_probe_complete = self.s2_anonymous_first or not bool(self.s2_key)
+        # S2 is task-local and auxiliary. A configured key is always used
+        # first; no username/email is required and the secret is never logged.
+        # All S2 HTTP attempts (including retries/future endpoints) share one
+        # request-start gate. Logical S2 searches are also serialized so a
+        # positive auxiliary result can stop later routine S2 work.
+        self._s2_access_mode = "key" if self.s2_key else "anonymous"
         self._s2_request_lock = threading.Lock()
+        self._s2_http_lock = threading.Lock()
+        self._s2_worker_lock = threading.Lock()
         self._s2_last_request_at = 0.0
+        self._s2_logical_searches = 0
+        self._s2_aux_satisfied = False
+        self._s2_disabled = False
         self.s2_fused = False
         self.oa_fused = False
         self.citation_map = {}
@@ -267,7 +272,9 @@ class SSciPipelineAgent:
                 "citation_audit_log": list(self.citation_audit_log),
                 "global_memory": self.global_memory,
                 "s2_access_mode": self._s2_access_mode,
-                "s2_probe_complete": bool(self._s2_probe_complete),
+                "s2_logical_searches": int(self._s2_logical_searches),
+                "s2_aux_satisfied": bool(self._s2_aux_satisfied),
+                "s2_disabled": bool(self._s2_disabled),
             }
 
     def load_state(self, state_dict):
@@ -286,16 +293,18 @@ class SSciPipelineAgent:
             self.citation_audit_log = list(state_dict.get("citation_audit_log", []))
             self.global_memory = state_dict.get("global_memory", self.global_memory)
             loaded_mode = str(state_dict.get("s2_access_mode") or "").strip().lower()
-            if self.s2_anonymous_first:
-                self._s2_access_mode = "anonymous"
-                self._s2_probe_complete = True
+            expected_mode = "key" if self.s2_key else "anonymous"
+            if loaded_mode in {"key", "anonymous", "blocked"}:
+                # A newly configured key supersedes a legacy anonymous checkpoint.
+                self._s2_access_mode = expected_mode if self.s2_key else loaded_mode
             else:
-                if loaded_mode in {"key", "anonymous", "blocked"}:
-                    self._s2_access_mode = loaded_mode
-                self._s2_probe_complete = bool(state_dict.get("s2_probe_complete", self._s2_probe_complete))
+                self._s2_access_mode = expected_mode
+            self._s2_logical_searches = max(0, int(state_dict.get("s2_logical_searches", self._s2_logical_searches)))
+            self._s2_aux_satisfied = bool(state_dict.get("s2_aux_satisfied", self._s2_aux_satisfied))
+            self._s2_disabled = bool(state_dict.get("s2_disabled", self._s2_disabled))
         logging.info(f"🔄 状态恢复成功: DS API={self.ds_calls}, 现有文献={len(self.citation_map)}")
 
-    def _requests_get_with_retry(self, url, headers=None, params=None, timeout=None):
+    def _requests_get_with_retry(self, url, headers=None, params=None, timeout=None, before_request=None):
         """GET with bounded, status-aware retry semantics.
 
         Authentication and query errors are returned immediately. Rate limits
@@ -308,6 +317,9 @@ class SSciPipelineAgent:
         last_error = None
         for attempt in range(attempts):
             self._check_cancelled()
+            if before_request is not None:
+                before_request()
+                self._check_cancelled()
             try:
                 response = requests.get(
                     url,
@@ -1628,51 +1640,68 @@ Return ONLY a valid JSON array of strings."""
         self.oa_calls = int(self.source_metrics['oa']['eligible_result_count'])
         return merged
 
-    def _s2_get(self, *, params, headers=None):
-        """Issue an S2 request with public-endpoint pacing when anonymous.
+    def _s2_before_http_attempt(self):
+        """Reserve one global S2 request-start slot and count the HTTP attempt.
 
-        SmartPaper is low-volume, so one anonymous S2 request at a time is
-        preferable to parallel bursts that trigger the shared public 429 pool.
-        OpenAlex remains independently concurrent.
+        The configured Semantic Scholar entitlement is cumulative across all
+        endpoints. This gate therefore lives below search/JIT logic and is
+        invoked before *every* HTTP attempt, including retries.
         """
-        normalized_headers = dict(headers or {})
-        anonymous = not bool(normalized_headers.get('x-api-key'))
-        if not anonymous:
-            return self._requests_get_with_retry(
-                f"{self.s2_api_url}/paper/search", headers=normalized_headers, params=params
-            )
         with self._s2_request_lock:
             elapsed = time.monotonic() - float(self._s2_last_request_at or 0.0)
-            wait = max(0.0, self.s2_anonymous_min_interval_seconds - elapsed)
+            wait = max(0.0, float(self.s2_min_interval_seconds) - elapsed)
             if wait > 0:
+                self._check_cancelled()
                 time.sleep(wait)
-            response = self._requests_get_with_retry(
-                f"{self.s2_api_url}/paper/search", headers={}, params=params
-            )
             self._s2_last_request_at = time.monotonic()
-            return response
+            self._source_metric_add('s2', 'request_count')
+
+    def _s2_get(self, *, params, headers=None, endpoint='/paper/search'):
+        """Issue one logical S2 request through the global 1-RPS gateway."""
+        normalized_headers = dict(headers or {})
+        path = '/' + str(endpoint or '/paper/search').lstrip('/')
+        with self._s2_http_lock:
+            return self._requests_get_with_retry(
+                f"{self.s2_api_url}{path}",
+                headers=normalized_headers,
+                params=params,
+                before_request=self._s2_before_http_attempt,
+            )
 
     def _s2_worker(self, term, offset):
+        # Serialize complete logical searches. This is deliberately separate
+        # from the HTTP-attempt gate: OpenAlex can still run concurrently, while
+        # queued S2 searches observe the positive-result/disable state before
+        # sending another request.
+        with self._s2_worker_lock:
+            return self._s2_worker_serial(term, offset)
+
+    def _s2_worker_serial(self, term, offset):
         self._check_cancelled()
+        if self._s2_aux_satisfied:
+            return []
+        if self._s2_disabled or self._s2_access_mode == 'blocked':
+            return []
+        if self._s2_logical_searches >= int(self.s2_max_logical_searches):
+            return []
+
         query = self._normalize_search_query(term)
         if not query:
             self._record_source_error('s2', 'S2_BAD_QUERY')
             return []
+        self._s2_logical_searches += 1
         params = {
             'query': query,
             'offset': offset,
-            'limit': 50,
+            'limit': int(self.s2_result_limit),
             'fields': 'paperId,title,authors,year,venue,abstract,externalIds',
         }
-        if self._s2_access_mode == 'blocked':
-            # The task already proved that both keyed and anonymous access are
-            # refused from this runtime. Avoid repeating doomed network calls;
-            # OpenAlex remains the safe literature fallback.
-            self._record_source_error('s2', 'S2_ACCESS_BLOCKED', self.source_metrics['s2'].get('last_status_code'))
-            return []
-        self._source_metric_add('s2', 'request_count')
         self._source_metric_set('s2', 'status', 'running')
-        logging.info('S2 search: term=%r normalized=%r offset=%s', term, query, offset)
+        logging.info(
+            'S2 search: term=%r normalized=%r offset=%s access_mode=%s logical=%s/%s limit=%s',
+            term, query, offset, self._s2_access_mode,
+            self._s2_logical_searches, self.s2_max_logical_searches, self.s2_result_limit,
+        )
         try:
             access_mode = self._s2_access_mode
             headers = {'x-api-key': self.s2_key} if access_mode == 'key' and self.s2_key else {}
@@ -1683,47 +1712,21 @@ Return ONLY a valid JSON array of strings."""
                 self._source_metric_set('s2', 'authenticated_status', status)
             else:
                 self._source_metric_set('s2', 'anonymous_status', status)
-            if status in (401, 403) and access_mode == 'key' and self.s2_key:
-                # A configured key can be stale/revoked while the public Graph
-                # endpoint remains usable. Probe anonymously exactly once per
-                # task; this reveals only status codes, never the secret.
-                with self._s2_probe_lock:
-                    if not self._s2_probe_complete:
-                        self._s2_probe_complete = True
-                        self._source_metric_add('s2', 'request_count')
-                        anonymous = self._s2_get(params=params, headers={})
-                        anonymous_status = int(anonymous.status_code)
-                        self._source_metric_set('s2', 'anonymous_status', anonymous_status)
-                        logging.warning(
-                            'S2 access diagnostic: key_present=true authenticated_status=%s anonymous_status=%s',
-                            status, anonymous_status,
-                        )
-                        if anonymous_status not in (401, 403):
-                            self._s2_access_mode = 'anonymous'
-                            self._source_metric_set('s2', 'access_mode', 'anonymous')
-                            response = anonymous
-                            status = anonymous_status
-                            self._source_metric_set('s2', 'last_status_code', status)
-                        else:
-                            self._s2_access_mode = 'blocked'
-                            self._source_metric_set('s2', 'access_mode', 'blocked')
-                    elif self._s2_access_mode == 'anonymous':
-                        self._source_metric_add('s2', 'request_count')
-                        response = self._s2_get(params=params, headers={})
-                        status = int(response.status_code)
-                        self._source_metric_set('s2', 'anonymous_status', status)
-                        self._source_metric_set('s2', 'last_status_code', status)
-                if self._s2_access_mode == 'blocked' or status in (401, 403):
-                    self._record_source_error('s2', 'S2_AUTH_ERROR', status)
-                    return []
-            elif status in (401, 403):
-                # Anonymous access itself is refused from this runtime.
-                if access_mode == 'anonymous':
-                    self._s2_access_mode = 'blocked'
-                    self._source_metric_set('s2', 'access_mode', 'blocked')
+
+            if status in (401, 403):
+                # With a configured key, do not silently switch to public access:
+                # a bad/revoked secret should be diagnosable. OpenAlex remains
+                # the paper-level fallback and generation continues.
+                self._s2_access_mode = 'blocked'
+                self._s2_disabled = True
+                self._source_metric_set('s2', 'access_mode', 'blocked')
                 self._record_source_error('s2', 'S2_AUTH_ERROR', status)
                 return []
             if status == 429:
+                # The bounded HTTP retry helper already honored Retry-After and
+                # sent every retry through the global gate. Stop S2 for this
+                # task after persistent rate limiting; OpenAlex continues.
+                self._s2_disabled = True
                 self._record_source_error('s2', 'S2_RATE_LIMIT', status)
                 return []
             if status == 400:
@@ -1776,7 +1779,10 @@ Return ONLY a valid JSON array of strings."""
                 })
             self._source_metric_add('s2', 'eligible_result_count', len(results))
             self._source_metric_set('s2', 'status', 'ok' if results else 'no_results')
-            if not results:
+            if results:
+                self._s2_aux_satisfied = True
+                self._source_metric_set('s2', 'last_safe_error', '')
+            else:
                 self._source_metric_set('s2', 'last_safe_error', 'S2_NO_RESULTS')
             return results
         except PipelineCancelledError:
@@ -1786,6 +1792,8 @@ Return ONLY a valid JSON array of strings."""
         except Exception as error:
             status = getattr(getattr(error, 'response', None), 'status_code', None)
             code = 'S2_RATE_LIMIT' if status == 429 else ('S2_AUTH_ERROR' if status in (401, 403) else 'S2_REQUEST_ERROR')
+            if status in (401, 403, 429):
+                self._s2_disabled = True
             self._record_source_error('s2', code, status, error)
         return []
 
@@ -1920,9 +1928,11 @@ Return ONLY a valid JSON array of strings."""
                 self.source_metrics[source]['status'] = 'not_started'
                 self.source_metrics[source]['authenticated_status'] = None
                 self.source_metrics[source]['anonymous_status'] = None
-            self._s2_access_mode = 'anonymous' if self.s2_anonymous_first else ('key' if self.s2_key else 'anonymous')
-            self._s2_probe_complete = self.s2_anonymous_first or not bool(self.s2_key)
+            self._s2_access_mode = 'key' if self.s2_key else 'anonymous'
             self._s2_last_request_at = 0.0
+            self._s2_logical_searches = 0
+            self._s2_aux_satisfied = False
+            self._s2_disabled = False
             self.source_metrics['s2']['key_present'] = bool(self.s2_key)
             self.source_metrics['s2']['access_mode'] = self._s2_access_mode
             self.s2_calls = self.oa_calls = 0
@@ -1947,7 +1957,7 @@ Return ONLY a valid JSON array of strings."""
             future_sources = {}
             with concurrent.futures.ThreadPoolExecutor(max_workers=self.lit_max_workers) as executor:
                 if not self.s2_fused:
-                    offset = 50 if second_page else 0
+                    offset = int(self.s2_result_limit) if second_page else 0
                     for term in terms:
                         future_sources[executor.submit(self._s2_worker, term, offset)] = 's2'
                 if not self.oa_fused:
@@ -3216,7 +3226,7 @@ Output only the revised Discussion body, without the '# Discussion' heading."""
         # Internal technical hyphens are allowed (GPT-4, PLS-SEM, COVID-19),
         # but a grammatical hyphen before lowercase prose is not part of the
         # acronym (AI-powered -> AI; LLM-assisted -> LLM).
-        return r'\b[A-Z]{2,}[A-Z0-9]*(?:-[A-Z0-9]+)*(?:s|es)?\b'
+        return r'(?<![A-Za-z0-9_])[A-Z]{2,}[A-Z0-9]*(?:-[A-Z0-9]+)*(?:s|es)?(?![A-Za-z0-9_])'
 
     @classmethod
     def _translation_acronym_tokens(cls, value):