A5-R3-R4C dormant ModelPolicy v2 execution seam

core/generation_runner.py

@@ -2888,11 +2888,53 @@ class GenerationRunner:
         routes = model.get("routes") if isinstance(model.get("routes"), dict) else {}
         if not routes:
             raise RuntimeError("trusted_task_policy_model_routes_missing")
-        # A5-R3-R3A closes snapshot identity only. It deliberately does not
-        # reinterpret fine-grained logical-node routes as the legacy coarse
-        # capability roles. Product execution remains fail-closed until the
-        # separate logical-node execution contract is implemented.
-        raise RuntimeError("trusted_task_policy_model_route_contract_not_ready")
+        if str(snapshot.get("model_policy_version") or "") not in {
+            "model-standard-v2", "model-flagship-v2"
+        }:
+            # Preserve the dormant R3A v1 fail-closed contract; only v2 may
+            # cross the new trusted logical-node execution seam.
+            raise RuntimeError("trusted_task_policy_model_route_contract_not_ready")
+        if str(model.get("provider_catalog_version") or "") != "deepseek-v4-concrete-v1":
+            raise RuntimeError("trusted_task_policy_provider_catalog_invalid")
+        return snapshot
+
+    @staticmethod
+    def _section_logical_node(mode: str, section_id: str, subsection_id: str | None = None) -> str | None:
+        """Map academic topology to a tier-neutral writer logical node."""
+        mode = str(mode or "").strip().lower()
+        section_id = str(section_id or "").strip().lower()
+        subsection_id = str(subsection_id or "").strip().lower()
+        if mode == "amj":
+            if section_id in {"introduction", "theory"}:
+                return "core_section_writing"
+            if section_id == "methodology":
+                return "methods_writing"
+            if section_id == "results":
+                return "quantitative_results_writing"
+            if section_id == "discussion":
+                if subsection_id in {"practical_implications", "limitations"}:
+                    return "ordinary_section_writing"
+                return "core_section_writing"
+            if section_id == "conclusion":
+                return "ordinary_section_writing"
+        elif mode == "asq":
+            if section_id in {"introduction", "theoretical_framing"}:
+                return "core_section_writing"
+            if section_id == "methods":
+                return "methods_writing"
+            if section_id == "findings":
+                if subsection_id == "emergent_model":
+                    return "core_section_writing"
+                return "evidence_rich_findings_narration"
+            if section_id == "discussion":
+                if subsection_id in {"boundaries", "practical_implications"}:
+                    return "ordinary_section_writing"
+                return "core_section_writing"
+            if section_id == "conclusion":
+                return "ordinary_section_writing"
+            if section_id == "appendix":
+                return None
+        raise RuntimeError(f"logical_node_mapping_missing:{mode}:{section_id}:{subsection_id}")
 
     def _run_impl(self) -> dict[str, Any]:
         req = self.request
@@ -2921,6 +2963,8 @@ class GenerationRunner:
         agent = SSciPipelineAgent(ds_key, s2_key)
         self.agent = agent
         agent.set_model_policy_snapshot(plan_policy_snapshot)
+        if trusted_task_policy_snapshot is not None:
+            agent.set_trusted_product_policy_snapshot(trusted_task_policy_snapshot)
         agent.set_cancel_checker(self.cancel_checker)
         # DeepSeek/search activity marks the executor as genuinely alive; the
         # heartbeat thread decides separately when to renew D1's lease.
@@ -3661,6 +3705,8 @@ class GenerationRunner:
         # glossary order deterministic and avoids sharing mutable writer state.
         translation_agent = SSciPipelineAgent(ds_key, s2_key)
         translation_agent.set_model_policy_snapshot(plan_policy_snapshot)
+        if trusted_task_policy_snapshot is not None:
+            translation_agent.set_trusted_product_policy_snapshot(trusted_task_policy_snapshot)
         self.translation_agent = translation_agent
         translation_agent.set_execution_progress_hook(lambda: self._progress_activity())
         translation_agent.set_cancel_checker(lambda: self.cancel_checker())
@@ -3902,6 +3948,9 @@ class GenerationRunner:
                 for word in ["abstract", "keyword", "conclusion", "references", "appendix"]
             )
             section_id = str(sec.get("section_id") or "")
