feat(agent): 在决策合成前增加低敏多 Agent 分歧摘要 (#1973)

* feat(agent): add low-sensitive disagreement summary

* fix(agent): avoid treating risk-clear signals as bullish

* fix(agent): align disagreement summary with runtime contracts

docs/CHANGELOG.md

@@ -8,6 +8,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/).
 > For user-friendly release highlights, see the [GitHub Releases](https://github.com/ZhuLinsen/daily_stock_analysis/releases) page.
 
 ## [Unreleased]
+- [改进] 为 multi-agent DecisionAgent 增加内部低敏分歧摘要输入管线，作为 #1904 P1 解释输出的前置 plumbing；不改变 public API、dashboard schema 或最终解释字段。
 - [改进] GitHub Actions 每日分析工作流补齐 TickFlow 数据源环境变量映射，并收敛 README 数据源稳定性说明到完整指南。
 - [修复] WebUI 启动时显式 `--host` / `--port` 不再被 `.env` 中的 `WEBUI_HOST` / `WEBUI_PORT` 覆盖，未传 CLI 参数时统一使用解析后的运行时配置。
 - [改进] GitHub Actions: 每日分析工作流（`00-daily-analysis.yml`）新增钉钉通知环境变量映射，支持在云端定时任务中直接使用钉钉机器人。

docs/full-guide.md

@@ -871,6 +871,12 @@ P3 在普通分析和 Agent 初始上下文中接入 `AnalysisContextPack` 低
 
 P3 当时不新增 API/Web/Bot 参数，不写入 history/task status/report metadata，不改变报告 JSON schema，也不把完整 pack 暴露到历史、通知或 Web。Agent 工具级复用 pack 数据和 P5 数据质量评分留给后续阶段。
 
+#### Multi-Agent 决策分歧摘要输入（Issue #1904 P1 plumbing）
+
+Multi-agent 在进入 `DecisionAgent` 前会构造内部低敏 `agent_disagreement_summary`，用于提示前序 Agent opinion 的方向分歧、风险 override 证据、风险 override 是否受当前 `AGENT_RISK_OVERRIDE` 配置启用，以及非关键阶段降级信息。该摘要只包含 agent name、signal、confidence、conflict type、decision path hint、低敏 risk control 状态和 degraded stage marker，不包含 reasoning、raw_data、原始错误文本、token 或私密 payload。
+
+该能力当前只是 `DecisionAgent` 的内部 Prompt 输入管线：摘要写入运行态 `ctx.meta`，不进入 Agent pre-fetched data，不新增 public API、Web/Desktop 展示、history/task status/report metadata、dashboard schema 或最终解释字段。`risk_level=high` 只作为风险证据，不会单独触发 override；summary 与最终 `_apply_risk_override()` 复用同一套 override 判断，并尊重 `AGENT_RISK_OVERRIDE=false`。非关键降级阶段沿用 orchestrator 的 `intel`、`risk` 和 specialist/skill agent 降级契约，避免把单一方向意见误描述成 multi-agent 共识。#1904 的用户可见最终解释输出仍属于后续阶段。
+
 #### AnalysisContextPack 低敏可见性（Issue #1389 P4）
 
 P4 新增 `report.details.analysis_context_pack_overview`，历史详情和 completed `/api/v1/analysis/status/{task_id}` 会从已持久化的 `context_snapshot` 返回同一份低敏 overview；同步分析响应也会读取本次已落库的 `analysis_history.context_snapshot` 提取 overview，因此 `SAVE_CONTEXT_SNAPSHOT=false` 时新记录不保证返回该字段。Web 端报告页在“策略点位”和“资讯”之后展示默认折叠的数据块摘要，折叠头部展示可用数、缺失数、非零的其他状态计数和触发来源，展开后展示数据块状态、来源、warning、missing reason、状态计数和新闻结果数。API 返回的 `details.context_snapshot` 会剥离顶层 `analysis_context_pack_overview`，避免透明度面板重复展示 raw snapshot。

docs/full-guide_EN.md

@@ -737,6 +737,12 @@ P3 injects a low-sensitivity `AnalysisContextPack` summary into regular analysis
 
 P3 itself did not add API/Web/Bot parameters, persist fields into history/task status/report metadata, change report JSON schemas, or expose the full pack through history, notifications, or Web surfaces. Agent tool-level reuse of pack data and P5 data-quality scoring are left to later phases.
 
+### Multi-Agent Decision Disagreement Summary Input (Issue #1904 P1 Plumbing)
+
+Before `DecisionAgent` runs, the multi-agent pipeline builds an internal low-sensitivity `agent_disagreement_summary` that summarizes directional disagreement across prior Agent opinions, risk-override evidence, whether risk override is enabled by the current `AGENT_RISK_OVERRIDE` setting, and non-critical stage degradation. The summary only contains agent name, signal, confidence, conflict type, decision path hint, low-sensitivity risk-control state, and degraded-stage markers. It does not include reasoning, raw data, raw error text, tokens, or private payloads.
+
+This is currently only internal Prompt input plumbing for `DecisionAgent`: the summary is stored in runtime `ctx.meta`, is not injected through Agent pre-fetched data, and does not add public API fields, Web/Desktop display, history/task-status/report metadata, dashboard schema, or final explanation fields. `risk_level=high` is risk evidence only and does not trigger override by itself; the summary and final `_apply_risk_override()` share the same override predicate and respect `AGENT_RISK_OVERRIDE=false`. Non-critical degraded stages reuse the orchestrator contract for `intel`, `risk`, and specialist/skill agents, so a remaining single directional opinion is not described as multi-agent consensus. User-visible final explanation output for #1904 remains a later phase.
+
 ### AnalysisContextPack Low-Sensitivity Visibility (Issue #1389 P4)
 
 P4 adds `report.details.analysis_context_pack_overview`. History detail and completed `/api/v1/analysis/status/{task_id}` responses read the same low-sensitivity overview from the persisted `context_snapshot`; sync analysis responses also extract the overview from the just-persisted `analysis_history.context_snapshot`, so new records do not guarantee this field when `SAVE_CONTEXT_SNAPSHOT=false`. The Web report page renders a collapsed data-block summary after Strategy and News, with available/missing counts, non-zero other status counts, and trigger source in the header and data-block status, source, warnings, missing reasons, status counts, and news result count after expansion. API `details.context_snapshot` strips the top-level `analysis_context_pack_overview` so the raw snapshot panel does not duplicate the public overview.

src/agent/agents/decision_agent.py

@@ -195,6 +195,12 @@ should sum to 100; all-zero means no effective signal and must not be faked.
                 parts.append(f"- [{rf.get('severity', 'medium')}] {rf.get('category', '')}: {rf.get('description', '')}")
             parts.append("")
 
+        disagreement_summary = ctx.meta.get("agent_disagreement_summary")
+        if isinstance(disagreement_summary, dict) and disagreement_summary:
+            parts.append("## Agent Disagreement Summary")
+            parts.append(json.dumps(disagreement_summary, ensure_ascii=False, default=str))
+            parts.append("")
+
         # Skill meta
         requested_skills = ctx.meta.get("skills_requested") or ctx.meta.get("strategies_requested")
         if requested_skills:

src/agent/disagreement.py

@@ -0,0 +1,183 @@
+# -*- coding: utf-8 -*-
+"""
+Low-sensitivity disagreement summary for multi-agent decision synthesis.
+
+This module intentionally exposes pure functions only.  The orchestrator owns
+when to compute the summary; DecisionAgent owns how to present it to the LLM.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Iterable
+from typing import Any, Dict, List
+
+from src.agent.protocols import AgentContext
+from src.agent.risk_override import build_risk_override_plan
+
+_BULLISH_SIGNALS = {"strong_buy", "buy"}
+_BEARISH_SIGNALS = {"strong_sell", "sell"}
+_RISK_AGENT_NAMES = {"risk"}
+_SUMMARY_STAGE_LIMIT = 8
+
+
+def build_agent_disagreement_summary(
+    ctx: AgentContext,
+    *,
+    risk_override_enabled: bool = True,
+) -> Dict[str, Any]:
+    """Build a structured, low-sensitivity summary of prior agent disagreement."""
+    buckets = {
+        "bullish_agents": [],
+        "bearish_agents": [],
+        "neutral_agents": [],
+    }
+
+    for opinion in ctx.opinions:
+        signal = _effective_signal(opinion.agent_name, opinion.signal)
+        agent_summary = _summarize_opinion(opinion.agent_name, signal, opinion.confidence)
+        if signal in _BULLISH_SIGNALS:
+            buckets["bullish_agents"].append(agent_summary)
+        elif signal in _BEARISH_SIGNALS:
+            buckets["bearish_agents"].append(agent_summary)
+        else:
+            buckets["neutral_agents"].append(agent_summary)
+
+    risk_override_plan = build_risk_override_plan(
+        ctx,
+        override_enabled=risk_override_enabled,
+    )
+    degraded_result = _build_degraded_result(ctx)
+    conflict_type = _classify_conflict_type(
+        buckets["bullish_agents"],
+        buckets["bearish_agents"],
+        buckets["neutral_agents"],
+        risk_override_plan.override_enabled and risk_override_plan.override_trigger_present,
+        degraded_result,
+    )
+
+    return {
+        **buckets,
+        "conflict_type": conflict_type,
+        "decision_path_hint": _decision_path_hint(conflict_type),
+        "risk_override_present": risk_override_plan.override_enabled
+        and risk_override_plan.override_trigger_present,
+        "risk_control": risk_override_plan.to_low_sensitivity_dict(),
+        "degraded_result": degraded_result,
+    }
+
+
+def _summarize_opinion(agent_name: str, signal: Any, confidence: Any) -> Dict[str, Any]:
+    """Keep only low-sensitivity opinion metadata for downstream synthesis."""
+    return {
+        "agent_name": str(agent_name or "unknown"),
+        "signal": _normalize_signal(signal),
+        "confidence": _safe_confidence(confidence),
+    }
+
+
+def _normalize_signal(signal: Any) -> str:
+    if not isinstance(signal, str):
+        return "hold"
+    normalized = signal.strip().lower()
+    if normalized in _BULLISH_SIGNALS or normalized in _BEARISH_SIGNALS or normalized == "hold":
+        return normalized
+    return "hold"
+
+
+def _effective_signal(agent_name: str, signal: Any) -> str:
+    normalized = _normalize_signal(signal)
+    if _is_risk_agent(agent_name) and normalized in _BULLISH_SIGNALS:
+        return "hold"
+    return normalized
+
+
+def _is_risk_agent(agent_name: str) -> bool:
+    return str(agent_name or "").strip().lower() in _RISK_AGENT_NAMES
+
+
+def _safe_confidence(confidence: Any) -> float:
+    try:
+        value = float(confidence)
+    except (TypeError, ValueError):
+        value = 0.0
+    return round(max(0.0, min(1.0, value)), 2)
+
+
+def _build_degraded_result(ctx: AgentContext) -> Dict[str, Any]:
+    stages = list(_iter_degraded_stages(ctx))
+    has_non_critical = any(stage.get("non_critical") is True for stage in stages)
+    return {
+        "present": bool(stages),
+        "non_critical_stage_present": has_non_critical,
+        "stages": stages[:_SUMMARY_STAGE_LIMIT],
+    }
+
+
+def _iter_degraded_stages(ctx: AgentContext) -> Iterable[Dict[str, Any]]:
+    source = ctx.meta.get("degraded_stages")
+    if not isinstance(source, list):
+        return
+
+    seen = set()
+    for item in source:
+        if not isinstance(item, dict):
+            continue
+        stage_name = str(item.get("stage_name") or "").strip()
+        status = str(item.get("status") or "").strip().lower()
+        if not stage_name or status != "failed":
+            continue
+        dedupe_key = (stage_name, status)
+        if dedupe_key in seen:
+            continue
+        seen.add(dedupe_key)
+        yield {
+            "stage_name": stage_name,
+            "status": status,
+            "non_critical": item.get("non_critical") is True,
+        }
+
+
+def _classify_conflict_type(
+    bullish_agents: List[Dict[str, Any]],
+    bearish_agents: List[Dict[str, Any]],
+    neutral_agents: List[Dict[str, Any]],
+    risk_override_present: bool,
+    degraded_result: Dict[str, Any],
+) -> str:
+    if risk_override_present:
+        return "risk_override"
+    if bullish_agents and bearish_agents:
+        return "mixed_directional_signals"
+    if degraded_result.get("present"):
+        if bullish_agents and not bearish_agents:
+            return "partial_bullish_with_degraded_inputs"
+        if bearish_agents and not bullish_agents:
+            return "partial_bearish_with_degraded_inputs"
+        return "degraded_only"
+    if bullish_agents and not bearish_agents:
+        return "aligned_bullish" if not neutral_agents else "bullish_with_neutral"
+    if bearish_agents and not bullish_agents:
+        return "aligned_bearish" if not neutral_agents else "bearish_with_neutral"
+    if neutral_agents:
+        return "aligned_neutral"
+    return "insufficient_opinions"
+
+
+def _decision_path_hint(conflict_type: str) -> str:
+    hints = {
+        "risk_override": "prioritize_risk_controls_and_cap_buy_signal",
+        "mixed_directional_signals": "explain_cross_agent_conflict_before_final_signal",
+        "degraded_only": "state_data_limitations_before_recommendation",
+        "partial_bullish_with_degraded_inputs": "state_degraded_inputs_before_any_bullish_lean",
+        "partial_bearish_with_degraded_inputs": "state_degraded_inputs_before_any_bearish_lean",
+        "aligned_bullish": "use_bullish_consensus_with_price_and_risk_checks",
+        "bullish_with_neutral": "lean_bullish_but_require_confirmation",
+        "aligned_bearish": "use_bearish_consensus_and_preserve_downside_controls",
+        "bearish_with_neutral": "lean_defensive_and_require_recovery_confirmation",
+        "aligned_neutral": "prefer_hold_watchlist_or_range_plan",
+        "insufficient_opinions": "prefer_conservative_hold_due_to_limited_agent_input",
+    }
+    return hints.get(conflict_type, "prefer_conservative_hold_due_to_mixed_inputs")
+
+
+__all__ = ["build_agent_disagreement_summary"]

src/agent/orchestrator.py

@@ -32,6 +32,8 @@ import time
 from dataclasses import dataclass, field
 from typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional
 
+from src.agent.chat_context import build_visible_chat_history
+from src.agent.disagreement import build_agent_disagreement_summary
 from src.agent.llm_adapter import LLMToolAdapter
 from src.agent.protocols import (
     AgentContext,
@@ -40,11 +42,11 @@ from src.agent.protocols import (
     StageStatus,
     normalize_decision_signal,
 )
+from src.agent.risk_override import build_risk_override_plan
 from src.agent.runner import parse_dashboard_json
 from src.agent.stock_scope import resolve_stock_scope
 from src.agent.stream_events import stream_event
 from src.agent.tools.registry import ToolRegistry
-from src.agent.chat_context import build_visible_chat_history
 from src.config import AGENT_MAX_STEPS_DEFAULT, get_config
 from src.report_language import normalize_report_language
 
@@ -55,6 +57,7 @@ logger = logging.getLogger(__name__)
 
 # Valid orchestrator modes (ordered by cost/depth)
 VALID_MODES = ("quick", "standard", "full", "specialist")
+NON_CRITICAL_BASE_STAGES = frozenset({"intel", "risk"})
 
 
 @dataclass
@@ -504,6 +507,9 @@ class AgentOrchestrator:
             if agent.agent_name == "decision" and getattr(self, "_skill_agent_names", None):
                 self._aggregate_skill_opinions(ctx)
 
+            if agent.agent_name == "decision":
+                self._prepare_decision_context(ctx)
+
             if progress_callback:
                 progress_callback(stream_event(
                     "stage_start",
@@ -569,11 +575,7 @@ class AgentOrchestrator:
             #   - intel / risk (standard support stages)
             #   - skill agents (specialist evaluation, optional)
             if result.status == StageStatus.FAILED:
-                non_critical = (
-                    agent.agent_name in ("intel", "risk")
-                    or agent.agent_name in getattr(self, "_skill_agent_names", set())
-                )
-                if not non_critical:
+                if not self._is_non_critical_stage(agent.agent_name):
                     logger.error("[Orchestrator] critical stage '%s' failed: %s", agent.agent_name, result.error)
                     return OrchestratorResult(
                         success=False,
@@ -583,6 +585,7 @@ class AgentOrchestrator:
                         tool_calls_log=all_tool_calls,
                     )
                 else:
+                    self._record_degraded_stage(ctx, agent.agent_name, result)
                     logger.warning("[Orchestrator] stage '%s' failed (non-critical, degrading): %s", agent.agent_name, result.error)
 
             index += 1
@@ -735,6 +738,41 @@ class AgentOrchestrator:
         """Compatibility wrapper for legacy tests/imports."""
         self._aggregate_skill_opinions(ctx)
 
+    def _prepare_decision_context(self, ctx: AgentContext) -> None:
+        """Populate low-sensitivity summaries consumed by DecisionAgent."""
+        ctx.meta["agent_disagreement_summary"] = build_agent_disagreement_summary(
+            ctx,
+            risk_override_enabled=getattr(self.config, "agent_risk_override", True),
+        )
+
+    def _record_degraded_stage(
+        self,
+        ctx: AgentContext,
+        agent_name: str,
+        result: StageResult,
+    ) -> None:
+        """Record a low-sensitivity degraded stage marker for downstream synthesis."""
+        if result.status != StageStatus.FAILED:
+            raise ValueError("degraded stage markers are only produced for failed stages")
+
+        degraded_stages = ctx.meta.setdefault("degraded_stages", [])
+        if not isinstance(degraded_stages, list):
+            degraded_stages = []
+            ctx.meta["degraded_stages"] = degraded_stages
+        degraded_stages.append({
+            "stage_name": agent_name,
+            "status": result.status.value,
+            "non_critical": self._is_non_critical_stage(agent_name),
+        })
+
+    def _is_non_critical_stage(self, agent_name: str) -> bool:
+        """Return whether a failed stage should degrade instead of aborting."""
+        normalized_name = str(agent_name or "").strip()
+        return (
+            normalized_name in NON_CRITICAL_BASE_STAGES
+            or normalized_name in getattr(self, "_skill_agent_names", set())
+        )
+
     # -----------------------------------------------------------------
     # Helpers
     # -----------------------------------------------------------------
@@ -1293,9 +1331,6 @@ class AgentOrchestrator:
         if ctx.get_data("risk_override_applied"):
             return
 
-        if not getattr(self.config, "agent_risk_override", True):
-            return
-
         dashboard = ctx.get_data("final_dashboard")
         if not isinstance(dashboard, dict):
             return
@@ -1303,22 +1338,16 @@ class AgentOrchestrator:
         risk_opinion = next((op for op in reversed(ctx.opinions) if op.agent_name == "risk"), None)
         risk_raw = risk_opinion.raw_data if risk_opinion and isinstance(risk_opinion.raw_data, dict) else {}
 
-        adjustment = str(risk_raw.get("signal_adjustment") or "").lower()
-        has_high_flag = any(str(flag.get("severity", "")).lower() == "high" for flag in ctx.risk_flags)
-        veto_buy = bool(risk_raw.get("veto_buy")) or adjustment == "veto" or has_high_flag
-
-        current_signal = normalize_decision_signal(dashboard.get("decision_type", "hold"))
-        new_signal = current_signal
-        if veto_buy and current_signal == "buy":
-            new_signal = "hold"
-        elif adjustment == "downgrade_one":
-            new_signal = _downgrade_signal(current_signal, steps=1)
-        elif adjustment == "downgrade_two":
-            new_signal = _downgrade_signal(current_signal, steps=2)
-
-        if new_signal == current_signal:
+        plan = build_risk_override_plan(
+            ctx,
+            current_signal=dashboard.get("decision_type", "hold"),
+            override_enabled=getattr(self.config, "agent_risk_override", True),
+        )
+        if not plan.will_apply or plan.target_signal is None or plan.current_signal is None:
             return
 
+        current_signal = plan.current_signal
+        new_signal = plan.target_signal
         dashboard["decision_type"] = new_signal
         dashboard["risk_warning"] = self._merge_risk_warning(
             dashboard.get("risk_warning"),
@@ -1368,7 +1397,8 @@ class AgentOrchestrator:
         ctx.set_data("risk_override_applied", {
             "from": current_signal,
             "to": new_signal,
-            "adjustment": adjustment or ("veto" if veto_buy else "none"),
+            "adjustment": plan.adjustment or ("veto" if plan.veto_buy else "none"),
+            "reason": plan.reason,
         })
 
         for opinion in reversed(ctx.opinions):
@@ -1383,8 +1413,8 @@ class AgentOrchestrator:
             "[Orchestrator] risk override applied: %s -> %s (adjustment=%s, high_flag=%s)",
             current_signal,
             new_signal,
-            adjustment or ("veto" if veto_buy else "none"),
-            has_high_flag,
+            plan.adjustment or ("veto" if plan.veto_buy else "none"),
+            plan.has_high_flag,
         )
 
     @staticmethod
@@ -1492,16 +1522,6 @@ def _extract_stock_code(text: str) -> str:
     return ""
 
 
-def _downgrade_signal(signal: str, steps: int = 1) -> str:
-    """Downgrade a dashboard decision signal by one or more levels."""
-    order = ["buy", "hold", "sell"]
-    try:
-        index = order.index(signal)
-    except ValueError:
-        return signal
-    return order[min(len(order) - 1, index + max(0, steps))]
-
-
 def _adjust_sentiment_score(score: int, signal: str) -> int:
     """Clamp sentiment score into the target band for the overridden signal."""
     bands = {

src/agent/risk_override.py

@@ -0,0 +1,147 @@
+# -*- coding: utf-8 -*-
+"""Shared risk override planning for the multi-agent pipeline."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any, Dict, Optional
+
+from src.agent.protocols import AgentContext, normalize_decision_signal
+
+
+_DOWNGRADE_STEPS = {
+    "downgrade_one": 1,
+    "downgrade_two": 2,
+}
+
+
+@dataclass(frozen=True)
+class RiskOverridePlan:
+    """Configuration-aware risk override decision shared by summary and executor."""
+
+    evidence_present: bool
+    override_enabled: bool
+    override_trigger_present: bool
+    veto_buy: bool
+    adjustment: str
+    has_high_flag: bool
+    risk_level_high: bool
+    current_signal: Optional[str]
+    target_signal: Optional[str]
+    will_apply: Optional[bool]
+    reason: str
+
+    def to_low_sensitivity_dict(self) -> Dict[str, Any]:
+        """Return a prompt-safe view that does not expose raw risk payloads."""
+        return {
+            "evidence_present": self.evidence_present,
+            "override_enabled": self.override_enabled,
+            "override_trigger_present": self.override_trigger_present,
+            "veto_buy": self.veto_buy,
+            "will_apply": self.will_apply,
+            "reason": self.reason,
+        }
+
+
+def build_risk_override_plan(
+    ctx: AgentContext,
+    *,
+    current_signal: Any = None,
+    override_enabled: bool = True,
+) -> RiskOverridePlan:
+    """Build the single source of truth for risk override decisions.
+
+    ``risk_level=high`` is risk evidence, but it is not by itself an override
+    trigger. Actual execution also depends on ``override_enabled`` and on the
+    final dashboard signal.
+    """
+    risk_raw = _latest_risk_raw(ctx)
+    adjustment = str(risk_raw.get("signal_adjustment") or "").strip().lower()
+    has_high_flag = any(
+        str(flag.get("severity", "")).strip().lower() == "high"
+        for flag in ctx.risk_flags
+        if isinstance(flag, dict)
+    )
+    risk_level_high = str(risk_raw.get("risk_level") or "").strip().lower() == "high"
+    veto_buy = bool(risk_raw.get("veto_buy")) or adjustment == "veto" or has_high_flag
+    has_downgrade = adjustment in _DOWNGRADE_STEPS
+    override_trigger_present = veto_buy or has_downgrade
+    evidence_present = override_trigger_present or risk_level_high
+
+    normalized_current = (
+        normalize_decision_signal(current_signal)
+        if isinstance(current_signal, str)
+        else None
+    )
+    target_signal = normalized_current
+    will_apply: Optional[bool]
+
+    if normalized_current is None:
+        will_apply = None
+    elif not override_enabled or not override_trigger_present:
+        will_apply = False
+    else:
+        if veto_buy and normalized_current == "buy":
+            target_signal = "hold"
+        elif has_downgrade:
+            target_signal = _downgrade_signal(
+                normalized_current,
+                steps=_DOWNGRADE_STEPS[adjustment],
+            )
+        will_apply = target_signal != normalized_current
+
+    return RiskOverridePlan(
+        evidence_present=evidence_present,
+        override_enabled=bool(override_enabled),
+        override_trigger_present=override_trigger_present,
+        veto_buy=veto_buy,
+        adjustment=adjustment,
+        has_high_flag=has_high_flag,
+        risk_level_high=risk_level_high,
+        current_signal=normalized_current,
+        target_signal=target_signal,
+        will_apply=will_apply,
+        reason=_risk_override_reason(
+            veto_buy=veto_buy,
+            adjustment=adjustment,
+            has_high_flag=has_high_flag,
+            risk_level_high=risk_level_high,
+        ),
+    )
+
+
+def _latest_risk_raw(ctx: AgentContext) -> Dict[str, Any]:
+    risk_opinion = next((op for op in reversed(ctx.opinions) if op.agent_name == "risk"), None)
+    if risk_opinion and isinstance(risk_opinion.raw_data, dict):
+        return risk_opinion.raw_data
+    return {}
+
+
+def _risk_override_reason(
+    *,
+    veto_buy: bool,
+    adjustment: str,
+    has_high_flag: bool,
+    risk_level_high: bool,
+) -> str:
+    if has_high_flag:
+        return "high_severity_flag"
+    if veto_buy:
+        return "risk_veto"
+    if adjustment in _DOWNGRADE_STEPS:
+        return adjustment
+    if risk_level_high:
+        return "high_risk_evidence"
+    return "none"
+
+
+def _downgrade_signal(signal: str, steps: int = 1) -> str:
+    order = ["buy", "hold", "sell"]
+    try:
+        index = order.index(signal)
+    except ValueError:
+        return signal
+    return order[min(len(order) - 1, index + max(0, steps))]
+
+
+__all__ = ["RiskOverridePlan", "build_risk_override_plan"]

tests/agent/test_disagreement.py

@@ -0,0 +1,408 @@
+# -*- coding: utf-8 -*-
+"""Tests for low-sensitivity multi-agent disagreement summaries."""
+
+import sys
+from types import SimpleNamespace
+from unittest.mock import MagicMock
+
+from src.agent.disagreement import build_agent_disagreement_summary
+from src.agent.protocols import AgentContext, AgentOpinion, StageResult, StageStatus
+
+
+def test_consensus_bullish_summary_is_low_sensitivity():
+    ctx = AgentContext(query="test", stock_code="600519")
+    ctx.add_opinion(
+        AgentOpinion(
+            agent_name="technical",
+            signal="buy",
+            confidence=0.82,
+            reasoning="secret reasoning",
+            raw_data={"token": "secret-token", "private_payload": "private position payload"},
+        )
+    )
+    ctx.add_opinion(AgentOpinion(agent_name="intel", signal="strong_buy", confidence=0.76))
+
+    summary = build_agent_disagreement_summary(ctx)
+    summary_text = str(summary)
+
+    assert summary["conflict_type"] == "aligned_bullish"
+    assert [item["agent_name"] for item in summary["bullish_agents"]] == ["technical", "intel"]
+    assert summary["bearish_agents"] == []
+    assert summary["risk_override_present"] is False
+    assert "secret reasoning" not in summary_text
+    assert "raw_data" not in summary_text
+    assert "secret-token" not in summary_text
+    assert "private position payload" not in summary_text
+
+
+def test_empty_opinions_are_conservative():
+    summary = build_agent_disagreement_summary(AgentContext())
+
+    assert summary["conflict_type"] == "insufficient_opinions"
+    assert summary["bullish_agents"] == []
+    assert summary["bearish_agents"] == []
+    assert summary["neutral_agents"] == []
+    assert summary["decision_path_hint"] == "prefer_conservative_hold_due_to_limited_agent_input"
+
+
+def test_mixed_directional_signals():
+    ctx = AgentContext(query="test", stock_code="600519")
+    ctx.add_opinion(AgentOpinion(agent_name="technical", signal="buy", confidence=0.72))
+    ctx.add_opinion(AgentOpinion(agent_name="intel", signal="sell", confidence=0.68))
+    ctx.add_opinion(AgentOpinion(agent_name="risk", signal="hold", confidence=0.66))
+
+    summary = build_agent_disagreement_summary(ctx)
+
+    assert summary["conflict_type"] == "mixed_directional_signals"
+    assert len(summary["bullish_agents"]) == 1
+    assert len(summary["bearish_agents"]) == 1
+    assert len(summary["neutral_agents"]) == 1
+
+
+def test_risk_agent_buy_signal_is_neutral_risk_clear_not_bullish():
+    ctx = AgentContext(query="test", stock_code="600519")
+    ctx.add_opinion(AgentOpinion(agent_name="technical", signal="buy", confidence=0.72))
+    ctx.add_opinion(
+        AgentOpinion(
+            agent_name="risk",
+            signal="buy",
+            confidence=0.66,
+            raw_data={"risk_level": "none", "private_payload": "private risk payload"},
+        )
+    )
+
+    summary = build_agent_disagreement_summary(ctx)
+    summary_text = str(summary)
+
+    assert [item["agent_name"] for item in summary["bullish_agents"]] == ["technical"]
+    assert [item["agent_name"] for item in summary["neutral_agents"]] == ["risk"]
+    assert summary["conflict_type"] != "aligned_bullish"
+    assert "risk_level" not in summary_text
+    assert "private risk payload" not in summary_text
+
+
+def test_high_severity_risk_flag_takes_override_priority():
+    ctx = AgentContext(query="test", stock_code="600519")
+    ctx.add_opinion(AgentOpinion(agent_name="technical", signal="buy", confidence=0.86))
+    ctx.add_risk_flag(category="regulatory", description="material investigation", severity="high")
+
+    summary = build_agent_disagreement_summary(ctx)
+
+    assert summary["risk_override_present"] is True
+    assert summary["risk_control"]["evidence_present"] is True
+    assert summary["risk_control"]["override_trigger_present"] is True
+    assert summary["conflict_type"] == "risk_override"
+    assert summary["decision_path_hint"] == "prioritize_risk_controls_and_cap_buy_signal"
+
+
+def test_risk_level_high_is_evidence_not_override_by_itself():
+    ctx = AgentContext(query="test", stock_code="600519")
+    ctx.add_opinion(AgentOpinion(agent_name="technical", signal="buy", confidence=0.86))
+    ctx.add_opinion(
+        AgentOpinion(
+            agent_name="risk",
+            signal="hold",
+            confidence=0.7,
+            raw_data={"risk_level": "high"},
+        )
+    )
+
+    summary = build_agent_disagreement_summary(ctx)
+
+    assert summary["risk_override_present"] is False
+    assert summary["risk_control"]["evidence_present"] is True
+    assert summary["risk_control"]["override_trigger_present"] is False
+    assert summary["conflict_type"] != "risk_override"
+    assert summary["decision_path_hint"] != "prioritize_risk_controls_and_cap_buy_signal"
+
+
+def test_disabled_risk_override_keeps_evidence_but_omits_override_hint():
+    ctx = AgentContext(query="test", stock_code="600519")
+    ctx.add_opinion(AgentOpinion(agent_name="technical", signal="buy", confidence=0.86))
+    ctx.add_opinion(
+        AgentOpinion(
+            agent_name="risk",
+            signal="sell",
+            confidence=0.9,
+            raw_data={"veto_buy": True},
+        )
+    )
+
+    summary = build_agent_disagreement_summary(ctx, risk_override_enabled=False)
+
+    assert summary["risk_override_present"] is False
+    assert summary["risk_control"]["evidence_present"] is True
+    assert summary["risk_control"]["override_enabled"] is False
+    assert summary["risk_control"]["override_trigger_present"] is True
+    assert summary["conflict_type"] != "risk_override"
+    assert summary["decision_path_hint"] != "prioritize_risk_controls_and_cap_buy_signal"
+
+
+def test_degraded_stage_summary_is_low_sensitivity():
+    ctx = AgentContext(query="test", stock_code="600519")
+    ctx.add_opinion(AgentOpinion(agent_name="technical", signal="hold", confidence=0.64))
+    ctx.meta["degraded_stages"] = [
+        {
+            "stage_name": "intel",
+            "status": "failed",
+            "non_critical": True,
+            "error": "raw failure text",
+            "private_payload": "private tool payload",
+        }
+    ]
+
+    summary = build_agent_disagreement_summary(ctx)
+    summary_text = str(summary)
+
+    assert summary["degraded_result"]["present"] is True
+    assert summary["degraded_result"]["non_critical_stage_present"] is True
+    assert summary["degraded_result"]["stages"] == [
+        {"stage_name": "intel", "status": "failed", "non_critical": True}
+    ]
+    assert "raw failure text" not in summary_text
+    assert "private tool payload" not in summary_text
+
+
+def test_degraded_reader_uses_only_failed_meta_records_and_dedupes():
+    ctx = AgentContext(query="test", stock_code="600519")
+    ctx.set_data("degraded_stages", [
+        {"stage_name": "risk", "status": "failed", "non_critical": True}
+    ])
+    ctx.meta["stage_results"] = [
+        {"stage_name": "intel", "status": "failed", "non_critical": True}
+    ]
+    ctx.set_data("stage_results", [
+        {"stage_name": "skill", "status": "failed", "non_critical": True}
+    ])
+    ctx.meta["degraded_stages"] = [
+        {"stage_name": "intel", "status": "failed", "non_critical": True},
+        {"stage_name": "intel", "status": "failed", "non_critical": True},
+        {"stage_name": "risk", "status": "timeout", "non_critical": True},
+        {"stage": "legacy_alias", "status": "failed", "non_critical": True},
+    ]
+
+    summary = build_agent_disagreement_summary(ctx)
+
+    assert summary["degraded_result"]["stages"] == [
+        {"stage_name": "intel", "status": "failed", "non_critical": True}
+    ]
+
+
+def test_directional_opinion_with_intel_failure_is_partial_not_bullish_consensus():
+    ctx = AgentContext(query="test", stock_code="600519")
+    ctx.add_opinion(AgentOpinion(agent_name="technical", signal="buy", confidence=0.74))
+    ctx.meta["degraded_stages"] = [
+        {"stage_name": "intel", "status": "failed", "non_critical": True}
+    ]
+
+    summary = build_agent_disagreement_summary(ctx)
+
+    assert summary["conflict_type"] == "partial_bullish_with_degraded_inputs"
+    assert summary["decision_path_hint"] == "state_degraded_inputs_before_any_bullish_lean"
+    assert summary["conflict_type"] != "aligned_bullish"
+    assert summary["decision_path_hint"] != "use_bullish_consensus_with_price_and_risk_checks"
+
+
+def test_directional_opinion_with_risk_failure_is_partial_not_bullish_consensus():
+    ctx = AgentContext(query="test", stock_code="600519")
+    ctx.add_opinion(AgentOpinion(agent_name="technical", signal="buy", confidence=0.74))
+    ctx.add_opinion(AgentOpinion(agent_name="intel", signal="hold", confidence=0.52))
+    ctx.meta["degraded_stages"] = [
+        {"stage_name": "risk", "status": "failed", "non_critical": True}
+    ]
+
+    summary = build_agent_disagreement_summary(ctx)
+
+    assert summary["conflict_type"] == "partial_bullish_with_degraded_inputs"
+    assert summary["degraded_result"]["non_critical_stage_present"] is True
+    assert summary["conflict_type"] != "aligned_bullish"
+
+
+def test_directional_opinion_with_specialist_failure_is_partial_and_non_critical():
+    ctx = AgentContext(query="test", stock_code="600519")
+    ctx.add_opinion(AgentOpinion(agent_name="technical", signal="sell", confidence=0.74))
+    ctx.add_opinion(AgentOpinion(agent_name="intel", signal="hold", confidence=0.52))
+    ctx.meta["degraded_stages"] = [
+        {"stage_name": "chan_theory", "status": "failed", "non_critical": True}
+    ]
+
+    summary = build_agent_disagreement_summary(ctx)
+
+    assert summary["conflict_type"] == "partial_bearish_with_degraded_inputs"
+    assert summary["decision_path_hint"] == "state_degraded_inputs_before_any_bearish_lean"
+    assert summary["degraded_result"]["non_critical_stage_present"] is True
+    assert summary["degraded_result"]["stages"] == [
+        {"stage_name": "chan_theory", "status": "failed", "non_critical": True}
+    ]
+
+
+def _mock_optional_litellm(monkeypatch):
+    monkeypatch.setitem(sys.modules, "litellm", MagicMock())
+
+
+def test_decision_agent_prompt_includes_disagreement_summary_when_present(monkeypatch):
+    _mock_optional_litellm(monkeypatch)
+    from src.agent.agents.decision_agent import DecisionAgent
+
+    ctx = AgentContext(query="test", stock_code="600519")
+    ctx.add_opinion(AgentOpinion(agent_name="technical", signal="buy", confidence=0.72))
+    ctx.add_opinion(AgentOpinion(agent_name="intel", signal="sell", confidence=0.68))
+    summary = build_agent_disagreement_summary(ctx)
+    ctx.meta["agent_disagreement_summary"] = summary
+
+    message = DecisionAgent(tool_registry=MagicMock(), llm_adapter=MagicMock()).build_user_message(ctx)
+
+    assert "## Agent Disagreement Summary" in message
+    assert "mixed_directional_signals" in message
+    assert "technical" in message
+
+
+def test_decision_agent_build_messages_injects_disagreement_summary_once(monkeypatch):
+    _mock_optional_litellm(monkeypatch)
+    from src.agent.agents.decision_agent import DecisionAgent
+
+    ctx = AgentContext(query="test", stock_code="600519")
+    ctx.add_opinion(AgentOpinion(agent_name="technical", signal="buy", confidence=0.72))
+    ctx.add_opinion(AgentOpinion(agent_name="intel", signal="sell", confidence=0.68))
+    ctx.set_data("realtime_quote", {"price": 123.45})
+    ctx.meta["agent_disagreement_summary"] = build_agent_disagreement_summary(ctx)
+
+    messages = DecisionAgent(tool_registry=MagicMock(), llm_adapter=MagicMock())._build_messages(ctx)
+    combined = "\n".join(str(message.get("content", "")) for message in messages)
+
+    assert combined.count("## Agent Disagreement Summary") == 1
+    assert combined.count("mixed_directional_signals") == 1
+    assert "[Pre-fetched: realtime_quote]" in combined
+    assert "[Pre-fetched: agent_disagreement_summary]" not in combined
+
+
+def test_decision_agent_prompt_omits_summary_when_context_lacks_it(monkeypatch):
+    _mock_optional_litellm(monkeypatch)
+    from src.agent.agents.decision_agent import DecisionAgent
+
+    ctx = AgentContext(query="test", stock_code="600519")
+    ctx.add_opinion(AgentOpinion(agent_name="technical", signal="buy", confidence=0.8))
+
+    message = DecisionAgent(tool_registry=MagicMock(), llm_adapter=MagicMock()).build_user_message(ctx)
+
+    assert "## Agent Opinions" in message
+    assert "## Agent Disagreement Summary" not in message
+
+
+def test_orchestrator_prepare_decision_context_sets_summary_without_running_agents(monkeypatch):
+    _mock_optional_litellm(monkeypatch)
+    from src.agent.orchestrator import AgentOrchestrator
+
+    ctx = AgentContext(query="test", stock_code="600519")
+    ctx.add_opinion(AgentOpinion(agent_name="technical", signal="buy", confidence=0.72))
+    ctx.add_opinion(AgentOpinion(agent_name="intel", signal="sell", confidence=0.68))
+    orchestrator = AgentOrchestrator(
+        tool_registry=MagicMock(),
+        llm_adapter=MagicMock(),
+        config=SimpleNamespace(agent_risk_override=True),
+    )
+
+    orchestrator._prepare_decision_context(ctx)
+
+    summary = ctx.meta.get("agent_disagreement_summary")
+    assert summary
+    assert summary["conflict_type"] == "mixed_directional_signals"
+    assert ctx.get_data("agent_disagreement_summary") is None
+
+
+def test_orchestrator_prepare_decision_context_respects_risk_override_config(monkeypatch):
+    _mock_optional_litellm(monkeypatch)
+    from src.agent.orchestrator import AgentOrchestrator
+
+    ctx = AgentContext(query="test", stock_code="600519")
+    ctx.add_opinion(AgentOpinion(agent_name="technical", signal="buy", confidence=0.72))
+    ctx.add_opinion(
+        AgentOpinion(
+            agent_name="risk",
+            signal="sell",
+            confidence=0.9,
+            raw_data={"veto_buy": True},
+        )
+    )
+    orchestrator = AgentOrchestrator(
+        tool_registry=MagicMock(),
+        llm_adapter=MagicMock(),
+        config=SimpleNamespace(agent_risk_override=False),
+    )
+
+    orchestrator._prepare_decision_context(ctx)
+
+    summary = ctx.meta.get("agent_disagreement_summary")
+    assert summary["risk_override_present"] is False
+    assert summary["risk_control"]["override_enabled"] is False
+    assert summary["risk_control"]["override_trigger_present"] is True
+    assert summary["conflict_type"] != "risk_override"
+
+
+def test_orchestrator_prepare_decision_context_propagates_summary_errors(monkeypatch):
+    _mock_optional_litellm(monkeypatch)
+    from src.agent import orchestrator as orchestrator_module
+    from src.agent.orchestrator import AgentOrchestrator
+
+    def raise_summary_error(*args, **kwargs):
+        raise RuntimeError("summary bug")
+
+    monkeypatch.setattr(orchestrator_module, "build_agent_disagreement_summary", raise_summary_error)
+    orchestrator = AgentOrchestrator(
+        tool_registry=MagicMock(),
+        llm_adapter=MagicMock(),
+        config=SimpleNamespace(agent_risk_override=True),
+    )
+
+    try:
+        orchestrator._prepare_decision_context(AgentContext(query="test", stock_code="600519"))
+    except RuntimeError as exc:
+        assert str(exc) == "summary bug"
+    else:
+        raise AssertionError("summary errors must not be swallowed")
+
+
+def test_orchestrator_records_specialist_failure_using_single_criticality_source(monkeypatch):
+    _mock_optional_litellm(monkeypatch)
+    from src.agent.orchestrator import AgentOrchestrator
+
+    ctx = AgentContext(query="test", stock_code="600519")
+    result = StageResult(stage_name="chan_theory", status=StageStatus.FAILED, error="raw error")
+    orchestrator = AgentOrchestrator(
+        tool_registry=MagicMock(),
+        llm_adapter=MagicMock(),
+        config=SimpleNamespace(agent_risk_override=True),
+    )
+    orchestrator._skill_agent_names = {"chan_theory"}
+
+    assert orchestrator._is_non_critical_stage("intel") is True
+    assert orchestrator._is_non_critical_stage("risk") is True
+    assert orchestrator._is_non_critical_stage("chan_theory") is True
+    assert orchestrator._is_non_critical_stage("technical") is False
+
+    orchestrator._record_degraded_stage(ctx, "chan_theory", result)
+
+    assert ctx.meta["degraded_stages"] == [
+        {"stage_name": "chan_theory", "status": "failed", "non_critical": True}
+    ]
+    summary = build_agent_disagreement_summary(ctx)
+    assert summary["degraded_result"]["non_critical_stage_present"] is True
+
+
+def test_orchestrator_rejects_non_failed_degraded_stage_markers(monkeypatch):
+    _mock_optional_litellm(monkeypatch)
+    from src.agent.orchestrator import AgentOrchestrator
+
+    orchestrator = AgentOrchestrator(
+        tool_registry=MagicMock(),
+        llm_adapter=MagicMock(),
+        config=SimpleNamespace(agent_risk_override=True),
+    )
+    result = StageResult(stage_name="intel", status=StageStatus.SKIPPED)
+
+    try:
+        orchestrator._record_degraded_stage(AgentContext(), "intel", result)
+    except ValueError as exc:
+        assert "failed stages" in str(exc)
+    else:
+        raise AssertionError("only failed stage results may produce degraded markers")

tests/test_multi_agent.py

@@ -868,6 +868,89 @@ class TestOrchestratorExecution(unittest.TestCase):
         result.meta["models_used"] = ["test/model"]
         return result
 
+    @staticmethod
+    def _decision_agent():
+        from src.agent.agents.decision_agent import DecisionAgent
+
+        return DecisionAgent(tool_registry=MagicMock(), llm_adapter=MagicMock())
+
+    @staticmethod
+    def _dashboard_json(decision_type="buy"):
+        return json.dumps({
+            "stock_name": "Test Stock",
+            "sentiment_score": 72,
+            "trend_prediction": "up",
+            "operation_advice": "buy",
+            "decision_type": decision_type,
+            "confidence_level": "Medium",
+            "dashboard": {
+                "phase_decision": {
+                    "phase_context": "regular",
+                    "action_window": "now",
+                    "immediate_action": "watch",
+                    "watch_conditions": [],
+                    "next_check_time": "next session",
+                    "confidence_reason": "test fixture",
+                    "data_limitations": [],
+                },
+                "core_conclusion": {
+                    "one_sentence": "test decision",
+                    "signal_type": "buy",
+                    "position_advice": {
+                        "no_position": "watch",
+                        "has_position": "hold",
+                    },
+                },
+            },
+            "analysis_summary": "test summary",
+            "key_points": ["technical fixture"],
+            "risk_warning": "",
+        }, ensure_ascii=False)
+
+    class _OpinionStage:
+        def __init__(
+            self,
+            agent_name,
+            *,
+            signal="hold",
+            confidence=0.5,
+            reasoning="fixture opinion",
+            raw_data=None,
+        ):
+            self.agent_name = agent_name
+            self.signal = signal
+            self.confidence = confidence
+            self.reasoning = reasoning
+            self.raw_data = raw_data or {}
+
+        def run(self, ctx, progress_callback=None, timeout_seconds=None):
+            ctx.add_opinion(AgentOpinion(
+                agent_name=self.agent_name,
+                signal=self.signal,
+                confidence=self.confidence,
+                reasoning=self.reasoning,
+                raw_data=self.raw_data,
+            ))
+            result = StageResult(stage_name=self.agent_name, status=StageStatus.COMPLETED)
+            result.meta["raw_text"] = self.reasoning
+            result.meta["models_used"] = ["test/model"]
+            return result
+
+    class _FailedStage:
+        def __init__(self, agent_name, error="stage failed"):
+            self.agent_name = agent_name
+            self.error = error
+
+        def run(self, ctx, progress_callback=None, timeout_seconds=None):
+            result = StageResult(
+                stage_name=self.agent_name,
+                status=StageStatus.FAILED,
+                error=self.error,
+            )
+            result.meta["raw_text"] = ""
+            result.meta["models_used"] = ["test/model"]
+            return result
+
     def test_prepare_agent_uses_default_constant_as_raise_threshold(self):
         orch = self._make_orchestrator()
         agent = MagicMock(agent_name="technical", max_steps=6)
@@ -940,6 +1023,230 @@ class TestOrchestratorExecution(unittest.TestCase):
         skill.run.assert_called_once()
         decision.run.assert_called_once()
 
+    def test_pipeline_summary_and_risk_override_share_disabled_override_contract(self):
+        orch = self._make_orchestrator(config=SimpleNamespace(agent_risk_override=False))
+        ctx = AgentContext(query="test", stock_code="600519")
+        captured_messages = []
+
+        def fake_run_agent_loop(messages, **kwargs):
+            captured_messages.append(messages)
+            return SimpleNamespace(
+                success=True,
+                content=self._dashboard_json(decision_type="buy"),
+                total_tokens=11,
+                tool_calls_log=[],
+                models_used=["test/model"],
+            )
+
+        technical = self._OpinionStage("technical", signal="buy", confidence=0.8)
+        risk = self._OpinionStage(
+            "risk",
+            signal="sell",
+            confidence=0.9,
+            raw_data={"veto_buy": True},
+        )
+        decision = self._decision_agent()
+
+        with patch.object(orch, "_build_agent_chain", return_value=[technical, risk, decision]):
+            with patch("src.agent.runner.parse_dashboard_json", side_effect=lambda raw: json.loads(raw)):
+                with patch("src.agent.agents.base_agent.run_agent_loop", side_effect=fake_run_agent_loop):
+                    result = orch._execute_pipeline(ctx, parse_dashboard=True)
+
+        self.assertTrue(result.success)
+        self.assertEqual(result.dashboard["decision_type"], "buy")
+        self.assertIsNone(ctx.get_data("risk_override_applied"))
+
+        combined = "\n".join(
+            str(message.get("content", ""))
+            for messages in captured_messages
+            for message in messages
+        )
+        self.assertEqual(combined.count("## Agent Disagreement Summary"), 1)
+        self.assertIn('"risk_override_present": false', combined)
+        self.assertIn('"override_enabled": false', combined)
+        self.assertIn('"override_trigger_present": true', combined)
+        self.assertNotIn('"conflict_type": "risk_override"', combined)
+        self.assertNotIn("[Pre-fetched: agent_disagreement_summary]", combined)
+
+    def test_pipeline_risk_level_high_is_evidence_not_runtime_override(self):
+        orch = self._make_orchestrator(config=SimpleNamespace(agent_risk_override=True))
+        ctx = AgentContext(query="test", stock_code="600519")
+        captured_messages = []
+
+        def fake_run_agent_loop(messages, **kwargs):
+            captured_messages.append(messages)
+            return SimpleNamespace(
+                success=True,
+                content=self._dashboard_json(decision_type="buy"),
+                total_tokens=11,
+                tool_calls_log=[],
+                models_used=["test/model"],
+            )
+
+        technical = self._OpinionStage("technical", signal="buy", confidence=0.8)
+        risk = self._OpinionStage(
+            "risk",
+            signal="sell",
+            confidence=0.9,
+            raw_data={"risk_level": "high"},
+        )
+        decision = self._decision_agent()
+
+        with patch.object(orch, "_build_agent_chain", return_value=[technical, risk, decision]):
+            with patch("src.agent.runner.parse_dashboard_json", side_effect=lambda raw: json.loads(raw)):
+                with patch("src.agent.agents.base_agent.run_agent_loop", side_effect=fake_run_agent_loop):
+                    result = orch._execute_pipeline(ctx, parse_dashboard=True)
+
+        self.assertTrue(result.success)
+        self.assertEqual(result.dashboard["decision_type"], "buy")
+        self.assertIsNone(ctx.get_data("risk_override_applied"))
+
+        combined = "\n".join(
+            str(message.get("content", ""))
+            for messages in captured_messages
+            for message in messages
+        )
+        self.assertEqual(combined.count("## Agent Disagreement Summary"), 1)
+        self.assertIn('"evidence_present": true', combined)
+        self.assertIn('"override_trigger_present": false', combined)
+        self.assertIn('"risk_override_present": false', combined)
+        self.assertNotIn('"conflict_type": "risk_override"', combined)
+
+    def test_pipeline_enabled_risk_veto_is_reflected_in_summary_and_final_dashboard(self):
+        orch = self._make_orchestrator(config=SimpleNamespace(agent_risk_override=True))
+        ctx = AgentContext(query="test", stock_code="600519")
+        captured_messages = []
+
+        def fake_run_agent_loop(messages, **kwargs):
+            captured_messages.append(messages)
+            return SimpleNamespace(
+                success=True,
+                content=self._dashboard_json(decision_type="buy"),
+                total_tokens=11,
+                tool_calls_log=[],
+                models_used=["test/model"],
+            )
+
+        technical = self._OpinionStage("technical", signal="buy", confidence=0.8)
+        risk = self._OpinionStage(
+            "risk",
+            signal="sell",
+            confidence=0.9,
+            raw_data={"veto_buy": True, "reasoning": "material risk"},
+        )
+        decision = self._decision_agent()
+
+        with patch.object(orch, "_build_agent_chain", return_value=[technical, risk, decision]):
+            with patch("src.agent.runner.parse_dashboard_json", side_effect=lambda raw: json.loads(raw)):
+                with patch("src.agent.agents.base_agent.run_agent_loop", side_effect=fake_run_agent_loop):
+                    result = orch._execute_pipeline(ctx, parse_dashboard=True)
+
+        self.assertTrue(result.success)
+        self.assertEqual(result.dashboard["decision_type"], "hold")
+        self.assertEqual(ctx.get_data("risk_override_applied"), {
+            "from": "buy",
+            "to": "hold",
+            "adjustment": "veto",
+            "reason": "risk_veto",
+        })
+
+        combined = "\n".join(
+            str(message.get("content", ""))
+            for messages in captured_messages
+            for message in messages
+        )
+        self.assertEqual(combined.count("## Agent Disagreement Summary"), 1)
+        self.assertIn('"conflict_type": "risk_override"', combined)
+        self.assertIn('"risk_override_present": true', combined)
+        self.assertIn('"override_enabled": true', combined)
+        self.assertIn('"override_trigger_present": true', combined)
+
+    def test_pipeline_degraded_directional_input_is_not_reported_as_consensus(self):
+        orch = self._make_orchestrator(config=SimpleNamespace(agent_risk_override=True))
+        ctx = AgentContext(query="test", stock_code="600519")
+        captured_messages = []
+
+        def fake_run_agent_loop(messages, **kwargs):
+            captured_messages.append(messages)
+            return SimpleNamespace(
+                success=True,
+                content=self._dashboard_json(decision_type="buy"),
+                total_tokens=11,
+                tool_calls_log=[],
+                models_used=["test/model"],
+            )
+
+        technical = self._OpinionStage("technical", signal="buy", confidence=0.8)
+        intel = self._FailedStage("intel", error="news source failed")
+        decision = self._decision_agent()
+
+        with patch.object(orch, "_build_agent_chain", return_value=[technical, intel, decision]):
+            with patch("src.agent.runner.parse_dashboard_json", side_effect=lambda raw: json.loads(raw)):
+                with patch("src.agent.agents.base_agent.run_agent_loop", side_effect=fake_run_agent_loop):
+                    result = orch._execute_pipeline(ctx, parse_dashboard=True)
+
+        self.assertTrue(result.success)
+        self.assertEqual(ctx.meta["degraded_stages"], [
+            {"stage_name": "intel", "status": "failed", "non_critical": True}
+        ])
+
+        combined = "\n".join(
+            str(message.get("content", ""))
+            for messages in captured_messages
+            for message in messages
+        )
+        self.assertEqual(combined.count("## Agent Disagreement Summary"), 1)
+        self.assertIn('"conflict_type": "partial_bullish_with_degraded_inputs"', combined)
+        self.assertIn('"decision_path_hint": "state_degraded_inputs_before_any_bullish_lean"', combined)
+        self.assertIn('"stage_name": "intel"', combined)
+        self.assertIn('"non_critical": true', combined)
+        self.assertNotIn('"conflict_type": "aligned_bullish"', combined)
+
+    def test_pipeline_specialist_failure_uses_runtime_non_critical_contract_in_summary(self):
+        orch = self._make_orchestrator(config=SimpleNamespace(agent_risk_override=True))
+        orch.mode = "specialist"
+        ctx = AgentContext(query="test", stock_code="600519")
+        captured_messages = []
+
+        def fake_run_agent_loop(messages, **kwargs):
+            captured_messages.append(messages)
+            return SimpleNamespace(
+                success=True,
+                content=self._dashboard_json(decision_type="sell"),
+                total_tokens=11,
+                tool_calls_log=[],
+                models_used=["test/model"],
+            )
+
+        technical = self._OpinionStage("technical", signal="sell", confidence=0.8)
+        intel = self._OpinionStage("intel", signal="hold", confidence=0.5)
+        risk = self._OpinionStage("risk", signal="hold", confidence=0.5)
+        specialist = self._FailedStage("chan_theory", error="specialist failed")
+        decision = self._decision_agent()
+
+        with patch.object(orch, "_build_agent_chain", return_value=[technical, intel, risk, decision]):
+            with patch.object(orch, "_build_specialist_agents", return_value=[specialist]):
+                with patch.object(orch, "_aggregate_skill_opinions", return_value=None):
+                    with patch("src.agent.runner.parse_dashboard_json", side_effect=lambda raw: json.loads(raw)):
+                        with patch("src.agent.agents.base_agent.run_agent_loop", side_effect=fake_run_agent_loop):
+                            result = orch._execute_pipeline(ctx, parse_dashboard=True)
+
+        self.assertTrue(result.success)
+        self.assertEqual(ctx.meta["degraded_stages"], [
+            {"stage_name": "chan_theory", "status": "failed", "non_critical": True}
+        ])
+
+        combined = "\n".join(
+            str(message.get("content", ""))
+            for messages in captured_messages
+            for message in messages
+        )
+        self.assertEqual(combined.count("## Agent Disagreement Summary"), 1)
+        self.assertIn('"conflict_type": "partial_bearish_with_degraded_inputs"', combined)
+        self.assertIn('"stage_name": "chan_theory"', combined)
+        self.assertIn('"non_critical_stage_present": true', combined)
+        self.assertIn('"non_critical": true', combined)
+
     def test_execute_pipeline_skips_stage_when_remaining_budget_below_minimum(self):
         orch = self._make_orchestrator(config=SimpleNamespace(agent_orchestrator_timeout_s=20))
         ctx = AgentContext(query="test", stock_code="600519", stock_name="贵州茅台")
@@ -2256,6 +2563,31 @@ class TestRiskOverride(unittest.TestCase):
         orch._apply_risk_override(ctx)
 
         self.assertEqual(dashboard["decision_type"], "buy")
+        self.assertIsNone(ctx.get_data("risk_override_applied"))
+
+    def test_risk_level_high_alone_does_not_override_buy_signal(self):
+        from src.agent.orchestrator import AgentOrchestrator
+
+        orch = AgentOrchestrator(
+            tool_registry=MagicMock(),
+            llm_adapter=MagicMock(),
+            config=SimpleNamespace(agent_risk_override=True),
+        )
+        ctx = AgentContext(query="test", stock_code="600519")
+        dashboard = self._make_dashboard()
+        ctx.set_data("final_dashboard", dashboard)
+        ctx.add_opinion(AgentOpinion(agent_name="decision", signal="buy", confidence=0.8, reasoning="base"))
+        ctx.add_opinion(AgentOpinion(
+            agent_name="risk",
+            signal="sell",
+            confidence=0.9,
+            raw_data={"risk_level": "high"},
+        ))
+
+        orch._apply_risk_override(ctx)
+
+        self.assertEqual(dashboard["decision_type"], "buy")
+        self.assertIsNone(ctx.get_data("risk_override_applied"))
 
 
 # ============================================================