+            section_logical_node = self._section_logical_node(
+                mode, section_id, sec.get("subsection_id")
+            )
             planned_items = [
                 item for item in list(citation_plan.get(display_name, []))
                 if agent.citation_source_role_eligible(
@@ -4224,6 +4273,7 @@ The previous attempt for this same manuscript unit was rejected by validation. R
                         agent.stream_deepseek(
                             sys_prompt, candidate_user_prompt, max_tokens=7000,
                             model_role="writing", thinking=False, allow_fallback=False,
+                            logical_node=section_logical_node,
                         ),
                         preview_prefix=full_draft_en + prefix,
                         preview_stage=(display_name if section_attempt == 1 else f"{display_name}:retry"),

core/meta_extractor.py

@@ -21,7 +21,8 @@ class MetaExtractor:
                 prompt, 
                 snippet, 
                 max_tokens=1000, 
-                model_name="deepseek-chat"
+                model_name="deepseek-chat",
+                logical_node='literature_parsing',
             )
             
             # 强制提取 JSON 部分，过滤模型可能的解释性前缀或后缀
@@ -49,4 +50,4 @@ class MetaExtractor:
 
 def extract_golden_metadata(agent, paper_texts: List[str]):
     extractor = MetaExtractor(agent)
-    return extractor.run_batch_extraction(paper_texts)
+    return extractor.run_batch_extraction(paper_texts)

core/model_gateway.py

@@ -1,10 +1,11 @@
 # -*- coding: utf-8 -*-
 """Provider-neutral model routing primitives for SmartPaper.
 
-A2 keeps the currently verified DeepSeek transport intact while moving provider
-selection out of academic business logic.  Providers are registered adapters;
-academic code asks for a capability (reasoning/writing/translation/fast/repair)
-and never needs to know the concrete vendor or model name.
+Legacy tasks continue to route by coarse capability.  A5-R3-R4B adds a
+separate, trusted product-policy seam: academic callers provide an explicit
+logical_node, while the frozen TaskPolicySnapshot supplies provider/model and
+execution settings.  Product routing never falls back to legacy capability
+routes.
 """
 from __future__ import annotations
 
@@ -40,14 +41,18 @@ class ModelRoute:
     fallback_providers: tuple[str, ...] = ()
 
 
-class ProviderNeutralModelGateway:
-    """Route capability calls to registered provider adapters.
+@dataclass(frozen=True)
+class TrustedProductRoute:
+    provider: str
+    model_id: str
+    thinking_enabled: bool
+    reasoning_effort: str | None
+    max_output_tokens: int
+    retry_budget: int
 
-    Adapter callables intentionally share the existing PipelineAgent request
-    signature.  This makes A2 a low-risk architectural seam: DeepSeek remains
-    the verified adapter today, while Gemini/Qwen/OpenAI adapters can be
-    registered later without changing manuscript generation code.
-    """
+
+class ProviderNeutralModelGateway:
+    """Route legacy capabilities or trusted product logical nodes to adapters."""
 
     def __init__(self, config: dict[str, Any] | None = None):
         cfg = dict(config or {})
@@ -61,6 +66,7 @@ class ProviderNeutralModelGateway:
             if isinstance(row, str):
                 provider = str(row or self.default_provider).strip().lower()
                 provider_role = cap
+                fallbacks = ()
             else:
                 item = dict(row or {})
                 provider = str(item.get("provider") or self.default_provider).strip().lower()
@@ -73,12 +79,15 @@ class ProviderNeutralModelGateway:
                     for name in raw_fallbacks
                     if str(name or "").strip()
                 )
-            if not isinstance(row, dict):
-                fallbacks = ()
-            self._routes[cap] = ModelRoute(provider=provider, provider_role=provider_role, fallback_providers=fallbacks)
+            self._routes[cap] = ModelRoute(
+                provider=provider,
+                provider_role=provider_role,
+                fallback_providers=fallbacks,
+            )
         self._call_adapters: dict[str, Callable[..., str]] = {}
         self._stream_adapters: dict[str, Callable[..., Iterable[str]]] = {}
         self._policy_snapshot: dict[str, Any] = {}
+        self._trusted_product_policy_snapshot: dict[str, Any] = {}
 
     def register_provider(
         self,
@@ -93,6 +102,7 @@ class ProviderNeutralModelGateway:
         self._call_adapters[name] = call_adapter
         self._stream_adapters[name] = stream_adapter
 
+    # Legacy config/request policy.  It remains a separate trust domain.
     def set_policy_snapshot(self, snapshot: dict[str, Any] | None) -> None:
         self._policy_snapshot = deepcopy(dict(snapshot or {}))
 
@@ -100,6 +110,19 @@ class ProviderNeutralModelGateway:
     def policy_snapshot(self) -> dict[str, Any]:
         return deepcopy(self._policy_snapshot)
 
+    def set_trusted_product_policy_snapshot(self, snapshot: dict[str, Any] | None) -> None:
+        """Attach a Worker/D1-verified TaskPolicySnapshot for product execution.
+
+        The runner validates task identity and snapshot SHA before calling this
+        setter.  The gateway still validates route shape and fails closed before
+        invoking any provider adapter.
+        """
+        self._trusted_product_policy_snapshot = deepcopy(dict(snapshot or {}))
+
+    @property
+    def trusted_product_policy_snapshot(self) -> dict[str, Any]:
+        return deepcopy(self._trusted_product_policy_snapshot)
+
     def _route(self, capability: str) -> ModelRoute:
         cap = str(capability or "reasoning").strip().lower() or "reasoning"
         policy_routes = dict(self._policy_snapshot.get("model_routes") or {})
@@ -114,7 +137,8 @@ class ProviderNeutralModelGateway:
                 provider=str(row.get("provider") or self.default_provider).strip().lower(),
                 provider_role=str(row.get("provider_role") or cap).strip().lower(),
                 fallback_providers=tuple(
-                    str(name or "").strip().lower() for name in raw_fallbacks
+                    str(name or "").strip().lower()
+                    for name in raw_fallbacks
                     if str(name or "").strip()
                 ),
             )
@@ -126,7 +150,118 @@ class ProviderNeutralModelGateway:
             name for name in (route.provider, *route.fallback_providers) if name
         ))
 
-    def call(self, *, capability: str, **kwargs) -> str:
+    def _trusted_product_context(self) -> tuple[dict[str, Any], dict[str, Any]] | None:
+        snapshot = self._trusted_product_policy_snapshot
+        if not snapshot:
+            return None
+        resolved = snapshot.get("resolved_policy")
+        if not isinstance(resolved, dict):
+            raise ModelRequestError("trusted_product_policy_invalid")
+        model = resolved.get("model")
+        if not isinstance(model, dict):
+            raise ModelRequestError("trusted_product_model_policy_missing")
+        routes = model.get("routes")
+        if not isinstance(routes, dict) or not routes:
+            raise ModelRequestError("trusted_product_model_routes_missing")
+        return resolved, routes
+
+    def _trusted_product_route(self, logical_node: str | None) -> tuple[TrustedProductRoute, str]:
+        context = self._trusted_product_context()
+        if context is None:
+            raise ModelRequestError("trusted_product_policy_missing")
+        resolved, routes = context
+        node = str(logical_node or "").strip()
+        if not node:
+            raise ModelRequestError("trusted_product_logical_node_required")
+        row = routes.get(node)
+        if not isinstance(row, dict):
+            raise ModelRequestError(f"trusted_product_route_missing:{node}")
+
+        provider = str(row.get("provider") or "").strip().lower()
+        model_id = str(row.get("model_id_or_alias") or "").strip()
+        if not provider or not model_id:
+            raise ModelRequestError(f"trusted_product_route_invalid:{node}")
+
+        thinking = row.get("thinking_enabled")
+        if not isinstance(thinking, bool):
+            raise ModelRequestError(f"trusted_product_route_thinking_invalid:{node}")
+        effort = row.get("reasoning_effort")
+        if thinking:
+            if str(effort or "").strip().lower() != "high":
+                raise ModelRequestError(f"trusted_product_route_effort_invalid:{node}")
+            effort = "high"
+        else:
+            if effort not in (None, ""):
+                raise ModelRequestError(f"trusted_product_route_effort_invalid:{node}")
+            effort = None
+
+        try:
+            max_output_tokens = int(row.get("max_output_tokens"))
+            retry_budget = int(row.get("retry_budget"))
+        except (TypeError, ValueError):
+            raise ModelRequestError(f"trusted_product_route_budget_invalid:{node}")
+        if max_output_tokens <= 0 or retry_budget < 0:
+            raise ModelRequestError(f"trusted_product_route_budget_invalid:{node}")
+        if row.get("escalation_model_id_or_alias") not in (None, ""):
+            raise ModelRequestError(f"trusted_product_route_escalation_forbidden:{node}")
+        if row.get("escalation_condition_code") not in (None, ""):
+            raise ModelRequestError(f"trusted_product_route_escalation_forbidden:{node}")
+
+        model_policy_version = str(
+            resolved.get("model_policy_version")
+            or snapshot_model_policy_version(self._trusted_product_policy_snapshot)
+            or ""
+        ).strip()
+        return TrustedProductRoute(
+            provider=provider,
+            model_id=model_id,
+            thinking_enabled=thinking,
+            reasoning_effort=effort,
+            max_output_tokens=max_output_tokens,
+            retry_budget=retry_budget,
+        ), model_policy_version
+
+    @staticmethod
+    def _effective_max_tokens(requested: Any, ceiling: int) -> int:
+        if requested is None:
+            return int(ceiling)
+        try:
+            value = int(requested)
+        except (TypeError, ValueError):
+            raise ModelRequestError("trusted_product_max_tokens_invalid")
+        if value <= 0:
+            raise ModelRequestError("trusted_product_max_tokens_invalid")
+        return min(value, int(ceiling))
+
+    def _product_payload(self, *, capability: str, logical_node: str | None, kwargs: dict[str, Any]):
+        route, model_policy_version = self._trusted_product_route(logical_node)
+        payload = dict(kwargs)
+        payload["model_role"] = str(capability or "reasoning").strip().lower() or "reasoning"
+        payload["model_name"] = route.model_id
+        payload["thinking"] = route.thinking_enabled
+        payload["reasoning_effort"] = route.reasoning_effort
+        payload["allow_fallback"] = False
+        payload["max_tokens"] = self._effective_max_tokens(
+            payload.get("max_tokens"), route.max_output_tokens
+        )
+        payload["retry_budget"] = route.retry_budget
+        payload["logical_node"] = str(logical_node or "").strip()
+        payload["model_policy_version"] = model_policy_version
+        payload["route_enforced"] = True
+        return route.provider, payload
+
+    def call(self, *, capability: str, logical_node: str | None = None, **kwargs) -> str:
+        if self._trusted_product_context() is not None:
+            provider, payload = self._product_payload(
+                capability=capability, logical_node=logical_node, kwargs=kwargs
+            )
+            adapter = self._call_adapters.get(provider)
+            if not callable(adapter):
+                raise ModelProviderUnavailableError(
+                    f"model_provider_unavailable:{provider}:{logical_node or capability}"
+                )
+            return adapter(**payload)
+
         route = self._route(capability)
         missing = []
         for provider in self._provider_sequence(route):
@@ -141,7 +276,19 @@ class ProviderNeutralModelGateway:
             f"model_provider_unavailable:{','.join(missing) or route.provider}:{capability}"
         )
 
-    def stream(self, *, capability: str, **kwargs):
+    def stream(self, *, capability: str, logical_node: str | None = None, **kwargs):
+        if self._trusted_product_context() is not None:
+            provider, payload = self._product_payload(
+                capability=capability, logical_node=logical_node, kwargs=kwargs
+            )
+            adapter = self._stream_adapters.get(provider)
+            if not callable(adapter):
+                raise ModelProviderUnavailableError(
+                    f"model_provider_unavailable:{provider}:{logical_node or capability}"
+                )
+            yield from adapter(**payload)
+            return
+
         route = self._route(capability)
         missing = []
         for provider in self._provider_sequence(route):
@@ -158,20 +305,17 @@ class ProviderNeutralModelGateway:
         )
 
 
+def snapshot_model_policy_version(snapshot: dict[str, Any] | None) -> str:
+    data = dict(snapshot or {})
+    return str(data.get("model_policy_version") or "").strip()
+
+
 def resolve_plan_policy_snapshot(
     app_config: dict[str, Any], request: dict[str, Any] | None = None
 ) -> dict[str, Any]:
-    """Freeze a config-driven plan policy without touching payment authority.
-
-    A2 only creates the interface.  Worker/D1 remain the authority for payment,
-    quota and entitlement; if the request does not yet carry a trusted plan_id,
-    the legacy-current policy preserves existing production behavior.
-    """
+    """Freeze a config-driven legacy plan policy without touching payment authority."""
     req = dict(request or {})
     policies = dict(app_config.get("plan_policies") or {})
-    # Never trust a browser/request plan label as an entitlement authority.
-    # A future Worker/D1 integration may inject a trusted entitlement snapshot;
-    # until then every task stays on the legacy-current production policy.
     entitlement = req.get("entitlement_snapshot") if isinstance(req.get("entitlement_snapshot"), dict) else {}
     requested = str(
         req.get("trusted_plan_id") or entitlement.get("plan_id") or ""

core/pipeline_agent.py

@@ -272,7 +272,7 @@ class SSciPipelineAgent:
         self, *, usage, model, role, attempt_index, streaming, thinking,
         reasoning_effort, system_prompt, user_prompt, started_at,
         completed_at, first_token_at=None, finish_reason="",
-        response_payload=None,
+        response_payload=None, logical_node="", model_policy_version="",
     ):
         if not self._model_usage_enabled():
             return
@@ -308,6 +308,9 @@ class SSciPipelineAgent:
                 metadata=sanitize_metadata({
                     "system_fingerprint": response_payload.get("system_fingerprint"),
                     "reasoning_effort": reasoning_effort,
+                    "logical_node": str(logical_node or ""),
+                    "model_policy_version": str(model_policy_version or ""),
+                    "resolved_model_id": str(response_payload.get("model") or model or ""),
                 }),
             )
             emit_usage_event(getattr(self, "_model_usage_sink", None), event)
@@ -851,8 +854,13 @@ class SSciPipelineAgent:
         requested_model,
         requested_thinking,
         allow_fallback,
+        retry_budget=None,
     ):
-        total = self.deepseek_max_retries
+        total = (
+            self.deepseek_max_retries
+            if retry_budget is None
+            else max(1, 1 + int(retry_budget))
+        )
         fallback_model = self.deepseek_models.get("fallback", "")
         for attempt in range(total):
             model = requested_model
@@ -877,7 +885,8 @@ class SSciPipelineAgent:
         thinking=None,
         reasoning_effort=None,
         allow_fallback=None,
-        response_format=None,
+        response_format=None, retry_budget=None, logical_node=None,
+        model_policy_version=None, route_enforced=False,
     ):
         if not str(getattr(self, "ds_key", "") or "").strip():
             raise DeepSeekAuthenticationError(
@@ -896,6 +905,15 @@ class SSciPipelineAgent:
             reasoning_effort=reasoning_effort,
             allow_fallback=allow_fallback,
         )
+        if route_enforced:
+            resolved_fallback = False
+            if not requested_thinking:
+                resolved_effort = None
+        attempt_total = (
+            self.deepseek_max_retries
+            if retry_budget is None
+            else max(1, 1 + int(retry_budget))
+        )
         tokens_est = self._count_tokens(str(user_prompt))
 
         logging.info(
@@ -913,6 +931,7 @@ class SSciPipelineAgent:
             requested_model,
             requested_thinking,
             resolved_fallback,
+            retry_budget=retry_budget,
         ):
             with self.lock:
                 self.ds_calls += 1
@@ -942,7 +961,7 @@ class SSciPipelineAgent:
                 logging.info(
                     "DeepSeek static attempt start | attempt=%s/%s | model=%s",
                     attempt + 1,
-                    self.deepseek_max_retries,
+                    attempt_total,
                     current_model,
                 )
                 response = requests.post(
@@ -1000,6 +1019,8 @@ class SSciPipelineAgent:
                     completed_at=completed_at,
                     finish_reason=choices[0].get("finish_reason"),
                     response_payload=result,
+                    logical_node=logical_node,
+                    model_policy_version=model_policy_version,
                 )
                 return content
 
@@ -1016,11 +1037,11 @@ class SSciPipelineAgent:
                 DeepSeekAPIError,
             ) as exc:
                 last_error = exc
-                is_last = attempt == self.deepseek_max_retries - 1
+                is_last = attempt == attempt_total - 1
                 logging.error(
                     "DeepSeek attempt %s/%s failed: %s",
                     attempt + 1,
-                    self.deepseek_max_retries,
+                    attempt_total,
                     exc,
                 )
                 self._touch_execution_progress("deepseek_static_retry")
@@ -1030,7 +1051,7 @@ class SSciPipelineAgent:
 
         raise DeepSeekAPIError(
             "DeepSeek API failed after "
-            f"{self.deepseek_max_retries} attempts: {last_error}"
+            f"{attempt_total} attempts: {last_error}"
         )
 
     def _stream_deepseek_transport(
@@ -1044,7 +1065,8 @@ class SSciPipelineAgent:
         reasoning_effort=None,
         temp=0.3,
         allow_fallback=None,
-        reasoning_only_recovery=False,
+        reasoning_only_recovery=False, retry_budget=None, logical_node=None,
+        model_policy_version=None, route_enforced=False,
     ):
         if not str(getattr(self, "ds_key", "") or "").strip():
             raise DeepSeekAuthenticationError(
@@ -1063,6 +1085,15 @@ class SSciPipelineAgent:
             reasoning_effort=reasoning_effort,
             allow_fallback=allow_fallback,
         )
+        if route_enforced:
+            resolved_fallback = False
+            if not requested_thinking:
+                resolved_effort = None
+        attempt_total = (
+            self.deepseek_max_retries
+            if retry_budget is None
+            else max(1, 1 + int(retry_budget))
+        )
         tokens_est = self._count_tokens(str(user_prompt))
 
         logging.info(
@@ -1080,6 +1111,7 @@ class SSciPipelineAgent:
             requested_model,
             requested_thinking,
             resolved_fallback,
+            retry_budget=retry_budget,
         ):
             with self.lock:
                 self.ds_calls += 1
@@ -1121,7 +1153,7 @@ class SSciPipelineAgent:
                 logging.info(
                     "DeepSeek stream attempt start | attempt=%s/%s | model=%s | idle_limit=%ss",
                     attempt + 1,
-                    self.deepseek_max_retries,
+                    attempt_total,
                     current_model,
                     self.deepseek_stream_idle_seconds,
                 )
@@ -1260,6 +1292,8 @@ class SSciPipelineAgent:
                     first_token_at=first_content_at,
                     finish_reason=stream_finish_reason,
                     response_payload=stream_response_payload,
+                    logical_node=logical_node,
+                    model_policy_version=model_policy_version,
                 )
                 return
 
@@ -1285,7 +1319,7 @@ class SSciPipelineAgent:
                 logging.error(
                     "DeepSeek stream attempt %s/%s failed: %s",
                     attempt + 1,
-                    self.deepseek_max_retries,
+                    attempt_total,
                     exc,
                 )
 
@@ -1296,24 +1330,28 @@ class SSciPipelineAgent:
                     ) from exc
 
                 self._touch_execution_progress("deepseek_stream_retry")
-                is_last = attempt == self.deepseek_max_retries - 1
+                is_last = attempt == attempt_total - 1
                 if is_last:
                     break
                 time.sleep(self._retry_delay(attempt))
 
         raise DeepSeekAPIError(
             "DeepSeek streaming failed after "
-            f"{self.deepseek_max_retries} attempts: {last_error}"
+            f"{attempt_total} attempts: {last_error}"
         )
 
     def set_model_policy_snapshot(self, snapshot):
         self.model_policy_snapshot = dict(snapshot or {})
         self.model_gateway.set_policy_snapshot(self.model_policy_snapshot)
 
+    def set_trusted_product_policy_snapshot(self, snapshot):
+        self.model_gateway.set_trusted_product_policy_snapshot(snapshot)
+
     def call_model(
         self, system_prompt, user_prompt, max_tokens=8000, temp=0.3,
         model_name=None, model_role="reasoning", thinking=None,
         reasoning_effort=None, allow_fallback=None, response_format=None,
+        logical_node=None,
     ):
         gateway = getattr(self, "model_gateway", None)
         if gateway is None:
@@ -1326,7 +1364,8 @@ class SSciPipelineAgent:
                 response_format=response_format,
             )
         return gateway.call(
-            capability=model_role, system_prompt=system_prompt, user_prompt=user_prompt,
+            capability=model_role, logical_node=logical_node,
+            system_prompt=system_prompt, user_prompt=user_prompt,
             max_tokens=max_tokens, temp=temp, model_name=model_name,
             thinking=thinking, reasoning_effort=reasoning_effort,
             allow_fallback=allow_fallback, response_format=response_format,
@@ -1335,7 +1374,7 @@ class SSciPipelineAgent:
     def stream_model(
         self, system_prompt, user_prompt, max_tokens=8000, model_name=None,
         model_role="reasoning", thinking=None, reasoning_effort=None, temp=0.3,
-        allow_fallback=None, reasoning_only_recovery=False,
+        allow_fallback=None, reasoning_only_recovery=False, logical_node=None,
     ):
         gateway = getattr(self, "model_gateway", None)
         if gateway is None:
@@ -1347,7 +1386,8 @@ class SSciPipelineAgent:
             )
             return
         yield from gateway.stream(
-            capability=model_role, system_prompt=system_prompt, user_prompt=user_prompt,
+            capability=model_role, logical_node=logical_node,
+            system_prompt=system_prompt, user_prompt=user_prompt,
             max_tokens=max_tokens, model_name=model_name, thinking=thinking,
             reasoning_effort=reasoning_effort, temp=temp, allow_fallback=allow_fallback,
             reasoning_only_recovery=reasoning_only_recovery,
@@ -1359,24 +1399,25 @@ class SSciPipelineAgent:
         self, system_prompt, user_prompt, max_tokens=8000, temp=0.3,
         model_name=None, model_role="reasoning", thinking=None,
         reasoning_effort=None, allow_fallback=None, response_format=None,
+        logical_node=None,
     ):
         return self.call_model(
             system_prompt, user_prompt, max_tokens=max_tokens, temp=temp,
             model_name=model_name, model_role=model_role, thinking=thinking,
             reasoning_effort=reasoning_effort, allow_fallback=allow_fallback,
-            response_format=response_format,
+            response_format=response_format, logical_node=logical_node,
         )
 
     def stream_deepseek(
         self, system_prompt, user_prompt, max_tokens=8000, model_name=None,
         model_role="reasoning", thinking=None, reasoning_effort=None, temp=0.3,
-        allow_fallback=None, reasoning_only_recovery=False,
+        allow_fallback=None, reasoning_only_recovery=False, logical_node=None,
     ):
         yield from self.stream_model(
             system_prompt, user_prompt, max_tokens=max_tokens, model_name=model_name,
             model_role=model_role, thinking=thinking, reasoning_effort=reasoning_effort,
             temp=temp, allow_fallback=allow_fallback,
-            reasoning_only_recovery=reasoning_only_recovery,
+            reasoning_only_recovery=reasoning_only_recovery, logical_node=logical_node,
         )
 
     def parse_user_core_literature(self, user_lit_text):
@@ -1398,6 +1439,7 @@ class SSciPipelineAgent:
                     model_role="reasoning",
                     thinking=True,
                     allow_fallback=False,
+                    logical_node='literature_parsing',
                 ) 
                 papers = self._extract_json_from_text(response)
                 if not papers: 
@@ -1461,7 +1503,7 @@ class SSciPipelineAgent:
         user_prompt = APP_CONFIG["prompt_templates"]["synthesis_user"].replace(
             "{topic}", topic).replace("{user_injection}", user_injection).replace(
             "{refs_text}", refs_text).replace("{directive}", APP_CONFIG['deep_synthesis_directive'])
-        return self.call_deepseek(sys_prompt, user_prompt, temp=0.2)
+        return self.call_deepseek(sys_prompt, user_prompt, temp=0.2, logical_node="literature_synthesis")
 
     def generate_anchors(
         self,
@@ -1504,6 +1546,7 @@ class SSciPipelineAgent:
                 model_role="reasoning",
                 thinking=False,
                 allow_fallback=False,
+                logical_node='literature_synthesis',
             )
 
         return _stream_direct_synthesis()
@@ -1562,6 +1605,7 @@ Return a concise PROVISIONAL RESEARCH LOGIC CONTRACT in plain English prose. It
                 model_role="reasoning",
                 thinking=False,
                 allow_fallback=False,
+                logical_node='research_logic',
             )
 
         return _stream_direct_research_logic()
@@ -1668,6 +1712,7 @@ Return a concise PROVISIONAL RESEARCH LOGIC CONTRACT in plain English prose. It
                 thinking=False,
                 allow_fallback=False,
                 response_format={"type": "json_object"},
+                logical_node='front_matter_writing',
             )
             try:
                 parsed = self._extract_json_from_text(response)
@@ -2630,6 +2675,7 @@ Return ONLY a valid JSON array of strings."""
                 user_prompt,
                 temp=0.1 + attempt * 0.1,
                 model_role="fast",
+                logical_node='search_jit',
             ))
             if queries and isinstance(queries, list) and len(queries) > 0:
                 final_queries = [str(q).strip() for q in queries if self._is_english_dominant(str(q).strip())]
@@ -3082,6 +3128,7 @@ Return ONLY a valid JSON array of strings."""
                     f"Topic: {topic}\nKeywords: {keywords}\nUsed: {sorted(all_used_terms)}",
                     temp=0.4,
                     model_role='fast',
+                    logical_node='search_jit',
                 )
                 parsed = self._extract_json_from_text(response)
                 terms = parsed if isinstance(parsed, list) else []
@@ -3287,6 +3334,7 @@ INPUT:
                     model_role='reasoning',
                     thinking=False,
                     allow_fallback=False,
+                    logical_node='citation_planning',
                 )
                 parsed = self._extract_json_from_text(raw)
                 parsed_rows = parsed.get('assignments') if isinstance(parsed, dict) else None
@@ -5243,6 +5291,7 @@ INPUT:
                 model_role='reasoning',
                 thinking=False,
                 allow_fallback=False,
+                logical_node='critical_citation_adjudication',
             )
             parsed = self._extract_json_from_text(response)
             if isinstance(parsed, dict):
@@ -5386,6 +5435,7 @@ Return exactly:
                         model_role="reasoning",
                         thinking=False,
                         allow_fallback=False,
+                        logical_node='ordinary_citation_qa',
                     )
                     parsed = self._extract_json_from_text(response)
                     last_error = None
@@ -5439,6 +5489,7 @@ Return exactly:
                         model_role="reasoning",
                         thinking=False,
                         allow_fallback=False,
+                        logical_node='ordinary_citation_qa',
                     )
                     parsed = self._extract_json_from_text(response)
                     by_id = index_items(parsed)
@@ -5911,6 +5962,7 @@ TEXT:
             model_role='writing',
             thinking=False,
             allow_fallback=False,
+            logical_node='ordinary_repair',
         )
         revised = self._remove_ai_meta_traces(revised)
         original_numbers = self._numeric_tokens(value)
@@ -5963,6 +6015,7 @@ TEXT:
             raw = self.call_deepseek(
                 'You are an independent senior management-journal reviewer. Be exact and conservative.',
                 prompt, max_tokens=800, temp=0.0, model_role='fast', thinking=False, allow_fallback=True,
+                logical_node='section_critique',
             )
         except PipelineCancelledError:
             raise
@@ -6037,6 +6090,7 @@ CURRENT UNIT:
             repaired = self.call_model(
                 'You are SmartPaper Adaptive Academic Repair. Preserve evidence truth and repair prose with Conciseness, Accuracy, and Rigor.',
                 prompt, max_tokens=7000, temp=0.15, model_role='repair', thinking=True, allow_fallback=True,
+                logical_node='ordinary_repair',
             )
         except PipelineCancelledError:
             raise
@@ -8059,6 +8113,7 @@ CURRENT UNIT:
                 json.dumps({'ir_version': self.CANONICAL_GLOSSARY_IR_VERSION, 'terms': payload}, ensure_ascii=False),
                 max_tokens=max(1200, min(5000, 120 + len(payload) * 90)),
                 temp=0.0, model_role='translation', thinking=False, allow_fallback=False,
+                logical_node='translation',
             )
             parsed = self._extract_json_from_text(response)
             items = parsed.get('terms', []) if isinstance(parsed, dict) else []
@@ -8321,6 +8376,7 @@ CURRENT UNIT:
                 json.dumps({'ir_version': self.CANONICAL_GLOSSARY_IR_VERSION, 'terms': payload}, ensure_ascii=False),
                 max_tokens=max(1200, min(4200, 180 + len(payload) * 120)),
                 temp=0.0, model_role='translation', thinking=False, allow_fallback=False,
+                logical_node='ordinary_translation_qa',
             )
             parsed = self._extract_json_from_text(response)
             items = parsed.get('terms', []) if isinstance(parsed, dict) else []
@@ -8453,6 +8509,7 @@ CURRENT UNIT:
                 json.dumps({'terms': missing}, ensure_ascii=False),
                 max_tokens=max(1200, min(5000, len(missing) * 80)),
                 temp=0.0, model_role='translation', thinking=False, allow_fallback=False,
+                logical_node='translation',
             )
             parsed = self._extract_json_from_text(response)
             items = parsed.get('terms', []) if isinstance(parsed, dict) else []
@@ -8600,6 +8657,7 @@ SECTION: {section_name}
         response = self.call_deepseek(
             'You are a bilingual academic translation equivalence auditor. Be conservative and evidence-preserving.',
             prompt, max_tokens=1200, temp=0.0, model_role='reasoning', thinking=False, allow_fallback=False,
+            logical_node='ordinary_translation_qa',
         )
         parsed = self._extract_json_from_text(response)
         ok = parsed.get('ok') if isinstance(parsed, dict) else None
@@ -8691,6 +8749,7 @@ If ok=false, codes MUST contain only values from allowed_issue_codes. For NEGATI
                 system_prompt, json.dumps(payload, ensure_ascii=False),
                 max_tokens=max(1200, min(3600, 650 + 260 * len(rows))),
                 temp=0.0, model_role='reasoning', thinking=False, allow_fallback=False,
+                logical_node='ordinary_translation_qa',
             )
             parsed = self._extract_json_from_text(response)
             returned = parsed.get('items', []) if isinstance(parsed, dict) else []
@@ -8962,6 +9021,7 @@ Translate headings and terminology naturally. Preserve standard variable abbrevi
             model_role='translation',
             thinking=False,
             allow_fallback=False,
+            logical_node='translation',
         )
         parsed = self._extract_json_from_text(response)
         items = parsed.get('terms', []) if isinstance(parsed, dict) else []
@@ -9151,6 +9211,7 @@ Translate headings and terminology naturally. Preserve standard variable abbrevi
             effective_prompt, marked_source,
             max_tokens=max(1800, min(9000, self._count_tokens(marked_source) * 2 + 1200)),
             temp=0.0, model_role='translation', thinking=False, allow_fallback=False,
+            logical_node='translation',
         )
         try:
             parsed = self._parse_marked_translation(result, expected_ids)
@@ -9336,6 +9397,7 @@ Each code must be one of: {sorted(allowed)}. Return one or more specific dimensi
             response = self.call_deepseek(
                 prompt, json.dumps(payload, ensure_ascii=False), max_tokens=900, temp=0.0,
                 model_role='reasoning', thinking=False, allow_fallback=False,
+                logical_node='ordinary_translation_qa',
             )
             parsed = self._extract_json_from_text(response)
             raw = parsed.get('delta_codes', []) if isinstance(parsed, dict) else []
@@ -9432,6 +9494,7 @@ Each code must be one of: {sorted(allowed)}. Return one or more specific dimensi
             effective_prompt, marked_source,
             max_tokens=max(1200, min(2600, self._count_tokens(marked_source) * 2 + 900)),
             temp=0.0, model_role='translation', thinking=False, allow_fallback=False,
+            logical_node='translation',
         )
         parsed = self._parse_marked_translation(result, [block_id])
         try:
@@ -9458,6 +9521,7 @@ Each code must be one of: {sorted(allowed)}. Return one or more specific dimensi
                     strict_prompt, marked_source,
                     max_tokens=max(1200, min(2600, self._count_tokens(marked_source) * 2 + 900)),
                     temp=0.0, model_role='translation', thinking=False, allow_fallback=False,
+                    logical_node='translation',
                 )
                 strict_parsed = self._parse_marked_translation(strict_result, [block_id])
                 restored = self._restore_translation_tokens(
@@ -9483,6 +9547,7 @@ Each code must be one of: {sorted(allowed)}. Return one or more specific dimensi
                     numeric_prompt, marked_source,
                     max_tokens=max(1200, min(2600, self._count_tokens(marked_source) * 2 + 900)),
                     temp=0.0, model_role='translation', thinking=False, allow_fallback=False,
+                    logical_node='translation',
                 )
                 retry_parsed = self._parse_marked_translation(retry_result, [block_id])
                 restored = self._restore_translation_tokens(
@@ -9856,6 +9921,7 @@ Decision must be CONFIRMED_BLOCKING or NOT_CONFIRMED. evidence_code must be one
                 system_prompt, json.dumps(payload, ensure_ascii=False),
                 max_tokens=max(1200, min(3200, 600 + 180 * len(model_rows))),
                 temp=0.0, model_role='reasoning', thinking=False, allow_fallback=False,
+                logical_node='critical_translation_adjudication',
             )
             parsed = self._extract_json_from_text(response)
             returned = parsed.get('items', []) if isinstance(parsed, dict) else []
@@ -10620,6 +10686,7 @@ SOURCE UNIT:
         translated = self.call_model(
             'You are SmartPaper Adaptive Translation. Produce faithful academic translation with no added or omitted claims.',
             prompt, max_tokens=8000, temp=0.1, model_role='translation', thinking=False, allow_fallback=True,
+            logical_node='translation',
         )
         translated = re.sub(r'<think>.*?</think>', '', str(translated or ''), flags=re.S).strip()
         if not translated:

core/qualitative_engine.py

@@ -733,6 +733,7 @@ Rules:
                     base_prompt,
                     max_tokens=8000, temp=0.0, model_role="reasoning",
                     thinking=False, allow_fallback=False, response_format={"type": "json_object"},
+                    logical_node='structured_json',
                 )
             else:
                 retry_prompt = base_prompt + (
@@ -753,6 +754,7 @@ Rules:
                     thinking=False,
                     allow_fallback=False,
                     response_format={"type": "json_object"},
+                    logical_node='structured_json',
                 )
 
             response = re.sub(

core/quantitative_engine.py

@@ -480,6 +480,7 @@ Keywords: {keywords}"""
                 model_role="fast",
                 thinking=False,
                 allow_fallback=True,
+                logical_node='structured_json',
             )
             parsed = agent._extract_json_from_text(raw)
             if not isinstance(parsed, dict):